You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucliu1108 (via GitHub)" <gi...@apache.org> on 2026/02/18 16:04:43 UTC
[PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
lucliu1108 opened a new pull request, #21508:
URL: https://github.com/apache/kafka/pull/21508
## Summary
This PR is an implementation of KIP-1251: https://cwiki.apache.org/confluence/display/KAFKA/KIP-1251%3A+Assignment+epochs+for+consumer+groups
This change introduces per-partition assignment epochs to relax the strict member epoch validation for consumer group offset commits. When receiving an offset commit request that includes the client-side member epoch and a member ID, we previously require checking
```
clientEpoch == brokerEpoch
```
for a valid offset commit, which could lead to false fencing.
We now allow a relaxed offset commit check using an assignment epoch for each assigned partition and each member,
```
assignmentEpoch <= clientEpoch <= brokerEpoch
```
This prevents false rejections of legitimate offset commits when a member's epoch is bumped but the client hasn't received the update yet.
## Changes
- Add `AssignmentEpochs` field to `TopicPartitions` in `ConsumerGroupCurrentMemberAssignmentValue` schema
- Add `assignedPartitionsWithEpochs` and `partitionsPendingRevocationWithEpochs` to `ConsumerGroupMember`
- Update `ConsumerGroup.validateOffsetCommit()` with relaxed validation logic
- Add `createAssignmentEpochValidator()` for per-partition epoch checks
- Update `GroupCoordinatorRecordHelpers` to serialize assignment epochs
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2835160412
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
For the `assignedPartitions`, it is part of the fields inherited from `ModernGroupMember` that `ShareGroupMember` also uses, so i left it there for the getter method.
Or another way is not to initialize it in the constructor (keep it null?) but maybe this is not good practice.
For the. `partitionsPendingRevocation`, there are 6 or 7 places it is directly called. Guess I could remove this method and do extra conversions for where its' called instead, or use a view.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #21508:
URL: https://github.com/apache/kafka/pull/21508#issuecomment-3924020702
cc @squah-confluent
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2827514018
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java:
##########
@@ -289,8 +289,8 @@ public static CoordinatorRecord newConsumerGroupCurrentAssignmentRecord(
.setMemberEpoch(member.memberEpoch())
.setPreviousMemberEpoch(member.previousMemberEpoch())
.setState(member.state().value())
- .setAssignedPartitions(toTopicPartitions(member.assignedPartitions()))
- .setPartitionsPendingRevocation(toTopicPartitions(member.partitionsPendingRevocation())),
+ .setAssignedPartitions(toTopicPartitions(member.assignedPartitionsWithEpochs()))
+ .setPartitionsPendingRevocation(toTopicPartitions(member.partitionsPendingRevocationWithEpochs())),
Review Comment:
newConsumerGroupCurrentAssignmentRecord now exclusively serializes member.assignedPartitionsWithEpochs()/partitionsPendingRevocationWithEpochs(). If those maps are empty while the Set-based assignment fields are populated (possible via existing Builder setters), this will write records with empty assignments and effectively lose the current assignment on disk. Consider adding a defensive fallback (e.g., derive epoch maps from assignedPartitions()/partitionsPendingRevocation() with a sensible default epoch) or validate/throw if the member is internally inconsistent before serializing.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -672,9 +672,23 @@ public CommitPartitionValidator validateOffsetCommit(
throw new UnsupportedVersionException("OffsetCommit version 9 or above must be used " +
"by members using the modern group protocol");
}
+ // For member using the classic protocol, use strict epoch validation.
+ if (member.useClassicProtocol()) {
+ validateMemberEpoch(memberEpoch, member.memberEpoch(), true);
+ return CommitPartitionValidator.NO_OP;
+ }
- validateMemberEpoch(memberEpoch, member.memberEpoch(), member.useClassicProtocol());
- return CommitPartitionValidator.NO_OP;
+ // For member using the consumer protocol
+ // Case 1: Strict epoch match
+ if (memberEpoch == member.memberEpoch()) {
+ return CommitPartitionValidator.NO_OP;
+ }
+ // Case 2:Client epoch > broker epoch, which is an invalid request
Review Comment:
Minor comment formatting: add a space after the colon in "Case 2:Client".
```suggestion
// Case 2: Client epoch > broker epoch, which is an invalid request
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -28,6 +28,9 @@
import org.apache.kafka.coordinator.group.modern.ModernGroupMember;
import java.util.ArrayList;
+import java.util.stream.Collectors;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
Review Comment:
Import order looks inconsistent with the rest of the module: java.util.stream.Collectors is currently placed between java.util.ArrayList and java.util.Collections. Other files keep java.util.* imports together and place java.util.stream.* afterwards (e.g., Utils.java:36-49). Please reorder these imports to match the established style (this may also be enforced by Checkstyle).
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -672,9 +672,23 @@ public CommitPartitionValidator validateOffsetCommit(
throw new UnsupportedVersionException("OffsetCommit version 9 or above must be used " +
"by members using the modern group protocol");
}
+ // For member using the classic protocol, use strict epoch validation.
+ if (member.useClassicProtocol()) {
+ validateMemberEpoch(memberEpoch, member.memberEpoch(), true);
+ return CommitPartitionValidator.NO_OP;
+ }
- validateMemberEpoch(memberEpoch, member.memberEpoch(), member.useClassicProtocol());
- return CommitPartitionValidator.NO_OP;
+ // For member using the consumer protocol
+ // Case 1: Strict epoch match
+ if (memberEpoch == member.memberEpoch()) {
+ return CommitPartitionValidator.NO_OP;
+ }
+ // Case 2:Client epoch > broker epoch, which is an invalid request
+ if (memberEpoch > member.memberEpoch()) {
+ throw new StaleMemberEpochException(String.format("The received member epoch %d is larger than "
+ + "the expected member epoch %d.", memberEpoch, member.memberEpoch()));
+ }
+ return createAssignmentEpochValidator(member, memberEpoch);
Review Comment:
The new relaxed offset-commit validation path (returning a per-partition validator when receivedMemberEpoch < broker member epoch) doesn’t appear to be covered by unit tests. It would be useful to add ConsumerGroupTest cases that (1) accept commits when assignmentEpoch <= receivedEpoch < brokerEpoch for assigned/pending-revocation partitions, and (2) reject commits when the partition isn’t assigned or when receivedEpoch < assignmentEpoch, to prevent regressions in the new behavior.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -205,6 +212,66 @@ public Builder setClassicMemberMetadata(ConsumerGroupMemberMetadataValue.Classic
return this;
}
+ public Builder setAssignedPartitionsWithEpochs(Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs) {
+ this.assignedPartitionsWithEpochs = assignedPartitionsWithEpochs;
+ this.assignedPartitions = assignedPartitionsWithEpochs.entrySet().stream()
+ .collect(Collectors.toMap(
+ Map.Entry::getKey,
+ e -> new HashSet<>(e.getValue().keySet())
+ ));
+ return this;
+ }
+
+ public Builder setPartitionsPendingRevocationWithEpochs(Map<Uuid, Map<Integer, Integer>> partitionsPendingRevocationWithEpochs) {
+ this.partitionsPendingRevocationWithEpochs = partitionsPendingRevocationWithEpochs;
+ this.partitionsPendingRevocation = partitionsPendingRevocationWithEpochs.entrySet().stream()
+ .collect(Collectors.toMap(
+ Map.Entry::getKey,
+ e -> new HashSet<>(e.getValue().keySet())
+ ));
+ return this;
+ }
Review Comment:
setAssignedPartitionsWithEpochs / setPartitionsPendingRevocationWithEpochs keep the Set-based fields in sync, but the Builder still also has the older setAssignedPartitions(...) and setPartitionsPendingRevocation(...) setters. Given GroupCoordinatorRecordHelpers now serializes assignedPartitionsWithEpochs()/partitionsPendingRevocationWithEpochs(), any code path that uses the older setters (without also setting the epoch maps) can drop epoch information and even persist empty assignments. Consider enforcing invariants in build() (e.g., derive epoch maps when missing), or deprecating/removing the Set-based setters to prevent constructing an internally inconsistent ConsumerGroupMember.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -672,9 +672,23 @@ public CommitPartitionValidator validateOffsetCommit(
throw new UnsupportedVersionException("OffsetCommit version 9 or above must be used " +
"by members using the modern group protocol");
}
+ // For member using the classic protocol, use strict epoch validation.
+ if (member.useClassicProtocol()) {
Review Comment:
Grammar in this comment is off: consider changing "For member using the classic protocol" to "For members using the classic protocol" (or "For a member using...") for clarity.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2842635990
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
So we could remove `assignedPartitions` from `ModernGroupMember` interface, and instead only add it as a field for `ShareGroupMember`? (Maybe in a future ticket)
Right now i will substitute all blocks where `ConsumerGroupMember#assignedPartitions` are called with the new `ConsumerGroupMember#assignedPartitionsWithEpochs` instead.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2838826022
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -672,9 +673,23 @@ public CommitPartitionValidator validateOffsetCommit(
throw new UnsupportedVersionException("OffsetCommit version 9 or above must be used " +
"by members using the modern group protocol");
}
+ // For members using the classic protocol, use strict epoch validation.
+ if (member.useClassicProtocol()) {
+ validateMemberEpoch(memberEpoch, member.memberEpoch(), true);
+ return CommitPartitionValidator.NO_OP;
+ }
- validateMemberEpoch(memberEpoch, member.memberEpoch(), member.useClassicProtocol());
- return CommitPartitionValidator.NO_OP;
+ // For member using the consumer protocol
Review Comment:
nit:
```suggestion
// For members using the consumer protocol, the epoch must either match the last epoch sent
// in a heartbeat or be greater than or equal to the partition's assignment epoch.
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -672,9 +673,23 @@ public CommitPartitionValidator validateOffsetCommit(
throw new UnsupportedVersionException("OffsetCommit version 9 or above must be used " +
"by members using the modern group protocol");
}
+ // For members using the classic protocol, use strict epoch validation.
Review Comment:
nit:
```suggestion
// For members using the classic protocol, use strict epoch validation.
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -63,9 +65,9 @@ public static class Builder {
private Set<String> subscribedTopicNames = Set.of();
private String subscribedTopicRegex = "";
private String serverAssignorName = null;
- private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
- private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
nit: Could you minimize the diff churn in this file by making the replacement methods/fields occupy the same position as the previous ones?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/CurrentAssignmentBuilder.java:
##########
@@ -299,47 +300,56 @@ private ConsumerGroupMember updateCurrentAssignment(
Map<Uuid, Set<Integer>> memberAssignedPartitions
) {
Set<Uuid> subscribedTopicIds = subscribedTopicIds();
-
// Reuse the original map if no topics need to be removed.
- Map<Uuid, Set<Integer>> newAssignedPartitions;
- Map<Uuid, Set<Integer>> newPartitionsPendingRevocation;
+ Map<Uuid, Map<Integer, Integer>> newAssignedPartitionsWithEpochs;
+ Map<Uuid, Map<Integer, Integer>> newPartitionsPendingRevocationWithEpochs;
+ boolean changed = false;
Review Comment:
Is this refactor necessary?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/CurrentAssignmentBuilder.java:
##########
@@ -370,10 +380,13 @@ private ConsumerGroupMember computeNextAssignment(
Set<Uuid> subscribedTopicIds = subscribedTopicIds();
boolean hasUnreleasedPartitions = false;
- Map<Uuid, Set<Integer>> newAssignedPartitions = new HashMap<>();
- Map<Uuid, Set<Integer>> newPartitionsPendingRevocation = new HashMap<>();
+ Map<Uuid, Map<Integer, Integer>> newAssignedPartitionsWithEpochs = new HashMap<>();
+ Map<Uuid, Map<Integer, Integer>> newPartitionsPendingRevocationWithEpochs = new HashMap<>();
Map<Uuid, Set<Integer>> newPartitionsPendingAssignment = new HashMap<>();
+ // Get existing epochs from member
+ Map<Uuid, Map<Integer, Integer>> existingAssignedEpochs = member.assignedPartitionsWithEpochs();
Review Comment:
Can we change the type of `memberAssignedPartitions` to `Map<topic id, Map<partition, assignment epoch>>` and work with epoch maps instead?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -672,9 +673,23 @@ public CommitPartitionValidator validateOffsetCommit(
throw new UnsupportedVersionException("OffsetCommit version 9 or above must be used " +
"by members using the modern group protocol");
}
+ // For members using the classic protocol, use strict epoch validation.
+ if (member.useClassicProtocol()) {
+ validateMemberEpoch(memberEpoch, member.memberEpoch(), true);
+ return CommitPartitionValidator.NO_OP;
+ }
- validateMemberEpoch(memberEpoch, member.memberEpoch(), member.useClassicProtocol());
- return CommitPartitionValidator.NO_OP;
+ // For member using the consumer protocol
+ // Case 1: Strict epoch match
Review Comment:
nit: Do these cases correspond to the KIP? I would drop the comments otherwise.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -190,21 +192,61 @@ public Builder setState(MemberState state) {
return this;
}
- public Builder setAssignedPartitions(Map<Uuid, Set<Integer>> assignedPartitions) {
- this.assignedPartitions = assignedPartitions;
+ public Builder setClassicMemberMetadata(ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata) {
+ this.classicMemberMetadata = classicMemberMetadata;
return this;
}
- public Builder setPartitionsPendingRevocation(Map<Uuid, Set<Integer>> partitionsPendingRevocation) {
- this.partitionsPendingRevocation = partitionsPendingRevocation;
+ public Builder setAssignedPartitionsWithEpochs(Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs) {
+ this.assignedPartitionsWithEpochs = assignedPartitionsWithEpochs;
return this;
}
- public Builder setClassicMemberMetadata(ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata) {
- this.classicMemberMetadata = classicMemberMetadata;
+ public Builder setPartitionsPendingRevocationWithEpochs(Map<Uuid, Map<Integer, Integer>> partitionsPendingRevocationWithEpochs) {
+ this.partitionsPendingRevocationWithEpochs = partitionsPendingRevocationWithEpochs;
+ return this;
+ }
+
+ /**
+ * Resets the assignment epochs to 0 for all assigned partitions.
+ * Used when a static member leaves, so that the rejoining member
Review Comment:
```suggestion
* Used when a static member leaves, so that the rejoining member's
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -672,9 +673,23 @@ public CommitPartitionValidator validateOffsetCommit(
throw new UnsupportedVersionException("OffsetCommit version 9 or above must be used " +
"by members using the modern group protocol");
}
+ // For members using the classic protocol, use strict epoch validation.
+ if (member.useClassicProtocol()) {
+ validateMemberEpoch(memberEpoch, member.memberEpoch(), true);
+ return CommitPartitionValidator.NO_OP;
+ }
- validateMemberEpoch(memberEpoch, member.memberEpoch(), member.useClassicProtocol());
- return CommitPartitionValidator.NO_OP;
+ // For member using the consumer protocol
+ // Case 1: Strict epoch match
+ if (memberEpoch == member.memberEpoch()) {
+ return CommitPartitionValidator.NO_OP;
+ }
+ // Case 2: Client epoch > broker epoch, which is an invalid request
+ if (memberEpoch > member.memberEpoch()) {
+ throw new StaleMemberEpochException(String.format("The received member epoch %d is larger than "
+ + "the expected member epoch %d.", memberEpoch, member.memberEpoch()));
+ }
+ return createAssignmentEpochValidator(member, memberEpoch);
Review Comment:
nit: Could we follow the streams implementation?
```suggestion
// Member epoch is older; validate against per-partition assignment epochs.
return createAssignmentEpochValidator(member, memberEpoch);
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -672,9 +673,23 @@ public CommitPartitionValidator validateOffsetCommit(
throw new UnsupportedVersionException("OffsetCommit version 9 or above must be used " +
"by members using the modern group protocol");
}
+ // For members using the classic protocol, use strict epoch validation.
+ if (member.useClassicProtocol()) {
+ validateMemberEpoch(memberEpoch, member.memberEpoch(), true);
+ return CommitPartitionValidator.NO_OP;
+ }
- validateMemberEpoch(memberEpoch, member.memberEpoch(), member.useClassicProtocol());
- return CommitPartitionValidator.NO_OP;
+ // For member using the consumer protocol
+ // Case 1: Strict epoch match
+ if (memberEpoch == member.memberEpoch()) {
+ return CommitPartitionValidator.NO_OP;
+ }
+ // Case 2: Client epoch > broker epoch, which is an invalid request
+ if (memberEpoch > member.memberEpoch()) {
+ throw new StaleMemberEpochException(String.format("The received member epoch %d is larger than "
+ + "the expected member epoch %d.", memberEpoch, member.memberEpoch()));
Review Comment:
nit: Could we follow the streams implementation?
```suggestion
throw new StaleMemberEpochException(String.format("Received member epoch %d is newer than "
+ "current member epoch %d.", memberEpoch, member.memberEpoch()));
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -1184,17 +1240,24 @@ public static ConsumerGroup fromClassicGroup(
// assignment of the classic group. All the members are put in the Stable state. If the classic
// group was in Preparing Rebalance or Completing Rebalance states, the classic members are
// asked to rejoin the group to re-trigger a rebalance or collect their assignments.
+ int memberEpoch = classicGroup.generationId();
+ // Convert assigned partitions to epochs map
+ Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = assignedPartitions.entrySet().stream()
+ .collect(Collectors.toMap(
+ Map.Entry::getKey,
+ e -> e.getValue().stream().collect(Collectors.toMap(p -> p, p -> memberEpoch))
+ ));
Review Comment:
nit: We could move this into a `Builder.setAssignedPartitions(assignedPartitions, assignmentEpoch)` convenience method.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -337,6 +406,50 @@ public Map<Uuid, Set<Integer>> partitionsPendingRevocation() {
return partitionsPendingRevocation;
}
+ /**
+ * @return The epoch-annotated assigned partitions map.
+ */
+ public Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs() {
+ return assignedPartitionsWithEpochs;
+ }
+
+ /**
+ * @return The epoch-annotated pending revocation partitions map.
+ */
+ public Map<Uuid, Map<Integer, Integer>> partitionsPendingRevocationWithEpochs() {
+ return partitionsPendingRevocationWithEpochs;
+ }
+
+ /**
+ * Gets the assignment epoch for a specific partition.
+ *
+ * @param topicId The topic UUID.
+ * @param partitionId The partition index.
+ * @return The epoch at which the partition was assigned, or null if not assigned.
+ */
+ public Integer getAssignmentEpoch(Uuid topicId, int partitionId) {
+ Map<Integer, Integer> partitionEpochs = assignedPartitionsWithEpochs.get(topicId);
+ if (partitionEpochs != null) {
+ return partitionEpochs.get(partitionId);
+ }
+ return null;
+ }
+
+ /**
+ * Gets the assignment epoch for a partition pending revocation.
+ *
+ * @param topicId The topic UUID.
+ * @param partitionId The partition index.
+ * @return The epoch at which the partition was assigned, or null if not pending revocation.
+ */
+ public Integer getPendingRevocationEpoch(Uuid topicId, int partitionId) {
Review Comment:
nit: The convention in this class is to omit `get` from method names.
```suggestion
public Integer pendingRevocationEpoch(Uuid topicId, int partitionId) {
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -190,21 +192,61 @@ public Builder setState(MemberState state) {
return this;
}
- public Builder setAssignedPartitions(Map<Uuid, Set<Integer>> assignedPartitions) {
- this.assignedPartitions = assignedPartitions;
+ public Builder setClassicMemberMetadata(ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata) {
+ this.classicMemberMetadata = classicMemberMetadata;
return this;
}
- public Builder setPartitionsPendingRevocation(Map<Uuid, Set<Integer>> partitionsPendingRevocation) {
- this.partitionsPendingRevocation = partitionsPendingRevocation;
+ public Builder setAssignedPartitionsWithEpochs(Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs) {
+ this.assignedPartitionsWithEpochs = assignedPartitionsWithEpochs;
return this;
}
- public Builder setClassicMemberMetadata(ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata) {
- this.classicMemberMetadata = classicMemberMetadata;
+ public Builder setPartitionsPendingRevocationWithEpochs(Map<Uuid, Map<Integer, Integer>> partitionsPendingRevocationWithEpochs) {
+ this.partitionsPendingRevocationWithEpochs = partitionsPendingRevocationWithEpochs;
+ return this;
+ }
+
+ /**
+ * Resets the assignment epochs to 0 for all assigned partitions.
+ * Used when a static member leaves, so that the rejoining member
+ * partitions will be assigned from epoch 0 to the new member ID.
+ * All commits using the old member ID will be fenced.
+ */
+ public Builder resetAssignedPartitionsEpochsToZero() {
+ if (this.assignedPartitionsWithEpochs.isEmpty()) {
+ return this;
+ }
+ Map<Uuid, Map<Integer, Integer>> resetEpochs = new HashMap<>();
+ for (Map.Entry<Uuid, Map<Integer, Integer>> entry : this.assignedPartitionsWithEpochs.entrySet()) {
+ Map<Integer, Integer> partitionEpochs = new HashMap<>();
+ for (Integer partitionId : entry.getValue().keySet()) {
+ partitionEpochs.put(partitionId, 0);
+ }
+ resetEpochs.put(entry.getKey(), Collections.unmodifiableMap(partitionEpochs));
+ }
+ this.assignedPartitionsWithEpochs = Collections.unmodifiableMap(resetEpochs);
return this;
}
+ private static Map<Uuid, Map<Integer, Integer>> assignmentWithEpochsFromTopicPartitions(
+ List<ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions> topicPartitions,
+ int defaultEpoch
+ ) {
Review Comment:
nit: This method should replace `Utils.assignmentFromTopicPartitions` and live in `Utils`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/CurrentAssignmentBuilder.java:
##########
@@ -299,47 +300,56 @@ private ConsumerGroupMember updateCurrentAssignment(
Map<Uuid, Set<Integer>> memberAssignedPartitions
) {
Set<Uuid> subscribedTopicIds = subscribedTopicIds();
-
Review Comment:
nit: stray newline change
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -337,6 +406,50 @@ public Map<Uuid, Set<Integer>> partitionsPendingRevocation() {
return partitionsPendingRevocation;
}
+ /**
+ * @return The epoch-annotated assigned partitions map.
+ */
+ public Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs() {
+ return assignedPartitionsWithEpochs;
+ }
+
+ /**
+ * @return The epoch-annotated pending revocation partitions map.
+ */
+ public Map<Uuid, Map<Integer, Integer>> partitionsPendingRevocationWithEpochs() {
+ return partitionsPendingRevocationWithEpochs;
+ }
+
+ /**
+ * Gets the assignment epoch for a specific partition.
+ *
+ * @param topicId The topic UUID.
+ * @param partitionId The partition index.
+ * @return The epoch at which the partition was assigned, or null if not assigned.
+ */
+ public Integer getAssignmentEpoch(Uuid topicId, int partitionId) {
Review Comment:
nit: The convention in this class is to omit `get` from method names.
```suggestion
public Integer assignmentEpoch(Uuid topicId, int partitionId) {
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -837,6 +852,47 @@ private void validateMemberEpoch(
}
}
+ /**
+ * Creates a validator that checks per-partition assignment epochs.
+ * A commit is rejected if the partition is not assigned to the member
+ * or if the received client-side epoch is older than the partition's assignment epoch(KIP-1251).
+ *
+ * @param member The consumer group member.
+ * @param receivedMemberEpoch The member epoch from the offset commit request.
+ * @return A validator that checks each partition's assignment epoch.
+ */
+ private CommitPartitionValidator createAssignmentEpochValidator(
+ ConsumerGroupMember member,
+ int receivedMemberEpoch
+ ) {
+ return (topicName, topicId, partitionId) -> {
+ // Check if the partition is in the assigned partitions.
+ // If not found in assigned, check partitions pending revocation.
+ Integer assignmentEpoch = member.getAssignmentEpoch(topicId, partitionId);
+ if (assignmentEpoch == null) {
+ assignmentEpoch = member.getPendingRevocationEpoch(topicId, partitionId);
+ }
+
+ // If client-side epoch != broker-side epoch, and the partition is not assigned to this member, reject.
+ if (assignmentEpoch == null) {
+ throw new StaleMemberEpochException(String.format(
+ "Partition %s-%d is not assigned or pending revocation for member %s. " +
+ "Committing unassigned partitions is only allowed when member epoch matches exactly " +
+ "(received: %d, current: %d).",
+ topicName, partitionId, member.memberId(), receivedMemberEpoch, member.memberEpoch()));
+ }
+
+ // If the received epoch is older than when this partition was assigned,
+ // It is a zombie commit and should be rejected.
+ if (receivedMemberEpoch < assignmentEpoch) {
+ throw new StaleMemberEpochException(
+ String.format("The received member epoch %d is older than the assignment epoch %d for partition %s-%d.",
+ receivedMemberEpoch, assignmentEpoch, topicName, partitionId)
+ );
+ }
+ };
+ }
Review Comment:
nit: Could we follow the streams implementation?
```suggestion
/**
* Creates a validator that checks if the received member epoch is valid for each partition's assignment epoch.
* A commit is rejected if the partition is not assigned to the member
* or if the received client-side epoch is older than the partition's assignment epoch (KIP-1251).
*
* @param member The member whose assignments are being validated.
* @param receivedMemberEpoch The received member epoch.
* @return A validator for per-partition validation.
*/
private CommitPartitionValidator createAssignmentEpochValidator(
ConsumerGroupMember member,
int receivedMemberEpoch
) {
return (topicName, topicId, partitionId) -> {
// Search for the partition in assigned partitions, then in partitions pending revocation
Integer assignmentEpoch = member.getAssignmentEpoch(topicId, partitionId);
if (assignmentEpoch == null) {
assignmentEpoch = member.getPendingRevocationEpoch(topicId, partitionId);
}
if (assignmentEpoch == null) {
throw new StaleMemberEpochException(String.format(
"Partition %s-%d is not assigned or pending revocation for member.",
topicName, partitionId));
}
if (receivedMemberEpoch < assignmentEpoch) {
throw new StaleMemberEpochException(
String.format("Received member epoch %d is older than assignment epoch %d for partition %s-%d.",
receivedMemberEpoch, assignmentEpoch, topicName, partitionId)
);
}
};
}
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/CurrentAssignmentBuilder.java:
##########
@@ -407,19 +422,35 @@ private ConsumerGroupMember computeNextAssignment(
!member.partitionsPendingRevocation().getOrDefault(topicId, Set.of()).contains(partitionId)
) || hasUnreleasedPartitions;
+ // Build epochs map for assigned partitions (preserve existing epochs)
if (!assignedPartitions.isEmpty()) {
- newAssignedPartitions.put(topicId, assignedPartitions);
+ Map<Integer, Integer> partitionEpochs = new HashMap<>();
+ for (Integer partitionId : assignedPartitions) {
+ partitionEpochs.put(partitionId, existingTopicEpochs.getOrDefault(partitionId, memberEpoch));
Review Comment:
The default is never used because `assignedPartitions` comes from `member.assignedPartitions()`.
```suggestion
partitionEpochs.put(partitionId, existingTopicEpochs.get(partitionId));
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
> I would leave `assignedPartitions` in place for now but would like to see it removed eventually.
Actually if we remove it now, we can call the new methods `assignedPartitions` and `partitionsPendingRevocation` without conflict and we don't need to invent a view to satisfy the base class.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/CurrentAssignmentBuilder.java:
##########
@@ -299,47 +300,56 @@ private ConsumerGroupMember updateCurrentAssignment(
Map<Uuid, Set<Integer>> memberAssignedPartitions
) {
Set<Uuid> subscribedTopicIds = subscribedTopicIds();
-
// Reuse the original map if no topics need to be removed.
- Map<Uuid, Set<Integer>> newAssignedPartitions;
- Map<Uuid, Set<Integer>> newPartitionsPendingRevocation;
+ Map<Uuid, Map<Integer, Integer>> newAssignedPartitionsWithEpochs;
+ Map<Uuid, Map<Integer, Integer>> newPartitionsPendingRevocationWithEpochs;
+ boolean changed = false;
+
if (subscribedTopicIds.isEmpty() && member.partitionsPendingRevocation().isEmpty()) {
- newAssignedPartitions = Map.of();
- newPartitionsPendingRevocation = memberAssignedPartitions;
+ newAssignedPartitionsWithEpochs = Map.of();
+ // Move all assigned to pending revocation with their epochs
+ newPartitionsPendingRevocationWithEpochs = new HashMap<>(member.assignedPartitionsWithEpochs());
+ changed = true;
} else {
- newAssignedPartitions = memberAssignedPartitions;
- newPartitionsPendingRevocation = new HashMap<>(member.partitionsPendingRevocation());
+ newAssignedPartitionsWithEpochs = new HashMap<>(member.assignedPartitionsWithEpochs());
+ newPartitionsPendingRevocationWithEpochs = new HashMap<>(member.partitionsPendingRevocationWithEpochs());
for (Map.Entry<Uuid, Set<Integer>> entry : memberAssignedPartitions.entrySet()) {
if (!subscribedTopicIds.contains(entry.getKey())) {
- if (newAssignedPartitions == memberAssignedPartitions) {
- newAssignedPartitions = new HashMap<>(memberAssignedPartitions);
- newPartitionsPendingRevocation = new HashMap<>(member.partitionsPendingRevocation());
+ changed = true;
+ Uuid topicId = entry.getKey();
+ Map<Integer, Integer> removedEpochs = newAssignedPartitionsWithEpochs.remove(topicId);
+ if (removedEpochs != null) {
+ newPartitionsPendingRevocationWithEpochs.merge(
+ topicId,
+ removedEpochs,
+ (existing, additional) -> {
+ Map<Integer, Integer> merged = new HashMap<>(existing);
+ merged.putAll(additional);
+ return merged;
+ }
+ );
}
- newAssignedPartitions.remove(entry.getKey());
- newPartitionsPendingRevocation.merge(
- entry.getKey(),
- entry.getValue(),
- (existing, additional) -> {
- existing = new HashSet<>(existing);
- existing.addAll(additional);
- return existing;
- }
- );
}
}
}
- if (newAssignedPartitions == memberAssignedPartitions) {
+ if (!changed) {
// If no partitions were removed, we can return the member as is.
return member;
}
+ Map<Uuid, Set<Integer>> newPartitionsPendingRevocation = newPartitionsPendingRevocationWithEpochs.entrySet().stream()
+ .collect(Collectors.toMap(
+ Map.Entry::getKey,
+ e -> e.getValue().keySet()
+ ));
Review Comment:
Can we try updating `ownsRevokedPartitions` to accept a `Map<topic id, Map<partition, assignment epoch>>` instead?
##########
group-coordinator/src/main/resources/common/message/ConsumerGroupCurrentMemberAssignmentValue.json:
##########
@@ -36,7 +36,10 @@
{ "name": "TopicId", "type": "uuid", "versions": "0+",
"about": "The topic Id." },
{ "name": "Partitions", "type": "[]int32", "versions": "0+",
- "about": "The partition Ids." }
+ "about": "The partition Ids." },
+ { "name": "AssignmentEpochs", "versions": "0+", "nullableVersions": "0+",
+ "taggedVersions": "0+", "tag": 0, "type": "[]int32", "default": null,
+ "about": "The epoch at which each partition was assigned to this member. Aligned with Partitions array. Used to validate offset commits (KIP-1251)." }
Review Comment:
nit: Could we follow the streams implementation?
```suggestion
"about": "The epoch at which the partition was assigned to the member. Used to fence zombie commits requests. Of the same length as 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2835160412
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
For the `assignedPartitions`, it is part of the fields inherited from `ModernGroupMember` that `ShareGroupMember` also uses, so i left it there for the getter method.
For the. `partitionsPendingRevocation`, there are 6 or 7 places it is directly called. Guess I could remove this method and do extra conversions for where its' called instead.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2834934664
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java:
##########
@@ -804,13 +804,19 @@ public static CoordinatorRecord newShareGroupStatePartitionMetadataRecord(
}
private static List<ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions> toTopicPartitions(
- Map<Uuid, Set<Integer>> topicPartitions
- ) {
- List<ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions> topics = new ArrayList<>(topicPartitions.size());
- topicPartitions.forEach((topicId, partitions) ->
+ Map<Uuid, Map<Integer, Integer>> topicPartitionsWithEpochs
+ ) {
+ List<ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions> topics = new ArrayList<>(topicPartitionsWithEpochs.size());
+ topicPartitionsWithEpochs.forEach((topicId, partitionEpochMap) -> {
+ List<Integer> partitionList = new ArrayList<>(partitionEpochMap.keySet());
+ List<Integer> epochList = partitionList.stream()
+ .map(partitionId -> partitionEpochMap.getOrDefault(partitionId, 0))
Review Comment:
why getOrDefault and not just get
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
I think this makes sense - are there so many places that require `assignedPartitions` and `partitionsPendingRevocation` that we cannot convert the calling context to use the map instead of the set? Or do we require it because of the sharing with the share group?
I wonder if it could be worth implementing a view on the keySet to avoid the second collection here, as it's probably one of the larger collections in the GC.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2843147002
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
Thanks!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2827683614
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -837,6 +851,46 @@ private void validateMemberEpoch(
}
}
+ /**
+ * Creates a validator that checks per-partition assignment epochs.
+ * A commit is rejected if the partition is not assigned to the member
+ * or if the received client-side epoch is older than the partition's assignment epoch(KIP-1251).
+ *
+ * @param member The consumer group member.
+ * @param receivedMemberEpoch The member epoch from the offset commit request.
+ * @return A validator that checks each partition's assignment epoch.
+ */
+ private CommitPartitionValidator createAssignmentEpochValidator(
+ ConsumerGroupMember member,
+ int receivedMemberEpoch
+ ) {
+ return (topicName, topicId, partitionId) -> {
+ // Check if the partition is in the assigned partitions.
+ // If not found in assigned, check partitions pending revocation.
+ Integer assignmentEpoch = member.getAssignmentEpoch(topicId, partitionId);
+ if (assignmentEpoch == null) {
+ assignmentEpoch = member.getPendingRevocationEpoch(topicId, partitionId);
+ }
+
+ // If the partition is not assigned to this member, reject.
+ if (assignmentEpoch == null) {
+ throw new StaleMemberEpochException(
+ String.format("Partition %s-%d is not assigned to member %s.",
Review Comment:
Above, we are allowing committing unassigned partitions if the "strict epoch match" passes, that is client-side member epoch = server-side member epoch.
Then this error message is probably not sufficient as a reason. We should also include that the client-side member epoch != server-side member epoch in this error.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4138,9 +4138,12 @@ private CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord>
ConsumerGroupMember member
) {
// We will write a member epoch of -2 for this departing static member.
+ // Assignment epochs are reset to 0 so when the static member rejoins, partitions
+ // are considered assigned from epoch 0 to the new member ID.
ConsumerGroupMember leavingStaticMember = new ConsumerGroupMember.Builder(member)
Review Comment:
It seems we also need to update the dynamic member leave code. We seem to set setAssignedPartitions there.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
Why are we adding assignedPartitionsWithEpochs in addition to assignedPartitions? This will get out of sync. I think we should just have the second, and replace all uses of `assignedPartitions` with `assignedPartitionsWithEpochs`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2843112113
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
I made the refactoring here: https://github.com/apache/kafka/pull/21557
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2838819751
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
> `assignedPartitions` is part of the fields inherited from `ModernGroupMember` that `ShareGroupMember` also uses
I'm no longer convinced `assignedPartitions` in `ModernGroupMember` is useful, considering we have three group types, consumer, share and streams, all wanting different assignment types. I would leave `assignedPartitions` in place for now but would like to see it removed eventually.
> `partitionsPendingRevocation`
All the callers look to be consumer group-specific, so we should be able to adjust them to use the map<partition, epoch>s without impacting share groups.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2829447490
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -837,6 +851,46 @@ private void validateMemberEpoch(
}
}
+ /**
+ * Creates a validator that checks per-partition assignment epochs.
+ * A commit is rejected if the partition is not assigned to the member
+ * or if the received client-side epoch is older than the partition's assignment epoch(KIP-1251).
+ *
+ * @param member The consumer group member.
+ * @param receivedMemberEpoch The member epoch from the offset commit request.
+ * @return A validator that checks each partition's assignment epoch.
+ */
+ private CommitPartitionValidator createAssignmentEpochValidator(
+ ConsumerGroupMember member,
+ int receivedMemberEpoch
+ ) {
+ return (topicName, topicId, partitionId) -> {
+ // Check if the partition is in the assigned partitions.
+ // If not found in assigned, check partitions pending revocation.
+ Integer assignmentEpoch = member.getAssignmentEpoch(topicId, partitionId);
+ if (assignmentEpoch == null) {
+ assignmentEpoch = member.getPendingRevocationEpoch(topicId, partitionId);
+ }
+
+ // If the partition is not assigned to this member, reject.
+ if (assignmentEpoch == null) {
+ throw new StaleMemberEpochException(
+ String.format("Partition %s-%d is not assigned to member %s.",
Review Comment:
I'm in favor of aligning with the streams implementation as much as possible. Streams has the same pattern and uses the message `Task %s-%d is not assigned or pending revocation for member.`. Shall we update the streams message too (in a separate PR maybe)?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2842681114
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
Yes. I would actually propose doing it now (in its own refactoring PR). Then we can rename `assignedPartitionsWithEpochs` and `partitionsPendingRevocationWithEpochs` to `assignedPartitions` and `partitionsPendingRevocation`. I think the "withEpochs" naming is clunky and it's cleaner if we can avoid it. By removing the base method, we no longer need to make the distinction for consumer groups.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 closed pull request #21508: KAFKA-20066: Assignment epochs for consumer groups
URL: https://github.com/apache/kafka/pull/21508
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2832233381
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -837,6 +851,46 @@ private void validateMemberEpoch(
}
}
+ /**
+ * Creates a validator that checks per-partition assignment epochs.
+ * A commit is rejected if the partition is not assigned to the member
+ * or if the received client-side epoch is older than the partition's assignment epoch(KIP-1251).
+ *
+ * @param member The consumer group member.
+ * @param receivedMemberEpoch The member epoch from the offset commit request.
+ * @return A validator that checks each partition's assignment epoch.
+ */
+ private CommitPartitionValidator createAssignmentEpochValidator(
+ ConsumerGroupMember member,
+ int receivedMemberEpoch
+ ) {
+ return (topicName, topicId, partitionId) -> {
+ // Check if the partition is in the assigned partitions.
+ // If not found in assigned, check partitions pending revocation.
+ Integer assignmentEpoch = member.getAssignmentEpoch(topicId, partitionId);
+ if (assignmentEpoch == null) {
+ assignmentEpoch = member.getPendingRevocationEpoch(topicId, partitionId);
+ }
+
+ // If the partition is not assigned to this member, reject.
+ if (assignmentEpoch == null) {
+ throw new StaleMemberEpochException(
+ String.format("Partition %s-%d is not assigned to member %s.",
Review Comment:
Sound good!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2833712464
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
I have removed all setters for `assignedPartitions` and `partitionsPendingRevocation`, but still keep the getters and as private fields to avoid redundant conversions in other methods that previously call these getters.
Since these private fields will be re-constructed every time the builder is called to set `assignedPartitionsWithEpochs` and `partitionsRevocationsWithEpochs`, the getters won't be async.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "lucliu1108 (via GitHub)" <gi...@apache.org>.
lucliu1108 commented on PR #21508:
URL: https://github.com/apache/kafka/pull/21508#issuecomment-3953471173
Hi @lucasbru @squah-confluent , thanks for the review!
I plan to break this large PR into 3:
1. https://github.com/apache/kafka/pull/21557: Move `assignmentPartitions` out of `ModernGroupMember` interface
2. https://github.com/apache/kafka/pull/21558: Change structure of `ConsumerGroupMember#assignedPartitions` to include epochs, and modify corresponding tests.
3. Update logic of offset fencing, and add corresponding tests.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2834977339
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/CurrentAssignmentBuilder.java:
##########
@@ -370,10 +380,13 @@ private ConsumerGroupMember computeNextAssignment(
Set<Uuid> subscribedTopicIds = subscribedTopicIds();
boolean hasUnreleasedPartitions = false;
- Map<Uuid, Set<Integer>> newAssignedPartitions = new HashMap<>();
- Map<Uuid, Set<Integer>> newPartitionsPendingRevocation = new HashMap<>();
+ Map<Uuid, Map<Integer, Integer>> newAssignedPartitionsWithEpochs = new HashMap<>();
+ Map<Uuid, Map<Integer, Integer>> newPartitionsPendingRevocationWithEpochs = new HashMap<>();
Map<Uuid, Set<Integer>> newPartitionsPendingAssignment = new HashMap<>();
+ // Get existing epochs from member
+ Map<Uuid, Map<Integer, Integer>> existingAssignedEpochs = member.assignedPartitionsWithEpochs();
Review Comment:
If an partition was pending revocation, and is added back to the target assignment of the member, it seems we want to keep the original assignment epoch. So we have to look at member.partitionPendingRevociations here as well, no?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20066: Assignment epochs for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #21508:
URL: https://github.com/apache/kafka/pull/21508#discussion_r2838926319
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupMember.java:
##########
@@ -66,6 +69,8 @@ public static class Builder {
private Map<Uuid, Set<Integer>> assignedPartitions = Map.of();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Map.of();
private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
+ private Map<Uuid, Map<Integer, Integer>> assignedPartitionsWithEpochs = Map.of();
Review Comment:
> I would leave `assignedPartitions` in place for now but would like to see it removed eventually.
Actually if we remove it now, we can call the new methods `assignedPartitions` and `partitionsPendingRevocation` without conflict and we won't need to invent a view to satisfy the base class.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org