Support rebalance-safe-commits with external commits #1425

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

Allow rebalance-safe-commits to be used in combination with external commits. External commits are commits done to some other system than the kafka broker, e.g. a relational database.

This new capability also supports the improved TransactionalProducer that will be introduced in zio-kafka 3.0.0.

--
Note: this PR initially also included the improved TransactionalProducer. This has been moved to #1434.

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.

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.

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

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

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?

zio-kafka/src/main/scala/zio/kafka/producer/TransactionalProducer.scala

Document the transactional producer introduced in #1425.

erikvanoosten · 2025-01-11T10:24:56Z

As discussed on Discord, we pivot this PR to support rebalance-safe-commits with external commits. The transactional producer changes are moved to a future PR.

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

The changes to the transactional producer are removed and will come back in a later PR. Make `RunloopAccess.withRunloopZIO` private again.

erikvanoosten · 2025-01-12T11:02:50Z

The changes in transactional producing have been pushed in the https://github.com/zio/zio-kafka/tree/transaction-rebalance-safe-commits-followup branch.

svroonland

👍

svroonland · 2025-01-12T20:12:57Z

So this can be merged and released in 2.10.0?

erikvanoosten · 2025-01-13T07:04:48Z

So this can be merged and released in 2.10.0?

Yes! Though that reminds me, we neer to change the version in the docs.

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 this to the 3.0.0 milestone Jan 5, 2025

erikvanoosten approved these changes Jan 5, 2025

View reviewed changes

erikvanoosten self-requested a review January 9, 2025 12:22

erikvanoosten reviewed Jan 9, 2025

View reviewed changes

zio-kafka/src/main/scala/zio/kafka/producer/TransactionalProducer.scala Outdated Show resolved Hide resolved

erikvanoosten added a commit that referenced this pull request Jan 9, 2025

Document transactions

cd107ab

Document the transactional producer introduced in #1425.

erikvanoosten mentioned this pull request Jan 9, 2025

Document transactions #1432

Draft

erikvanoosten added a commit that referenced this pull request Jan 9, 2025

Document transactions

8233bc0

Document the transactional producer introduced in #1425.

erikvanoosten force-pushed the transaction-rebalance-safe-commits branch from 545743c to c22a55f Compare January 11, 2025 10:18

erikvanoosten changed the title ~~Transactional committing and producing with rebalanceSafeCommits~~ Support rebalance-safe-commits with external commits Jan 11, 2025

erikvanoosten removed this from the 3.0.0 milestone Jan 11, 2025

erikvanoosten self-requested a review January 11, 2025 14:12

svroonland and others added 9 commits January 12, 2025 11:36

Transactional committing and producing with rebalanceSafeCommits

abbf811

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

Simplify stream

833314e

Alternative implementation

414175f

Fixup

8a34c12

Formatting

ed96dd9

Provide Consumer upon creation of TransactionalProducer

a857bc6

Random transaction ID per consumer

933e50e

Pivot to allowing registration of external commits

064e710

The changes to the transactional producer are removed and will come back in a later PR. Make `RunloopAccess.withRunloopZIO` private again.

Add a test

afc01ee

erikvanoosten force-pushed the transaction-rebalance-safe-commits branch from c22a55f to afc01ee Compare January 12, 2025 10:38

Fix logging in test

442c970

Add documentation

ef2d73d

svroonland commented Jan 12, 2025

View reviewed changes

fix version in docs

e60eca1

erikvanoosten approved these changes Jan 13, 2025

View reviewed changes

erikvanoosten merged commit c5da73e into master Jan 13, 2025
9 checks passed

erikvanoosten deleted the transaction-rebalance-safe-commits branch January 13, 2025 07:41

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support rebalance-safe-commits with external commits #1425

Support rebalance-safe-commits with external commits #1425

svroonland commented Dec 29, 2024 •

edited by erikvanoosten

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 commented Jan 11, 2025

erikvanoosten commented Jan 12, 2025

svroonland left a comment

svroonland commented Jan 12, 2025

erikvanoosten commented Jan 13, 2025

Support rebalance-safe-commits with external commits #1425

Support rebalance-safe-commits with external commits #1425

Conversation

svroonland commented Dec 29, 2024 • edited by erikvanoosten 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 commented Jan 11, 2025

erikvanoosten commented Jan 12, 2025

svroonland left a comment

Choose a reason for hiding this comment

svroonland commented Jan 12, 2025

erikvanoosten commented Jan 13, 2025

svroonland commented Dec 29, 2024 •

edited by erikvanoosten

Loading

erikvanoosten commented Jan 4, 2025 •

edited

Loading

erikvanoosten commented Jan 5, 2025 •

edited

Loading

erikvanoosten left a comment •

edited

Loading