MINOR: Strengthen tests for replaying streams group assignment configs (#23872)

Follow-up to #23165. Test-only change.
-
`testStreamsGroupEpochDoesNotIncreaseWhenEveryAssignmentConfigIsAtItsDefault`
now starts the group with
non-default assignment configs, so it fails if replaying a record
without configs does not reset them to the
defaults.
- Add
`testStreamsGroupEpochIncreasesWhenAnAssignmentConfigIsNotAtItsDefault`:
after the same replay, a group with
`num.standby.replicas=2` still bumps the group epoch, since the replayed
defaults differ from the current configs.
- Add `AssignmentConfigsImplTest` cases for `fromRecord` with an empty
list and with an unknown key.
- Use `AssignmentConfigsImpl.DEFAULT` instead of `new
AssignmentConfigsImpl(0, List.of())` in `GroupMetadataManagerTest`.

Reviewers: Lucas Brutschy <lbrutschy@confluent.io>

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

@@ -19867,7 +19867,7 @@ public class GroupMetadataManagerTest {
                 .withTargetAssignmentTimestamp(12345L)
                 .withMetadataHash(groupMetadataHash)
                 .withValidatedTopologyEpoch(0)
-                .withLastAssignmentConfigs(new AssignmentConfigsImpl(0, List.of()))
+                .withLastAssignmentConfigs(AssignmentConfigsImpl.DEFAULT)
             )
             .build();
 
@@ -19937,7 +19937,7 @@ public class GroupMetadataManagerTest {
                 .withTargetAssignmentEpoch(10)
                 .withMetadataHash(groupMetadataHash)
                 .withValidatedTopologyEpoch(0)
-                .withLastAssignmentConfigs(new AssignmentConfigsImpl(0, List.of()))
+                .withLastAssignmentConfigs(AssignmentConfigsImpl.DEFAULT)
             )
             .build();
 
@@ -20458,7 +20458,7 @@ public class GroupMetadataManagerTest {
                 .withTargetAssignmentEpoch(10)
                 .withMetadataHash(groupMetadataHash)
                 .withValidatedTopologyEpoch(0)
-                .withLastAssignmentConfigs(new AssignmentConfigsImpl(0, List.of()))
+                .withLastAssignmentConfigs(AssignmentConfigsImpl.DEFAULT)
             )
             .build();
 
@@ -24420,6 +24420,8 @@ public class GroupMetadataManagerTest {
                 .withTopology(StreamsTopology.fromHeartbeatRequest(topology))
                 .withValidatedTopologyEpoch(0)
                 .withMetadataHash(metadataHash)
+                // Not the defaults, so the replay below must reset them.
+                .withLastAssignmentConfigs(new AssignmentConfigsImpl(1, List.of()))
             )
             .build();
 
@@ -24465,6 +24467,100 @@ public class GroupMetadataManagerTest {
         assertEquals(10, group.groupEpoch());
     }
 
+    @Test
+    public void testStreamsGroupEpochIncreasesWhenAnAssignmentConfigIsNotAtItsDefault() {
+        // A record without assignment configs replays as the defaults, so if a config is set to a non-default value,
+        // the stored configs differ from the current ones and the heartbeat must bump the group epoch.
+        String groupId = "fooup";
+        String memberId = Uuid.randomUuid().toString();
+        String subtopology1 = "subtopology1";
+        String fooTopicName = "foo";
+        Uuid fooTopicId = Uuid.randomUuid();
+
+        Topology topology = new Topology().setSubtopologies(List.of(
+            new Subtopology().setSubtopologyId(subtopology1).setSourceTopics(List.of(fooTopicName))
+        ));
+
+        CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+            .addTopic(fooTopicId, fooTopicName, 6)
+            .buildCoordinatorMetadataImage();
+
+        long metadataHash = computeGroupHash(Map.of(fooTopicName, computeTopicHash(fooTopicName, metadataImage)));
+
+        MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+        GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+            .withStreamsGroupTaskAssignors(List.of(assignor))
+            .withMetadataImage(metadataImage)
+            .withStreamsGroup(new StreamsGroupBuilder(groupId, 10)
+                .withMember(streamsGroupMemberBuilderWithDefaults(memberId)
+                    .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
+                    .setMemberEpoch(10)
+                    .setPreviousMemberEpoch(9)
+                    .setAssignedTasks(TaskAssignmentTestUtil.mkTasksTupleWithCommonEpoch(TaskRole.ACTIVE, 10,
+                        TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3, 4, 5)))
+                    .build())
+                .withTargetAssignment(memberId, TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE,
+                    TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3, 4, 5)))
+                .withTargetAssignmentEpoch(10)
+                .withTopology(StreamsTopology.fromHeartbeatRequest(topology))
+                .withValidatedTopologyEpoch(0)
+                .withMetadataHash(metadataHash)
+                // Equal to the group config, so the epoch bumps only if the replay below resets them to the defaults.
+                .withLastAssignmentConfigs(new AssignmentConfigsImpl(2, List.of()))
+            )
+            .build();
+
+        Properties newConfig = new Properties();
+        newConfig.put(GroupConfig.STREAMS_NUM_STANDBY_REPLICAS_CONFIG, "2");
+        context.groupConfigManager.updateGroupConfig(groupId, newConfig);
+
+        // Replay a metadata record without assignment configs, as written by a 4.1 broker.
+        context.replay(CoordinatorRecord.record(
+            new StreamsGroupMetadataKey().setGroupId(groupId),
+            new ApiMessageAndVersion(
+                new StreamsGroupMetadataValue()
+                    .setEpoch(10)
+                    .setMetadataHash(metadataHash)
+                    .setValidatedTopologyEpoch(0)
+                    .setLastAssignmentConfigs(null),
+                (short) 0
+            )
+        ));
+
+        StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
+        assertEquals(AssignmentConfigsImpl.DEFAULT, group.lastAssignmentConfigs());
+
+        assignor.prepareGroupAssignment(
+            Map.of(memberId, TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE,
+                TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3, 4, 5)))
+        );
+
+        CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+            new StreamsGroupHeartbeatRequestData()
+                .setGroupId(groupId)
+                .setMemberId(memberId)
+                .setMemberEpoch(10)
+                .setActiveTasks(List.of(new StreamsGroupHeartbeatRequestData.TaskIds()
+                    .setSubtopologyId(subtopology1)
+                    .setPartitions(List.of(0, 1, 2, 3, 4, 5))))
+                .setStandbyTasks(List.of())
+                .setWarmupTasks(List.of()));
+
+        StreamsGroupMetadataValue metadataValue = result.records().stream()
+            .filter(record -> record.key() instanceof StreamsGroupMetadataKey)
+            .map(record -> (StreamsGroupMetadataValue) record.value().message())
+            .findFirst()
+            .orElseThrow(() -> new AssertionError("Expected a StreamsGroupMetadata record"));
+        assertEquals(11, metadataValue.epoch());
+        assertEquals(
+            AssignmentConfigsImpl.DEFAULT.withNumStandbyReplicas(2),
+            AssignmentConfigsImpl.fromRecord(metadataValue.lastAssignmentConfigs())
+        );
+        assertEquals(11, result.response().data().memberEpoch());
+        assertEquals(11, group.groupEpoch());
+        assertEquals(AssignmentConfigsImpl.DEFAULT.withNumStandbyReplicas(2), group.lastAssignmentConfigs());
+    }
+
     @Test
     public void testStreamsGroupEpochShouldNotIncreaseWithAcceptableRecoveryLagConfigChange() {
         String groupId = "fooup";

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImplTest.java

@@ -45,6 +45,23 @@ public class AssignmentConfigsImplTest {
         assertEquals(AssignmentConfigsImpl.DEFAULT, AssignmentConfigsImpl.fromRecord(null));
     }
 
+    @Test
+    void testFromRecordWithEmptyConfigs() {
+        assertEquals(AssignmentConfigsImpl.DEFAULT, AssignmentConfigsImpl.fromRecord(List.of()));
+    }
+
+    @Test
+    void testFromRecordIgnoresUnknownConfigs() {
+        // A newer broker may record configs that this broker does not know.
+        assertEquals(
+            AssignmentConfigsImpl.DEFAULT.withNumStandbyReplicas(1),
+            AssignmentConfigsImpl.fromRecord(List.of(
+                lastAssignmentConfig("unknown.config", "value"),
+                lastAssignmentConfig("num.standby.replicas", "1")
+            ))
+        );
+    }
+
     @Test
     void testFromRecordWithoutRackAwareAssignmentTags() {
         // The tags are only recorded when any are configured.