New commit-listeners are lost in case of commit failure

[dforsl]

We've for some time had an issue that when temporary network errors occur, it seems like one or more partitions sometime simply stop processing. We've observed the following:

1. Partition X starts the commit procedure
2. Partition Y then also wants to commit, but a commit is already in progress so the offset is "enqueued"
3. The current commit, with X's offset, fails and will be retried by ~merging the offsets with any new offsets that were queued for commit whilst the attempt was made.
4. A new commit-attempt is made, with offsets for both X and Y and succeeds. The callback for X is executed, but not the one for Y. Typically, the flow that wanted to commit Y's offset is stuck waiting.

Expected Behavior

The Mono from ReceiverOffset.commit should complete after 4 above.

Actual Behavior

It never completes.

Possible Solution

In CommittableBatch.restoreOffsets, we are passed CommitArgs to restore. Offsets are "merged" with any new that have been added, whilst callbackEmitters are simply overwritten. A quick look suggests that merging the lists should solve the issue, but there might be other things I've missed of course.

• Reactor version(s) used: reactor-kafka 1.1.1.RELEASE