jeffkbkim commented on code in PR #14122:
URL: https://github.com/apache/kafka/pull/14122#discussion_r1309021495
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -9119,4 +8742,257 @@ public void
testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
HeartbeatResponseData heartbeatResponse =
context.sendGenericGroupHeartbeat(heartbeatRequest);
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(),
heartbeatResponse.errorCode());
}
+
+ private static void assertNoOrEmptyResult(List<ExpiredTimeout<Void,
Record>> timeouts) {
+ assertTrue(timeouts.size() <= 1);
+ timeouts.forEach(timeout -> assertEquals(EMPTY_RESULT,
timeout.result));
+ }
+
+ private List<String> verifyGenericGroupJoinResponses(
+ List<JoinResult> joinResults,
+ int expectedSuccessCount,
+ Errors expectedFailure
+ ) {
+ int successCount = 0;
+ List<String> memberIds = new ArrayList<>();
+ for (JoinResult joinResult : joinResults) {
+ if (!joinResult.joinFuture.isDone()) {
+ fail("All responseFutures should be completed.");
+ }
+ try {
+ JoinGroupResponseData joinResponse =
joinResult.joinFuture.get();
+ if (joinResponse.errorCode() == Errors.NONE.code()) {
+ successCount++;
+ } else {
+ assertEquals(expectedFailure.code(),
joinResponse.errorCode());
+ }
+ memberIds.add(joinResponse.memberId());
+ } catch (Exception e) {
+ fail("Unexpected exception: " + e.getMessage());
+ }
+ }
+
+ assertEquals(expectedSuccessCount, successCount);
+ return memberIds;
+ }
+
+ private static JoinGroupRequestProtocolCollection toProtocols(String...
protocolNames) {
+ JoinGroupRequestProtocolCollection protocols = new
JoinGroupRequestProtocolCollection(0);
+ List<String> topicNames = Arrays.asList("foo", "bar", "baz");
+ for (int i = 0; i < protocolNames.length; i++) {
+ protocols.add(new JoinGroupRequestProtocol()
+ .setName(protocolNames[i])
+ .setMetadata(ConsumerProtocol.serializeSubscription(new
ConsumerPartitionAssignor.Subscription(
+ Collections.singletonList(topicNames.get(i %
topicNames.size())))).array())
+ );
+ }
+ return protocols;
+ }
+
+ private static Record newGroupMetadataRecord(
+ String groupId,
+ GroupMetadataValue value,
+ MetadataVersion metadataVersion
+ ) {
+ return new Record(
+ new ApiMessageAndVersion(
+ new GroupMetadataKey()
+ .setGroup(groupId),
+ (short) 2
+ ),
+ new ApiMessageAndVersion(
+ value,
+ metadataVersion.groupMetadataValueVersion()
+ )
+ );
+ }
+
+ private static Record newGroupMetadataRecordWithCurrentState(
+ GenericGroup group,
+ MetadataVersion metadataVersion
+ ) {
+ return RecordHelpers.newGroupMetadataRecord(group,
group.groupAssignment(), metadataVersion);
+ }
+
+ private static class JoinGroupRequestBuilder {
Review Comment:
@mumrah that makes sense. I will do this as a follow up PR as
https://github.com/apache/kafka/pull/14147 is built on this PR
--
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
