[FLINK-36906] Optimize the logic for determining if a split is finished

[xiaochen-zhou]

When determining if a split is finished, process only the partitions with data from the current fetch instead of all partitions. This reduces unnecessary partition checks and improves performance and resource utilization.

[xiaochen-zhou]

Friendly ping, do you have time to take a look @AHeise 🙏 ?

[AHeise]

Afaik this doesn't work and was the main reason for #100. If you last message is a transaction marker, then you would never check the stop condition on that partition at the point in time.

I'll trigger the CI which should fail for the test that was specifically added for that scenario.

I'll leave this PR open untli we figured out if it can indeed be improved. Please double check the linked PR and the respective ticket.

[xiaochen-zhou]

Afaik this doesn't work and was the main reason for #100. If you last message is a transaction marker, then you would never check the stop condition on that partition at the point in time.

I'll trigger the CI which should fail for the test that was specifically added for that scenario.

I'll leave this PR open untli we figured out if it can indeed be improved. Please double check the linked PR and the respective ticket.

OK.