jolshan commented on code in PR #15212:
URL: https://github.com/apache/kafka/pull/15212#discussion_r1457917435
##########
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:
What is the affect of throwing this error? Do we also block removing the
rest of the partitions?
--
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
