@Override
                        public void run() {
                            // An event to flush the batch is pushed to the front of the queue
                            // to ensure that the linger time is respected.
                            enqueueFirst(new CoordinatorInternalEvent("FlushBatch", tp, () -> {
                                if (this.isCancelled()) return;
                                withActiveContextOrThrow(tp, CoordinatorContext::flushCurrentBatch);
                            }));
                        }
Allocates a new batch if none already exists.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java