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.cachedBufferSize).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):
Using `Supplier<Long>` necessiates synchronization, as the `cachedBufferSize` value is accessed by different threads

Code after:
        Time time,
        Timer timer,
        Duration defaultWriteTimeout,
        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 {