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

Code before:
        CoordinatorMetrics coordinatorMetrics,
        Serializer<U> serializer,
        Compression compression,
        OptionalInt appendLingerMs,
        ExecutorService executorService,
        Supplier<Integer> appendMaxBufferSizeSupplier
    ) {
        this.logPrefix = logPrefix;
        this.log = logContext.logger(CoordinatorRuntime.class);
        this.time = time;
        this.timer = timer;
        this.defaultWriteTimeout = defaultWriteTimeout;
        this.coordinators = new ConcurrentHashMap<>();
        this.processor = processor;
        this.partitionWriter = partitionWriter;
        this.loader = loader;
        this.coordinatorShardBuilderSupplier = coordinatorShardBuilderSupplier;
        this.runtimeMetrics = runtimeMetrics;
        this.coordinatorMetrics = coordinatorMetrics;
        this.serializer = serializer;
        this.compression = compression;
        this.appendLingerMs = appendLingerMs;
        this.executorService = executorService;
        this.appendMaxBufferSizeSupplier = appendMaxBufferSizeSupplier;
        this.runtimeMetrics.registerAppendBufferSizeGauge(
            () -> coordinators.values().stream().mapToLong(c -> c.bufferSupplier.size()).sum()
        );
    }

    /**
     * Throws a NotCoordinatorException exception if the runtime is not
     * running.
     */
    private void throwIfNotRunning() {
        if (!isRunning.get()) {
            throw Errors.NOT_COORDINATOR.exception();
        }
    }

    /**
     * Enqueues a new event at the end of the processing queue.
     *
     * @param event The event.
     * @throws NotCoordinatorException If the event processor is closed.
     */
    private void enqueueLast(CoordinatorEvent event) {
        try {
            processor.enqueueLast(event);
        } catch (RejectedExecutionException ex) {
            throw new NotCoordinatorException("Can't accept an event because the processor is closed", ex);
        }

Review comment (performance):
This appears to create an implicit call chain between `CoordinatorRuntime` and `CoordinatorRuntimeMetricsImpl`. Perhaps, `CoordinatorRuntimeMetricsImpl` could maintain a `AtomicLong` variable, and we could update its value via `freeCurrentBatch`. For example:
```java
            if (currentBatch.builder.buffer().capacity() <= maxBufferSize) {
                var before = bufferSupplier.size();
                bufferSupplier.release(currentBatch.builder.buffer());
                runtimeMetrics.recordAppendBufferSize(bufferSupplier.size() - before);
            } else if (currentBatch.buffer.capacity() <= maxBufferSize) {
                var before = bufferSupplier.size();
                bufferSupplier.release(currentBatch.buffer);
                runtimeMetrics.recordAppendBufferSize(bufferSupplier.size() - before);
            } else {
                runtimeMetrics.recordAppendBufferDiscarded();
            }
```

Code after:
        CoordinatorRuntimeMetrics runtimeMetrics,
        CoordinatorMetrics coordinatorMetrics,
        Serializer<U> serializer,
        Compression compression,
        OptionalInt appendLingerMs,
        ExecutorService executorService,
        Supplier<Integer> cachedBufferMaxBytesSupplier
    ) {
        this.logPrefix = logPrefix;
        this.log = logContext.logger(CoordinatorRuntime.class);
        this.time = time;
        this.timer = timer;
        this.defaultWriteTimeout = defaultWriteTimeout;
        this.coordinators = new ConcurrentHashMap<>();
        this.processor = processor;
        this.partitionWriter = partitionWriter;
        this.loader = loader;
        this.coordinatorShardBuilderSupplier = coordinatorShardBuilderSupplier;
        this.runtimeMetrics = runtimeMetrics;
        this.coordinatorMetrics = coordinatorMetrics;
        this.serializer = serializer;
        this.compression = compression;
        this.appendLingerMs = appendLingerMs;
        this.executorService = executorService;
        this.cachedBufferMaxBytesSupplier = cachedBufferMaxBytesSupplier;
        this.runtimeMetrics.registerBufferCacheSizeGauge(
            () -> coordinators.values().stream().mapToLong(c -> c.cachedBufferSize.get()).sum()
        );
    }

    /**
     * Throws a NotCoordinatorException exception if the runtime is not
     * running.
     */
    private void throwIfNotRunning() {
        if (!isRunning.get()) {
            throw Errors.NOT_COORDINATOR.exception();
        }
    }

    /**
     * Enqueues a new event at the end of the processing queue.
     *
     * @param event The event.
     * @throws NotCoordinatorException If the event processor is closed.
     */
    private void enqueueLast(CoordinatorEvent event) {
        try {
            processor.enqueueLast(event);
        } catch (RejectedExecutionException ex) {
            throw new NotCoordinatorException("Can't accept an event because the processor is closed", ex);