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
                );

                long offset = partitionWriter.append(
                    tp,
                    VerificationGuard.SENTINEL,
                    MemoryRecords.withEndTransactionMarker(
                        time.milliseconds(),
                        producerId,
                        producerEpoch,
                        new EndTransactionMarker(
                            result == TransactionResult.COMMIT ? ControlRecordType.COMMIT : ControlRecordType.ABORT,
                            coordinatorEpoch
                        )
                    )
                );
                coordinator.updateLastWrittenOffset(offset);

                deferredEventQueue.add(offset, event);
            } catch (Throwable t) {
                coordinator.revertLastWrittenOffset(prevLastWrittenOffset);
                event.complete(t);
            }
        }
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.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java