You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "brandboat (via GitHub)" <gi...@apache.org> on 2024/07/04 16:33:31 UTC

[PR] KAFKA-17081 Tweak GroupCoordinatorConfig: re-introduce local attributes and validation [kafka]

brandboat opened a new pull request, #16524:
URL: https://github.com/apache/kafka/pull/16524

   related to https://issues.apache.org/jira/browse/KAFKA-17081
   
   address comments in https://github.com/apache/kafka/pull/16458#issuecomment-2206822223
   
   ### Committer Checklist (excluded from commit message)
   - [ ] Verify design and implementation 
   - [ ] Verify test coverage and CI build status
   - [ ] Verify documentation (including upgrade notes)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-17081 Tweak GroupCoordinatorConfig: re-introduce local attributes and validation [kafka]

Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #16524:
URL: https://github.com/apache/kafka/pull/16524#discussion_r1667247308


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -213,82 +214,151 @@ public class GroupCoordinatorConfig {
      */
     public static final int CLASSIC_GROUP_NEW_MEMBER_JOIN_TIMEOUT_MS = 5 * 60 * 1000;
 
-    private final AbstractConfig config;
+    private final int numThreads;
+    private final int appendLingerMs;
+    private final int consumerGroupSessionTimeoutMs;
+    private final int consumerGroupHeartbeatIntervalMs;
+    private final int consumerGroupMaxSize;
+    private final List<ConsumerGroupPartitionAssignor> consumerGroupAssignors;
+    private final int offsetsTopicSegmentBytes;
+    private final int offsetMetadataMaxSize;
+    private final int classicGroupMaxSize;
+    private final int classicGroupInitialRebalanceDelayMs;
+    private final int classicGroupMinSessionTimeoutMs;
+    private final int classicGroupMaxSessionTimeoutMs;
+    private final long offsetsRetentionCheckIntervalMs;
+    private final long offsetsRetentionMs;
+    private final int offsetCommitTimeoutMs;
+    private final ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy;
+    private final CompressionType offsetTopicCompressionType;
+    private final int offsetsLoadBufferSize;
+    private final int offsetsTopicPartitions;
+    private final short offsetsTopicReplicationFactor;
+    private final short offsetCommitRequiredAcks;
+    private final int consumerGroupMinSessionTimeoutMs;
+    private final int consumerGroupMaxSessionTimeoutMs;
+    private final int consumerGroupMinHeartbeatIntervalMs;
+    private final int consumerGroupMaxHeartbeatIntervalMs;
 
     public GroupCoordinatorConfig(AbstractConfig config) {
-        this.config = config;
+        this.numThreads = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_NUM_THREADS_CONFIG);
+        this.appendLingerMs = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG);
+        this.consumerGroupSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG);
+        this.consumerGroupHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG);
+        this.consumerGroupMaxSize = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG);
+        this.consumerGroupAssignors = config.getConfiguredInstances(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, ConsumerGroupPartitionAssignor.class);
+        this.offsetsTopicSegmentBytes = config.getInt(GroupCoordinatorConfig.OFFSETS_TOPIC_SEGMENT_BYTES_CONFIG);
+        this.offsetMetadataMaxSize = config.getInt(GroupCoordinatorConfig.OFFSET_METADATA_MAX_SIZE_CONFIG);
+        this.classicGroupMaxSize = config.getInt(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG);
+        this.classicGroupInitialRebalanceDelayMs = config.getInt(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG);
+        this.classicGroupMinSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG);
+        this.classicGroupMaxSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG);
+        this.offsetsRetentionCheckIntervalMs = config.getLong(GroupCoordinatorConfig.OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG);
+        this.offsetsRetentionMs = config.getInt(GroupCoordinatorConfig.OFFSETS_RETENTION_MINUTES_CONFIG) * 60L * 1000L;
+        this.offsetCommitTimeoutMs = config.getInt(GroupCoordinatorConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG);
+        this.consumerGroupMigrationPolicy = ConsumerGroupMigrationPolicy.parse(

Review Comment:
   All comments are addressed in the latest commit, thanks



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-17081 Tweak GroupCoordinatorConfig: re-introduce local attributes and validation [kafka]

Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16524:
URL: https://github.com/apache/kafka/pull/16524#discussion_r1667195347


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -213,82 +214,151 @@ public class GroupCoordinatorConfig {
      */
     public static final int CLASSIC_GROUP_NEW_MEMBER_JOIN_TIMEOUT_MS = 5 * 60 * 1000;
 
-    private final AbstractConfig config;
+    private final int numThreads;
+    private final int appendLingerMs;
+    private final int consumerGroupSessionTimeoutMs;
+    private final int consumerGroupHeartbeatIntervalMs;
+    private final int consumerGroupMaxSize;
+    private final List<ConsumerGroupPartitionAssignor> consumerGroupAssignors;
+    private final int offsetsTopicSegmentBytes;
+    private final int offsetMetadataMaxSize;
+    private final int classicGroupMaxSize;
+    private final int classicGroupInitialRebalanceDelayMs;
+    private final int classicGroupMinSessionTimeoutMs;
+    private final int classicGroupMaxSessionTimeoutMs;
+    private final long offsetsRetentionCheckIntervalMs;
+    private final long offsetsRetentionMs;
+    private final int offsetCommitTimeoutMs;
+    private final ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy;
+    private final CompressionType offsetTopicCompressionType;
+    private final int offsetsLoadBufferSize;
+    private final int offsetsTopicPartitions;
+    private final short offsetsTopicReplicationFactor;
+    private final short offsetCommitRequiredAcks;
+    private final int consumerGroupMinSessionTimeoutMs;
+    private final int consumerGroupMaxSessionTimeoutMs;
+    private final int consumerGroupMinHeartbeatIntervalMs;
+    private final int consumerGroupMaxHeartbeatIntervalMs;
 
     public GroupCoordinatorConfig(AbstractConfig config) {
-        this.config = config;
+        this.numThreads = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_NUM_THREADS_CONFIG);
+        this.appendLingerMs = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG);
+        this.consumerGroupSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG);
+        this.consumerGroupHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG);
+        this.consumerGroupMaxSize = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG);
+        this.consumerGroupAssignors = config.getConfiguredInstances(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, ConsumerGroupPartitionAssignor.class);
+        this.offsetsTopicSegmentBytes = config.getInt(GroupCoordinatorConfig.OFFSETS_TOPIC_SEGMENT_BYTES_CONFIG);
+        this.offsetMetadataMaxSize = config.getInt(GroupCoordinatorConfig.OFFSET_METADATA_MAX_SIZE_CONFIG);
+        this.classicGroupMaxSize = config.getInt(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG);
+        this.classicGroupInitialRebalanceDelayMs = config.getInt(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG);
+        this.classicGroupMinSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG);
+        this.classicGroupMaxSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG);
+        this.offsetsRetentionCheckIntervalMs = config.getLong(GroupCoordinatorConfig.OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG);
+        this.offsetsRetentionMs = config.getInt(GroupCoordinatorConfig.OFFSETS_RETENTION_MINUTES_CONFIG) * 60L * 1000L;
+        this.offsetCommitTimeoutMs = config.getInt(GroupCoordinatorConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG);
+        this.consumerGroupMigrationPolicy = ConsumerGroupMigrationPolicy.parse(

Review Comment:
   Please add UT to ensure the incorrect value can cause in contructor.



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -213,82 +214,151 @@ public class GroupCoordinatorConfig {
      */
     public static final int CLASSIC_GROUP_NEW_MEMBER_JOIN_TIMEOUT_MS = 5 * 60 * 1000;
 
-    private final AbstractConfig config;
+    private final int numThreads;
+    private final int appendLingerMs;
+    private final int consumerGroupSessionTimeoutMs;
+    private final int consumerGroupHeartbeatIntervalMs;
+    private final int consumerGroupMaxSize;
+    private final List<ConsumerGroupPartitionAssignor> consumerGroupAssignors;
+    private final int offsetsTopicSegmentBytes;
+    private final int offsetMetadataMaxSize;
+    private final int classicGroupMaxSize;
+    private final int classicGroupInitialRebalanceDelayMs;
+    private final int classicGroupMinSessionTimeoutMs;
+    private final int classicGroupMaxSessionTimeoutMs;
+    private final long offsetsRetentionCheckIntervalMs;
+    private final long offsetsRetentionMs;
+    private final int offsetCommitTimeoutMs;
+    private final ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy;
+    private final CompressionType offsetTopicCompressionType;
+    private final int offsetsLoadBufferSize;
+    private final int offsetsTopicPartitions;
+    private final short offsetsTopicReplicationFactor;
+    private final short offsetCommitRequiredAcks;
+    private final int consumerGroupMinSessionTimeoutMs;
+    private final int consumerGroupMaxSessionTimeoutMs;
+    private final int consumerGroupMinHeartbeatIntervalMs;
+    private final int consumerGroupMaxHeartbeatIntervalMs;
 
     public GroupCoordinatorConfig(AbstractConfig config) {
-        this.config = config;
+        this.numThreads = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_NUM_THREADS_CONFIG);
+        this.appendLingerMs = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG);
+        this.consumerGroupSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG);
+        this.consumerGroupHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG);
+        this.consumerGroupMaxSize = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG);
+        this.consumerGroupAssignors = config.getConfiguredInstances(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, ConsumerGroupPartitionAssignor.class);

