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

Code before:
     * @param runtimeMetrics                    The runtime metrics.
     * @param coordinatorMetrics                The coordinator metrics.
     * @param serializer                        The serializer.
     * @param compression                       The compression codec.
     * @param appendLingerMs                    The append linger time in ms.
     * @param executorService                   The executor service.
     * @param appendMaxBufferSizeSupplier       The append max buffer size supplier.
     */
    @SuppressWarnings("checkstyle:ParameterNumber")
    private CoordinatorRuntime(
        String logPrefix,
        LogContext logContext,
        CoordinatorEventProcessor processor,
        PartitionWriter partitionWriter,
        CoordinatorLoader<U> loader,
        CoordinatorShardBuilderSupplier<S, U> coordinatorShardBuilderSupplier,
        Time time,
        Timer timer,
        Duration defaultWriteTimeout,
        CoordinatorRuntimeMetrics runtimeMetrics,
        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.get()).sum()
        );
    }

    /**
     * Throws a NotCoordinatorException exception if the runtime is not

Review comment (refactor):
nit: Could we move this parameter up, just after `appendLingerMs`? So the `append...` parameters are grouped together. The same for the assignment in the constructor.

Code after:
     * @param timer                             The system timer.
     * @param defaultWriteTimeout               The write operation timeout.
     * @param runtimeMetrics                    The runtime metrics.
     * @param coordinatorMetrics                The coordinator metrics.
     * @param serializer                        The serializer.
     * @param compression                       The compression codec.
     * @param appendLingerMs                    The append linger time in ms.
     * @param executorService                   The executor service.
     * @param cachedBufferMaxBytesSupplier      The cached buffer max bytes supplier.
     */
    @SuppressWarnings("checkstyle:ParameterNumber")
    private CoordinatorRuntime(
        String logPrefix,
        LogContext logContext,
        CoordinatorEventProcessor processor,
        PartitionWriter partitionWriter,
        CoordinatorLoader<U> loader,
        CoordinatorShardBuilderSupplier<S, U> coordinatorShardBuilderSupplier,
        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()
        );
    }