Kafka - transactional.id for horizontally scalable consume-transform-produce

[kerkhofsd]

Given a (Quarkus) application which implements a consume-transform-produce pattern in Kafka using smallrye-reactive-messaging.
I want the application to scale (horizontally).

As per my understanding, in this scenario the transactional.id value should contain the partition id to be able to:

• protect against transactions from zombie instances consuming from the same partition.
• scale the amount of producers (consume-transform-producer) to a maximum of the number of partitions.

I found the same problem statement described in detail in following SO post: https://stackoverflow.com/questions/50335227/how-to-pick-a-kafka-transaction-id

Looking a bit how Spring Kafka provides this functionality, it looks to me like previously they had a producer instance per partition and working with a transactional.id prefix, resulting in transactional.id: <transactionIdPrefix>.<group.id>.<topic>.<partition> . Even more, it looks like they're using a different approach for zombie fencing: "fetch-offset-request fencing" (ref: https://docs.spring.io/spring-kafka/reference/kafka/exactly-once.html )

Using smallrye-reactive-messaging, can I achieve a horizontally scalable consume-transform-produce loop? How should I configure the transactional.id to achieve this?

Thanks in advance for any input.

[ozangunalp]

Thanks for the very relevant question. That is something we need to document more.

First of all, the Kafka connector EOP support is already on KIP-447.
So you do not need a transactional.id suffixed with the partition id.

Reactive Messaging sets up transactional producer and calls initTransactions when it sees a transactional.id set. And in Quarkus when a KafkaTransactions is used, the transactional.id for that channel configuration will have a default value of ${quarkus.application.name}-${channelName}, which is of course configurable.

For horizontal scaling, this needs to be unique between your processor instances, to avoid fencing. However, it does not have to be the same between instance restarts. For example, in Kubernetes, you can suffix the transactional.id with the pod name, which is a 5 chars random string. If you go through scaling/instance restarts a lot you need to think of adjusting the transactional.id.expiration.ms to cleanup state in the broker.

For vertical scaling, it is a little more complicated, and limited in Reactive Messaging. In theory, you could have used the concurrency of your EOP consumer to run multiple transactions concurrently in a single instance. BUT this is not currently possible because concurrent transactions will run on the same Kafka producer instance, which of course is not possible.

Instead what needs to be done to support this is to back the KafkaTransactions outgoing channel with a kafka-producer-per-topic-partition producer, creating and managing a Kafka producer per topic-partition. This shouldn't be very complicated.

Lastly, note that concurrent transactions, whether they are vertically or horizontally scaled, still need to write to separate topic-partitions. This is why @Blocking(ordered = false) would not work with Kafka transactions.

[kerkhofsd]

Thank you very much, that clarifies a lot.

One follow up question on your remark:

Lastly, note that concurrent transactions, whether they are vertically or horizontally scaled, still need to write to separate topic-partitions. This is why @Blocking(ordered = false) would not work with Kafka transactions.

Given the application scales horizontally, is it safe to assume that each Kafka producer (/ KafkaTransactions outgoing channel) will automatically be publishing to a different, available, partition? Or is this something that requires additional effort / configuration to achieve?
Any insights or pointers to relevant documentation for this would be greatly appreciated!

[ozangunalp]

No that is not something that can be done automatically. Ideally your input topic and output topic need to be co-partitioned, and your application need to send processed records to the same partition.

[kerkhofsd]

Understood, thanks for the swift responses!