jolshan commented on code in PR #15212:
URL: https://github.com/apache/kafka/pull/15212#discussion_r1461110485
##########
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:
Ack -- so we would essentially get stuck and this would
1) prevent further issues
2) probably alert the user as the consumer could not continue
--
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
