jeffkbkim commented on code in PR #13870:
URL: https://github.com/apache/kafka/pull/13870#discussion_r1259096806
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -2377,156 +2606,2165 @@ public void testOnNewMetadataImage() {
// Verify the groups.
Arrays.asList("group1", "group2", "group3", "group4").forEach(groupId
-> {
- ConsumerGroup group =
context.groupMetadataManager.getOrMaybeCreateConsumerGroup(groupId, false);
+ ConsumerGroup group = context.groupMetadataManager
+ .getOrMaybeCreateConsumerGroup(groupId, false);
assertTrue(group.hasMetadataExpired(context.time.milliseconds()));
});
Arrays.asList("group5").forEach(groupId -> {
- ConsumerGroup group =
context.groupMetadataManager.getOrMaybeCreateConsumerGroup(groupId, false);
+ ConsumerGroup group = context.groupMetadataManager
+ .getOrMaybeCreateConsumerGroup(groupId, false);
assertFalse(group.hasMetadataExpired(context.time.milliseconds()));
});
// Verify image.
assertEquals(image, context.groupMetadataManager.image());
}
- private <T> void assertUnorderedListEquals(
- List<T> expected,
- List<T> actual
- ) {
- assertEquals(new HashSet<>(expected), new HashSet<>(actual));
- }
+ @Test
+ public void testJoinGroupShouldReceiveErrorIfGroupOverMaxSize() throws
Exception {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withGenericGroupMaxSize(10)
+ .build();
- private void assertResponseEquals(
- ConsumerGroupHeartbeatResponseData expected,
- ConsumerGroupHeartbeatResponseData actual
- ) {
- if (!responseEquals(expected, actual)) {
- assertionFailure()
- .expected(expected)
- .actual(actual)
- .buildAndThrow();
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withDefaultProtocolTypeAndProtocols()
+ .withReason("exceed max group size")
+ .build();
+
+ for (int i = 0; i < 10; i++) {
+ CompletableFuture<JoinGroupResponseData> responseFuture;
+ if (i == 0) {
+ responseFuture = context.sendGenericGroupJoin(
+ request,
+ false,
+ false,
+ new ExpectedGenericGroupResult(Errors.NONE, true)
+ );
+ } else {
+ responseFuture = context.sendGenericGroupJoin(request);
+ }
+ assertFalse(responseFuture.isDone());
}
+ CompletableFuture<JoinGroupResponseData> responseFuture =
context.sendGenericGroupJoin(request);
+ assertTrue(responseFuture.isDone());
+ assertEquals(Errors.GROUP_MAX_SIZE_REACHED.code(),
responseFuture.get(5, TimeUnit.SECONDS).errorCode());
}
- private boolean responseEquals(
- ConsumerGroupHeartbeatResponseData expected,
- ConsumerGroupHeartbeatResponseData actual
- ) {
- if (expected.throttleTimeMs() != actual.throttleTimeMs()) return false;
- if (expected.errorCode() != actual.errorCode()) return false;
- if (!Objects.equals(expected.errorMessage(), actual.errorMessage()))
return false;
- if (!Objects.equals(expected.memberId(), actual.memberId())) return
false;
- if (expected.memberEpoch() != actual.memberEpoch()) return false;
- if (expected.shouldComputeAssignment() !=
actual.shouldComputeAssignment()) return false;
- if (expected.heartbeatIntervalMs() != actual.heartbeatIntervalMs())
return false;
- // Unordered comparison of the assignments.
- return responseAssignmentEquals(expected.assignment(),
actual.assignment());
- }
+ @Test
+ public void testDynamicMembersJoinGroupWithMaxSizeAndRequiredKnownMember()
{
+ boolean requiredKnownMemberId = true;
+ int groupMaxSize = 10;
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withGenericGroupMaxSize(groupMaxSize)
+ .withGenericGroupInitialRebalanceDelayMs(50)
+ .build();
- private boolean responseAssignmentEquals(
- ConsumerGroupHeartbeatResponseData.Assignment expected,
- ConsumerGroupHeartbeatResponseData.Assignment actual
- ) {
- if (expected == actual) return true;
- if (expected == null) return false;
- if (actual == null) return false;
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withDefaultProtocolTypeAndProtocols()
+ .build();
- if (!Objects.equals(fromAssignment(expected.pendingTopicPartitions()),
fromAssignment(actual.pendingTopicPartitions())))
- return false;
+ // First round of join requests. Generate member ids. All requests
will be accepted
+ // as the group is still Empty.
+ List<CompletableFuture<JoinGroupResponseData>> responseFutures = new
ArrayList<>();
+ for (int i = 0; i < groupMaxSize + 1; i++) {
+ if (i == 0) {
+ responseFutures.add(context.sendGenericGroupJoin(
+ request,
+ requiredKnownMemberId,
+ false,
+ new ExpectedGenericGroupResult(Errors.NONE, true)
+ ));
+ } else {
+ responseFutures.add(context.sendGenericGroupJoin(request,
requiredKnownMemberId));
+ }
+ }
- return
Objects.equals(fromAssignment(expected.assignedTopicPartitions()),
fromAssignment(actual.assignedTopicPartitions()));
+ GenericGroup group =
context.groupMetadataManager.getOrMaybeCreateGenericGroup(
+ "group-id",
+ false
+ );
+
+ List<String> memberIds =
verifyGenericGroupJoinResponses(responseFutures, 0, Errors.MEMBER_ID_REQUIRED);
+ assertEquals(groupMaxSize + 1, memberIds.size());
+ assertEquals(0, group.size());
+ assertTrue(group.isInState(EMPTY));
+ assertEquals(groupMaxSize + 1, group.numPendingJoinMembers());
+
+ // Second round of join requests with the generated member ids.
+ // One of them will fail, reaching group max size.
+ responseFutures = new ArrayList<>();
+ for (String memberId : memberIds) {
+ request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(memberId)
+ .withDefaultProtocolTypeAndProtocols()
+ .build();
+
+ responseFutures.add(context.sendGenericGroupJoin(request,
requiredKnownMemberId));
+ }
+
+ // Advance clock by group initial rebalance delay to complete first
inital delayed join.
+ // This will extend the initial rebalance as new members have joined.
+ context.timer.advanceClock(50);
+ // Advance clock by group initial rebalance delay to complete second
inital delayed join.
+ // Since there are no new members that joined since the previous
delayed join,
+ // the join group phase will complete.
+ context.timer.advanceClock(50);
+
+ verifyGenericGroupJoinResponses(responseFutures, groupMaxSize,
Errors.GROUP_MAX_SIZE_REACHED);
+ assertEquals(groupMaxSize, group.size());
+ assertEquals(0, group.numPendingJoinMembers());
+ assertTrue(group.isInState(COMPLETING_REBALANCE));
+
+
+ // Members that were accepted can rejoin while others are rejected in
CompletingRebalance state.
+ responseFutures = new ArrayList<>();
+ for (String memberId : memberIds) {
+ request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(memberId)
+ .withDefaultProtocolTypeAndProtocols()
+ .build();
+
+ responseFutures.add(context.sendGenericGroupJoin(request,
requiredKnownMemberId));
+ }
+
+ verifyGenericGroupJoinResponses(responseFutures, groupMaxSize,
Errors.GROUP_MAX_SIZE_REACHED);
}
- private Map<Uuid, Set<Integer>> fromAssignment(
- List<ConsumerGroupHeartbeatResponseData.TopicPartitions> assignment
- ) {
- if (assignment == null) return null;
+ @Test
+ public void
testDynamicMembersJoinGroupWithMaxSizeAndNotRequiredKnownMember() {
+ boolean requiredKnownMemberId = false;
+ int groupMaxSize = 10;
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withGenericGroupMaxSize(groupMaxSize)
+ .withGenericGroupInitialRebalanceDelayMs(50)
+ .build();
- Map<Uuid, Set<Integer>> assignmentMap = new HashMap<>();
- assignment.forEach(topicPartitions -> {
- assignmentMap.put(topicPartitions.topicId(), new
HashSet<>(topicPartitions.partitions()));
- });
- return assignmentMap;
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withDefaultProtocolTypeAndProtocols()
+ .build();
+
+ // First round of join requests. This will trigger a rebalance.
+ List<CompletableFuture<JoinGroupResponseData>> responseFutures = new
ArrayList<>();
+ responseFutures.add(context.sendGenericGroupJoin(
+ request,
+ requiredKnownMemberId,
+ true,
+ new ExpectedGenericGroupResult(Errors.NONE, true)
+ ));
+ for (int i = 0; i < groupMaxSize; i++) {
+ responseFutures.add(context.sendGenericGroupJoin(request,
requiredKnownMemberId));
+ }
+
+ GenericGroup group =
context.groupMetadataManager.getOrMaybeCreateGenericGroup(
+ "group-id",
+
+ false
+ );
+
+ assertEquals(groupMaxSize, group.size());
+ assertEquals(groupMaxSize, group.numAwaitingJoinResponse());
+ assertTrue(group.isInState(PREPARING_REBALANCE));
+
+ // Advance clock by group initial rebalance delay to complete first
inital delayed join.
+ // This will extend the initial rebalance as new members have joined.
+ context.timer.advanceClock(50);
+ // Advance clock by group initial rebalance delay to complete second
inital delayed join.
+ // Since there are no new members that joined since the previous
delayed join,
+ // we will complete the rebalance.
+ context.timer.advanceClock(50);
+
+ List<String> memberIds =
verifyGenericGroupJoinResponses(responseFutures, groupMaxSize,
Errors.GROUP_MAX_SIZE_REACHED);
+
+ // Members that were accepted can rejoin while others are rejected in
CompletingRebalance state.
+ responseFutures = new ArrayList<>();
+ for (String memberId : memberIds) {
+ request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(memberId)
+ .withDefaultProtocolTypeAndProtocols()
+ .build();
+
+ responseFutures.add(context.sendGenericGroupJoin(request,
requiredKnownMemberId));
+ }
+
+ verifyGenericGroupJoinResponses(responseFutures, 10,
Errors.GROUP_MAX_SIZE_REACHED);
+ assertEquals(groupMaxSize, group.size());
+ assertEquals(0, group.numAwaitingJoinResponse());
+ assertTrue(group.isInState(COMPLETING_REBALANCE));
}
- private void assertRecordsEquals(
- List<Record> expectedRecords,
- List<Record> actualRecords
- ) {
- try {
- assertEquals(expectedRecords.size(), actualRecords.size());
+ @Test
+ public void testStaticMembersJoinGroupWithMaxSize() {
+ int groupMaxSize = 10;
- for (int i = 0; i < expectedRecords.size(); i++) {
- Record expectedRecord = expectedRecords.get(i);
- Record actualRecord = actualRecords.get(i);
- assertRecordEquals(expectedRecord, actualRecord);
+ List<String> groupInstanceIds = IntStream.range(0, groupMaxSize + 1)
+ .mapToObj(i -> "instance-id-" + i)
+ .collect(Collectors.toList());
+
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withGenericGroupMaxSize(groupMaxSize)
+ .withGenericGroupInitialRebalanceDelayMs(50)
+ .build();
+
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withDefaultProtocolTypeAndProtocols()
+ .build();
+
+ // First round of join requests. The group metadata manager will
prepare a rebalance.
+ List<CompletableFuture<JoinGroupResponseData>> responseFutures = new
ArrayList<>();
+ for (int i = 0; i < groupMaxSize + 1; i++) {
+ String instanceId = groupInstanceIds.get(i);
+ request = request.setGroupInstanceId(instanceId);
+ if (i == 0) {
+ responseFutures.add(context.sendGenericGroupJoin(
+ request,
+ false,
+ true,
+ new ExpectedGenericGroupResult(Errors.NONE, true)
+ ));
+ } else {
+ responseFutures.add(context.sendGenericGroupJoin(request));
}
- } catch (AssertionFailedError e) {
- assertionFailure()
- .expected(expectedRecords)
- .actual(actualRecords)
- .buildAndThrow();
}
- }
- private void assertRecordEquals(
- Record expected,
- Record actual
- ) {
- try {
- assertApiMessageAndVersionEquals(expected.key(), actual.key());
- assertApiMessageAndVersionEquals(expected.value(), actual.value());
- } catch (AssertionFailedError e) {
- assertionFailure()
- .expected(expected)
- .actual(actual)
- .buildAndThrow();
+ GenericGroup group =
context.groupMetadataManager.getOrMaybeCreateGenericGroup(
+ "group-id",
+ false
+ );
+
+ assertEquals(groupMaxSize, group.size());
+ assertEquals(groupMaxSize, group.numAwaitingJoinResponse());
+ assertTrue(group.isInState(PREPARING_REBALANCE));
+
+ // Advance clock by group initial rebalance delay to complete first
inital delayed join.
+ // This will extend the initial rebalance as new members have joined.
+ context.timer.advanceClock(50);
+ // Advance clock by group initial rebalance delay to complete second
inital delayed join.
+ // Since there are no new members that joined since the previous
delayed join,
+ // we will complete the rebalance.
+ context.timer.advanceClock(50);
+
+ List<String> memberIds =
verifyGenericGroupJoinResponses(responseFutures, groupMaxSize,
Errors.GROUP_MAX_SIZE_REACHED);
+
+ // Members which were accepted can rejoin, others are rejected, while
+ // completing rebalance
+ responseFutures = new ArrayList<>();
+ for (int i = 0; i < groupMaxSize; i++) {
+ String memberId = memberIds.get(i);
+ String instanceId = groupInstanceIds.get(i);
+
+ request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(memberId)
+ .withGroupInstanceId(instanceId)
+ .withDefaultProtocolTypeAndProtocols()
+ .build();
+
+ responseFutures.add(context.sendGenericGroupJoin(request));
}
+
+ verifyGenericGroupJoinResponses(responseFutures, groupMaxSize,
Errors.GROUP_MAX_SIZE_REACHED);
+ assertEquals(groupMaxSize, group.size());
+ assertEquals(0, group.numAwaitingJoinResponse());
+ assertTrue(group.isInState(COMPLETING_REBALANCE));
}
- private void assertApiMessageAndVersionEquals(
- ApiMessageAndVersion expected,
- ApiMessageAndVersion actual
- ) {
- if (expected == actual) return;
+ @Test
+ public void testDynamicMembersCanRejoinGroupWithMaxSizeWhileRebalancing() {
+ boolean requiredKnownMemberId = true;
+ int groupMaxSize = 10;
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withGenericGroupMaxSize(groupMaxSize)
+ .withGenericGroupInitialRebalanceDelayMs(50)
+ .build();
- assertEquals(expected.version(), actual.version());
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withDefaultProtocolTypeAndProtocols()
+ .build();
- if (actual.message() instanceof
ConsumerGroupCurrentMemberAssignmentValue) {
- // The order of the topics stored in
ConsumerGroupCurrentMemberAssignmentValue is not
- // always guaranteed. Therefore, we need a special comparator.
- ConsumerGroupCurrentMemberAssignmentValue expectedValue =
- (ConsumerGroupCurrentMemberAssignmentValue) expected.message();
- ConsumerGroupCurrentMemberAssignmentValue actualValue =
- (ConsumerGroupCurrentMemberAssignmentValue) actual.message();
+ List<CompletableFuture<JoinGroupResponseData>> responseFutures = new
ArrayList<>();
+ // First round of join requests. Generate member ids.
+ for (int i = 0; i < groupMaxSize + 1; i++) {
+ if (i == 0) {
+ responseFutures.add(context.sendGenericGroupJoin(
+ request,
+ requiredKnownMemberId,
+ false,
+ new ExpectedGenericGroupResult(Errors.NONE, true)
+ ));
+ } else {
+ responseFutures.add(context.sendGenericGroupJoin(request,
requiredKnownMemberId));
+ }
+ }
- assertEquals(expectedValue.memberEpoch(),
actualValue.memberEpoch());
- assertEquals(expectedValue.previousMemberEpoch(),
actualValue.previousMemberEpoch());
- assertEquals(expectedValue.targetMemberEpoch(),
actualValue.targetMemberEpoch());
- assertEquals(expectedValue.error(), actualValue.error());
- assertEquals(expectedValue.metadataVersion(),
actualValue.metadataVersion());
- assertEquals(expectedValue.metadataBytes(),
actualValue.metadataBytes());
+ GenericGroup group =
context.groupMetadataManager.getOrMaybeCreateGenericGroup(
+ "group-id",
+ false
+ );
- // We transform those to Maps before comparing them.
-
assertEquals(fromTopicPartitions(expectedValue.assignedPartitions()),
- fromTopicPartitions(actualValue.assignedPartitions()));
-
assertEquals(fromTopicPartitions(expectedValue.partitionsPendingRevocation()),
-
fromTopicPartitions(actualValue.partitionsPendingRevocation()));
-
assertEquals(fromTopicPartitions(expectedValue.partitionsPendingAssignment()),
-
fromTopicPartitions(actualValue.partitionsPendingAssignment()));
- } else {
- assertEquals(expected.message(), actual.message());
+ assertEquals(0, group.size());
+ assertEquals(groupMaxSize + 1, group.numPendingJoinMembers());
+ assertTrue(group.isInState(EMPTY));
+
+ List<String> memberIds =
verifyGenericGroupJoinResponses(responseFutures, 0, Errors.MEMBER_ID_REQUIRED);
+ assertEquals(groupMaxSize + 1, memberIds.size());
+
+ // Second round of join requests with the generated member ids.
+ // One of them will fail, reaching group max size.
+ responseFutures = new ArrayList<>();
+ for (String memberId : memberIds) {
+
responseFutures.add(context.sendGenericGroupJoin(request.setMemberId(memberId),
requiredKnownMemberId));
}
- }
- private Map<Uuid, Set<Integer>> fromTopicPartitions(
- List<ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions>
assignment
- ) {
- Map<Uuid, Set<Integer>> assignmentMap = new HashMap<>();
- assignment.forEach(topicPartitions -> {
- assignmentMap.put(topicPartitions.topicId(), new
HashSet<>(topicPartitions.partitions()));
- });
- return assignmentMap;
+ assertEquals(groupMaxSize, group.size());
+ assertEquals(groupMaxSize, group.numAwaitingJoinResponse());
+ assertTrue(group.isInState(PREPARING_REBALANCE));
+
+ // Members can rejoin while rebalancing
+ responseFutures = new ArrayList<>();
+ for (String memberId : memberIds) {
+ request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(memberId)
+ .withDefaultProtocolTypeAndProtocols()
+ .build();
+
+ responseFutures.add(context.sendGenericGroupJoin(request,
requiredKnownMemberId));
+ }
+
+ // Advance clock by group initial rebalance delay to complete first
inital delayed join.
+ // This will extend the initial rebalance as new members have joined.
+ context.timer.advanceClock(50);
+ // Advance clock by group initial rebalance delay to complete second
inital delayed join.
+ // Since there are no new members that joined since the previous
delayed join,
+ // we will complete the rebalance.
+ context.timer.advanceClock(50);
+
+ verifyGenericGroupJoinResponses(responseFutures, groupMaxSize,
Errors.GROUP_MAX_SIZE_REACHED);
+ assertEquals(groupMaxSize, group.size());
+ assertEquals(0, group.numAwaitingJoinResponse());
+ assertTrue(group.isInState(COMPLETING_REBALANCE));
}
+
+ @Test
+ public void
testLastJoiningMembersAreKickedOutWhenRejoiningGroupWithMaxSize() {
+ int groupMaxSize = 10;
+
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withGenericGroupMaxSize(groupMaxSize)
+ .withGenericGroupInitialRebalanceDelayMs(50)
+ .build();
+
+ // Create a group and add members that exceed the group max size.
+ GenericGroup group =
context.groupMetadataManager.getOrMaybeCreateGenericGroup(
+ "group-id",
+ true
+ );
+
+ List<String> memberIds = IntStream.range(0, groupMaxSize + 2)
+ .mapToObj(i -> group.generateMemberId("client-id",
Optional.empty()))
+ .collect(Collectors.toList());
+
+ memberIds.forEach(memberId -> {
+ JoinGroupRequestProtocolCollection protocols = new
JoinGroupRequestProtocolCollection();
+ protocols.add(new JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(new byte[0]));
+
+ group.add(
+ new GenericGroupMember(
+ memberId,
+ Optional.empty(),
+ "client-id",
+ "client-host",
+ 10000,
+ 5000,
+ "consumer",
+ protocols
+ )
+ );
+ });
+
+ context.groupMetadataManager.prepareRebalance(group, "test");
+
+ List<CompletableFuture<JoinGroupResponseData>> responseFutures = new
ArrayList<>();
+ for (String memberId : memberIds) {
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(memberId)
+ .withDefaultProtocolTypeAndProtocols()
+ .withRebalanceTimeoutMs(10000)
+ .build();
+
+ responseFutures.add(context.sendGenericGroupJoin(request));
+ }
+
+ assertEquals(groupMaxSize, group.size());
+ assertEquals(groupMaxSize, group.numAwaitingJoinResponse());
+ assertTrue(group.isInState(PREPARING_REBALANCE));
+
+ // Advance clock by rebalance timeout to complete join phase.
+ context.timer.advanceClock(10000);
+
+ verifyGenericGroupJoinResponses(responseFutures, groupMaxSize,
Errors.GROUP_MAX_SIZE_REACHED);
+
+ assertEquals(groupMaxSize, group.size());
+ assertTrue(group.isInState(COMPLETING_REBALANCE));
+
+ memberIds.subList(groupMaxSize, groupMaxSize + 2)
+ .forEach(memberId -> assertFalse(group.hasMemberId(memberId)));
+
+ memberIds.subList(0, groupMaxSize)
+ .forEach(memberId -> assertTrue(group.hasMemberId(memberId)));
+ }
+
+ @Test
+ public void testJoinGroupSessionTimeoutTooSmall() throws Exception {
+ int minSessionTimeout = 50;
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withGenericGroupMinSessionTimeoutMs(minSessionTimeout)
+ .build();
+
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withSessionTimeoutMs(minSessionTimeout - 1)
+ .build();
+
+ CompletableFuture<JoinGroupResponseData> responseFuture =
context.sendGenericGroupJoin(request);
+ assertTrue(responseFuture.isDone());
+ assertEquals(Errors.INVALID_SESSION_TIMEOUT.code(),
responseFuture.get(5, TimeUnit.SECONDS).errorCode());
+ }
+
+ @Test
+ public void testJoinGroupSessionTimeoutTooLarge() throws Exception {
+ int maxSessionTimeout = 50;
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withGenericGroupMaxSessionTimeoutMs(maxSessionTimeout)
+ .build();
+
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withSessionTimeoutMs(maxSessionTimeout + 1)
+ .build();
+
+ CompletableFuture<JoinGroupResponseData> responseFuture =
context.sendGenericGroupJoin(request);
+ assertTrue(responseFuture.isDone());
+ assertEquals(Errors.INVALID_SESSION_TIMEOUT.code(),
responseFuture.get(5, TimeUnit.SECONDS).errorCode());
+ }
+
+ @Test
+ public void testJoinGroupUnknownConsumerNewGroup() throws Exception {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId("member-id")
+ .build();
+
+ JoinGroupResponseData response =
context.joinGenericGroupAsDynamicMember(request);
+
+ assertEquals(Errors.GROUP_ID_NOT_FOUND.code(), response.errorCode());
+
+ // Static member
+ request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId("member-id")
+ .withGroupInstanceId("group-instance-id")
+ .build();
+
+ CompletableFuture<JoinGroupResponseData> responseFuture =
context.sendGenericGroupJoin(request);
+ assertTrue(responseFuture.isDone());
+ assertEquals(Errors.GROUP_ID_NOT_FOUND.code(), responseFuture.get(5,
TimeUnit.SECONDS).errorCode());
+ }
+
+ @Test
+ public void testGenericGroupJoinInconsistentProtocolType() throws
Exception {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ JoinGroupRequestProtocolCollection protocols = new
JoinGroupRequestProtocolCollection(0);
+ protocols.add(new JoinGroupRequestProtocol().setName("range"));
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withDefaultProtocolTypeAndProtocols()
+ .build();
+
+ JoinGroupResponseData response =
context.joinGenericGroupAsDynamicMember(request);
+ assertEquals(Errors.NONE.code(), response.errorCode());
+
+ request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withProtocolType("connect")
+ .withProtocols(protocols)
+ .build();
+
+ CompletableFuture<JoinGroupResponseData> responseFuture =
context.sendGenericGroupJoin(request);
+ assertTrue(responseFuture.isDone());
+ assertEquals(Errors.INCONSISTENT_GROUP_PROTOCOL.code(),
responseFuture.get(5, TimeUnit.SECONDS).errorCode());
+ }
+
+ @Test
+ public void testJoinGroupWithEmptyProtocolType() {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ JoinGroupRequestProtocolCollection protocols = new
JoinGroupRequestProtocolCollection(0);
+ protocols.add(new JoinGroupRequestProtocol().setName("range"));
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withProtocolType("")
+ .withProtocols(protocols)
+ .build();
+
+ JoinGroupResponseData response =
context.joinGenericGroupAsDynamicMember(request);
+ assertEquals(Errors.INCONSISTENT_GROUP_PROTOCOL.code(),
response.errorCode());
+ }
+
+ @Test
+ public void testJoinGroupWithEmptyGroupProtocol() {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ JoinGroupRequestProtocolCollection protocols = new
JoinGroupRequestProtocolCollection(0);
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withProtocolType("consumer")
+ .withProtocols(protocols)
+ .build();
+
+ JoinGroupResponseData response =
context.joinGenericGroupAsDynamicMember(request);
+ assertEquals(Errors.INCONSISTENT_GROUP_PROTOCOL.code(),
response.errorCode());
+ }
+
+ @Test
+ public void testNewMemberJoinExpiration() throws Exception {
+ // This tests new member expiration during a protracted rebalance. We
first create a
+ // group with one member which uses a large value for session timeout
and rebalance timeout.
+ // We then join with one new member and let the rebalance hang while
we await the first member.
+ // The new member join timeout expires and its JoinGroup request is
failed.
+
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withDefaultProtocolTypeAndProtocols()
+ .withSessionTimeoutMs(5000 +
context.genericGroupNewMemberJoinTimeoutMs)
+ .withRebalanceTimeoutMs(2 *
context.genericGroupNewMemberJoinTimeoutMs)
+ .build();
+
+ JoinGroupResponseData firstResponse =
context.joinGenericGroupAsDynamicMember(request);
+ String firstMemberId = firstResponse.memberId();
+ assertEquals(Errors.NONE.code(), firstResponse.errorCode());
+
+ GenericGroup group =
context.groupMetadataManager.getOrMaybeCreateGenericGroup(
+ "group-id",
+ false
+ );
+
+ assertNotNull(group);
+ assertEquals(0,
group.allMembers().stream().filter(GenericGroupMember::isNew).count());
+
+ // Send second join group request for a new dynamic member.
+ CompletableFuture<JoinGroupResponseData> secondResponseFuture =
context.sendGenericGroupJoin(request
+ .setSessionTimeoutMs(5000)
+ .setRebalanceTimeoutMs(5000));
+
+ assertFalse(secondResponseFuture.isDone());
+
+ assertEquals(2, group.allMembers().size());
+ assertEquals(1,
group.allMembers().stream().filter(GenericGroupMember::isNew).count());
+
+ GenericGroupMember newMember =
group.allMembers().stream().filter(GenericGroupMember::isNew).findFirst().get();
+ assertNotEquals(firstMemberId, newMember.memberId());
+
+ // Advance clock by new member join timeout to expire the second
member.
+ context.timer.advanceClock(context.genericGroupNewMemberJoinTimeoutMs);
Review Comment:
there's a +1 on all advance clocks in scala. i haven't actually looked but
assumed that it's in place due to how the purgatory works.
the java timer implementation does not require a +1
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
