Repository navigation
feat(ReloadingProducer): Add delivery futures - #591
Merged
Merged
Conversation
Add ReloadingKafkaProducer::produce_record_with_delivery, which returns a future that resolves once the broker acknowledged the message. Messages buffered during a swap carry their sender through the replay, and messages purged with the old client resolve with an error. Taskbroker needs this to only mark retries and deadletters as done once they are actually in Kafka. ref STREAM-1964
untitaker
marked this pull request as draft
October 28, 2026 12:48
Gives ReloadingKafkaProducer the same interface as AsyncKafkaProducer, so hosts can switch over without changing how they produce.
Make produce_record_with_delivery private so AsyncProducer is the one way to wait for delivery. Always box the delivery opaque, which removes the unsafe IntoOpaque impl. Report the actual error when replaying a buffered message fails, instead of a generic one.
…ucer-delivery-futures # Conflicts: # rust-arroyo/src/backends/kafka/reloading_producer.rs
From rdkafka 0.38, flush returns once the broker has the messages, while the polling thread may not have served their delivery reports yet. A reload could then swap out a client with reports still pending, which made test_old_queue_is_empty_when_the_swap_happens flaky on 0.38. Also wait for in_flight_count to reach zero, which is what 0.37's flush did.
untitaker
marked this pull request as ready for review
November 3, 2026 10:14
evanh
approved these changes
Nov 3, 2026
evanh
left a comment
evanh
left a comment
Member
There was a problem hiding this comment.
I think this looks good, one question though.
| } | ||
| // Delivered, but the polling thread has not served every | ||
| // report yet. Give it a moment. | ||
| Ok(()) => std::thread::sleep(remaining.min(Duration::from_millis(1))), |
Member
There was a problem hiding this comment.
Do we want to time out if the records have been delivered but we are waiting for the polling thread? Should we be more gracious with the timeout in that case?
Member
Author
There was a problem hiding this comment.
yeah that's a bug, fixed it
Once flush has succeeded, the messages are with the broker. Running out of time while the polling thread serves their reports is not a failed drain: the reports are still served by the polling thread or when the client is dropped. Returning an error there made drain() report a timeout and claim queued messages would be dropped.
lvthanh03
approved these changes
Nov 3, 2026
…ucer-delivery-futures # Conflicts: # rust-arroyo/src/backends/kafka/producer.rs
mwarkentin
pushed a commit
that referenced
this pull request
Nov 5, 2026
…am-2030-fix-flaky-reloading-producer-test Brings in #591 (delivery futures) and #600 (QueueFull when the reload buffer is full). The only conflict was drain()'s signature: it keeps this branch's bool return and takes main's Client<C>. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01QUYJLCL2crYejXVqdCdBL9
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
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.
Makes
ReloadingKafkaProducerimplementAsyncProducer<KafkaPayload>, the same interface asAsyncKafkaProducer. The returned future resolves once the broker acknowledged the message, or with the reason it was not delivered.Taskbroker waits for delivery before it marks retries and deadletters as done in its store.
produce_recordonly tells you the message was enqueued, and a reload can purge it from the old client, so Taskbroker can't move to the reloading producer without this.How it works:
deliverycall, and all other callbacks are forwarded.KafkaProducernow accepts contexts with anyDeliveryOpaque. Existing callers are unaffected becauseBaseRecord's opaque defaults to().ReloadingKafkaProducernow implements bothProducerandAsyncProducer, so code that imports both traits has to callproducewith the trait path.ref STREAM-1964
Also fixes
KafkaProducer::flush_forto wait untilin_flight_countis zero. From rdkafka 0.38,flushreturns once the broker has the messages, while their delivery reports may not have been served yet, so a reload could swap out a client with reports still pending. This madetest_old_queue_is_empty_when_the_swap_happensflaky on 0.38. With this, delivery futures for messages on the old client resolve before the swap.