feat(producer): Add an AsyncProducer that returns Future #476
Conversation
This file contains hidden or 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
Using this producer allows client code to wait for the future to complete, and capture the resulting error. Under the hood the FutureProducer creates a oneshot channel that it uses to pass the results from the produce.
untitaker
approved these changes
Jul 24, 2025
It's not clear to me that's what we want. If we do that, then anyone using the strategy will wait for their message to be flushed from the buffer, which is different from what is happening now. |
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.
Create a new
AsyncProducerthat returns aFutureon produce. Under the hood, the AsyncProducer is using therdkafka::FutureProducerwhich uses channels to send the results from the delivery callback. This is also using thesend_resultfunction, which will error immediately if the message can't be put into the local rdkafka queue (e.g. if the queue is full).