You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dongnuo123 (via GitHub)" <gi...@apache.org> on 2024/03/24 03:44:15 UTC
[PR] [Draft] Add type check to classic group timeout operations [kafka]
dongnuo123 opened a new pull request, #15587:
URL: https://github.com/apache/kafka/pull/15587
When implementing the group type conversion from a classic group to a consumer group, if the replay of conversion records fails, the group should be reverted back including its timeouts.
A possible solution is to keep all the classic group timeouts and add a type check to the timeout operations. If the group is successfully upgraded, it won't be able to pass the type check and its operations will be executed without actually doing anything; if the group upgrade fails, the group map will be reverted and the timeout operations will be executed as is.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Draft] Add type check to classic group timeout operations [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15587:
URL: https://github.com/apache/kafka/pull/15587#discussion_r1551577818
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2814,7 +2820,9 @@ private CoordinatorResult<Void, Record> tryCompleteInitialRebalanceElseSchedule(
int delayMs,
int remainingMs
) {
- if (group.newMemberAdded() && remainingMs != 0) {
+ if (!containsClassicGroup(group.groupId())) {
+ log.info("Group {} is null or not a classic group, skipping the initial rebalance stage.", group.groupId());
+ } else if (group.newMemberAdded() && remainingMs != 0) {
Review Comment:
Same comment.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2451,6 +2451,8 @@ private CoordinatorResult<Void, Record> completeClassicGroupJoin(
if (group.isInState(DEAD)) {
log.info("Group {} is dead, skipping rebalance stage.", groupId);
+ } else if (!containsClassicGroup(group.groupId())) {
+ log.info("Group {} is null or not a classic group, skipping rebalance stage.", groupId);
Review Comment:
I find this check a little unexpected here because the method received a `ClassicGroup`. It seems that `completeClassicGroupJoin` is called from two different contexts. In the first one, it is called in the regular processing of the request so the `ClassicGroup` is available and we know that it is fine. In the second one, it is called from an expired timer. In this case, we need to validate that the group still exists. Therefore, I wonder whether we should add overload to `completeClassicGroupJoin` which takes the group id and looks up the group based on it. We could then use it in the "timer context". What do you think?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2963,7 +2971,9 @@ private CoordinatorResult<Void, Record> expirePendingSync(
ClassicGroup group,
int generationId
) {
- if (generationId != group.generationId()) {
+ if (!containsClassicGroup(group.groupId())) {
Review Comment:
It looks like that we could change the param here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2553,6 +2555,10 @@ private CoordinatorResult<Void, Record> expireClassicGroupMemberHeartbeat(
log.info("Received notification of heartbeat expiration for member {} after group {} " +
"had already been unloaded or deleted.",
memberId, group.groupId());
+ } else if (!containsClassicGroup(group.groupId())) {
+ log.info("Received notification of heartbeat expiration for member {} after group {} " +
+ "had already been deleted or upgraded.",
+ memberId, group.groupId());
Review Comment:
I have a similar comment for this one. However, in this case, we could just replace the `group` argument by `groupId` and do the lookup here.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add type check to classic group timeout operations [kafka]
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #15587:
URL: https://github.com/apache/kafka/pull/15587#discussion_r1557990553
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2415,6 +2415,22 @@ private CoordinatorResult<Void, Record> classicGroupJoinExistingMember(
return EMPTY_RESULT;
}
+ /**
+ * An overload of {@link GroupMetadataManager#completeClassicGroupJoin(ClassicGroup)} used as
+ * timeout operation. It additionally looks up the group by the id and checks the group type.
+ * completeClassicGroupJoin will only be called if the group is CLASSIC.
+ */
+ private CoordinatorResult<Void, Record> completeClassicGroupJoin(String groupId) {
+ ClassicGroup group;
+ try {
+ group = getOrMaybeCreateClassicGroup(groupId, false);
+ } catch (UnknownMemberIdException | GroupIdNotFoundException exception) {
Review Comment:
The existing coordinator completes responses with UNKNOWN_MEMBER_ID instead of GROUP_ID_NOT_FOUND for join and sync. DeleteGroup and DeleteOffset requests are completed with GROUP_ID_NOT_FOUND if the group does not exist.
So just returning null and handling them separately for each API might be correct.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add type check to classic group timeout operations [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on PR #15587:
URL: https://github.com/apache/kafka/pull/15587#issuecomment-2037838999
Need to add unit tests for the change when the group upgrade/downgrade conversion are finished.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add type check to classic group timeout operations [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15587:
URL: https://github.com/apache/kafka/pull/15587#discussion_r1555019741
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2415,6 +2415,20 @@ private CoordinatorResult<Void, Record> classicGroupJoinExistingMember(
return EMPTY_RESULT;
}
+ /**
+ * An overload of {@link GroupMetadataManager#completeClassicGroupJoin(ClassicGroup)} used as
+ * timeout operation. It additionally looks up the group by the id and checks the group type.
+ * completeClassicGroupJoin will only be called if the group is CLASSIC.
+ */
+ private CoordinatorResult<Void, Record> completeClassicGroupJoin(String groupId) {
+ if (containsClassicGroup(groupId)) {
+ return completeClassicGroupJoin(getOrMaybeCreateClassicGroup(groupId, false));
Review Comment:
Yes I agree. Let me use the try catch
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add type check to classic group timeout operations [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #15587:
URL: https://github.com/apache/kafka/pull/15587
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add type check to classic group timeout operations [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15587:
URL: https://github.com/apache/kafka/pull/15587#discussion_r1558986456
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2415,6 +2415,22 @@ private CoordinatorResult<Void, Record> classicGroupJoinExistingMember(
return EMPTY_RESULT;
}
+ /**
+ * An overload of {@link GroupMetadataManager#completeClassicGroupJoin(ClassicGroup)} used as
+ * timeout operation. It additionally looks up the group by the id and checks the group type.
+ * completeClassicGroupJoin will only be called if the group is CLASSIC.
+ */
+ private CoordinatorResult<Void, Record> completeClassicGroupJoin(String groupId) {
+ ClassicGroup group;
+ try {
+ group = getOrMaybeCreateClassicGroup(groupId, false);
+ } catch (UnknownMemberIdException | GroupIdNotFoundException exception) {
Review Comment:
Filed https://issues.apache.org/jira/browse/KAFKA-16503.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add type check to classic group timeout operations [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15587:
URL: https://github.com/apache/kafka/pull/15587#discussion_r1557570573
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2415,6 +2415,22 @@ private CoordinatorResult<Void, Record> classicGroupJoinExistingMember(
return EMPTY_RESULT;
}
+ /**
+ * An overload of {@link GroupMetadataManager#completeClassicGroupJoin(ClassicGroup)} used as
+ * timeout operation. It additionally looks up the group by the id and checks the group type.
+ * completeClassicGroupJoin will only be called if the group is CLASSIC.
+ */
+ private CoordinatorResult<Void, Record> completeClassicGroupJoin(String groupId) {
+ ClassicGroup group;
+ try {
+ group = getOrMaybeCreateClassicGroup(groupId, false);
+ } catch (UnknownMemberIdException | GroupIdNotFoundException exception) {
Review Comment:
I actually wonder whether raising `GroupIdNotFoundException` in `getOrMaybeCreateClassicGroup` is correct. It is likely not expected on the join-group and the sync-group paths. @jeffkbkim What do you think?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add type check to classic group timeout operations [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15587:
URL: https://github.com/apache/kafka/pull/15587#discussion_r1553499653
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2415,6 +2415,20 @@ private CoordinatorResult<Void, Record> classicGroupJoinExistingMember(
return EMPTY_RESULT;
}
+ /**
+ * An overload of {@link GroupMetadataManager#completeClassicGroupJoin(ClassicGroup)} used as
+ * timeout operation. It additionally looks up the group by the id and checks the group type.
+ * completeClassicGroupJoin will only be called if the group is CLASSIC.
+ */
+ private CoordinatorResult<Void, Record> completeClassicGroupJoin(String groupId) {
+ if (containsClassicGroup(groupId)) {
+ return completeClassicGroupJoin(getOrMaybeCreateClassicGroup(groupId, false));
Review Comment:
I am not a fan of this pattern because you effectively have to look up the group twice. One option would be to use a try..catch to catch the exception thrown by getOrMaybeCreateClassicGroup. Another option would be to 1) do the lookup, 2) verify non-null and group type and return if it fails.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2415,6 +2415,20 @@ private CoordinatorResult<Void, Record> classicGroupJoinExistingMember(
return EMPTY_RESULT;
}
+ /**
+ * An overload of {@link GroupMetadataManager#completeClassicGroupJoin(ClassicGroup)} used as
+ * timeout operation. It additionally looks up the group by the id and checks the group type.
+ * completeClassicGroupJoin will only be called if the group is CLASSIC.
+ */
+ private CoordinatorResult<Void, Record> completeClassicGroupJoin(String groupId) {
+ if (containsClassicGroup(groupId)) {
+ return completeClassicGroupJoin(getOrMaybeCreateClassicGroup(groupId, false));
+ } else {
+ log.info("Group {} is null or not a classic group, skipping rebalance stage.", groupId);
Review Comment:
I wonder if we could use `debug` here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2805,31 +2826,36 @@ private CoordinatorResult<Void, Record> maybeCompleteJoinElseSchedule(
* Try to complete the join phase of the initial rebalance.
* Otherwise, extend the rebalance.
*
- * @param group The group under initial rebalance.
+ * @param groupId The group under initial rebalance.
*
* @return The coordinator result that will be appended to the log.
*/
private CoordinatorResult<Void, Record> tryCompleteInitialRebalanceElseSchedule(
- ClassicGroup group,
+ String groupId,
int delayMs,
int remainingMs
) {
- if (group.newMemberAdded() && remainingMs != 0) {
- // A new member was added. Extend the delay.
- group.setNewMemberAdded(false);
- int newDelayMs = Math.min(classicGroupInitialRebalanceDelayMs, remainingMs);
- int newRemainingMs = Math.max(remainingMs - delayMs, 0);
-
- timer.schedule(
- classicGroupJoinKey(group.groupId()),
- newDelayMs,
- TimeUnit.MILLISECONDS,
- false,
- () -> tryCompleteInitialRebalanceElseSchedule(group, newDelayMs, newRemainingMs)
- );
+ if (containsClassicGroup(groupId)) {
Review Comment:
ditto.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2533,45 +2547,52 @@ private void schedulePendingSync(ClassicGroup group) {
group.rebalanceTimeoutMs(),
TimeUnit.MILLISECONDS,
false,
- () -> expirePendingSync(group, group.generationId()));
+ () -> expirePendingSync(group.groupId(), group.generationId()));
}
/**
* Invoked when the heartbeat operation is expired from the timer. Possibly remove the member and
* try complete the join phase.
*
- * @param group The group.
+ * @param groupId The group id.
* @param memberId The member id.
*
* @return The coordinator result that will be appended to the log.
*/
private CoordinatorResult<Void, Record> expireClassicGroupMemberHeartbeat(
- ClassicGroup group,
+ String groupId,
String memberId
) {
- if (group.isInState(DEAD)) {
- log.info("Received notification of heartbeat expiration for member {} after group {} " +
- "had already been unloaded or deleted.",
- memberId, group.groupId());
- } else if (group.isPendingMember(memberId)) {
- log.info("Pending member {} in group {} has been removed after session timeout expiration.",
- memberId, group.groupId());
-
- return removePendingMemberAndUpdateClassicGroup(group, memberId);
- } else if (!group.hasMemberId(memberId)) {
- log.debug("Member {} has already been removed from the group.", memberId);
- } else {
- ClassicGroupMember member = group.member(memberId);
- if (!member.hasSatisfiedHeartbeat()) {
- log.info("Member {} in group {} has failed, removing it from the group.",
- member.memberId(), group.groupId());
+ if (containsClassicGroup(groupId)) {
+ ClassicGroup group = getOrMaybeCreateClassicGroup(groupId, false);
+ if (group.isInState(DEAD)) {
+ log.info("Received notification of heartbeat expiration for member {} after group {} " +
+ "had already been unloaded or deleted.",
+ memberId, group.groupId());
+ } else if (group.isPendingMember(memberId)) {
+ log.info("Pending member {} in group {} has been removed after session timeout expiration.",
+ memberId, group.groupId());
+
+ return removePendingMemberAndUpdateClassicGroup(group, memberId);
+ } else if (!group.hasMemberId(memberId)) {
+ log.debug("Member {} has already been removed from the group.", memberId);
+ } else {
+ ClassicGroupMember member = group.member(memberId);
+ if (!member.hasSatisfiedHeartbeat()) {
+ log.info("Member {} in group {} has failed, removing it from the group.",
+ member.memberId(), group.groupId());
- return removeMemberAndUpdateClassicGroup(
- group,
- member,
- "removing member " + member.memberId() + " on heartbeat expiration."
- );
+ return removeMemberAndUpdateClassicGroup(
+ group,
+ member,
+ "removing member " + member.memberId() + " on heartbeat expiration."
+ );
+ }
}
+ } else {
+ log.info("Received notification of heartbeat expiration for member {} after group {} " +
Review Comment:
nit: debug?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2533,45 +2547,52 @@ private void schedulePendingSync(ClassicGroup group) {
group.rebalanceTimeoutMs(),
TimeUnit.MILLISECONDS,
false,
- () -> expirePendingSync(group, group.generationId()));
+ () -> expirePendingSync(group.groupId(), group.generationId()));
}
/**
* Invoked when the heartbeat operation is expired from the timer. Possibly remove the member and
* try complete the join phase.
*
- * @param group The group.
+ * @param groupId The group id.
* @param memberId The member id.
*
* @return The coordinator result that will be appended to the log.
*/
private CoordinatorResult<Void, Record> expireClassicGroupMemberHeartbeat(
- ClassicGroup group,
+ String groupId,
String memberId
) {
- if (group.isInState(DEAD)) {
- log.info("Received notification of heartbeat expiration for member {} after group {} " +
- "had already been unloaded or deleted.",
- memberId, group.groupId());
- } else if (group.isPendingMember(memberId)) {
- log.info("Pending member {} in group {} has been removed after session timeout expiration.",
- memberId, group.groupId());
-
- return removePendingMemberAndUpdateClassicGroup(group, memberId);
- } else if (!group.hasMemberId(memberId)) {
- log.debug("Member {} has already been removed from the group.", memberId);
- } else {
- ClassicGroupMember member = group.member(memberId);
- if (!member.hasSatisfiedHeartbeat()) {
- log.info("Member {} in group {} has failed, removing it from the group.",
- member.memberId(), group.groupId());
+ if (containsClassicGroup(groupId)) {
Review Comment:
nit: Same comment here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2954,38 +2980,42 @@ private void removeSyncExpiration(ClassicGroup group) {
/**
* Expire pending sync.
*
- * @param group The group.
+ * @param groupId The group id.
* @param generationId The generation when the pending sync was originally scheduled.
*
* @return The coordinator result that will be appended to the log.
* */
private CoordinatorResult<Void, Record> expirePendingSync(
- ClassicGroup group,
+ String groupId,
int generationId
) {
- if (generationId != group.generationId()) {
- log.error("Received unexpected notification of sync expiration for {} with an old " +
- "generation {} while the group has {}.", group.groupId(), generationId, group.generationId());
- } else {
- if (group.isInState(DEAD) || group.isInState(EMPTY) || group.isInState(PREPARING_REBALANCE)) {
- log.error("Received unexpected notification of sync expiration after group {} already " +
- "transitioned to {} state.", group.groupId(), group.stateAsString());
- } else if (group.isInState(COMPLETING_REBALANCE) || group.isInState(STABLE)) {
- if (!group.hasReceivedSyncFromAllMembers()) {
- Set<String> pendingSyncMembers = new HashSet<>(group.allPendingSyncMembers());
- pendingSyncMembers.forEach(memberId -> {
- group.remove(memberId);
- timer.cancel(classicGroupHeartbeatKey(group.groupId(), memberId));
- });
-
- log.debug("Group {} removed members who haven't sent their sync requests: {}",
- group.groupId(), pendingSyncMembers);
-
- return prepareRebalance(group, "Removing " + pendingSyncMembers + " on pending sync request expiration");
+ if (containsClassicGroup(groupId)) {
Review Comment:
ditto.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2805,31 +2826,36 @@ private CoordinatorResult<Void, Record> maybeCompleteJoinElseSchedule(
* Try to complete the join phase of the initial rebalance.
* Otherwise, extend the rebalance.
*
- * @param group The group under initial rebalance.
+ * @param groupId The group under initial rebalance.
*
* @return The coordinator result that will be appended to the log.
*/
private CoordinatorResult<Void, Record> tryCompleteInitialRebalanceElseSchedule(
- ClassicGroup group,
+ String groupId,
int delayMs,
int remainingMs
) {
- if (group.newMemberAdded() && remainingMs != 0) {
- // A new member was added. Extend the delay.
- group.setNewMemberAdded(false);
- int newDelayMs = Math.min(classicGroupInitialRebalanceDelayMs, remainingMs);
- int newRemainingMs = Math.max(remainingMs - delayMs, 0);
-
- timer.schedule(
- classicGroupJoinKey(group.groupId()),
- newDelayMs,
- TimeUnit.MILLISECONDS,
- false,
- () -> tryCompleteInitialRebalanceElseSchedule(group, newDelayMs, newRemainingMs)
- );
+ if (containsClassicGroup(groupId)) {
+ ClassicGroup group = getOrMaybeCreateClassicGroup(groupId, false);
+ if (group.newMemberAdded() && remainingMs != 0) {
+ // A new member was added. Extend the delay.
+ group.setNewMemberAdded(false);
+ int newDelayMs = Math.min(classicGroupInitialRebalanceDelayMs, remainingMs);
+ int newRemainingMs = Math.max(remainingMs - delayMs, 0);
+
+ timer.schedule(
+ classicGroupJoinKey(group.groupId()),
+ newDelayMs,
+ TimeUnit.MILLISECONDS,
+ false,
+ () -> tryCompleteInitialRebalanceElseSchedule(group.groupId(), newDelayMs, newRemainingMs)
+ );
+ } else {
+ // No more time remaining. Complete the join phase.
+ return completeClassicGroupJoin(group);
+ }
} else {
- // No more time remaining. Complete the join phase.
- return completeClassicGroupJoin(group);
+ log.info("Group {} is null or not a classic group, skipping the initial rebalance stage.", groupId);
Review Comment:
ditto.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Draft] Add type check to classic group timeout operations [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15587:
URL: https://github.com/apache/kafka/pull/15587#discussion_r1552050811
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2451,6 +2451,8 @@ private CoordinatorResult<Void, Record> completeClassicGroupJoin(
if (group.isInState(DEAD)) {
log.info("Group {} is dead, skipping rebalance stage.", groupId);
+ } else if (!containsClassicGroup(group.groupId())) {
+ log.info("Group {} is null or not a classic group, skipping rebalance stage.", groupId);
Review Comment:
Yeah I agree with it. I'll try to change the params and overload them if they are also called in the regular context
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add type check to classic group timeout operations [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15587:
URL: https://github.com/apache/kafka/pull/15587#discussion_r1557570573
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2415,6 +2415,22 @@ private CoordinatorResult<Void, Record> classicGroupJoinExistingMember(
return EMPTY_RESULT;
}
+ /**
+ * An overload of {@link GroupMetadataManager#completeClassicGroupJoin(ClassicGroup)} used as
+ * timeout operation. It additionally looks up the group by the id and checks the group type.
+ * completeClassicGroupJoin will only be called if the group is CLASSIC.
+ */
+ private CoordinatorResult<Void, Record> completeClassicGroupJoin(String groupId) {
+ ClassicGroup group;
+ try {
+ group = getOrMaybeCreateClassicGroup(groupId, false);
+ } catch (UnknownMemberIdException | GroupIdNotFoundException exception) {
Review Comment:
I actually wonder whether raising `GroupIdNotFoundException` in `getOrMaybeCreateClassicGroup` is correct. It is likely not expected on the join-group and the sync-group paths. @jeffkbkim What do you think? We don't have to address this in 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Add type check to classic group timeout operations [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15587:
URL: https://github.com/apache/kafka/pull/15587#discussion_r1558005191
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2415,6 +2415,22 @@ private CoordinatorResult<Void, Record> classicGroupJoinExistingMember(
return EMPTY_RESULT;
}
+ /**
+ * An overload of {@link GroupMetadataManager#completeClassicGroupJoin(ClassicGroup)} used as
+ * timeout operation. It additionally looks up the group by the id and checks the group type.
+ * completeClassicGroupJoin will only be called if the group is CLASSIC.
+ */
+ private CoordinatorResult<Void, Record> completeClassicGroupJoin(String groupId) {
+ ClassicGroup group;
+ try {
+ group = getOrMaybeCreateClassicGroup(groupId, false);
+ } catch (UnknownMemberIdException | GroupIdNotFoundException exception) {
Review Comment:
This makes sense. My concern is more on the join-group/sync-group paths. We use `getOrMaybeCreateClassicGroup` there and don't seem to convert `GroupIdNotFoundException`. Do we need to?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org