fix(blq): Fix forward loop in demoted namespaces #504

lvthanh03 · 2025-10-24T01:00:49Z

Checks whether topics that taskbroker is consuming from and producing to is the same before forwarding tasks.

Added metrics for forwarded tasks and fixed inflated completed tasks metrics.

Also handles edge case where all tasks are forwarded and batch is empty.

Tested and working in sandbox.

Refs STREAM-279

linear · 2025-10-24T01:00:52Z

STREAM-279 Introduce a runtime option to forward the content of a namespace to the `long` namespace

codecov · 2025-10-24T01:04:03Z

Codecov Report

❌ Patch coverage is 90.47619% with 10 lines in your changes missing coverage. Please review.
✅ Project coverage is 88.83%. Comparing base (7a40dce) to head (31833ef).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
src/kafka/inflight_activation_batcher.rs	91.13%	7 Missing ⚠️
src/upkeep.rs	40.00%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #504      +/-   ##
==========================================
+ Coverage   88.80%   88.83%   +0.03%     
==========================================
  Files          20       20              
  Lines        5787     5867      +80     
==========================================
+ Hits         5139     5212      +73     
- Misses        648      655       +7

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

src/kafka/inflight_activation_batcher.rs

evanh

Can you add some tests for the issues you mentioned?

src/kafka/inflight_activation_writer.rs

fpacifici · 2025-10-24T20:41:22Z

src/kafka/inflight_activation_batcher.rs

 pub struct InflightActivationBatcher {
    batch: Vec<InflightActivation>,
    batch_size: usize,
+    forwarded_count: usize,


THis is just to report metrics properly in case some tasks are demoted, right ?

I also used this in flush so that offsets can be committed for forwarded tasks even if batch is empty

src/upkeep.rs

src/kafka/inflight_activation_batcher.rs

markstory · 2025-10-27T15:21:01Z

src/kafka/utils.rs

+/// * `Err(DecodeError)` - If the protobuf cannot be decoded
+pub fn modify_activation_namespace(activation_bytes: &[u8]) -> Result<Vec<u8>, prost::DecodeError> {
+    let mut activation = TaskActivation::decode(activation_bytes)?;
+    activation.namespace += DEFAULT_DEMOTED_NAMESPACE_SUFFIX;


This isn't a safe change to make. We won't have task definitions bound to these generated namespaces and tasks will fail as the tasks are undefined.

I've switched to headers to achieve the same results

src/upkeep.rs

evanh · 2025-10-28T15:25:49Z

src/kafka/inflight_activation_batcher.rs

+            let results = join_all(sends).await;
+            let success_count = results.iter().filter(|r| r.is_ok()).count();
+
+            metrics::histogram!("consumer.forwarded_rows").record(success_count as f64);


Can you add a metric for failed produces as well please.

evanh · 2025-10-28T15:27:46Z

src/kafka/inflight_activation_batcher.rs

+                    // Already forwarded, fall through to add to batch
+                }
+                Err(_) => {
+                    // Decode error, fall through to add to batch to handle in upkeep


We should have a metric at least for this case, so we can tell if something weird is happening.

markstory · 2025-10-28T16:01:24Z

src/kafka/inflight_activation_batcher.rs

+                    return Ok(());
+                }
+                Ok(None) => {
+                    // Already forwarded, fall through to add to batch


We should have a metric here as well, as this shouldn't happen in normal scenarios.

markstory · 2025-10-28T16:02:23Z

src/kafka/inflight_activation_batcher.rs

+        if !self.batch.is_empty() {
+            metrics::histogram!("consumer.batch_rows").record(self.batch.len() as f64);
+            metrics::histogram!("consumer.batch_bytes").record(self.batch_size as f64);
+        }


Why not always make the metrics? The zeros are meaningful.

markstory · 2025-10-28T16:07:02Z

src/kafka/inflight_activation_batcher.rs

+            let sends = self.forward_batch.iter().map(|payload| {
+                self.producer.send(
+                    FutureRecord::<(), Vec<u8>>::to(&topic).payload(payload),
+                    Timeout::After(Duration::from_millis(self.config.send_timeout_ms)),
+                )
+            });


One concern I have with demoted namespaces, is that the producer cluster won't always have a long topic deployed on it. For example, if a broker is connected to kafka-events and we need to demote to the long/limited topic, that topic only exists on kafka-small. The current logic will result in the broker failing to produce.

I think we would need to expand the application configuration to have broker configuration for the demoted namespace flow. That would also imply that the topic we demote to is also fixed and could be removed from runtime config.

markstory · 2025-10-28T16:07:25Z

src/kafka/inflight_activation_batcher.rs

+                )
+            });
+
+            let results = join_all(sends).await;


Much better 👏

markstory · 2025-10-28T16:13:31Z

src/kafka/utils.rs

