You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "apoorvmittal10 (via GitHub)" <gi...@apache.org> on 2024/08/07 15:07:47 UTC
[PR] KAFKA-17288: Removed tracking partition member epoch (KIP-932) [kafka]
apoorvmittal10 opened a new pull request, #16828:
URL: https://github.com/apache/kafka/pull/16828
Partition epochs are tracked for consumer groups where epoch is the current assigned member epoch. As share groups have partitions shared hence maintaing the partition epochs is not required.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-17288: Removed tracking partition member epoch (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16828:
URL: https://github.com/apache/kafka/pull/16828#discussion_r1710310089
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -677,6 +705,73 @@ private void maybeRemovePartitionEpoch(
}
}
+ /**
+ * 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.
+ */
+ void removePartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int expectedEpoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull != null) {
+ 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 {
+ throw new IllegalStateException(
+ String.format("Cannot remove the epoch %d from %s because it does not have any epoch",
+ expectedEpoch, topicId));
+ }
+ });
+ });
+ }
+
+ /**
+ * Adds the partitions epoch based on the provided assignment.
+ *
+ * @param assignment The assignment.
+ * @param epoch The new epoch.
+ * @throws IllegalStateException if the partition already has an epoch assigned.
+ * package-private for testing.
+ */
+ void addPartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int epoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull == null) {
+ partitionsOrNull = new TimelineHashMap<>(snapshotRegistry, assignedPartitions.size());
+ }
+ for (Integer partitionId : assignedPartitions) {
+ Integer prevValue = partitionsOrNull.put(partitionId, epoch);
Review Comment:
As per my understanding, `currentPartitionEpoch` is backed by `TimelineHashMap` where soft state is updated here but if exception occurs then the transaction is not persisted and gets overriden on next successful update. This was an existing code in ConsumerGroup.java which has been put back in current PR, we initially moved this code to ModernGroup so both Share and Consumer groups can use but later realized that Share Groups do not need this functionality.
--
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-17288: Removed tracking partition member epoch (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16828:
URL: https://github.com/apache/kafka/pull/16828#discussion_r1711290190
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -677,6 +705,73 @@ private void maybeRemovePartitionEpoch(
}
}
+ /**
+ * 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.
+ */
+ void removePartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int expectedEpoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull != null) {
+ 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 {
+ throw new IllegalStateException(
+ String.format("Cannot remove the epoch %d from %s because it does not have any epoch",
+ expectedEpoch, topicId));
+ }
+ });
+ });
+ }
+
+ /**
+ * Adds the partitions epoch based on the provided assignment.
+ *
+ * @param assignment The assignment.
+ * @param epoch The new epoch.
+ * @throws IllegalStateException if the partition already has an epoch assigned.
+ * package-private for testing.
+ */
+ void addPartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int epoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull == null) {
+ partitionsOrNull = new TimelineHashMap<>(snapshotRegistry, assignedPartitions.size());
+ }
+ for (Integer partitionId : assignedPartitions) {
+ Integer prevValue = partitionsOrNull.put(partitionId, epoch);
Review Comment:
> If so, even though we throw exception here, the method "put" still changes the partitionOrNull.
Yes, that's right. But it's a soft update which gets upadted when CoordinatorRuntime either does rollback or commit. So I don't think it should concern us. Adding @dajac for more context.
--
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-17288: Removed tracking partition member epoch (KIP-932) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #16828:
URL: https://github.com/apache/kafka/pull/16828
--
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-17288: Removed tracking partition member epoch (KIP-932) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16828:
URL: https://github.com/apache/kafka/pull/16828#discussion_r1711588078
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -677,6 +705,73 @@ private void maybeRemovePartitionEpoch(
}
}
+ /**
+ * 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.
+ */
+ void removePartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int expectedEpoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull != null) {
+ 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 {
+ throw new IllegalStateException(
+ String.format("Cannot remove the epoch %d from %s because it does not have any epoch",
+ expectedEpoch, topicId));
+ }
+ });
+ });
+ }
+
+ /**
+ * Adds the partitions epoch based on the provided assignment.
+ *
+ * @param assignment The assignment.
+ * @param epoch The new epoch.
+ * @throws IllegalStateException if the partition already has an epoch assigned.
+ * package-private for testing.
+ */
+ void addPartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int epoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull == null) {
+ partitionsOrNull = new TimelineHashMap<>(snapshotRegistry, assignedPartitions.size());
+ }
+ for (Integer partitionId : assignedPartitions) {
+ Integer prevValue = partitionsOrNull.put(partitionId, epoch);
Review Comment:
> Yes, that's right. But it's a soft update which gets upadted when CoordinatorRuntime either does rollback or commit. So I don't think it should concern us.
thanks for explanation. I overlook the `SnapshotRegistry` :(
--
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-17288: Removed tracking partition member epoch (KIP-932) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16828:
URL: https://github.com/apache/kafka/pull/16828#discussion_r1710544091
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -677,6 +705,73 @@ private void maybeRemovePartitionEpoch(
}
}
+ /**
+ * 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.
+ */
+ void removePartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int expectedEpoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull != null) {
+ 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 {
+ throw new IllegalStateException(
+ String.format("Cannot remove the epoch %d from %s because it does not have any epoch",
+ expectedEpoch, topicId));
+ }
+ });
+ });
+ }
+
+ /**
+ * Adds the partitions epoch based on the provided assignment.
+ *
+ * @param assignment The assignment.
+ * @param epoch The new epoch.
+ * @throws IllegalStateException if the partition already has an epoch assigned.
+ * package-private for testing.
+ */
+ void addPartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int epoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull == null) {
+ partitionsOrNull = new TimelineHashMap<>(snapshotRegistry, assignedPartitions.size());
+ }
+ for (Integer partitionId : assignedPartitions) {
+ Integer prevValue = partitionsOrNull.put(partitionId, epoch);
Review Comment:
Sorry that I'm in traveling so I can't check it on my computer. The following description is my guess.
It seems to me partitionOrNull should not be updated by "epoch" if it already has an epoch assigned, right? If so, even though we throw exception here, the method "put" still changes the partitionOrNull.
--
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-17288: Removed tracking partition member epoch (KIP-932) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16828:
URL: https://github.com/apache/kafka/pull/16828#discussion_r1710128342
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -677,6 +705,73 @@ private void maybeRemovePartitionEpoch(
}
}
+ /**
+ * 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.
+ */
+ void removePartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int expectedEpoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull != null) {
+ 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 {
+ throw new IllegalStateException(
+ String.format("Cannot remove the epoch %d from %s because it does not have any epoch",
+ expectedEpoch, topicId));
+ }
+ });
+ });
+ }
+
+ /**
+ * Adds the partitions epoch based on the provided assignment.
+ *
+ * @param assignment The assignment.
+ * @param epoch The new epoch.
+ * @throws IllegalStateException if the partition already has an epoch assigned.
+ * package-private for testing.
+ */
+ void addPartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int epoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull == null) {
+ partitionsOrNull = new TimelineHashMap<>(snapshotRegistry, assignedPartitions.size());
+ }
+ for (Integer partitionId : assignedPartitions) {
+ Integer prevValue = partitionsOrNull.put(partitionId, epoch);
Review Comment:
BTW, I have the same question for `removePartitionEpochs`
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -677,6 +705,73 @@ private void maybeRemovePartitionEpoch(
}
}
+ /**
+ * 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.
+ */
+ void removePartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int expectedEpoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull != null) {
+ 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 {
+ throw new IllegalStateException(
+ String.format("Cannot remove the epoch %d from %s because it does not have any epoch",
+ expectedEpoch, topicId));
+ }
+ });
+ });
+ }
+
+ /**
+ * Adds the partitions epoch based on the provided assignment.
+ *
+ * @param assignment The assignment.
+ * @param epoch The new epoch.
+ * @throws IllegalStateException if the partition already has an epoch assigned.
+ * package-private for testing.
+ */
+ void addPartitionEpochs(
+ Map<Uuid, Set<Integer>> assignment,
+ int epoch
+ ) {
+ assignment.forEach((topicId, assignedPartitions) -> {
+ currentPartitionEpoch.compute(topicId, (__, partitionsOrNull) -> {
+ if (partitionsOrNull == null) {
+ partitionsOrNull = new TimelineHashMap<>(snapshotRegistry, assignedPartitions.size());
+ }
+ for (Integer partitionId : assignedPartitions) {
+ Integer prevValue = partitionsOrNull.put(partitionId, epoch);
Review Comment:
We don't create new `partitionsOrNull` when updating `partitionsOrNull`, so should we use `putIfAbsent` to replace `put` to avoid overriding the epoch when the partition is still owned by another epoch?
for another, should we view `addPartitionEpochs` as transaction op? that means we update the `currentPartitionEpoch` only if all elements in `assignment` are valid.
--
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-17288: Removed tracking partition member epoch (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #16828:
URL: https://github.com/apache/kafka/pull/16828#issuecomment-2276668780
@omkreddy @dajac @chia7712 Unrelated tests failure in the PR.
--
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-17288: Removed tracking partition member epoch (KIP-932) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16828:
URL: https://github.com/apache/kafka/pull/16828#issuecomment-2282921102
all failed tests pass on my local. will merge it
--
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