Re: [PR] KAFKA-19276: Trigger rebalance for streams config update [WIP] [kafka]

[via GitHub]

UladzislauBlok commented on code in PR #19967:
URL: https://github.com/apache/kafka/pull/19967#discussion_r2190622312


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java:
##########
@@ -75,6 +75,12 @@ public final class GroupConfig extends AbstractConfig {
 
     public static final String STREAMS_NUM_STANDBY_REPLICAS_CONFIG = 
"streams.num.standby.replicas";
 
+    public static final String STREAMS_ASSIGNOR_NAME_CONFIG = 
"streams.assignor.name";

Review Comment:
   STREAMS_SESSION_TIMEOUT_MS_CONFIG
   STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG
   Are not part of ticket description. Should we also trigger re-balance if 
they were changed? 



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/ConfigUpdateEvent.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.coordinator.group;
+
+public enum ConfigUpdateEvent {
+    // Client Actions Required

Review Comment:
   Here client == subscriber / listener, so any place in code where we want to 
react to config update



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -462,6 +465,11 @@ GroupMetadataManager build() {
      */
     private final TimelineHashMap<String, Group> groups;
 
+    /**
+     * The classic and consumer groups keyed by their name.
+     */
+    private final Map<String, List<GroupConfigListener>> groupConfigListeners;

Review Comment:
   We have it in both places, as we want to have list of listeners (observers) 
to set flag (in config manager) / check (here) if config was changed
   See diagram:
   
   ![Untitled-2025-07-07-1859 
excalidraw](https://github.com/user-attachments/assets/78366968-7f8f-4be9-8936-8d43c36d1c5a)
   



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfigManager.java:
##########
@@ -57,6 +62,8 @@ public void updateGroupConfig(String groupId, Properties 
newGroupConfig) {
             newGroupConfig
         );
         configMap.put(groupId, newConfig);
+        listenerMap.getOrDefault(groupId, Collections.emptyList())

Review Comment:
   True, I used Collections.emptyList() as it more verbal. Could fix it



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -462,6 +465,11 @@ GroupMetadataManager build() {
      */
     private final TimelineHashMap<String, Group> groups;
 
+    /**
+     * The classic and consumer groups keyed by their name.

Review Comment:
   Right, my bad. I'll describe it better



-- 
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-unsubscr...@kafka.apache.org

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