dajac commented on code in PR #22969:
URL: https://github.com/apache/kafka/pull/22969#discussion_r3812682964
##########
coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java:
##########
@@ -697,12 +697,28 @@ private void flushCurrentBatch() {
if (currentBatch.builder.numRecords() == 0) {
// The only way we can get here is if append() has
failed in an unexpected
// way and left an empty batch. Try to clean it up.
- log.debug("Tried to flush an empty batch for {}.", tp);
+ log.warn("Tried to flush an empty batch for {}.", tp);
// There should not be any deferred events attached to
the batch. We fail
// the batch just in case. As a side effect,
coordinator state is also
- // reverted, but there should be no changes since the
batch was empty.
- failCurrentBatch(new IllegalStateException("Record
batch was empty"));
- return;
+ // reverted. There are (at least) two cases to
consider which can result in
+ // revertible coordinator state changes:
+ // 1. The previous failed write operation was
non-replaying and updated
+ // coordinator state directly. We can't persist
its records because they
+ // weren't added to the batch successfully, so we
must revert the
+ // coordinator state changes to keep the
coordinator state in sync with
+ // the log.
+ // 2. The current write operation is non-replaying
and has updated
+ // coordinator state directly. After this flush,
it will append its
+ // records to the log. We must revert state
changes due to case 1, which
+ // means we must also throw to avoid appending
records to the log. We
+ // pick a retriable exception, so that a previous
failed write operation
+ // doesn't cause a fatal error.
+ //
+ // NB: The handling of empty batches is still not
fully correct. If there
+ // is no attempt to flush the empty batch before
appending new records,
+ // we end up keeping coordinator state changes
from case 1 without
+ // corresponding records written to the log.
+ throw Errors.COORDINATOR_NOT_AVAILABLE.exception();
Review Comment:
Reverted the commit: [Revert "fixup: throw when failing empty
batch"](https://github.com/apache/kafka/pull/22969/commits/0b8620619a95232a0375007b512883edfdeb8d9d)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
