dajac commented on code in PR #20417:
URL: https://github.com/apache/kafka/pull/20417#discussion_r2359111312
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2466,31 +2476,30 @@ private CoordinatorResult<Void, CoordinatorRecord>
classicGroupJoinToConsumerGro
} else {
targetAssignmentEpoch = group.assignmentEpoch();
targetAssignment =
group.targetAssignment(updatedMember.memberId(), updatedMember.instanceId());
-
}
- // 3. Reconcile the member's assignment with the target assignment if
the member is not
- // fully reconciled yet.
- updatedMember = maybeReconcile(
- groupId,
- updatedMember,
- group::currentPartitionEpoch,
- targetAssignmentEpoch,
- targetAssignment,
- toTopicPartitions(subscription.ownedPartitions(), metadataImage),
- records
- );
+ // 3. If it's not a downgrade triggered by static member replacement
with a different subscription,
+ // reconcile the member's assignment with the target assignment if the
member is not fully reconciled yet.
+ if (!bumpGroupEpoch || !downgrade) {
+ updatedMember = maybeReconcile(
+ groupId,
+ updatedMember,
+ group::currentPartitionEpoch,
+ targetAssignmentEpoch,
+ targetAssignment,
+ toTopicPartitions(subscription.ownedPartitions(),
metadataImage),
+ records
+ );
+ }
// 4. Maybe downgrade the consumer group if the last static member
using the
// consumer protocol is replaced by the joining static member.
- ConsumerGroupMember existingStaticMemberOrNull =
group.staticMember(request.groupInstanceId());
- boolean downgrade = existingStaticMemberOrNull != null &&
- validateOnlineDowngradeWithReplacedMember(group,
existingStaticMemberOrNull);
if (downgrade) {
Review Comment:
now that we use `downgrade` in a few places, i wonder if we could use an if
(downgrade) .. else .. instead of gating all the ifs. have you considered it?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2466,31 +2476,30 @@ private CoordinatorResult<Void, CoordinatorRecord>
classicGroupJoinToConsumerGro
} else {
targetAssignmentEpoch = group.assignmentEpoch();
targetAssignment =
group.targetAssignment(updatedMember.memberId(), updatedMember.instanceId());
-
}
- // 3. Reconcile the member's assignment with the target assignment if
the member is not
- // fully reconciled yet.
- updatedMember = maybeReconcile(
- groupId,
- updatedMember,
- group::currentPartitionEpoch,
- targetAssignmentEpoch,
- targetAssignment,
- toTopicPartitions(subscription.ownedPartitions(), metadataImage),
- records
- );
+ // 3. If it's not a downgrade triggered by static member replacement
with a different subscription,
+ // reconcile the member's assignment with the target assignment if the
member is not fully reconciled yet.
+ if (!bumpGroupEpoch || !downgrade) {
Review Comment:
I am not sure to understand why we need `!bumpGroupEpoch` here. Couldn't we
need to reconcile event if the group epoch has not changed?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
