jolshan commented on code in PR #15212:
URL: https://github.com/apache/kafka/pull/15212#discussion_r1459490703
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -875,30 +875,43 @@ private void maybeRemovePartitionEpoch(
ConsumerGroupMember oldMember
) {
if (oldMember != null) {
- removePartitionEpochs(oldMember.assignedPartitions());
- removePartitionEpochs(oldMember.partitionsPendingRevocation());
+ removePartitionEpochs(oldMember.assignedPartitions(),
oldMember.memberEpoch());
+ removePartitionEpochs(oldMember.partitionsPendingRevocation(),
oldMember.memberEpoch());
}
}
/**
* Removes the partition epochs based on the provided assignment.
*
* @param assignment The assignment.
+ * @param expectedEpoch The expected epoch.
+ * @throws IllegalStateException if the epoch does not match the expected
one.
+ * package-private for testing.
*/
- private void removePartitionEpochs(
- Map<Uuid, Set<Integer>> assignment
+ void removePartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int expectedEpoch
) {
assignment.forEach((topicId, assignedPartitions) -> {
currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
if (partitionsOrNull != null) {
- assignedPartitions.forEach(partitionsOrNull::remove);
+ assignedPartitions.forEach(partitionId -> {
+ Integer prevValue =
partitionsOrNull.remove(partitionId);
+ if (prevValue != expectedEpoch) {
+ throw new IllegalStateException(
+ String.format("Cannot remove the epoch %d from
%s-%s because the partition is " +
+ "still owned at a different epoch %d",
expectedEpoch, topicId, partitionId, prevValue));
+ }
+ });
if (partitionsOrNull.isEmpty()) {
return null;
} else {
return partitionsOrNull;
}
} else {
- return null;
+ throw new IllegalStateException(
Review Comment:
Thanks for stepping through this. So the operation/request would need to be
retried? Do we expect it to recover on its own when metadata updates for
example?
--
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
