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

Code before:
         * @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.
         */
        private void append(
            long producerId,
            short producerEpoch,
            VerificationGuard verificationGuard,
            List<U> records,
            boolean replay,
            boolean isAtomic,
            DeferredEvent event
        ) {
            if (state != CoordinatorState.ACTIVE) {
                throw new IllegalStateException("Coordinator must be active to append records");
            }

            if (records.isEmpty()) {
                handleEmptyRecords(event);
            } else {
                // If the records are not empty, first, they are applied to the state machine,
                // second, they are appended to the opened batch.
                long currentTimeMs = time.milliseconds();

                // If the current write operation is transactional, the current batch
                // is written before proceeding with it.
                if (producerId != RecordBatch.NO_PRODUCER_ID) {
                    isAtomic = true;
                    // If flushing fails, we don't catch the exception in order to let
                    // the caller fail the current operation.
                    flushCurrentBatch();
                }

                // Allocate a new batch if none exists.
                maybeAllocateNewBatch(
                    producerId,
                    producerEpoch,
                    verificationGuard,
                    currentTimeMs
                );

                // Prepare the records.
                List<SimpleRecord> recordsToAppend = new ArrayList<>(records.size());
                for (U record : records) {

Review comment (suggestion):
We should move the comment out of handleEmptyRecords.
```
            // If the records are empty, it was a read operation after all. The response can be
            // returned once any pending write operations complete.
```

Code after:
         * @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.
         */
        private void append(
            long producerId,
            short producerEpoch,
            VerificationGuard verificationGuard,
            List<U> records,
            boolean replay,
            boolean isAtomic,
            DeferredEvent event
        ) {
            if (state != CoordinatorState.ACTIVE) {
                throw new IllegalStateException("Coordinator must be active to append records");
            }

            if (records.isEmpty()) {
                // If the records are empty, it was a read operation after all. In this case,
                // the response can be returned once any pending write operations complete.
                waitForPendingWrites(event);
            } else {
                // If the records are not empty, first, they are applied to the state machine,
                // second, they are appended to the opened batch.
                long currentTimeMs = time.milliseconds();

                // If the current write operation is transactional, the current batch
                // is written before proceeding with it.
                if (producerId != RecordBatch.NO_PRODUCER_ID) {
                    isAtomic = true;
                    // If flushing fails, we don't catch the exception in order to let
                    // the caller fail the current operation.
                    flushCurrentBatch();
                }

                // Allocate a new batch if none exists.
                maybeAllocateNewBatch(
                    producerId,
                    producerEpoch,
                    verificationGuard,
                    currentTimeMs
                );

                // Prepare the records.
                List<SimpleRecord> recordsToAppend = new ArrayList<>(records.size());
                for (U record : records) {
                    recordsToAppend.add(new SimpleRecord(