Repository: apache/kafka
File: coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java (Java)

Code before:
                        // Thrown NotCoordinatorException to fail the operation that
                        // triggered the write. We use NotCoordinatorException to be
                        // consistent with the transition to FAILED.
                        throw Errors.NOT_COORDINATOR.exception();
                    }

                    // Add all the pending deferred events to the deferred event queue.
                    deferredEventQueue.add(offset, currentBatch.deferredEvents);

                    // Free up the current batch.
                    freeCurrentBatch();
                } catch (Throwable t) {
                    log.error("Writing records to {} failed due to: {}.", tp, t.getMessage());
                    failCurrentBatch(t);
                    // We rethrow the exception for the caller to handle it too.
                    throw t;
                }
            }
        }

        /**
         * Flushes the current batch if it is transactional or if it has passed the append linger time.
         */
        private void maybeFlushCurrentBatch(long currentTimeMs) {
            if (currentBatch != null) {
                if (currentBatch.builder.isTransactional() || (currentBatch.appendTimeMs - currentTimeMs) >= appendLingerMs) {
                    flushCurrentBatch();
                }
            }
        }

        /**
         * Fails the current batch, reverts to the snapshot to the base/start offset of the
         * batch, fails all the associated events.
         */
        private void failCurrentBatch(Throwable t) {
            if (currentBatch != null) {
                coordinator.revertLastWrittenOffset(currentBatch.baseOffset);
                currentBatch.deferredEvents.complete(t);
                freeCurrentBatch();
            }
        }

        /**
         * Allocates a new batch if none already exists.
         */
        private void maybeAllocateNewBatch(
            long producerId,
            short producerEpoch,
            VerificationGuard verificationGuard,
            long currentTimeMs

Review comment (performance):
@majialoong and I were discussing the condition `(currentBatch.appendTimeMs - currentTimeMs) >= appendLingerMs`. The correct version seems to be `(currentTimeMs - currentBatch.appendTimeMs) >= appendLingerMs`. Or we can remove it, since a `lingerTimeoutTask` already exists

WDYT?

Code after:
                        // Thrown NotCoordinatorException to fail the operation that
                        // triggered the write. We use NotCoordinatorException to be
                        // consistent with the transition to FAILED.
                        throw Errors.NOT_COORDINATOR.exception();
                    }

                    // Add all the pending deferred events to the deferred event queue.
                    deferredEventQueue.add(offset, currentBatch.deferredEvents);

                    // Free up the current batch.
                    freeCurrentBatch();
                } catch (Throwable t) {
                    log.error("Writing records to {} failed due to: {}.", tp, t.getMessage());
                    failCurrentBatch(t);
                    // We rethrow the exception for the caller to handle it too.
                    throw t;
                }
            }
        }

        /**
         * Flushes the current batch if it is transactional, if it has passed the append linger time, or if it is full.
         */
        private void maybeFlushCurrentBatch(long currentTimeMs) {
            if (currentBatch != null) {
                if (currentBatch.builder.isTransactional() || (currentBatch.appendTimeMs - currentTimeMs) >= appendLingerMs || !currentBatch.builder.hasRoomFor(0)) {
                    flushCurrentBatch();
                }
            }
        }

        /**
         * Fails the current batch, reverts to the snapshot to the base/start offset of the
         * batch, fails all the associated events.
         */
        private void failCurrentBatch(Throwable t) {
            if (currentBatch != null) {
                coordinator.revertLastWrittenOffset(currentBatch.baseOffset);
                currentBatch.deferredEvents.complete(t);
                freeCurrentBatch();
            }
        }

        /**
         * Allocates a new batch if none already exists.
         */
        private void maybeAllocateNewBatch(
            long producerId,
            short producerEpoch,
            VerificationGuard verificationGuard,
            long currentTimeMs