dajac commented on code in PR #15212: URL: https://github.com/apache/kafka/pull/15212#discussion_r1460376759
########## 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: The operation would fail and the client would decide what to do. In this case, we would not recover because we would likely hit the same bug again. I suppose that we could recover in different situations. Overall, the point of the change is to fail when the state is incorrect instead of silently accepting it. This will help us to detect those issues. -- 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