KAFKA-19720: Regex subscription should be empty for classic members joining mixed group

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

@@ -3025,10 +3025,14 @@ private ConsumerGroupMember getOrMaybeSubscribeStaticConsumerGroupMember(
                 }
 
                 // Copy the member but with its new member id.
-                ConsumerGroupMember newMember = new ConsumerGroupMember.Builder(existingStaticMemberOrNull, memberId)
+                ConsumerGroupMember.Builder memberBuilder = new ConsumerGroupMember.Builder(existingStaticMemberOrNull, memberId)
                     .setMemberEpoch(0)
-                    .setPreviousMemberEpoch(0)
-                    .build();
+                    .setPreviousMemberEpoch(0);
+                if (useClassicProtocol) {
+                    // Regex subscription is not supported for classic member.
+                    memberBuilder.setSubscribedTopicRegex("");
+                }
+                ConsumerGroupMember newMember = memberBuilder.build();
 
                 // Generate the records to replace the member. We don't care about the regular expression
                 // here because it is taken care of later after the static membership replacement.

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

@@ -20985,6 +20985,75 @@ foooTopicName, computeTopicHash(foooTopicName, new KRaftCoordinatorMetadataImage
         );
     }
 
+    @Test
+    public void testConsumerMemberWithRegexReplacedByClassicMember() {
+        String groupId = "fooup";
+        String instanceId = "instance-id";
+        String memberId1 = Uuid.randomUuid().toString();
+        String memberId2 = Uuid.randomUuid().toString();
+
+        Uuid fooTopicId = Uuid.randomUuid();
+        String fooTopicName = "foo";
+        Uuid barTopicId = Uuid.randomUuid();
+        String barTopicName = "bar";
+
+        CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+            .addTopic(fooTopicId, fooTopicName, 6)
+            .addTopic(barTopicId, barTopicName, 3)
+            .buildCoordinatorMetadataImage(12345L);
+
+        GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+            .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(new NoOpPartitionAssignor()))
+            .withMetadataImage(metadataImage)
+            .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+                .withMember(new ConsumerGroupMember.Builder(memberId1)
+                    .setInstanceId(instanceId)
+                    .setState(MemberState.STABLE)
+                    .setMemberEpoch(10)
+                    .setPreviousMemberEpoch(10)
+                    .setClientId(DEFAULT_CLIENT_ID)
+                    .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+                    .setRebalanceTimeoutMs(5000)
+                    .setSubscribedTopicRegex("foo*")
+                    .setServerAssignorName(NoOpPartitionAssignor.NAME)
+                    .setAssignedPartitions(mkAssignment(
+                        mkTopicAssignment(fooTopicId, 0, 1, 2)))
+                    .build())
+                .withMember(new ConsumerGroupMember.Builder(memberId2)
+                    .setState(MemberState.STABLE)
+                    .setMemberEpoch(10)
+                    .setPreviousMemberEpoch(10)
+                    .setClientId(DEFAULT_CLIENT_ID)
+                    .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+                    .setRebalanceTimeoutMs(5000)
+                    .setSubscribedTopicRegex("foo*")
+                    .setServerAssignorName(NoOpPartitionAssignor.NAME)
+                    .setAssignedPartitions(mkAssignment(
+                        mkTopicAssignment(fooTopicId, 3, 4, 5)))
+                    .build())
+                .withAssignment(memberId1, mkAssignment(
+                    mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)))
+                .withAssignmentEpoch(10))
+            .build();
+
+        // Member 1 is replaced by a classic member with the same instance id.
+        context.sendClassicGroupJoin(
+            new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
+                .withGroupId(groupId)
+                .withMemberId(UNKNOWN_MEMBER_ID)
+                .withGroupInstanceId(instanceId)
+                .withProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+                .withDefaultProtocolTypeAndProtocols()
+                .build()
+        );
+
+        // The new member should have no regex subscription.
+        assertEquals(
+            "",
+            context.groupMetadataManager.consumerGroup(groupId).staticMember(instanceId).subscribedTopicRegex()
+        );
+    }
+
     @Test
     public void testConsumerGroupMemberJoinsWithRegexWithTopicAuthorizationFailure() {
         String groupId = "fooup";