Review Comment:
   @brandboat Have you added test for `CONSUMER_GROUP_ASSIGNORS_CONFIG`?



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -213,82 +214,151 @@ public class GroupCoordinatorConfig {
      */
     public static final int CLASSIC_GROUP_NEW_MEMBER_JOIN_TIMEOUT_MS = 5 * 60 * 1000;
 
-    private final AbstractConfig config;
+    private final int numThreads;
+    private final int appendLingerMs;
+    private final int consumerGroupSessionTimeoutMs;
+    private final int consumerGroupHeartbeatIntervalMs;
+    private final int consumerGroupMaxSize;
+    private final List<ConsumerGroupPartitionAssignor> consumerGroupAssignors;
+    private final int offsetsTopicSegmentBytes;
+    private final int offsetMetadataMaxSize;
+    private final int classicGroupMaxSize;
+    private final int classicGroupInitialRebalanceDelayMs;
+    private final int classicGroupMinSessionTimeoutMs;
+    private final int classicGroupMaxSessionTimeoutMs;
+    private final long offsetsRetentionCheckIntervalMs;
+    private final long offsetsRetentionMs;
+    private final int offsetCommitTimeoutMs;
+    private final ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy;
+    private final CompressionType offsetTopicCompressionType;
+    private final int offsetsLoadBufferSize;
+    private final int offsetsTopicPartitions;
+    private final short offsetsTopicReplicationFactor;
+    private final short offsetCommitRequiredAcks;
+    private final int consumerGroupMinSessionTimeoutMs;
+    private final int consumerGroupMaxSessionTimeoutMs;
+    private final int consumerGroupMinHeartbeatIntervalMs;
+    private final int consumerGroupMaxHeartbeatIntervalMs;
 
     public GroupCoordinatorConfig(AbstractConfig config) {
-        this.config = config;
+        this.numThreads = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_NUM_THREADS_CONFIG);
+        this.appendLingerMs = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG);
+        this.consumerGroupSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG);
+        this.consumerGroupHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG);
+        this.consumerGroupMaxSize = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG);
+        this.consumerGroupAssignors = config.getConfiguredInstances(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, ConsumerGroupPartitionAssignor.class);