+    let mut activation = TaskActivation::decode(activation_bytes)?;
+    if activation.headers.get("forwarded").map(|v| v.as_str()) == Some("true") {
+        return Ok(None);
+    }


I like the safety property that this provides. Would it be possible to simplify this and avoid deserializing and mutating the activation? We'll have the namespace on the InflightActivation, wouldn't skipping forwarding when the task's namespace matches the demoted namespace option work provide the same safety against loops?

I think that wouldn't provide safety against loops since all brokers share the same runtime config, however something similar I think would work is to compare between TASKBROKER_LONG_TOPIC and the topic we are trying to forward tasks to. If they are the same, we skip the forward.

I think that wouldn't provide safety against loops since all brokers share the same runtime config

Each replica in a processing pool does have the same runtime config, but each pool has configuration that is separate from other pools. We could have config to demote tasks in the default pool, and not have that config present on the long or limited pools. Comparing the topic names as we talked about today offline is another solution that would prevent loops.

lvthanh03 · 2025-10-28T19:10:45Z

Switched to comparing topic values to determine whether the task needs forwarding. Also added more metrics for visibility.

markstory

Looks good to me. We can address the demoted topic cluster connection in a follow up pull request.

markstory · 2025-10-28T20:51:54Z

src/kafka/inflight_activation_batcher.rs

+        println!("kafka_topic: {:?}", config.kafka_topic);
+        println!("kafka_long_topic: {:?}", config.kafka_long_topic);


Some stray prints.

fix(blq): Modify namespace to "long" in forwarded TaskActivation

8a12967

lvthanh03 requested a review from a team as a code owner October 24, 2025 01:00

comments

367c34f

lvthanh03 marked this pull request as draft October 24, 2025 01:29

lvthanh03 removed the request for review from a team October 24, 2025 01:29

lvthanh03 added 2 commits October 23, 2025 21:54

Record forwarded_rows and allow flush

d7c6fa1

flush Writer if batch is empty

724bae8

lvthanh03 marked this pull request as ready for review October 24, 2025 02:48

lvthanh03 requested a review from a team October 24, 2025 02:48

seer-by-sentry bot reviewed Oct 24, 2025

View reviewed changes

src/kafka/inflight_activation_batcher.rs Outdated Show resolved Hide resolved

evanh reviewed Oct 24, 2025

View reviewed changes

Add tests for modifying namespaces

631489b

lvthanh03 changed the title ~~fix(blq): Modify namespace to "long" in forwarded TaskActivation~~ fix(blq): Modify namespace for forwarded TaskActivation in demoted namespaces Oct 24, 2025

lvthanh03 commented Oct 24, 2025

View reviewed changes

src/kafka/inflight_activation_writer.rs Show resolved Hide resolved

fpacifici reviewed Oct 24, 2025

View reviewed changes

markstory reviewed Oct 27, 2025

View reviewed changes

lvthanh03 added 3 commits October 27, 2025 11:42

Try btaching forwards

dbe860e

Only batch forwards

5224077

upkeep metrics

24c9f4d

lvthanh03 changed the title ~~fix(blq): Modify namespace for forwarded TaskActivation in demoted namespaces~~ fix(blq): Tag forwarded activations in demoted namespaces Oct 27, 2025

lvthanh03 added 2 commits October 27, 2025 14:20

import order

ccbe793

Add flush results tests

45b83d8

evanh reviewed Oct 28, 2025

View reviewed changes

Add more metrics for forwarding tasks

d9fb271

markstory reviewed Oct 28, 2025

View reviewed changes

lvthanh03 added 2 commits October 28, 2025 13:26

compare topics instead of tagging

4af1359

forward logic fix

2aed06d

lvthanh03 changed the title ~~fix(blq): Tag forwarded activations in demoted namespaces~~ fix(blq): Fix forward loop in demoted namespaces Oct 28, 2025

Produce zeros for consumer.batch_rows and batch_bytes metrics

31833ef

markstory approved these changes Oct 28, 2025

View reviewed changes

lvthanh03 merged commit 85c72bc into main Oct 29, 2025
27 of 28 checks passed

lvthanh03 deleted the tony/taskactivation-namespace branch October 29, 2025 14:44

		println!("kafka_topic: {:?}", config.kafka_topic);
		println!("kafka_long_topic: {:?}", config.kafka_long_topic);

Uh oh!

fix(blq): Fix forward loop in demoted namespaces #504

fix(blq): Fix forward loop in demoted namespaces #504

Uh oh!

Conversation

lvthanh03 commented Oct 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

linear bot commented Oct 24, 2025

Uh oh!

codecov bot commented Oct 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

evanh left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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!

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!

Choose a reason for hiding this comment

Uh oh!

lvthanh03 commented Oct 28, 2025

Uh oh!

markstory left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

lvthanh03 commented Oct 24, 2025 •

edited

Loading

codecov bot commented Oct 24, 2025 •

edited

Loading