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

Code before:
                            }));
                        }
                    });
                    CoordinatorRuntime.this.timer.add(lingerTimeoutTask.get());
                }

                currentBatch = new CoordinatorBatch(
                    log,
                    prevLastWrittenOffset,
                    currentTimeMs,
                    maxBatchSize,
                    verificationGuard,
                    buffer,
                    builder,
                    lingerTimeoutTask
                );
            }
        }

        /**
         * Handles the case where append() is called with no records.
         *
         * @param event             The event that must be completed when the 
         *                          records are written.
         */
        private void handleEmptyRecords(DeferredEvent event) {
            // If the records are empty, it was a read operation after all. In this case,
            // the response can be returned directly iff there are no pending write operations;
            // otherwise, the read needs to wait on the last write operation to be completed.
            if (currentBatch != null && currentBatch.builder.numRecords() > 0) {
                currentBatch.deferredEvents.add(event);
            } else {
                if (coordinator.lastCommittedOffset() < coordinator.lastWrittenOffset()) {
                    deferredEventQueue.add(coordinator.lastWrittenOffset(), DeferredEventCollection.of(log, event));
                } else {
                    event.complete(null);
                }
            }
        }

        /**
         * Appends records to the log and replay them to the state machine.
         *
         * @param producerId        The producer id.
         * @param producerEpoch     The producer epoch.
         * @param verificationGuard The verification guard.
         * @param records           The records to append.
         * @param replay            A boolean indicating whether the records
         *                          must be replayed or not.
         * @param isAtomic          A boolean indicating whether the records
         *                          must be written atomically or not.

Review comment (suggestion):
We can rename this to `waitForPendingWrites` or similar and update the javadoc.

Code after:
                            }));
                        }
                    });
                    CoordinatorRuntime.this.timer.add(lingerTimeoutTask.get());
                }

                currentBatch = new CoordinatorBatch(
                    log,
                    prevLastWrittenOffset,
                    currentTimeMs,
                    maxBatchSize,
                    verificationGuard,
                    buffer,
                    builder,
                    lingerTimeoutTask
                );
            }
        }

        /**
         * Completes the given event once all pending writes are completed. 
         *
         * @param event             The event to complete once all pending 
         *                          writes are completed.
         */
        private void waitForPendingWrites(DeferredEvent event) {
            if (currentBatch != null && currentBatch.builder.numRecords() > 0) {
                currentBatch.deferredEvents.add(event);
            } else {
                if (coordinator.lastCommittedOffset() < coordinator.lastWrittenOffset()) {
                    deferredEventQueue.add(coordinator.lastWrittenOffset(), DeferredEventCollection.of(log, event));
                } else {
                    event.complete(null);
                }
            }
        }

        /**
         * Appends records to the log and replay them to the state machine.
         *
         * @param producerId        The producer id.
         * @param producerEpoch     The producer epoch.
         * @param verificationGuard The verification guard.
         * @param records           The records to append.
         * @param replay            A boolean indicating whether the records
         *                          must be replayed or not.
         * @param isAtomic          A boolean indicating whether the records
         *                          must be written atomically or not.
         * @param event             The event that must be completed when the
         *                          records are written.
         */