dajac commented on code in PR #13870:
URL: https://github.com/apache/kafka/pull/13870#discussion_r1243806647
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -874,4 +1072,1338 @@ 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.
+ // TODO: this needs to be checked in conjunction with empty group
offsets.
+// if (groups.containsKey(groupId)) {
+// throw new IllegalStateException("Unexpected unload of active
group " + groupId +
+// "while loading partition " + topicPartition);
+// }
+ } 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<CompletableFuture<Void>, Record> genericGroupJoin(
+ RequestContext context,
+ JoinGroupRequestData request,
+ CompletableFuture<JoinGroupResponseData> responseFuture
+ ) {
+ CoordinatorResult<CompletableFuture<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;
+ }
+
+ String joinReason = request.reason();
+ if (joinReason == null || joinReason.isEmpty()) {
+ joinReason = "not provided";
+ }
+
+ if (!acceptJoiningMember(group, memberId)) {
+ group.remove(memberId);
+ responseFuture.complete(new JoinGroupResponseData()
+ .setMemberId(UNKNOWN_MEMBER_ID)
+ .setErrorCode(Errors.GROUP_MAX_SIZE_REACHED.code())
+ );
+
+ } else if (isUnknownMember) {
+ result = genericGroupJoinNewMember(
+ context,
+ request,
+ group,
+ joinReason,
+ responseFuture
+ );
+ } else {
+ result = genericGroupJoinExistingMember(
+ context,
+ request,
+ group,
+ joinReason,
+ responseFuture
+ );
+ }
+
+ // Attempt to complete join group phase. We do not complete
+ // the join group phase if this is the initial rebalance.
+ if (group.isInState(PREPARING_REBALANCE) &&
+ group.hasAllMembersJoined() &&
+ group.generationId() != 0
+ ) {
+ // The only two cases where we produce records to append are
when:
+ // 1) A new static member replaces an existing member
during Stable state.
+ // 2) The group already completed the join phase and
transitioned to Empty state.
+ // Therefore, we should not be in PreparingRebalance state.
+ if (result != EMPTY_RESULT) {
+ throw new IllegalStateException("There are records to
append but we are attempting to" +
+ "complete the join phase.");
+ }
Review Comment:
How about the following? We keep track whether the group was newly created
in a boolean. When we get the result from those methods, we check if the group
is new, if it is, we check if the result has at least one record. If it does
not, we recreate it while adding an empty record for the group.
--
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
