feat(batch): add batch_size_bytes flush trigger

[enochtangg]

Summary

Adds batch_size_bytes to Batch, so a batch also closes once its payloads add up to a byte limit. Today a batch can only be closed by count or time, so its size in bytes depends entirely on message size. Whichever of count, bytes or time is hit first closes the batch.

• Only the Rust adapter supports it. The Python Arroyo adapter raises an error if it's set in code. (A value set only in deployment YAML isn't caught there, because that adapter never applies step overrides. That gap already existed.)
• It can be set in the deployment config's steps_config, like batch_size and batch_timedelta.
• Setting it alone isn't valid. A count or time limit is still required.

How it works

Each message's payload size is added to a running total as it's appended to the batch, and the batch flushes once the total reaches the limit. A batch goes over the limit by at most one message.

Measuring needs the GIL, but the consumer thread already holds it for as long as it runs, so this doesn't wait on anything. Nothing is measured when no byte limit is set.

Limitation

Only raw bytes payloads are counted. If Batch comes after a Parser or Map, the payloads are Python objects, they count as 0, and the byte limit never fires. The existing streams.pipeline.batch.size_bytes metric shows 0 in that case.

Testing

• Rust unit tests for the byte limit, bytes inside PyAnyMessage, payloads that aren't bytes, no measuring without a limit, and the step flushing through submit/poll.
• Python tests for config override, validation, and the Python adapter rejecting the setting.
• Ran it locally against Kafka. Batches closed exactly where expected, for example at 878 messages / 200,184 bytes with 228-byte messages and a 200,000-byte limit.
• Throughput: 500k messages through a DevNull sink, 7 runs each, about 90,700 msg/s without a limit and 90,400 msg/s with one. The difference is within the noise between runs.

🤖 Generated with Claude Code

[evanh]

This functionally makes sense to me but is a very complicated change to add `flush if bytes > max_bytes). I would maybe ask the AI to see if it can make the change more minimal.

[enochtangg]

@evanh Thanks, I managed to simplify it quite a bit. The extra complexity came from avoiding GIL acquisition per message, so we was only measuring bytes every 32 to 256 messages. In practice this probably isn't needed today since the consumer thread already holds the GIL while it runs. To simply, we now check each message and the size is added to a running total, and the batch flushes once the total reaches the limit.