@Override
        public void run() {
            try {
                // Get the context of the coordinator or fail if the coordinator is not in active state.
                withActiveContextOrThrow(tp, context -> {
                    // Execute the operation.
                    result = op.generateRecordsAndResult(context.coordinator.coordinator());

                    // Append the records and replay them to the state machine.
                    context.append(
                        producerId,
                        producerEpoch,
                        verificationGuard,
                        result.records(),
                        result.replayRecords(),
                        result.isAtomic(),
                        this
                    );

                    // If the operation is not done, create an operation timeout.
                    if (!future.isDone()) {
                        operationTimeout = new OperationTimeout(tp, this, writeTimeout.toMillis());
                        timer.add(operationTimeout);
                    }
                });
            } catch (Throwable t) {
                complete(t);
            }
        }
Called by the CoordinatorEventProcessor when the event is executed.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java