Kafka Merge is getting stall because of intermediate partition offset state

[akrambek]

Describe the bug

This is a current understanding of the issue

zilla: offset <= stable <= latest
which means offset <= stable, stable <= latest, offset <= latest
kafka: offset < stable <= high water mark
which means offset < stable, stable <= high water mark, offset < high water mark

Due to this invariant sometimes we may get into a state where the fix to send a flush when stableOffset or latestOffset has advanced (with no records) after records have been sent or caught up to the latest offset so at least one FLUSH was already sent.