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

Code before:
                                recordToReplay
                            );
                        }

                        currentBatch.builder.append(recordToAppend);
                        currentBatch.nextOffset++;
                    } catch (Throwable t) {
                        log.error("Replaying record {} to {} failed due to: {}.", recordToReplay, tp, t.getMessage());

                        // Add the event to the list of pending events associated with the last
                        // batch in order to fail it too.
                        currentBatch.deferredEvents.add(event);

                        // If an exception is thrown, we fail the entire batch. Exceptions should be
                        // really exceptional in this code path and they would usually be the results
                        // of bugs preventing records to be replayed.
                        failCurrentBatch(t);

                        return;
                    }
                }

                // Add the event to the list of pending events associated with the batch.
                currentBatch.deferredEvents.add(event);

                if (isAtomic && !currentBatch.builder.hasRoomFor(0)) {
                    flushCurrentBatch();
                } else {
                    // Write the current batch if it is transactional or if the linger timeout
                    // has expired.
                    // If flushing fails, we don't catch the exception in order to let
                    // the caller fail the current operation. 
                    maybeFlushCurrentBatch(currentTimeMs);
                }
            }
        }

        /**
         * Completes a transaction.
         *
         * @param producerId        The producer id.
         * @param producerEpoch     The producer epoch.
         * @param coordinatorEpoch  The coordinator epoch of the transaction coordinator.
         * @param result            The transaction result.
         * @param event             The event that must be completed when the
         *                          control record is written.
         */
        private void completeTransaction(
            long producerId,
            short producerEpoch,
            int coordinatorEpoch,

Review comment (performance):
```
                    // Flush the batch if its uncompressed size is larger than the max batch size.
                    // It would get flushed on the next append() anyway.
```

Code after:
                        // of bugs preventing records to be replayed.
                        failCurrentBatch(t);

                        return;
                    }
                }

                // Add the event to the list of pending events associated with the batch.
                currentBatch.deferredEvents.add(event);

                // Write the current batch if it is transactional, if the linger timeout
                // has expired, or if it is full. 
                // If flushing fails, we don't catch the exception in order to let
                // the caller fail the current operation.
                maybeFlushCurrentBatch(currentTimeMs);
            }
        }

        /**
         * Completes a transaction.
         *
         * @param producerId        The producer id.
         * @param producerEpoch     The producer epoch.
         * @param coordinatorEpoch  The coordinator epoch of the transaction coordinator.
         * @param result            The transaction result.
         * @param event             The event that must be completed when the
         *                          control record is written.
         */
        private void completeTransaction(
            long producerId,
            short producerEpoch,
            int coordinatorEpoch,
            TransactionResult result,
            DeferredEvent event
        ) {
            if (state != CoordinatorState.ACTIVE) {
                throw new IllegalStateException("Coordinator must be active to complete a transaction");
            }

            // The current batch must be written before the transaction marker is written
            // in order to respect the order.
            flushCurrentBatch();

            long prevLastWrittenOffset = coordinator.lastWrittenOffset();
            try {
                coordinator.replayEndTransactionMarker(
                    producerId,
                    producerEpoch,
                    result
                );