KafkaStreamPublisher: Fix ArrayIndexOutOfBoundsException

[nbauernfeind]

Fixes #4516

Since we append to these chunks off of the update-graph, we only flush when we need to acquire a new set of chunks. Note that StreamPublisherBase#flush does not check to see if the chunk has .size() > 0 but IntChunkColumnSource.addChunk does ensure that you are not adding empty chunks. (Noting that flush() is what happens first during an update graph cycle for the blink table adapter.)

We attempt to be clever to only allocate a new chunk if we are going to write a row to it. To do this, we pre-decrement a counter remaining and if the chunk is full we flush. Note this check if (--remaining == 0) { is an off-by-one as we have not yet written the row we are counting. That isn't a big deal though; that just means we never actually fill the chunks. The AIOBE actually comes from recalculating the new remaining whenever we flush -- we are not accounting for the row that is about to be written.

This causes us to leave the method with remaining == 1 but the chunks are already full.