private void flushCurrentBatch() {
            if (currentBatch != null) {
                try {
                    // Write the records to the log and update the last written offset.
                    long offset = partitionWriter.append(
                        tp,
                        currentBatch.verificationGuard,
                        currentBatch.builder.build()
                    );
                    coordinator.updateLastWrittenOffset(offset);

                    if (offset != currentBatch.nextOffset) {
                        log.error("The state machine of the coordinator {} is out of sync with the underlying log. " +
                            "The last written offset returned is {} while the coordinator expected {}. The coordinator " +
                            "will be reloaded in order to re-synchronize the state machine.",
                            tp, offset, currentBatch.nextOffset);
                        // Transition to FAILED state to unload the state machine and complete
                        // exceptionally all the pending operations.
                        transitionTo(CoordinatorState.FAILED);
                        // Transition to LOADING to trigger the restoration of the state.
                        transitionTo(CoordinatorState.LOADING);
                        // 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.
                    for (DeferredEvent event : currentBatch.deferredEvents) {
                        deferredEventQueue.add(offset, event);
                    }

                    // 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 (or pending) batch to the log. When the batch is written
locally, a new snapshot is created in the snapshot registry and the events
associated with the batch are added to the deferred event queue.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java