dajac commented on code in PR #15212:
URL: https://github.com/apache/kafka/pull/15212#discussion_r1458570521
##########
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:
It will fail the entire write operation so we would rollback the entire
state to the state of the last written offset. Let's take the
ConsumerGroupHeartbeat as an example.
Here is the common flow:
1) A write operation is create to handle the request.
2) The operation handles the request, this generate the response and the
records to apply to the state machine.
3) The records are replayed to the state machine.
4) The records are written to the log.
5) The response is returned when the records are replicated.
The exception here would basically break step 3) so the entire operation
would be failed and the HB would return an unknown server error
(IllegalStateException). When the operation fails, the state machine is rolled
back to the state before the operation.
--
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