Review Comment:
   For another, could you make `consumerGroupAssignors` immutable?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-17081 Tweak GroupCoordinatorConfig: re-introduce local attributes and validation [kafka]

Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16524:
URL: https://github.com/apache/kafka/pull/16524#discussion_r1665921015


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -213,82 +214,151 @@ public class GroupCoordinatorConfig {
      */
     public static final int CLASSIC_GROUP_NEW_MEMBER_JOIN_TIMEOUT_MS = 5 * 60 * 1000;
 
-    private final AbstractConfig config;
+    private final int numThreads;
+    private final int appendLingerMs;
+    private final int consumerGroupSessionTimeoutMs;
+    private final int consumerGroupHeartbeatIntervalMs;
+    private final int consumerGroupMaxSize;
+    private final List<ConsumerGroupPartitionAssignor> consumerGroupAssignors;
+    private final int offsetsTopicSegmentBytes;
+    private final int offsetMetadataMaxSize;
+    private final int classicGroupMaxSize;
+    private final int classicGroupInitialRebalanceDelayMs;
+    private final int classicGroupMinSessionTimeoutMs;
+    private final int classicGroupMaxSessionTimeoutMs;
+    private final long offsetsRetentionCheckIntervalMs;
+    private final long offsetsRetentionMs;
+    private final int offsetCommitTimeoutMs;
+    private final ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy;
+    private final CompressionType offsetTopicCompressionType;
+    private final int offsetsLoadBufferSize;
+    private final int offsetsTopicPartitions;
+    private final short offsetsTopicReplicationFactor;
+    private final short offsetCommitRequiredAcks;
+    private final int consumerGroupMinSessionTimeoutMs;
+    private final int consumerGroupMaxSessionTimeoutMs;
+    private final int consumerGroupMinHeartbeatIntervalMs;
+    private final int consumerGroupMaxHeartbeatIntervalMs;
 
     public GroupCoordinatorConfig(AbstractConfig config) {
-        this.config = config;
+        this.numThreads = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_NUM_THREADS_CONFIG);
+        this.appendLingerMs = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG);
+        this.consumerGroupSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG);
+        this.consumerGroupHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG);
+        this.consumerGroupMaxSize = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG);
+        this.consumerGroupAssignors = config.getConfiguredInstances(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, ConsumerGroupPartitionAssignor.class);

