Transactional committing and producing with rebalanceSafeCommits #1425

svroonland · 2024-12-29T09:10:10Z

This simplifies the setup for transactional producing and committing, no longer requiring a custom rebalance listener or the use of partitionedAssignmentStream, see ConsumerSpec.

Note that the transaction ID is no longer used for fencing since https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics, the mechanism relies on the consumer group metadata instead.

NOTE This change is not binary compatible

Just a POC for now, this was the easiest interface change I could think of

erikvanoosten · 2025-01-03T20:10:54Z

This is nice. Some questions:

Can we somehow guarantee that rebalanceSafeCommits is enabled when we use this new transactional producer?
If we simply roll this out, will applications with the existing transactional producers break? Maybe not, but if that is so, I propose we either make this backward compatible, or we move this to zio-kafka 3.

erikvanoosten · 2025-01-04T08:33:42Z

The updated TransactionProducer is not binary backward compatible; it requires a Consumer in the environment, so the ZIO type is different. It will be source compatible for many applications because many will have a Consumer available anyway.

However, this is a strange set-up. The Consumer and the TransactionProducer are tied together; they must connect to the same Kafka broker cluster. Therefore it makes more sense to produce them together, or at least let one refer to the other.

I propose that we change the construction of TransactionProducers such that they require a Consumer.
This has some advantages:

Method TransactionProducer.createTransaction stays backward compatible because its type doesn't change.
We could even make it fully backward compatible: if a consumer is provided we do the new thing pioneered in this PR, if its not provided, nothing changes. When a Consumer is provided could also force rebalanceSafeCommits to true.
We may even be able to check that the provided Consumer connects to the same broker as the TransactionProducer that is being constructed.

Re. method Consumer.registerOffsetsCommittedInTransaction, this is an internal method and should not be exposed too easily. We could do this by creating a separate trait (e.g. something like ConsumerInternal) and let ConsumerLive extend both traits. This is not the best way, we can probably think of something better.

erikvanoosten · 2025-01-04T13:15:03Z

What is Consumer.markCommittedInTransaction actually used for? I don't see where it is used.

svroonland · 2025-01-05T10:31:54Z

Agreed, providing the Consumer upon TransactionalProducer creation is better. I think it's fine to disregard backwards compatibility here, since the old method was very complicated and not well supported, this is such an improvement in usability.

If we want to check the settings, we could have a Consumer.settings method to retrieve the settings with which the Consumer was created. Is that what you have in mind as well?

There is no Consumer.markCommittedInTransaction, only Committer.markCommittedInTransaction. I'm not sure how we can hide this method, but agree that this is not ideal. In an earlier commit this was done with an internal method on the OffsetBatch, I guess we could revert to that.

…ce-safe-commits

erikvanoosten · 2025-01-05T10:54:24Z

I think it's fine to disregard backwards compatibility here...

Yeah, that is a good point. We'd better make very good release notes then :)

we could have a Consumer.settings method ... Is that what you have in mind as well?

Not necessarily, but I guess that is what is needed indeed.

There is no Consumer.markCommittedInTransaction, only Committer.markCommittedInTransaction.

Check, I get it now. Well done.

erikvanoosten

Great work! Some nitpicks in the comments.

BTW, enforcing rebalanceSafeCommits in the consumer would still be nice.

erikvanoosten · 2025-01-05T11:15:44Z

zio-kafka/src/main/scala/zio/kafka/consumer/Consumer.scala

+  override def registerOffsetsCommittedInTransaction(
+    offsetBatch: OffsetBatch
+  ): Task[Unit] =
+    runloopAccess.withRunloopZIO(true)(runloop => runloop.registerOffsetsCommittedInTransaction(offsetBatch))


We can consider adding method registerOffsetsCommittedInTransaction to runloopAccess. This would be more in line with how ConsumerLive uses RunloopAccess in other places.

erikvanoosten · 2025-01-05T11:16:58Z

zio-kafka/src/main/scala/zio/kafka/consumer/Consumer.scala

+  /**
+   * Used internally by the [[zio.kafka.producer.TransactionalProducer]]
+   */


Suggested change

/**

* Used internally by the [[zio.kafka.producer.TransactionalProducer]]

*/

/**

* This is not a public API.

* It is used internally by the [[zio.kafka.producer.TransactionalProducer]].

*/

