jeffkbkim commented on code in PR #13870:
URL: https://github.com/apache/kafka/pull/13870#discussion_r1254999681
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -874,4 +1087,1348 @@ public void replay(
consumerGroup.updateMember(newMember);
}
}
+
+ /**
+ * Replays GroupMetadataKey/Value to update the soft state of
+ * the generic group.
+ *
+ * @param key A GroupMetadataKey key.
+ * @param value A GroupMetadataValue record.
+ */
+ public void replay(
+ GroupMetadataKey key,
+ GroupMetadataValue value,
+ short version
+ ) {
+ String groupId = key.group();
+
+ if (value == null) {
+ // Tombstone. Group should not be added.
+ } else {
+ List<GenericGroupMember> loadedMembers = new ArrayList<>();
+ for (GroupMetadataValue.MemberMetadata member : value.members()) {
+ int rebalanceTimeout = version == 0 ? member.sessionTimeout()
: member.rebalanceTimeout();
+
+ JoinGroupRequestProtocolCollection supportedProtocols = new
JoinGroupRequestProtocolCollection();
+ supportedProtocols.add(new JoinGroupRequestProtocol()
+ .setName(value.protocol())
+ .setMetadata(member.subscription()));
+
+ GenericGroupMember loadedMember = new GenericGroupMember(
+ member.memberId(),
+ Optional.ofNullable(member.groupInstanceId()),
+ member.clientId(),
+ member.clientHost(),
+ rebalanceTimeout,
+ member.sessionTimeout(),
+ value.protocolType(),
+ supportedProtocols,
+ member.assignment()
+ );
+
+ loadedMembers.add(loadedMember);
+ }
+
+ String protocolType = value.protocolType();
+
+ GenericGroup genericGroup = new GenericGroup(
+ this.logContext,
+ groupId,
+ loadedMembers.isEmpty() ? EMPTY : STABLE,
+ time,
+ value.generation(),
+ protocolType == null || protocolType.isEmpty() ?
Optional.empty() : Optional.of(protocolType),
+ Optional.ofNullable(value.protocol()),
+ Optional.ofNullable(value.leader()),
+ value.currentStateTimestamp() == -1 ? Optional.empty() :
Optional.of(value.currentStateTimestamp())
+ );
+
+ loadedMembers.forEach(member -> {
+ genericGroup.add(member, null);
+ log.info("Loaded member {} in group {} with generation {}.",
+ member.memberId(), groupId, genericGroup.generationId());
+ });
+
+ genericGroup.setSubscribedTopics(
+ genericGroup.computeSubscribedTopics()
+ );
+ }
+ }
+
+ /**
+ * Handle a JoinGroupRequest.
+ *
+ * @param context The request context.
+ * @param request The actual JoinGroup request.
+ *
+ * @return The result that contains records to append if the join group
phase completes.
+ */
+ public CoordinatorResult<Void, Record> genericGroupJoin(
+ RequestContext context,
+ JoinGroupRequestData request,
+ CompletableFuture<JoinGroupResponseData> responseFuture
+ ) {
+ CoordinatorResult<Void, Record> result = EMPTY_RESULT;
+
+ String groupId = request.groupId();
+ String memberId = request.memberId();
+ int sessionTimeoutMs = request.sessionTimeoutMs();
+
+ if (sessionTimeoutMs < genericGroupMinSessionTimeoutMs ||
+ sessionTimeoutMs > genericGroupMaxSessionTimeoutMs
+ ) {
+ responseFuture.complete(new JoinGroupResponseData()
+ .setMemberId(memberId)
+ .setErrorCode(Errors.INVALID_SESSION_TIMEOUT.code())
+ );
+ } else {
+ boolean isUnknownMember = memberId.equals(UNKNOWN_MEMBER_ID);
+ // Group is created if it does not exist and the member id is
UNKNOWN. if member
+ // is specified but group does not exist, request is rejected with
GROUP_ID_NOT_FOUND
+ GenericGroup group;
+ try {
+ group = getOrMaybeCreateGenericGroup(groupId, isUnknownMember);
+ } catch (Throwable t) {
+ responseFuture.complete(new JoinGroupResponseData()
+ .setMemberId(memberId)
+ .setErrorCode(Errors.forException(t).code())
+ );
+ return EMPTY_RESULT;
+ }
+
+ CoordinatorResult<Void, Record> newGroupResult = EMPTY_RESULT;
+ if (group.isNew()) {
Review Comment:
i misunderstood, moved to using a local variable
--
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