Review Comment:
   could you add test to make sure incorrect value can cause error since CONSUMER_GROUP_ASSIGNORS_CONFIG` has no validator?



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -213,82 +214,151 @@ public class GroupCoordinatorConfig {
      */
     public static final int CLASSIC_GROUP_NEW_MEMBER_JOIN_TIMEOUT_MS = 5 * 60 * 1000;
 
-    private final AbstractConfig config;
+    private final int numThreads;
+    private final int appendLingerMs;
+    private final int consumerGroupSessionTimeoutMs;
+    private final int consumerGroupHeartbeatIntervalMs;
+    private final int consumerGroupMaxSize;
+    private final List<ConsumerGroupPartitionAssignor> consumerGroupAssignors;
+    private final int offsetsTopicSegmentBytes;
+    private final int offsetMetadataMaxSize;
+    private final int classicGroupMaxSize;
+    private final int classicGroupInitialRebalanceDelayMs;
+    private final int classicGroupMinSessionTimeoutMs;
+    private final int classicGroupMaxSessionTimeoutMs;
+    private final long offsetsRetentionCheckIntervalMs;
+    private final long offsetsRetentionMs;
+    private final int offsetCommitTimeoutMs;
+    private final ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy;
+    private final CompressionType offsetTopicCompressionType;
+    private final int offsetsLoadBufferSize;
+    private final int offsetsTopicPartitions;
+    private final short offsetsTopicReplicationFactor;
+    private final short offsetCommitRequiredAcks;
+    private final int consumerGroupMinSessionTimeoutMs;
+    private final int consumerGroupMaxSessionTimeoutMs;
+    private final int consumerGroupMinHeartbeatIntervalMs;
+    private final int consumerGroupMaxHeartbeatIntervalMs;
 
     public GroupCoordinatorConfig(AbstractConfig config) {
-        this.config = config;
+        this.numThreads = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_NUM_THREADS_CONFIG);
+        this.appendLingerMs = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG);
+        this.consumerGroupSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG);
+        this.consumerGroupHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG);
+        this.consumerGroupMaxSize = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG);
+        this.consumerGroupAssignors = config.getConfiguredInstances(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, ConsumerGroupPartitionAssignor.class);
+        this.offsetsTopicSegmentBytes = config.getInt(GroupCoordinatorConfig.OFFSETS_TOPIC_SEGMENT_BYTES_CONFIG);
+        this.offsetMetadataMaxSize = config.getInt(GroupCoordinatorConfig.OFFSET_METADATA_MAX_SIZE_CONFIG);
+        this.classicGroupMaxSize = config.getInt(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG);
+        this.classicGroupInitialRebalanceDelayMs = config.getInt(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG);
+        this.classicGroupMinSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG);
+        this.classicGroupMaxSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG);
+        this.offsetsRetentionCheckIntervalMs = config.getLong(GroupCoordinatorConfig.OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG);
+        this.offsetsRetentionMs = config.getInt(GroupCoordinatorConfig.OFFSETS_RETENTION_MINUTES_CONFIG) * 60L * 1000L;
+        this.offsetCommitTimeoutMs = config.getInt(GroupCoordinatorConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG);
+        this.consumerGroupMigrationPolicy = ConsumerGroupMigrationPolicy.parse(
+                config.getString(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG));
+        this.offsetTopicCompressionType = Optional.ofNullable(config.getInt(GroupCoordinatorConfig.OFFSETS_TOPIC_COMPRESSION_CODEC_CONFIG))

Review Comment:
   ditto. please add test. Also, maybe we should add validator to the def of `OFFSETS_TOPIC_COMPRESSION_CODEC_CONFIG`



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -213,82 +214,151 @@ public class GroupCoordinatorConfig {
      */
     public static final int CLASSIC_GROUP_NEW_MEMBER_JOIN_TIMEOUT_MS = 5 * 60 * 1000;
 
-    private final AbstractConfig config;
+    private final int numThreads;
+    private final int appendLingerMs;
+    private final int consumerGroupSessionTimeoutMs;
+    private final int consumerGroupHeartbeatIntervalMs;
+    private final int consumerGroupMaxSize;
+    private final List<ConsumerGroupPartitionAssignor> consumerGroupAssignors;
+    private final int offsetsTopicSegmentBytes;
+    private final int offsetMetadataMaxSize;
+    private final int classicGroupMaxSize;
+    private final int classicGroupInitialRebalanceDelayMs;
+    private final int classicGroupMinSessionTimeoutMs;
+    private final int classicGroupMaxSessionTimeoutMs;
+    private final long offsetsRetentionCheckIntervalMs;
+    private final long offsetsRetentionMs;
+    private final int offsetCommitTimeoutMs;
+    private final ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy;
+    private final CompressionType offsetTopicCompressionType;
+    private final int offsetsLoadBufferSize;
+    private final int offsetsTopicPartitions;
+    private final short offsetsTopicReplicationFactor;
+    private final short offsetCommitRequiredAcks;
+    private final int consumerGroupMinSessionTimeoutMs;
+    private final int consumerGroupMaxSessionTimeoutMs;
+    private final int consumerGroupMinHeartbeatIntervalMs;
+    private final int consumerGroupMaxHeartbeatIntervalMs;
 
     public GroupCoordinatorConfig(AbstractConfig config) {
-        this.config = config;
+        this.numThreads = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_NUM_THREADS_CONFIG);
+        this.appendLingerMs = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG);
+        this.consumerGroupSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG);
+        this.consumerGroupHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG);
+        this.consumerGroupMaxSize = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG);
+        this.consumerGroupAssignors = config.getConfiguredInstances(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, ConsumerGroupPartitionAssignor.class);
+        this.offsetsTopicSegmentBytes = config.getInt(GroupCoordinatorConfig.OFFSETS_TOPIC_SEGMENT_BYTES_CONFIG);
+        this.offsetMetadataMaxSize = config.getInt(GroupCoordinatorConfig.OFFSET_METADATA_MAX_SIZE_CONFIG);
+        this.classicGroupMaxSize = config.getInt(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG);
+        this.classicGroupInitialRebalanceDelayMs = config.getInt(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG);
+        this.classicGroupMinSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG);
+        this.classicGroupMaxSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG);
+        this.offsetsRetentionCheckIntervalMs = config.getLong(GroupCoordinatorConfig.OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG);
+        this.offsetsRetentionMs = config.getInt(GroupCoordinatorConfig.OFFSETS_RETENTION_MINUTES_CONFIG) * 60L * 1000L;
+        this.offsetCommitTimeoutMs = config.getInt(GroupCoordinatorConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG);
+        this.consumerGroupMigrationPolicy = ConsumerGroupMigrationPolicy.parse(

Review Comment:
   ditto



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -39,6 +39,7 @@
 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.require;
 
 /**
  * The group coordinator configurations.

Review Comment:
   please add comments why we create local attributes



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-17081 Tweak GroupCoordinatorConfig: re-introduce local attributes and validation [kafka]

Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #16524:
URL: https://github.com/apache/kafka/pull/16524#discussion_r1666951993


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -213,82 +214,151 @@ public class GroupCoordinatorConfig {
      */
     public static final int CLASSIC_GROUP_NEW_MEMBER_JOIN_TIMEOUT_MS = 5 * 60 * 1000;
 
-    private final AbstractConfig config;
+    private final int numThreads;
+    private final int appendLingerMs;
+    private final int consumerGroupSessionTimeoutMs;
+    private final int consumerGroupHeartbeatIntervalMs;
+    private final int consumerGroupMaxSize;
+    private final List<ConsumerGroupPartitionAssignor> consumerGroupAssignors;
+    private final int offsetsTopicSegmentBytes;
+    private final int offsetMetadataMaxSize;
+    private final int classicGroupMaxSize;
+    private final int classicGroupInitialRebalanceDelayMs;
+    private final int classicGroupMinSessionTimeoutMs;
+    private final int classicGroupMaxSessionTimeoutMs;
+    private final long offsetsRetentionCheckIntervalMs;
+    private final long offsetsRetentionMs;
+    private final int offsetCommitTimeoutMs;
+    private final ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy;
+    private final CompressionType offsetTopicCompressionType;
+    private final int offsetsLoadBufferSize;
+    private final int offsetsTopicPartitions;
+    private final short offsetsTopicReplicationFactor;
+    private final short offsetCommitRequiredAcks;
+    private final int consumerGroupMinSessionTimeoutMs;
+    private final int consumerGroupMaxSessionTimeoutMs;
+    private final int consumerGroupMinHeartbeatIntervalMs;
+    private final int consumerGroupMaxHeartbeatIntervalMs;
 
     public GroupCoordinatorConfig(AbstractConfig config) {
-        this.config = config;
+        this.numThreads = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_NUM_THREADS_CONFIG);
+        this.appendLingerMs = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG);
+        this.consumerGroupSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG);
+        this.consumerGroupHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG);
+        this.consumerGroupMaxSize = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG);
+        this.consumerGroupAssignors = config.getConfiguredInstances(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, ConsumerGroupPartitionAssignor.class);
+        this.offsetsTopicSegmentBytes = config.getInt(GroupCoordinatorConfig.OFFSETS_TOPIC_SEGMENT_BYTES_CONFIG);
+        this.offsetMetadataMaxSize = config.getInt(GroupCoordinatorConfig.OFFSET_METADATA_MAX_SIZE_CONFIG);
+        this.classicGroupMaxSize = config.getInt(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG);
+        this.classicGroupInitialRebalanceDelayMs = config.getInt(GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG);
+        this.classicGroupMinSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG);
+        this.classicGroupMaxSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG);
+        this.offsetsRetentionCheckIntervalMs = config.getLong(GroupCoordinatorConfig.OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG);
+        this.offsetsRetentionMs = config.getInt(GroupCoordinatorConfig.OFFSETS_RETENTION_MINUTES_CONFIG) * 60L * 1000L;
+        this.offsetCommitTimeoutMs = config.getInt(GroupCoordinatorConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG);
+        this.consumerGroupMigrationPolicy = ConsumerGroupMigrationPolicy.parse(

Review Comment:
   Thanks ! I fixed all of them.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-17081 Tweak GroupCoordinatorConfig: re-introduce local attributes and validation [kafka]

Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #16524:
URL: https://github.com/apache/kafka/pull/16524


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-17081 Tweak GroupCoordinatorConfig: re-introduce local attributes and validation [kafka]

Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #16524:
URL: https://github.com/apache/kafka/pull/16524#discussion_r1667234615


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -213,82 +214,151 @@ public class GroupCoordinatorConfig {
      */
     public static final int CLASSIC_GROUP_NEW_MEMBER_JOIN_TIMEOUT_MS = 5 * 60 * 1000;
 
-    private final AbstractConfig config;
+    private final int numThreads;
+    private final int appendLingerMs;
+    private final int consumerGroupSessionTimeoutMs;
+    private final int consumerGroupHeartbeatIntervalMs;
+    private final int consumerGroupMaxSize;
+    private final List<ConsumerGroupPartitionAssignor> consumerGroupAssignors;
+    private final int offsetsTopicSegmentBytes;
+    private final int offsetMetadataMaxSize;
+    private final int classicGroupMaxSize;
+    private final int classicGroupInitialRebalanceDelayMs;
+    private final int classicGroupMinSessionTimeoutMs;
+    private final int classicGroupMaxSessionTimeoutMs;
+    private final long offsetsRetentionCheckIntervalMs;
+    private final long offsetsRetentionMs;
+    private final int offsetCommitTimeoutMs;
+    private final ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy;
+    private final CompressionType offsetTopicCompressionType;
+    private final int offsetsLoadBufferSize;
+    private final int offsetsTopicPartitions;
+    private final short offsetsTopicReplicationFactor;
+    private final short offsetCommitRequiredAcks;
+    private final int consumerGroupMinSessionTimeoutMs;
+    private final int consumerGroupMaxSessionTimeoutMs;
+    private final int consumerGroupMinHeartbeatIntervalMs;
+    private final int consumerGroupMaxHeartbeatIntervalMs;
 
     public GroupCoordinatorConfig(AbstractConfig config) {
-        this.config = config;
+        this.numThreads = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_NUM_THREADS_CONFIG);
+        this.appendLingerMs = config.getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG);
+        this.consumerGroupSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG);
+        this.consumerGroupHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG);
+        this.consumerGroupMaxSize = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG);
+        this.consumerGroupAssignors = config.getConfiguredInstances(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, ConsumerGroupPartitionAssignor.class);

Review Comment:
   > @brandboat Have you added test for CONSUMER_GROUP_ASSIGNORS_CONFIG?
   
   I think It's already there before I submit this pr.
   
   > For another, could you make consumerGroupAssignors immutable?
   
   Will do



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org