Add docs

Author: UladzislauBlok

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java

@@ -25,6 +25,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Properties;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -211,7 +212,7 @@ default boolean shouldExpire() {
     }
 
     /**
-     * TODO Add docs
+     * Listener method to trigger some action if config was changed.
      */
-    default void onGroupConfigChanged() {}
+    default void onGroupConfigChanged(Properties updatedProperties) {}
 }

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfigManager.java

@@ -54,7 +54,9 @@ public void updateGroupConfig(String groupId, Properties newGroupConfig) {
             throw new InvalidRequestException("Group name can't be empty.");
         }
 
-        Optional.ofNullable(listener).ifPresent(listener -> listener.onGroupConfigUpdate(groupId));
+        if (listener != null) {
+            listener.onGroupConfigUpdate(groupId, newGroupConfig);
+        }
 
         final GroupConfig newConfig = GroupConfig.fromProps(
             defaultConfig.originals(),

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

@@ -194,6 +194,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.OptionalInt;
+import java.util.Properties;
 import java.util.Set;
 import java.util.SortedMap;
 import java.util.concurrent.CompletableFuture;
@@ -1913,7 +1914,7 @@ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> stream
         // to persist the change, and bump the group epoch later.
         boolean bumpGroupEpoch = hasStreamsMemberMetadataChanged(groupId, member, updatedMember, records);
 
-        // todo comment
+        // Check if StreamsGroup has flag rebalanceRequired equals to true
         if (group.rebalanceRequired()) {
             bumpGroupEpoch = true;
         }
@@ -8372,10 +8373,12 @@ public List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseT
     }
 
     /**
-     * TODO docs
+     * Trigger {@link org.apache.kafka.coordinator.group.Group#onGroupConfigChanged} for selected group id
+     *
+     * @see org.apache.kafka.coordinator.group.Group#onGroupConfigChanged(Properties)
      */
-    public void onGroupConfigUpdate(String groupId) {
-        Optional.ofNullable(groups.get(groupId)).ifPresent(Group::onGroupConfigChanged);
+    public void onGroupConfigUpdate(String groupId, Properties updatedConfig) {
+        Optional.ofNullable(groups.get(groupId)).ifPresent(group -> group.onGroupConfigChanged(updatedConfig));
     }
 
     /**

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java

@@ -51,10 +51,12 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Properties;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static org.apache.kafka.coordinator.group.GroupConfig.STREAMS_NUM_STANDBY_REPLICAS_CONFIG;
 import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.ASSIGNING;
 import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.DEAD;
 import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.EMPTY;
@@ -72,6 +74,10 @@ public class StreamsGroup implements Group {
      */
     private static final String PROTOCOL_TYPE = "streams";
 
+    private static final Set<String> PROPERTIES_TO_TIGGER_REBALANCE = Set.of(
+            STREAMS_NUM_STANDBY_REPLICAS_CONFIG
+    );
+
     public enum StreamsGroupState {
         EMPTY("Empty"),
         NOT_READY("NotReady"),
@@ -197,7 +203,7 @@ public static class DeadlineAndEpoch {
     private final TimelineObject<Optional<ConfiguredTopology>> configuredTopology;
 
     /**
-     * TODO docs
+     * The flag that group should be rebalanced
      */
     private final AtomicBoolean rebalanceRequired;
 
@@ -834,14 +840,24 @@ public boolean isInStates(Set<String> statesFilter, long committedOffset) {
         return statesFilter.contains(state.get(committedOffset).toLowerCaseString());
     }
 
+    /**
+     * Check if rebalance is required, after config was changed
+     *
+     * @param updatedProperties Properties that was updated
+     */
     @Override
-    public void onGroupConfigChanged() {
-        rebalanceRequired.set(true); // FIXME check config
+    public void onGroupConfigChanged(Properties updatedProperties) {
+        PROPERTIES_TO_TIGGER_REBALANCE.forEach(property -> {
+            if (updatedProperties.containsKey(property)) {
+                rebalanceRequired.set(true);
+            }
+        });
     }
 
     /**
-     * TODO docs
-     * @return
+     * Check if rebalance is required. Reset the flag, set it false.
+     *
+     * @return true - if rebalance is required
      */
     public boolean rebalanceRequired() {
         return rebalanceRequired.getAndSet(false);