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

Code before:
import static org.apache.kafka.common.config.ConfigDef.Type.INT;
import static org.apache.kafka.common.config.ConfigDef.Type.LIST;
import static org.apache.kafka.common.config.ConfigDef.Type.LONG;
import static org.apache.kafka.common.config.ConfigDef.Type.SHORT;
import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
import static org.apache.kafka.common.utils.Utils.maybeCloseQuietly;
import static org.apache.kafka.common.utils.Utils.require;

/**
 * The group coordinator configurations.
 * This configuration utilizes several local variables instead of calling AbstractConfig#get.... as all configs here
 * are static and non-dynamic, with some being accessed extremely frequently (e.g., offsets.commit.timeout.ms).
 * Using local variable is advantageous as it avoids the overhead of repeatedly looking up these configurations in AbstractConfig.
 */
public class GroupCoordinatorConfig {
    ///
    /// Group coordinator configs
    ///
    public static final String GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG = "group.coordinator.rebalance.protocols";
    public static final String GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DOC = "The list of enabled rebalance protocols." +
            "The " + Group.GroupType.STREAMS + " rebalance protocol is in early access and therefore must not be used in production.";
    public static final List<String> GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DEFAULT = List.of(
        Group.GroupType.CLASSIC.toString(),
        Group.GroupType.CONSUMER.toString(),
        Group.GroupType.STREAMS.toString());
    public static final List<String> GROUP_COORDINATOR_REBALANCE_PROTOCOLS_ALLOWED = List.of(
        Group.GroupType.CLASSIC.toString(),
        Group.GroupType.CONSUMER.toString(),
        Group.GroupType.STREAMS.toString()
    );
    public static final String GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG = "group.coordinator.append.linger.ms";
    public static final String GROUP_COORDINATOR_APPEND_LINGER_MS_DOC = "The duration in milliseconds that the coordinator will " +
        "wait for writes to accumulate before flushing them to disk. Increasing this value improves write efficiency and batch size, " +
        "but also increases the response latency for requests, as the coordinator must wait for batches to be flushed to " +
        "disk before completing request processing. Transactional writes are not accumulated.";
    public static final int GROUP_COORDINATOR_APPEND_LINGER_MS_DEFAULT = 5;

    public static final String GROUP_COORDINATOR_NUM_THREADS_CONFIG = "group.coordinator.threads";
    public static final String GROUP_COORDINATOR_NUM_THREADS_DOC = "The number of threads used by the group coordinator.";
    public static final int GROUP_COORDINATOR_NUM_THREADS_DEFAULT = 4;

    public static final String OFFSETS_LOAD_BUFFER_SIZE_CONFIG = "offsets.load.buffer.size";
    public static final int OFFSETS_LOAD_BUFFER_SIZE_DEFAULT = 5 * 1024 * 1024;
    public static final String OFFSETS_LOAD_BUFFER_SIZE_DOC = "Batch size for reading from the offsets segments when loading group metadata " +
        " into the cache (soft-limit, overridden if records are too large).";

    public static final String OFFSET_COMMIT_TIMEOUT_MS_CONFIG = "offsets.commit.timeout.ms";
    public static final int OFFSET_COMMIT_TIMEOUT_MS_DEFAULT = 5000;
    public static final String OFFSET_COMMIT_TIMEOUT_MS_DOC = "Offset commit will be delayed until all replicas for the offsets topic receive the commit " +
        "or this timeout is reached. This is similar to the producer request timeout. This is applied to all the writes made by the coordinator.";


Review comment (suggestion):
I think we could reuse the `GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DEFAULT` constant since they are exactly the same.

Code after:
import static org.apache.kafka.common.config.ConfigDef.Type.INT;
import static org.apache.kafka.common.config.ConfigDef.Type.LIST;
import static org.apache.kafka.common.config.ConfigDef.Type.LONG;
import static org.apache.kafka.common.config.ConfigDef.Type.SHORT;
import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
import static org.apache.kafka.common.utils.Utils.maybeCloseQuietly;
import static org.apache.kafka.common.utils.Utils.require;

/**
 * The group coordinator configurations.
 * This configuration utilizes several local variables instead of calling AbstractConfig#get.... as all configs here
 * are static and non-dynamic, with some being accessed extremely frequently (e.g., offsets.commit.timeout.ms).
 * Using local variable is advantageous as it avoids the overhead of repeatedly looking up these configurations in AbstractConfig.
 */
public class GroupCoordinatorConfig {
    ///
    /// Group coordinator configs
    ///
    public static final String GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG = "group.coordinator.rebalance.protocols";
    public static final String GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DOC = "The list of enabled rebalance protocols." +
            "The " + Group.GroupType.STREAMS + " rebalance protocol is in early access and therefore must not be used in production.";
    public static final List<String> GROUP_COORDINATOR_REBALANCE_PROTOCOLS_DEFAULT = List.of(
        Group.GroupType.CLASSIC.toString(),
        Group.GroupType.CONSUMER.toString(),
        Group.GroupType.STREAMS.toString());
    public static final String GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG = "group.coordinator.append.linger.ms";
    public static final String GROUP_COORDINATOR_APPEND_LINGER_MS_DOC = "The duration in milliseconds that the coordinator will " +
        "wait for writes to accumulate before flushing them to disk. Increasing this value improves write efficiency and batch size, " +
        "but also increases the response latency for requests, as the coordinator must wait for batches to be flushed to " +
        "disk before completing request processing. Transactional writes are not accumulated.";
    public static final int GROUP_COORDINATOR_APPEND_LINGER_MS_DEFAULT = 5;

    public static final String GROUP_COORDINATOR_NUM_THREADS_CONFIG = "group.coordinator.threads";
    public static final String GROUP_COORDINATOR_NUM_THREADS_DOC = "The number of threads used by the group coordinator.";
    public static final int GROUP_COORDINATOR_NUM_THREADS_DEFAULT = 4;

    public static final String OFFSETS_LOAD_BUFFER_SIZE_CONFIG = "offsets.load.buffer.size";
    public static final int OFFSETS_LOAD_BUFFER_SIZE_DEFAULT = 5 * 1024 * 1024;
    public static final String OFFSETS_LOAD_BUFFER_SIZE_DOC = "Batch size for reading from the offsets segments when loading group metadata " +
        " into the cache (soft-limit, overridden if records are too large).";

    public static final String OFFSET_COMMIT_TIMEOUT_MS_CONFIG = "offsets.commit.timeout.ms";
    public static final int OFFSET_COMMIT_TIMEOUT_MS_DEFAULT = 5000;
    public static final String OFFSET_COMMIT_TIMEOUT_MS_DOC = "Offset commit will be delayed until all replicas for the offsets topic receive the commit " +
        "or this timeout is reached. This is similar to the producer request timeout. This is applied to all the writes made by the coordinator.";

    public static final String OFFSETS_TOPIC_PARTITIONS_CONFIG = "offsets.topic.num.partitions";
    public static final int OFFSETS_TOPIC_PARTITIONS_DEFAULT = 50;
    public static final String OFFSETS_TOPIC_PARTITIONS_DOC = "The number of partitions for the offset commit topic (should not change after deployment).";

    public static final String OFFSETS_TOPIC_SEGMENT_BYTES_CONFIG = "offsets.topic.segment.bytes";