Repository navigation
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}
| let destination = TopicOrPartition::Topic(Topic::new(record.topic)); | ||
| let payload = KafkaPayload::new( | ||
| record.key.map(|key| key.to_bytes().to_vec()), | ||
| record.headers.map(Headers::from), | ||
| record.payload.map(|payload| payload.to_bytes().to_vec()), |
There was a problem hiding this comment.
Bug: When buffering messages, produce_record discards the partition and timestamp from the BaseRecord, causing messages to be sent to incorrect partitions upon replay.
Severity: HIGH
Suggested Fix
Modify produce_record to check if record.partition is set. If it is, create a TopicOrPartition::Partition instance with the topic and partition index. Also, update KafkaPayload or the buffering mechanism to preserve the record.timestamp so it can be correctly applied when the message is replayed from the buffer.
Prompt for AI Agent
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.
Location: rust-arroyo/src/backends/kafka/reloading_producer.rs#L432-L436
Potential issue: The `produce_record` function, when buffering messages during a
configuration reload, incorrectly handles `BaseRecord` data. It always creates a
`TopicOrPartition::Topic` variant, ignoring any partition assignment specified in
`record.partition`. It should instead create a `TopicOrPartition::Partition` variant
when a partition is provided. Additionally, the `KafkaPayload` created does not include
the `record.timestamp`, causing this information to be silently discarded. When the
buffered messages are later replayed, they are sent to random partitions instead of
their intended destination, and the original timestamp is lost. This breaks
partition-based routing and ordering guarantees for producers that specify a partition.
Did we get this right? 👍 / 👎 to inform future reviews.
There was a problem hiding this comment.
This seems valid? Don't we need the partition?
![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 branch has not been deployed
2 participants
...and expose validate_topic.
These additional changes are needed for porting Relay from
arroyo::KafkaProducertoarroyo::ReloadingKafkaProducer.ref STREAM-1834