erikvanoosten · 2025-01-05T11:35:44Z

zio-kafka/src/main/scala/zio/kafka/consumer/internal/LiveCommitter.scala

+    TopicPartition,
+    OffsetAndMetadata
+  ] => Task[Unit] = offsets =>
+    committedOffsetsRef.modify(_.addCommits(Chunk(Commit(java.lang.System.nanoTime(), offsets, null)))).unit


Let's put this below val commit to keep the order of the trait.

Suggested change

committedOffsetsRef.modify(_.addCommits(Chunk(Commit(java.lang.System.nanoTime(), offsets, null)))).unit

committedOffsetsRef

.modify {

// The continuation promise can be `null` because this commit is not handled by the consumer,

// and transactional commits have no completion handler anyway.

_.addCommits(Chunk(Commit(java.lang.System.nanoTime(), offsets, null)))

}

.unit

erikvanoosten · 2025-01-05T11:40:45Z

zio-kafka-test/src/test/scala/zio/kafka/consumer/ConsumerSpec.scala

-                            ConsumerConfig.MAX_POLL_RECORDS_CONFIG -> "200"
-                          ),
-                          rebalanceListener = transactionalRebalanceListener(streamCompleteOnRebalanceRef)
+                ZIO.scoped {


Would this extra scoped fix the spurious test failures?

svroonland added 5 commits December 29, 2024 10:09

Transactional committing and producing with rebalanceSafeCommits

c55e092

Just a POC for now, this was the easiest interface change I could think of

Simplify stream

666c13b

Alternative implementation

daab5cc

Fixup

98de633

Formatting

901a536

svroonland marked this pull request as ready for review January 2, 2025 10:03

svroonland requested a review from erikvanoosten as a code owner January 2, 2025 10:03

svroonland added 2 commits January 4, 2025 17:05

Provide Consumer upon creation of TransactionalProducer

9426c99

Random transaction ID per consumer

9c3c039

Merge remote-tracking branch 'origin/master' into transaction-rebalan…

545743c

…ce-safe-commits

svroonland added this to the 3.0.0 milestone Jan 5, 2025

erikvanoosten approved these changes Jan 5, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Transactional committing and producing with rebalanceSafeCommits #1425

Transactional committing and producing with rebalanceSafeCommits #1425

svroonland commented Dec 29, 2024 •

edited

Loading

erikvanoosten commented Jan 3, 2025

erikvanoosten commented Jan 4, 2025 •

edited

Loading

erikvanoosten commented Jan 4, 2025

svroonland commented Jan 5, 2025

erikvanoosten commented Jan 5, 2025 •

edited

Loading

erikvanoosten left a comment •

edited

Loading

erikvanoosten Jan 5, 2025

erikvanoosten Jan 5, 2025

erikvanoosten Jan 5, 2025

erikvanoosten Jan 5, 2025

-  /**
-   * Used internally by the [[zio.kafka.producer.TransactionalProducer]]
-   */
+  /**
+   * This is not a public API.
+   * It is used internally by the [[zio.kafka.producer.TransactionalProducer]].
+   */

Transactional committing and producing with rebalanceSafeCommits #1425

Are you sure you want to change the base?

Transactional committing and producing with rebalanceSafeCommits #1425

Conversation

svroonland commented Dec 29, 2024 • edited Loading

erikvanoosten commented Jan 3, 2025

erikvanoosten commented Jan 4, 2025 • edited Loading

erikvanoosten commented Jan 4, 2025

svroonland commented Jan 5, 2025

erikvanoosten commented Jan 5, 2025 • edited Loading

erikvanoosten left a comment • edited Loading

Choose a reason for hiding this comment

erikvanoosten Jan 5, 2025

Choose a reason for hiding this comment

erikvanoosten Jan 5, 2025

Choose a reason for hiding this comment

erikvanoosten Jan 5, 2025

Choose a reason for hiding this comment

erikvanoosten Jan 5, 2025

Choose a reason for hiding this comment

svroonland commented Dec 29, 2024 •

edited

Loading

erikvanoosten commented Jan 4, 2025 •

edited

Loading

erikvanoosten commented Jan 5, 2025 •

edited

Loading

erikvanoosten left a comment •

edited

Loading