Add ReloadingKafkaProducer by untitaker · Pull Request #573 · getsentry/arroyo

untitaker · 2026-09-14T16:43:34Z

POC of a kafka producer that can change config at runtime. It is built entirely outside of KafkaProducer so the POC is easier to review can be safely rolled out, but I'm not sure that's a good idea (after all, relay will probably be tied to KafkaProducer).

The basic idea of hot-reloading is that whenever a new config is provided, the producer is swapped out for a new one and the old one is drained in a bg thread. Bursts of config updates are debounced because only one swapping+draining operation can happen at a time. Newer config updates overwrite old ones.

produce() waits for the old producer to finish draining iff the produced message has a partition key. A partition key implies strict ordering of messages per-partition and in order to preserve these semantics, it's technically more correct to wait for the old producer to drain before using the new producer. In practice Relay sets partition keys for every message regardless of ordering requirements, so there's also a producer setting to opt out of blocking drain again.

Config is fully hidden in Arroyo, users are expected to pass opaque config blobs from options-automator. These config blobs are ultimately managed by the streaming team in a central location.

ref STREAM-1804

linear-code · 2026-09-14T17:42:53Z

STREAM-1804

untitaker · 2026-09-16T12:30:58Z

+}
+
+impl ArroyoProducer<KafkaPayload> for ReloadingKafkaProducer {
+    fn produce(


joris: this method is now blocking, we need to make it non-blocking (but not async) again. wait for drain in bg too

untitaker · 2026-09-16T12:56:06Z

+    /// means rollout started, not that the config is live; watch
+    /// [`ReloadingKafkaProducer::generation`]. Broker work happens on the worker.
+    pub fn push_config(&self, blob: &[u8]) -> Result<ReloadOutcome, ReloadError> {
+        let config = ConfigBlob::parse(blob)?.producer_config(&self.inner.selector)?;


we need to add metrics here for failing to parse configs, tagged too

cursor

Stale Bugbot comment from a previous run.

evanh

Conceptually all makes sense to me!

evanh · 2026-09-18T18:07:51Z

+    /// Client construction does no I/O; this prevents swapping in an
+    /// unreachable producer.
+    fn probe(&self, candidate: &KafkaProducer) -> Result<(), KafkaError> {
+        let Some(timeout) = self.settings.probe_timeout else {


Should we skip the probe if the timeout isn't set? Or block indefinitely?

i hate both, i made it non-optional

cursor

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 9267fba. Configure here.}

cursor · 2026-09-21T09:37:11Z

+}
+
+struct Inner {
+    /// Read on every produce, taken for writing only to start buffering and


Pending gauge stuck after drop

Low Severity

ShutdownOnDrop sets shutdown and wakes the worker but never updates config_reload_pending. The worker also returns immediately on shutdown without calling record_pending. Dropping a producer that still has a desired config leaves the gauge at 1.

Additional Locations (1)

rust-arroyo/src/backends/kafka/reloading_producer.rs#L392-L394

^{Reviewed by Cursor Bugbot for commit 9267fba. Configure here.}

yes i considered handling this but thought it was too much work. we'll have to deal with it at query time anyway since processes can die by e.g. OOM too. in the end we will be able to see an absence of datapoints.

sentry · 2026-09-21T09:38:50Z

+        let producer = producer(&servers);
+
+        let outcome = producer.push_config(&blob(&servers, "all")).unwrap();
+


Bug: A race condition can cause a re-pushed configuration to be silently dropped after a failed rollout attempt, even though push_config returns Accepted.
_{Severity: HIGH}

Suggested Fix

Add a sequence or version number to the desired config. This will allow the worker to distinguish a new push of the same configuration from the one it just failed to apply, preventing incorrect deduplication.

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#L1000 Potential issue: A race condition in the `run_worker` method can cause a configuration update to be silently dropped. If a config rollout fails (e.g., due to a failed probe) and the same configuration is pushed again before the worker re-acquires its lock, the worker will incorrectly deduplicate the request. It clears the `desired` config, believing it has already processed it. However, the `push_config` call will have returned `Accepted` to the caller, leading to a state mismatch where the caller believes the config is queued for application, but the producer remains on the old configuration. This is explicitly acknowledged as a "known gap" in the code comments.

_{Did we get this right? 👍 / 👎 to inform future reviews.}

there are more bugs like this, adding a seq counter only fixes one of them. i want to punt on that rabbithole for now

evanh

Seems good to me, minus the couple followup TODOs. Nice work!

lvthanh03 · 2026-09-21T16:46:45Z

+            if let Err(error) = producer.produce(&destination, payload) {
+                metrics::counter!(
+                    "arroyo.producer.config_reload_replay_failed",
+                    "topic" => self.selector.topic().to_owned(),
+                    "producer_name" => self.producer_name.clone(),
+                )
+                .increment(1);
+                tracing::error!(
+                    topic = %self.selector.topic(),
+                    %error,
+                    "could not replay a buffered message onto the new client"


Does that mean we lose the message during replay if produce returns an error? Should we retry here?

yeah unfortunately (note that this is only for semantic partitioning, if you set ignore_key_ordering, the buffer is disabled entirely). rdkafka retries internally already though.

lvthanh03 · 2026-09-21T17:02:59Z

+
+        if purged > 0 {
+            metrics::counter!(
+                "arroyo.producer.config_reload_purged_messages",


Should we drop the messages outright when drain times out? Could we retain the old client and retry draining, though that would further delay the reload.

same as in the other comment, rdkafka already ought to retry on network errors, so i think the draining retry won't do much

untitaker added 2 commits September 14, 2026 16:44

wip

e054458

add ignore_key_ordeirng

dc7d7d0

fix example for ignore_key_ordering

edf2bb1

untitaker commented Sep 16, 2026

View reviewed changes

untitaker marked this pull request as ready for review September 17, 2026 14:58

untitaker requested review from a team as code owners September 17, 2026 14:58

cursor Bot reviewed Sep 17, 2026

View reviewed changes

Comment thread rust-arroyo/src/backends/kafka/reloading_producer.rs

evanh reviewed Sep 18, 2026

View reviewed changes

untitaker added 2 commits September 21, 2026 11:27

implement buffer, add metrics

b1e02ec

make probe_timeout non-optiona;l

9267fba

cursor Bot reviewed Sep 21, 2026

View reviewed changes

sentry Bot reviewed Sep 21, 2026

View reviewed changes

untitaker requested a review from evanh September 21, 2026 09:43

evanh approved these changes Sep 21, 2026

View reviewed changes

untitaker merged commit eb7cce6 into main Sep 21, 2026
23 checks passed

untitaker deleted the reloading-producer branch September 21, 2026 15:53

lvthanh03 reviewed Sep 21, 2026

View reviewed changes

		let producer = producer(&servers);

		let outcome = producer.push_config(&blob(&servers, "all")).unwrap();

Conversation

untitaker commented Sep 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

linear-code Bot commented Sep 14, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cursor Bot left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

evanh left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cursor Bot left a comment

Choose a reason for hiding this comment

Uh oh!

cursor Bot Sep 21, 2026

Choose a reason for hiding this comment

Pending gauge stuck after drop

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sentry Bot Sep 21, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

evanh left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

lvthanh03 Sep 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

untitaker commented Sep 14, 2026 •

edited

Loading

cursor Bot left a comment •

edited

Loading

lvthanh03 Sep 21, 2026 •

edited

Loading