This looks clean. One thing I couldn't figure out is how we make sure that some of the partition receivers don't buffer unbounded messages while waiting for stragglers.

Hmm, I thought I had that figured out, but now that I look again, I was wrong.

I'll add a per-partition queuing limit.

This looks clean. One thing I couldn't figure out is how we make sure that some of the partition receivers don't buffer unbounded messages while waiting for stragglers.

Hmm, I thought I had that figured out, but now that I look again, I was wrong.

I'll add a per-partition queuing limit.

Can this affect the connector even without this new feature, in the sense that there's nothing preventing partition readers from accumulating unbounded data?

This looks clean. One thing I couldn't figure out is how we make sure that some of the partition receivers don't buffer unbounded messages while waiting for stragglers.

Hmm, I thought I had that figured out, but now that I look again, I was wrong.
I'll add a per-partition queuing limit.

Can this affect the connector even without this new feature, in the sense that there's nothing preventing partition readers from accumulating unbounded data?

I took another look.

We account data as buffered as soon as it comes in on the partition reader thread, so the total amount buffered is limited by the buffer limit plus however long it takes the controller to tell us to pause. So this will limit the amount buffered in either case.

However, that means there's another complication: we need to make sure that every partition can buffer at least one record, even if the overall buffers are filled, because otherwise we can be full of buffered records that can't be input to the circuit.

Also, I think that the idea of records that are buffered but can't be input to the circuit will cause livelock in the controller, which will continuously try to start a step since we've told it that there is data buffered.

So, there might need to be further distinction between the modes:

• In synchronized mode, we need to do per-partition backpressure and we need to (somehow) account records as buffered only when they are available to be input to the circuit.
• In non-synchronized mode, we can use the existing strategies.

I'll continue to think about this today. Thanks for asking probing questions.