Repository navigation
feat(producer): Support custom contexts and BaseRecord in the reloading producer - #578
Merged
Merged
Conversation
…ng producer `ReloadingKafkaProducer` was fixed to `KafkaProducer<ProducerContext>` and only exposed `produce(KafkaPayload)`. Hosts that build a `BaseRecord` and carry their own rdkafka context, such as Relay, could not use it without giving up their delivery callbacks and paying a copy per message. Make it generic over the context and clone that context into every client it builds, including the ones installed by later reloads, so callbacks keep firing across a swap. Add `produce_record`, which forwards to the live client without copying on the fast path and only converts to an owned `KafkaPayload` while a reload is buffering.
Hosts that validate their topics against the broker at startup, such as Relay, could not do so for a topic on the reloading producer. Delegate to whichever client is live.
![sentry[bot]](https://avatars.githubusercontent.com/in/12637?s=60&v=4){kind=small}
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Want reviews to match your repository better? Bugbot Learning can learn team-specific rules from PR activity. A team admin can enable Learning in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit cbf0ab7. Configure here.
`produce_record` built a `TopicOrPartition::Topic` unconditionally, so a record produced to an explicit partition while a reload was draining the old client came back out of the buffer without one. Replay then picked a partition from the key instead of the one asked for. Carry the partition through, rejecting a value that does not fit `Partition`'s `u16` rather than wrapping it into an unrelated partition. Also express `produce` in terms of `produce_record` so the buffering path exists once. `to_base_record` borrows from both arguments, so this adds no copies. An explicit `timestamp` is still dropped by the buffer; `KafkaPayload` has nowhere to keep one. Documented on `produce_record`.
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.
...and expose validate_topic.
These additional changes are needed for porting Relay from
arroyo::KafkaProducertoarroyo::ReloadingKafkaProducer.ref STREAM-1834