Convert commitAsync callback handling to ZIO sooner #1404
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…cala Co-authored-by: Erik van Oosten <e.vanoosten@grons.nl>
…cala Co-authored-by: Erik van Oosten <e.vanoosten@grons.nl>
…cala Co-authored-by: Erik van Oosten <e.vanoosten@grons.nl>
With rebalanceSafeCommits, we expect all pulled records to be processed and committed.
Helps understanding the logic here
…-rebalance-listener-file
svroonland
changed the base branch from
separate-rebalance-listener-file
to
master
zio-kafka/src/main/scala/zio/kafka/consumer/internal/Runloop.scala
Outdated
Show resolved
Hide resolved
zio-kafka/src/main/scala/zio/kafka/consumer/internal/Runloop.scala
Outdated
Show resolved
Hide resolved
zio-kafka/src/main/scala/zio/kafka/consumer/internal/LiveCommitter.scala
Outdated
Show resolved
Hide resolved
zio-kafka/src/main/scala/zio/kafka/consumer/internal/LiveCommitter.scala
Outdated
Show resolved
Hide resolved
zio-kafka/src/main/scala/zio/kafka/consumer/internal/LiveCommitter.scala
Outdated
Show resolved
Hide resolved
svroonland
commented
Nov 27, 2024
zio-kafka/src/main/scala/zio/kafka/consumer/internal/Runloop.scala
Outdated
Show resolved
Hide resolved
zio-kafka/src/main/scala/zio/kafka/consumer/internal/Runloop.scala
Outdated
Show resolved
Hide resolved
…tter.scala Co-authored-by: Erik van Oosten <e.vanoosten@grons.nl>
…cala Co-authored-by: Erik van Oosten <e.vanoosten@grons.nl>
zio-kafka/src/main/scala/zio/kafka/consumer/internal/Runloop.scala
Outdated
Show resolved
Hide resolved
svroonland
changed the title
svroonland
changed the title
erikvanoosten
approved these changes
Dec 6, 2024
| private def commitAsyncZIO( | ||
| consumer: ByteArrayKafkaConsumer, | ||
| offsets: Map[TopicPartition, OffsetAndMetadata], | ||
| doOnComplete: Either[Exception, Map[TopicPartition, OffsetAndMetadata]] => UIO[Unit] |
There was a problem hiding this comment.
In this PR we now always provide a callback to commitAsync, even when the given list of offsets is empty (so a callback is not neccesary). During a rebalance, this method is called a lot (every 100ms) and it is likely that most of these calls do not need a callback.
WDYT of making doOnComplete an Option, and then when it's None we pass null to commitAsync like we did before?
There was a problem hiding this comment.
On the other hand, doing this removes the ability to react to errors... 🤔
There was a problem hiding this comment.
Ah yeah I totally missed this. But I don't think it would make a large performance difference if it's only once every 100 ms.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
KafkaConsumer's
commitAsynctakes a callback, which we program against with complicated followup code. This PR attempts to convert everything to ZIO's earlier on, making chaining followup effects easier to reason about.As this changes some functionality around locking and same / single threads, here's a summary of what do we need to ensure:
commitAsync. InRunloop.runthis is done usingConsumerAccess. In the rebalance coordinator (while rebalancing) we already have the lock as we're callingpoll()so no need for extra locking.Runloop.runwe get this for free by guaranteeing exclusive access. In the rebalance coordinator apoll()call is in the middle of being executed and we need to callcommitAsyncon the same thread as the rebalance listener is invoked.Anything that is not calling commitAsync is free to run on any thread as executed by the default ZIO runtime.