MINOR: Small refactor in GroupMetadataManager (apache#19090)

The code in GroupMetadataManager to request metadata refresh got pretty
ugly with the addition of share and stream groups. It seems preferable
to put the method in the base class.

Reviewers: Andrew Schofield <[email protected]>

Author: dajac

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

@@ -195,4 +195,9 @@ void validateOffsetFetch(
      * @return The number of members.
      */
     int numMembers();
+
+    /**
+     * Requests a metadata refresh.
+     */
+    void requestMetadataRefresh();
 }

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

@@ -4786,11 +4786,8 @@ public void onNewMetadataImage(MetadataImage newImage, MetadataDelta delta) {
         });
         allGroupIds.forEach(groupId -> {
             Group group = groups.get(groupId);
-            if (group != null && (group.type() == CONSUMER || group.type() == SHARE)) {
-                ((ModernGroup<?>) group).requestMetadataRefresh();
-            }
-            if (group != null && (group.type() == STREAMS)) {
-                ((StreamsGroup) group).requestMetadataRefresh();
+            if (group != null) {
+                group.requestMetadataRefresh();
             }
         });
     }

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java

@@ -342,6 +342,14 @@ public int numMembers() {
         return members.size();
     }
 
+    /**
+     * Requests a metadata refresh.
+     */
+    @Override
+    public void requestMetadataRefresh() {
+        // This does not apply to classic groups.
+    }
+
     /**
      * Used to identify whether the given member is the leader of this group.
      *

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/ModernGroup.java

@@ -414,6 +414,7 @@ public void setMetadataRefreshDeadline(
     /**
      * Requests a metadata refresh.
      */
+    @Override
     public void requestMetadataRefresh() {
         this.metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
     }

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

@@ -602,6 +602,7 @@ public void setMetadataRefreshDeadline(
     /**
      * Requests a metadata refresh.
      */
+    @Override
     public void requestMetadataRefresh() {
         this.metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
     }