chickenchickenlove commented on code in PR #22245: URL: https://github.com/apache/kafka/pull/22245#discussion_r3325276589
########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupStaticMemberGroupMetadataManagerTest.java: ########## @@ -0,0 +1,1910 @@ +/* + * 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; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.errors.FencedInstanceIdException; +import org.apache.kafka.common.errors.FencedMemberEpochException; +import org.apache.kafka.common.errors.GroupIdNotFoundException; +import org.apache.kafka.common.errors.GroupMaxSizeReachedException; +import org.apache.kafka.common.errors.UnknownMemberIdException; +import org.apache.kafka.common.errors.UnreleasedInstanceIdException; +import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData; +import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData; +import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest; +import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord; +import org.apache.kafka.coordinator.common.runtime.CoordinatorResult; +import org.apache.kafka.coordinator.common.runtime.MetadataImageBuilder; +import org.apache.kafka.coordinator.common.runtime.MockCoordinatorTimer; +import org.apache.kafka.coordinator.group.StreamsGroupTestUtil.StreamsTopicFixture; +import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataKey; +import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue; +import org.apache.kafka.coordinator.group.generated.StreamsGroupMetadataKey; +import org.apache.kafka.coordinator.group.generated.StreamsGroupMetadataValue; +import org.apache.kafka.coordinator.group.streams.MemberState; +import org.apache.kafka.coordinator.group.streams.MockTaskAssignor; +import org.apache.kafka.coordinator.group.streams.StreamsCoordinatorRecordHelpers; +import org.apache.kafka.coordinator.group.streams.StreamsGroup; +import org.apache.kafka.coordinator.group.streams.StreamsGroupBuilder; +import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult; +import org.apache.kafka.coordinator.group.streams.StreamsGroupMember; +import org.apache.kafka.coordinator.group.streams.StreamsTopology; +import org.apache.kafka.coordinator.group.streams.TasksTuple; +import org.apache.kafka.coordinator.group.streams.TasksTupleWithEpochs; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Stream; + +import static org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH; +import static org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH; +import static org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH; +import static org.apache.kafka.coordinator.group.Assertions.assertRecordsEquals; +import static org.apache.kafka.coordinator.group.Assertions.assertResponseEquals; +import static org.apache.kafka.coordinator.group.GroupMetadataManager.groupSessionTimeoutKey; +import static org.apache.kafka.coordinator.group.GroupMetadataManagerTestContext.DEFAULT_CLIENT_ADDRESS; +import static org.apache.kafka.coordinator.group.GroupMetadataManagerTestContext.DEFAULT_CLIENT_ID; +import static org.apache.kafka.coordinator.group.GroupMetadataManagerTestContext.DEFAULT_PROCESS_ID; +import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.getDefaultAssignmentConfigs; +import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.resetAssignedTasksEpochsToZero; +import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.staticHeartbeat; +import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.staticJoinHeartbeat; +import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.streamsGroupMemberBuilderWithDefaults; +import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.streamsTopicFixture; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class StreamsGroupStaticMemberGroupMetadataManagerTest { + + private static final int DEFAULT_MEMBER_EPOCH = 10; + private static final int DEFAULT_GROUP_EPOCH = 10; + + + @Test + public void testUnknownStaticMemberLeaveStreamsGroup() { + String groupId = "streams-group"; + + String activeMemberId = "active-member"; + String activeInstanceId = "active-instance"; + + String unknownMemberId = "unknown-member-id"; + String unknownInstanceId = "unknown-instance-id"; + + StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 1); + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky"))) + .withMetadataImage(topic.metadataImage()) + .withStreamsGroup(new StreamsGroupBuilder(groupId, DEFAULT_GROUP_EPOCH) + .withMember(streamsGroupMemberBuilderWithDefaults(activeMemberId, activeInstanceId) + .setMemberEpoch(DEFAULT_MEMBER_EPOCH) + .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1) + .build()) + .withTargetAssignmentEpoch(DEFAULT_GROUP_EPOCH) + .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology())) + .withValidatedTopologyEpoch(0) + .withMetadataHash(topic.metadataHash()) + .withLastAssignmentConfigs(getDefaultAssignmentConfigs())) + .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT) + .build(); + + assertThrows( + UnknownMemberIdException.class, + () -> context.streamsGroupHeartbeat(staticHeartbeat( + groupId, + unknownMemberId, + unknownInstanceId, + LEAVE_GROUP_STATIC_MEMBER_EPOCH + )) + ); + } + + @Test + public void testStreamsStaticJoinWithNewInstanceAtMaxSizeThrowsGroupMaxSizeReached() { + // With max.size=2 already reached, + // joining with a new static instanceId must throw GroupMaxSizeReachedException. + int streamsGroupMaxSize = 2; + String groupId = "streams-group"; + + String activeMemberId = "active-member"; + String activeInstanceId = "active-instance"; + + String leftMemberId = "left-member"; + String leftInstanceId = "left-instance"; + + String newMemberId = "new-member"; + String newInstanceId = "new-instance"; + + StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 1); + + MockTaskAssignor assignor = new MockTaskAssignor("sticky"); + assignor.prepareGroupAssignment(Map.of(activeMemberId, TasksTuple.EMPTY)); + + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withStreamsGroupTaskAssignors(List.of(assignor)) + .withMetadataImage(topic.metadataImage()) + .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, streamsGroupMaxSize) + .withStreamsGroup(new StreamsGroupBuilder(groupId, DEFAULT_GROUP_EPOCH) + .withMember(streamsGroupMemberBuilderWithDefaults(activeMemberId, activeInstanceId) + .setMemberEpoch(DEFAULT_MEMBER_EPOCH) + .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1) + .build()) + .withMember(streamsGroupMemberBuilderWithDefaults(leftMemberId, leftInstanceId) + .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH) + .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1) + .build()) + .withTargetAssignmentEpoch(DEFAULT_GROUP_EPOCH) + .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology())) + .withMetadataHash(topic.metadataHash()) + .withValidatedTopologyEpoch(0) + ) + .build(); + + assertThrows( + GroupMaxSizeReachedException.class, + () -> context.streamsGroupHeartbeat(staticJoinHeartbeat( + groupId, + newMemberId, + newInstanceId, + topic + )) + ); + + } + + @Test + public void testStreamsStaticRejoinWithLeaveGroupStaticEpochAtMaxSizeSucceeds() { + // If a static member is in leave epoch (-2), + // rejoining with the same memberId/instanceId is allowed even at max size. + int streamsGroupMaxSize = 2; + String groupId = "streams-group"; + + String activeMemberId = "active-member"; + String activeInstanceId = "active-instance"; + + String leftMemberId = "left-member"; + String leftInstanceId = "left-instance"; + String newJoinMemberId = "new-join-member"; + + StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 1); + + MockTaskAssignor assignor = new MockTaskAssignor("sticky"); + assignor.prepareGroupAssignment(Map.of( + activeMemberId, TasksTuple.EMPTY, + leftMemberId, TasksTuple.EMPTY + )); + + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withStreamsGroupTaskAssignors(List.of(assignor)) + .withMetadataImage(topic.metadataImage()) + .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, streamsGroupMaxSize) + .withStreamsGroup(new StreamsGroupBuilder(groupId, DEFAULT_GROUP_EPOCH) + .withMember(streamsGroupMemberBuilderWithDefaults(activeMemberId, activeInstanceId) + .setMemberEpoch(DEFAULT_MEMBER_EPOCH) + .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1) + .build()) + .withMember(streamsGroupMemberBuilderWithDefaults(leftMemberId, leftInstanceId) + .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH) + .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1) + .build()) + .withTargetAssignmentEpoch(DEFAULT_GROUP_EPOCH) + .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology())) + .withMetadataHash(topic.metadataHash()) + .withValidatedTopologyEpoch(0) + .withLastAssignmentConfigs(getDefaultAssignmentConfigs()) + ) + .build(); + + assertDoesNotThrow(() -> context.streamsGroupHeartbeat(staticJoinHeartbeat( + groupId, + newJoinMemberId, + leftInstanceId, + topic + ))); + + } + + @Test + public void testStreamsStaticJoinWithUnreleasedInstanceThrowsUnreleasedInstanceIdAtMaxSize() { + // If an active static member (epoch=10) still owns the instanceId, + // a different memberId joining with that instanceId must fail even at max size. + int streamsGroupMaxSize = 1; + String groupId = "streams-group"; + + String activeMemberId = "active-member"; + String activeInstanceId = "active-instance"; + + String newMemberId = "new-member"; + + StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 1); + + MockTaskAssignor assignor = new MockTaskAssignor("sticky"); + assignor.prepareGroupAssignment(Map.of(activeMemberId, TasksTuple.EMPTY)); + + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withStreamsGroupTaskAssignors(List.of(assignor)) + .withMetadataImage(topic.metadataImage()) + .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, streamsGroupMaxSize) + .withStreamsGroup(new StreamsGroupBuilder(groupId, DEFAULT_GROUP_EPOCH) + .withMember(streamsGroupMemberBuilderWithDefaults(activeMemberId, activeInstanceId) + .setMemberEpoch(DEFAULT_MEMBER_EPOCH) + .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1) + .build()) + .withTargetAssignmentEpoch(DEFAULT_GROUP_EPOCH) + .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology())) + .withMetadataHash(topic.metadataHash()) + .withValidatedTopologyEpoch(0) + .withLastAssignmentConfigs(getDefaultAssignmentConfigs()) + ) + .build(); + + assertThrows( + UnreleasedInstanceIdException.class, + () -> context.streamsGroupHeartbeat(staticJoinHeartbeat( + groupId, + newMemberId, + activeInstanceId, + topic + )) + ); + } + + @ParameterizedTest + @MethodSource("staticMemberReusedInstanceErrorCases") + public void testStaticMemberSendHeartbeatWithVariousEpochThenThrowError( + int heartbeatMemberEpoch, + Class<? extends Exception> expectedException + ) { + // same instance id is reused with mismatched member identity/epoch. + String groupId = "streams-group"; + + String activeMemberId = "active-member"; + String activeInstanceId = "active-instance"; + String newMemberId = "new-member"; + + StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 1); + + MockTaskAssignor assignor = new MockTaskAssignor("sticky"); + assignor.prepareGroupAssignment(Map.of(activeMemberId, TasksTuple.EMPTY)); + + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withStreamsGroupTaskAssignors(List.of(assignor)) + .withMetadataImage(topic.metadataImage()) + .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, 2) + .withStreamsGroup(new StreamsGroupBuilder(groupId, DEFAULT_GROUP_EPOCH) + .withMember(streamsGroupMemberBuilderWithDefaults(activeMemberId, activeInstanceId) + .setMemberEpoch(DEFAULT_MEMBER_EPOCH) + .setPreviousMemberEpoch(DEFAULT_MEMBER_EPOCH - 1) + .build()) + .withTargetAssignmentEpoch(DEFAULT_GROUP_EPOCH) + .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology())) + .withMetadataHash(topic.metadataHash()) + .withValidatedTopologyEpoch(0) + .withLastAssignmentConfigs(getDefaultAssignmentConfigs()) + ) + .build(); + + assertThrows( + expectedException, + () -> context.streamsGroupHeartbeat(staticHeartbeat( + groupId, + newMemberId, + activeInstanceId, + heartbeatMemberEpoch + )) + ); + } + + private static Stream<Arguments> staticMemberReusedInstanceErrorCases() { + return Stream.of( + Arguments.of(0, UnreleasedInstanceIdException.class), // static member try to join when static member already existed, then throw UnreleasedInstanceIdException. + Arguments.of(1000, FencedInstanceIdException.class) // static member try to send bigger epoch when static member already existed, then throw FencedInstanceIdException. + ); + } + + @Test + public void testStaticMemberJoinThenRevokeAndReceiveTasks() { + int enoughMaxSize = 100; + testStaticMemberJoinThenRevokeAndReceiveTasksWith2Members(enoughMaxSize); + } + + @Test + public void testStaticMemberJoinThenRevokeAndReceiveTasksInMaxSizeBoundary() { + int boundarySize = 2; + testStaticMemberJoinThenRevokeAndReceiveTasksWith2Members(boundarySize); + } + + private void testStaticMemberJoinThenRevokeAndReceiveTasksWith2Members(int maxSize) { + String groupId = "fooup"; + + String memberId1 = Uuid.randomUuid().toString(); + String memberId2 = Uuid.randomUuid().toString(); + String otherMemberId2 = Uuid.randomUuid().toString(); + + String instanceId1 = Uuid.randomUuid().toString(); + String instanceId2 = Uuid.randomUuid().toString(); + + StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4); + + MockTaskAssignor assignor = new MockTaskAssignor("sticky"); + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_MAX_SIZE_CONFIG, maxSize) + .withStreamsGroupTaskAssignors(List.of(assignor)) + .withMetadataImage(topic.metadataImage()) + .withStreamsGroup(new StreamsGroupBuilder(groupId, 10) + .withMember(StreamsGroupTestUtil.streamsGroupMemberBuilderWithDefaults(memberId1) + .setInstanceId(instanceId1) + .setMemberEpoch(10) + .setPreviousMemberEpoch(9) + .setAssignedTasks(topic.assignedTasks(10, 0, 1, 2, 3)) + .build()) + .withTargetAssignment(memberId1, topic.targetAssignment(0, 1, 2, 3)) + .withTargetAssignmentEpoch(10) + .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology())) + .withMetadataHash(topic.metadataHash()) + .withValidatedTopologyEpoch(0) + ) + .build(); + + // Next target assignment after member2 joins. + assignor.prepareGroupAssignment(Map.of( + memberId1, topic.targetAssignment(0, 1), + memberId2, topic.targetAssignment(2, 3) + )); + + // 1) Static member2 joins. It gets no active tasks yet because member1 still owns them. + CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> joinResult = context.streamsGroupHeartbeat( + staticJoinHeartbeat(groupId, memberId2, instanceId2, topic) + ); + + assertResponseEquals( + new StreamsGroupHeartbeatResponseData() + .setMemberId(memberId2) + .setMemberEpoch(11) + .setHeartbeatIntervalMs(5000) + .setTaskOffsetIntervalMs(60000) + .setActiveTasks(List.of()) + .setWarmupTasks(List.of()) + .setStandbyTasks(List.of()), + joinResult.response().data() + ); + + // 2) member1 receives revocation instruction: keep only [0,1]. + CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> revokeInstructionResult = context.streamsGroupHeartbeat( + new StreamsGroupHeartbeatRequestData() + .setGroupId(groupId) + .setInstanceId(instanceId1) + .setMemberId(memberId1) + .setMemberEpoch(10) + ); + + assertResponseEquals( + new StreamsGroupHeartbeatResponseData() + .setMemberId(memberId1) + .setMemberEpoch(10) + .setHeartbeatIntervalMs(5000) + .setTaskOffsetIntervalMs(60000) + .setActiveTasks(topic.responseTasks(0, 1)) + .setWarmupTasks(List.of()) + .setStandbyTasks(List.of()), + revokeInstructionResult.response().data() + ); + + // 3) member1 acknowledges revocation by reporting owned active tasks [0,1]. + CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> revokeAckResult = context.streamsGroupHeartbeat( + staticHeartbeat(groupId, memberId1, instanceId1, 10) + .setActiveTasks(List.of( + new StreamsGroupHeartbeatRequestData.TaskIds() + .setSubtopologyId("subtopology1") + .setPartitions(List.of(0, 1)) + )) + .setStandbyTasks(List.of()) + .setWarmupTasks(List.of()) + ); + + assertResponseEquals( + new StreamsGroupHeartbeatResponseData() + .setMemberId(memberId1) + .setMemberEpoch(11) + .setHeartbeatIntervalMs(5000) + .setTaskOffsetIntervalMs(60000) + .setStatus(List.of()), + revokeAckResult.response().data() + ); + + // 4) member2 heartbeats again and now receives [2,3]. + CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> member2ReceiveResult = context.streamsGroupHeartbeat( + staticHeartbeat(groupId, memberId2, instanceId2, 11) + ); + + assertResponseEquals( + new StreamsGroupHeartbeatResponseData() + .setMemberId(memberId2) + .setMemberEpoch(11) + .setHeartbeatIntervalMs(5000) + .setTaskOffsetIntervalMs(60000) + .setActiveTasks(topic.responseTasks(2, 3)) + .setWarmupTasks(List.of()) + .setStandbyTasks(List.of()), + member2ReceiveResult.response().data() + ); + + // 5) member2 leave. + CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> member2LeaveResult = context.streamsGroupHeartbeat( + staticHeartbeat(groupId, memberId2, instanceId2, LEAVE_GROUP_STATIC_MEMBER_EPOCH) + ); + + assertResponseEquals( + new StreamsGroupHeartbeatResponseData() + .setMemberId(memberId2) + .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH) + .setStatus(List.of()) + .setActiveTasks(null) + .setWarmupTasks(null) + .setStandbyTasks(null) + .setHeartbeatIntervalMs(0), + member2LeaveResult.response().data() + ); + + // 6) member2 re-join with other memberId. + CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> member2rejoinResult = context.streamsGroupHeartbeat( + staticHeartbeat(groupId, otherMemberId2, instanceId2, JOIN_GROUP_MEMBER_EPOCH) + ); + + assertResponseEquals( + new StreamsGroupHeartbeatResponseData() + .setMemberId(otherMemberId2) + .setMemberEpoch(11) + .setHeartbeatIntervalMs(5000) + .setTaskOffsetIntervalMs(60000) + .setActiveTasks(topic.responseTasks(2, 3)) + .setWarmupTasks(List.of()) + .setStandbyTasks(List.of()), + member2rejoinResult.response().data() + ); + } + + @Test + public void testStaticMemberRejoinWithUpdatedProcessIdBumpsStreamsGroupEpoch() { + String groupId = "fooup"; + int groupEpoch = DEFAULT_GROUP_EPOCH; + int bumpedGroupEpoch = groupEpoch + 1; + + String oldMemberId = Uuid.randomUuid().toString(); + String rejoinMemberId = Uuid.randomUuid().toString(); + String instanceId = Uuid.randomUuid().toString(); + + String oldProcessId = "old-process-id"; + String newProcessId = "new-process-id"; + + StreamsTopicFixture topic = streamsTopicFixture("subtopology1", "foo", 4); + TasksTupleWithEpochs assignedTasks = topic.assignedTasks(groupEpoch, 0, 1, 2, 3); + TasksTuple targetAssignment = topic.targetAssignment(0, 1, 2, 3); + + MockTaskAssignor assignor = new MockTaskAssignor("sticky"); + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withStreamsGroupTaskAssignors(List.of(assignor)) + .withMetadataImage(topic.metadataImage()) + .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch) + .withMember(streamsGroupMemberBuilderWithDefaults(oldMemberId, instanceId) + .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH) + .setPreviousMemberEpoch(groupEpoch) + .setProcessId(oldProcessId) + .setAssignedTasks(assignedTasks) + .build()) + .withTargetAssignment(oldMemberId, targetAssignment) + .withTargetAssignmentEpoch(groupEpoch) + .withTopology(StreamsTopology.fromHeartbeatRequest(topic.topology())) + .withValidatedTopologyEpoch(0) + .withMetadataHash(topic.metadataHash()) + .withLastAssignmentConfigs(getDefaultAssignmentConfigs())) + .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_DEFAULT) + .build(); + + assignor.prepareGroupAssignment(Map.of(rejoinMemberId, targetAssignment)); + + CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat( + staticJoinHeartbeat(groupId, rejoinMemberId, instanceId, newProcessId) + ); + + assertEquals(rejoinMemberId, result.response().data().memberId()); + assertEquals(bumpedGroupEpoch, result.response().data().memberEpoch()); + + CoordinatorRecord metadataRecord = result.records().stream() + .filter(record -> record.key() instanceof StreamsGroupMetadataKey) + .findFirst() + .orElse(null); + + assertNotNull(metadataRecord, "Expected a StreamsGroupMetadata record when static member config changes."); + StreamsGroupMetadataValue metadataValue = (StreamsGroupMetadataValue) metadataRecord.value().message(); + assertEquals(bumpedGroupEpoch, metadataValue.epoch()); + + assertTrue(result.records().contains( + StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord( + groupId, + bumpedGroupEpoch, + topic.metadataHash(), + 0, + getDefaultAssignmentConfigs() + ) + )); + } + + @Test + public void testStaticMemberRejoinWithUpdatedProcessIdBumpsStreamsGroupEpoch1() throws UnknownHostException { Review Comment: You are right. I missed it, My bad...! Fixed it! -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
