Repository: apache/kafka
File: group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java (Java)

Code before:

            CoordinatorEventProcessor processor = new MultiThreadedEventProcessor(
                logContext,
                "group-coordinator-event-processor-",
                config.numThreads(),
                time,
                coordinatorRuntimeMetrics
            );

            CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
                new CoordinatorRuntime.Builder<GroupCoordinatorShard, CoordinatorRecord>()
                    .withTime(time)
                    .withTimer(timer)
                    .withLogPrefix(logPrefix)
                    .withLogContext(logContext)
                    .withEventProcessor(processor)
                    .withPartitionWriter(writer)
                    .withLoader(loader)
                    .withCoordinatorShardBuilderSupplier(supplier)
                    .withWriteTimeout(Duration.ofMillis(config.offsetCommitTimeoutMs()))
                    .withCoordinatorRuntimeMetrics(coordinatorRuntimeMetrics)
                    .withCoordinatorMetrics(groupCoordinatorMetrics)
                    .withSerializer(new GroupCoordinatorRecordSerde())
                    .withCompression(Compression.of(config.offsetTopicCompressionType()).build())
                    .withAppendLingerMs(config.appendLingerMs())
                    .withExecutorService(Executors.newFixedThreadPool(config.numExecutorThreads()))
                    .withCachedBufferMaxBytesSupplier(config::cachedBufferMaxBytes)
                    .build();

            return new GroupCoordinatorService(
                logContext,
                config,
                runtime,
                groupCoordinatorMetrics,
                groupConfigManager,
                persister,
                timer,
                partitionMetadataClient
            );
        }
    }

    /**
     * The logger.
     */
    private final Logger log;

    /**
     * The group coordinator configurations.
     */
    private final GroupCoordinatorConfig config;

Review comment (suggestion):
While we are here, would it be possible to give a proper name to the threads used here? We could follow the name we use for the config so administrators can connect the dots.

Code after:
                    .withAuthorizerPlugin(authorizerPlugin);

            CoordinatorEventProcessor processor = new MultiThreadedEventProcessor(
                logContext,
                "group-coordinator-event-processor-",
                config.numThreads(),
                time,
                coordinatorRuntimeMetrics
            );

            CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
                new CoordinatorRuntime.Builder<GroupCoordinatorShard, CoordinatorRecord>()
                    .withTime(time)
                    .withTimer(timer)
                    .withLogPrefix(logPrefix)
                    .withLogContext(logContext)
                    .withEventProcessor(processor)
                    .withPartitionWriter(writer)
                    .withLoader(loader)
                    .withCoordinatorShardBuilderSupplier(supplier)
                    .withWriteTimeout(Duration.ofMillis(config.offsetCommitTimeoutMs()))
                    .withCoordinatorRuntimeMetrics(coordinatorRuntimeMetrics)
                    .withCoordinatorMetrics(groupCoordinatorMetrics)
                    .withSerializer(new GroupCoordinatorRecordSerde())
                    .withCompression(Compression.of(config.offsetTopicCompressionType()).build())
                    .withAppendLingerMs(config.appendLingerMs())
                    .withExecutorService(Executors.newFixedThreadPool(
                        config.numBackgroundThreads(),
                        ThreadUtils.createThreadFactory("group-coordinator-background-%d", false)
                    ))
                    .withCachedBufferMaxBytesSupplier(config::cachedBufferMaxBytes)
                    .build();

            return new GroupCoordinatorService(
                logContext,
                config,
                runtime,
                groupCoordinatorMetrics,
                groupConfigManager,
                persister,
                timer,
                partitionMetadataClient
            );
        }
    }

    /**
     * The logger.
     */
    private final Logger log;