feat(consumer): Add memory-based batching limit to Rust consumer

[onewland]

Add a --max-batch-size-calculation CLI flag to the Rust consumer that controls how --max-batch-size is interpreted: either as a message count (rows, the default) or as total encoded byte size (bytes).

The existing --max-batch-size parameter limits batches by number of messages, but this is insufficient to prevent Kubernetes OOMs during backlogs because message sizes can be unpredictable (particularly in eap_items across item types)

With --max-batch-size-calculation=bytes, operators can set --max-batch-size to a byte threshold (e.g. 100000000 for ~100MB) and the Reduce accumulator will flush when the total encoded byte size reaches that limit. Using a single flag instead of a separate --max-batch-size-bytes parameter avoids ambiguity about which limit applies.

Changes:

• Python CLI/config: New --max-batch-size-calculation click option (rows/bytes) on rust_consumer, wired through ConsumerConfig
• Rust config: BatchSizeCalculation enum with #[serde(rename_all = "lowercase")], deserialized from the JSON config
• Rust types: BytesInsertBatch tracks num_bytes (set from encoded_rows.len() at creation, summed on merge)
• Rust factory: Both JSONEachRow and RowBinary Reduce steps use byte-based compute_batch_size when Bytes is configured

Validation:

• tested locally that ~10KB messages will batch correctly with a 50KB-specified batch size or batches of 10 rows.

Next steps:

• emit a metric for what we're currently accumulating in batch sizes
• try setting --max-batch-size-calculation on a few s4s2 and de consumers to a reasonable value of that metric
• replace this for all consumers in all regions? it seems safer but also like kind of a tedious process; maybe eap + outcomes is good enough

https://linear.app/getsentry/issue/EAP-460/add-max-batch-size-bytes-to-consumer

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: RowBinary path never sets num_bytes, breaking byte-based batching
  • In make_rust_processor_row_binary, the per-message BytesInsertBatch<Vec<T>> now calls .with_num_bytes(...) using summed item_type_metrics.bytes_processed, so byte-based Reduce batching can flush correctly.

Or push these changes by commenting:

@cursor push 810e8ee99a

Preview (810e8ee99a)

diff --git a/rust_snuba/src/strategies/processor.rs b/rust_snuba/src/strategies/processor.rs
--- a/rust_snuba/src/strategies/processor.rs
+++ b/rust_snuba/src/strategies/processor.rs
@@ -213,7 +213,14 @@
             }
         }
 
+        let num_bytes = transformed
+            .item_type_metrics
+            .as_ref()
+            .map(|metrics| metrics.bytes_processed.values().sum())
+            .unwrap_or(0);
+
         let mut payload = BytesInsertBatch::from_rows(transformed.rows)
+            .with_num_bytes(num_bytes)
             .with_message_timestamp(timestamp)
             .with_commit_log_offsets(CommitLogOffsets(BTreeMap::from([(
                 partition.index,

_{This Bugbot Autofix run was free. To enable autofix for future PRs, go to the Cursor dashboard.}

[volokluev]

can you add a test?

[onewland]

can you add a test?

yeah, it's not ready yet because we really should add it to the RowBinary writer (as the bot comment[s] said)

[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.}

[kylemumma]

LGTM