[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550: URL: https://github.com/apache/kafka/pull/13550#discussion_r1165947809 ## clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java: ## @@ -16,6 +16,13 @@ */ package org.apache.kafka.clients.consumer.internals; +import org.apache.kafka.clients.consumer.internals.Utils.PartitionComparator; Review Comment: this is "optimized" by the IDE -- 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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1165959573
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java:
##
@@ -153,43 +158,46 @@ private boolean allSubscriptionsEqual(Set
allTopics,
List ownedPartitions = new ArrayList<>();
consumerToOwnedPartitions.put(consumer, ownedPartitions);
-// Only consider this consumer's owned partitions as valid if it
is a member of the current highest
-// generation, or it's generation is not present but we have not
seen any known generation so far
-if (memberData.generation.isPresent() &&
memberData.generation.get() >= maxGeneration
+if (memberData.generation.isPresent() &&
memberData.generation.get() >= maxGeneration - 1
|| !memberData.generation.isPresent() && maxGeneration ==
DEFAULT_GENERATION) {
-// If the current member's generation is higher, all the
previously owned partitions are invalid
-if (memberData.generation.isPresent() &&
memberData.generation.get() > maxGeneration) {
Review Comment:
we don't need this because the maxGeneration has already been computed
previously, so we know the set of valid members.
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1165960472
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java:
##
@@ -153,43 +158,45 @@ private boolean allSubscriptionsEqual(Set
allTopics,
List ownedPartitions = new ArrayList<>();
consumerToOwnedPartitions.put(consumer, ownedPartitions);
-// Only consider this consumer's owned partitions as valid if it
is a member of the current highest
-// generation, or it's generation is not present but we have not
seen any known generation so far
-if (memberData.generation.isPresent() &&
memberData.generation.get() >= maxGeneration
+if (memberData.generation.isPresent() &&
memberData.generation.get() >= maxGeneration - 1
|| !memberData.generation.isPresent() && maxGeneration ==
DEFAULT_GENERATION) {
-// If the current member's generation is higher, all the
previously owned partitions are invalid
-if (memberData.generation.isPresent() &&
memberData.generation.get() > maxGeneration) {
-allPreviousPartitionsToOwner.clear();
-partitionsWithMultiplePreviousOwners.clear();
-for (String droppedOutConsumer :
membersOfCurrentHighestGeneration) {
-
consumerToOwnedPartitions.get(droppedOutConsumer).clear();
-}
-
-membersOfCurrentHighestGeneration.clear();
-maxGeneration = memberData.generation.get();
-}
-
-membersOfCurrentHighestGeneration.add(consumer);
for (final TopicPartition tp : memberData.partitions) {
-// filter out any topics that no longer exist or aren't
part of the current subscription
if (allTopics.contains(tp.topic())) {
String otherConsumer =
allPreviousPartitionsToOwner.put(tp, consumer);
if (otherConsumer == null) {
// this partition is not owned by other consumer
in the same generation
ownedPartitions.add(tp);
} else {
-log.error("Found multiple consumers {} and {}
claiming the same TopicPartition {} in the "
-+ "same generation {}, this will be
invalidated and removed from their previous assignment.",
- consumer, otherConsumer, tp,
maxGeneration);
-
consumerToOwnedPartitions.get(otherConsumer).remove(tp);
-partitionsWithMultiplePreviousOwners.add(tp);
+int memberGeneration =
memberData.generation.orElse(DEFAULT_GENERATION);
+int otherMemberGeneration =
subscriptions.get(otherConsumer).generationId().orElse(DEFAULT_GENERATION);
+
+if (memberGeneration == otherMemberGeneration) {
+if
(subscriptions.get(otherConsumer).generationId().orElse(DEFAULT_GENERATION) ==
memberData.generation.orElse(DEFAULT_GENERATION)) {
+log.error("Found multiple consumers {} and
{} claiming the same TopicPartition {} in the "
++ "same generation {}, this will
be invalidated and removed from their previous assignment.",
+consumer, otherConsumer, tp,
maxGeneration);
+
partitionsWithMultiplePreviousOwners.add(tp);
+}
+
consumerToOwnedPartitions.get(otherConsumer).remove(tp);
+allPreviousPartitionsToOwner.put(tp, consumer);
+continue;
+}
+
+if (memberGeneration > otherMemberGeneration) {
+log.warn("Found multiple consumers {} and {}
claiming the same TopicPartition {} in " +
Review Comment:
I feel... duplicated ownership should be a warning.
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1169135808
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##
@@ -835,6 +835,7 @@ public void handle(SyncGroupResponse syncResponse,
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another
rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
+resetStateAndGeneration("member missed the rebalance",
true);
Review Comment:
Hey @dajac thanks for the review. I think in case of the member misses a
generation, we want to make sure the owned partitions are revoked (due to
generation reset). Regardlessly, it should still rejoin with its current
partitions and should continue to hold on to its partition if it is only 1
generation behind. If it is 1+ generations behind, circle back to the beginning
of my response, we want to make sure they are revoked because the partition
might have already been reassigned.
This makes me think that this will happen regularly in medium to large
groups. -> I think this might not be as uncommon as what we think, especially
with a large consumer group deployed to multiple pods, considering the pods can
be staled before sending out syncGroup, while another consumer in a different
pod tries to join the group.
I hope i'm answering your questions there, I apologize if I misunderstood
anything.
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1170094810
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##
@@ -835,6 +835,7 @@ public void handle(SyncGroupResponse syncResponse,
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another
rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
+resetStateAndGeneration("member missed the rebalance",
true);
Review Comment:
Hmm, I think the partitions are only lost during the onJoinPrepare, what I'm
thinking is this:
https://github.com/apache/kafka/blob/61530d68ce83467de6190a52da37b3c0af84f0ef/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java#L507
The gist is: on one of the 4 exception thrown in join/sync group, it should
immediately re-send the join request. Are you thinking about the how client
handles the illegal generation error? I think it is only being thrown during
sync group and heartbeat. So I think by just resetting the generation shouldn't
immediately causes revocation.
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1170094810
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##
@@ -835,6 +835,7 @@ public void handle(SyncGroupResponse syncResponse,
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another
rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
+resetStateAndGeneration("member missed the rebalance",
true);
Review Comment:
Hmm, I think the partitions are only lost during the onJoinPrepare, what I'm
thinking is this:
https://github.com/apache/kafka/blob/61530d68ce83467de6190a52da37b3c0af84f0ef/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java#L507-L511
The gist is: on one of the 4 exception thrown in join/sync group, it should
immediately re-send the join request. Are you thinking about the how client
handles the illegal generation error? I think it is only being thrown during
sync group and heartbeat. So I think by just resetting the generation shouldn't
immediately causes revocation.
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1170098942
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##
@@ -835,6 +835,7 @@ public void handle(SyncGroupResponse syncResponse,
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another
rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
+resetStateAndGeneration("member missed the rebalance",
true);
Review Comment:
Added to my previous comment, I think you will need to set the
`needsOnJoinPrepare` to true to go through the revocation as pointed out here:
https://github.com/apache/kafka/blob/61530d68ce83467de6190a52da37b3c0af84f0ef/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java#L821
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1170599794
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##
@@ -835,6 +835,7 @@ public void handle(SyncGroupResponse syncResponse,
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another
rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
+resetStateAndGeneration("member missed the rebalance",
true);
Review Comment:
Offline discussion with @hachikuji - it seems like what we want is to
revoke the old partitions but resend these partitions on the subsequent join.
RN, join only sends out the assigned 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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1171712519
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##
@@ -835,6 +839,8 @@ public void handle(SyncGroupResponse syncResponse,
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another
rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
+savePartitionAndGenerationState();
Review Comment:
saving the generation and partition state prior to resetting 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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1171780417
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##
@@ -835,6 +839,8 @@ public void handle(SyncGroupResponse syncResponse,
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another
rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
+savePartitionAndGenerationState();
Review Comment:
i think the ration for resetting the generation is to ensure member's
partitions get revoke during the onJoinPrepare, i'm trying to think is there's
a better way to do this...
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1172072549
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##
@@ -835,6 +839,8 @@ public void handle(SyncGroupResponse syncResponse,
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another
rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
+savePartitionAndGenerationState();
Review Comment:
I'm not sure what's the better way to invoke `onPartitionsLost` w/o
resetting the generation. The alternative is to use a flag, but that's rather
unclean.
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1176980400
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##
@@ -835,6 +839,8 @@ public void handle(SyncGroupResponse syncResponse,
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another
rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
+savePartitionAndGenerationState();
Review Comment:
This is discarded in the latest commit
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639 - A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1176980677
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##
@@ -835,6 +835,7 @@ public void handle(SyncGroupResponse syncResponse,
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another
rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
+resetStateAndGeneration("member missed the rebalance",
true);
Review Comment:
this is discarded in the latest commit
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639: A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1177893519
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java:
##
@@ -149,47 +148,51 @@ private boolean allSubscriptionsEqual(Set
allTopics,
}
MemberData memberData = memberData(subscription);
+maxGeneration = Math.max(maxGeneration,
memberData.generation.orElse(DEFAULT_GENERATION));
List ownedPartitions = new ArrayList<>();
consumerToOwnedPartitions.put(consumer, ownedPartitions);
-// Only consider this consumer's owned partitions as valid if it
is a member of the current highest
-// generation, or it's generation is not present but we have not
seen any known generation so far
-if (memberData.generation.isPresent() &&
memberData.generation.get() >= maxGeneration
-|| !memberData.generation.isPresent() && maxGeneration ==
DEFAULT_GENERATION) {
-
-// If the current member's generation is higher, all the
previously owned partitions are invalid
-if (memberData.generation.isPresent() &&
memberData.generation.get() > maxGeneration) {
-allPreviousPartitionsToOwner.clear();
-partitionsWithMultiplePreviousOwners.clear();
-for (String droppedOutConsumer :
membersOfCurrentHighestGeneration) {
-
consumerToOwnedPartitions.get(droppedOutConsumer).clear();
-}
-
-membersOfCurrentHighestGeneration.clear();
-maxGeneration = memberData.generation.get();
-}
+// the member has a valid generation, so we can consider its owned
partitions if it has the highest
+// generation amongst
+for (final TopicPartition tp : memberData.partitions) {
+if (allTopics.contains(tp.topic())) {
+String otherConsumer =
allPreviousPartitionsToOwner.put(tp, consumer);
+if (otherConsumer == null) {
+// this partition is not owned by other consumer in
the same generation
+ownedPartitions.add(tp);
+} else {
+final int memberGeneration =
memberData.generation.orElse(DEFAULT_GENERATION);
+final int otherMemberGeneration =
subscriptions.get(otherConsumer).generationId().orElse(DEFAULT_GENERATION);
+
+if (memberGeneration == otherMemberGeneration) {
+if
(subscriptions.get(otherConsumer).generationId().orElse(DEFAULT_GENERATION) ==
memberData.generation.orElse(DEFAULT_GENERATION)) {
+log.error("Found multiple consumers {} and {}
claiming the same TopicPartition {} in the "
++ "same generation {}, this will be
invalidated and removed from their previous assignment.",
+consumer, otherConsumer, tp,
memberGeneration);
+partitionsWithMultiplePreviousOwners.add(tp);
Review Comment:
that seems like the case, reference to the snippet here:
```
for (TopicPartition doublyClaimedPartition :
partitionsWithMultiplePreviousOwners) {
if (ownedPartitions.contains(doublyClaimedPartition)) {
log.error("Found partition {} still claimed as owned
by consumer {}, despite being claimed by multiple "
+ "consumers already in the same
generation. Removing it from the ownedPartitions",
doublyClaimedPartition, consumer);
ownedPartitions.remove(doublyClaimedPartition);
}
```
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639: A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1177895473
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java:
##
@@ -149,47 +148,51 @@ private boolean allSubscriptionsEqual(Set
allTopics,
}
MemberData memberData = memberData(subscription);
+maxGeneration = Math.max(maxGeneration,
memberData.generation.orElse(DEFAULT_GENERATION));
List ownedPartitions = new ArrayList<>();
consumerToOwnedPartitions.put(consumer, ownedPartitions);
-// Only consider this consumer's owned partitions as valid if it
is a member of the current highest
-// generation, or it's generation is not present but we have not
seen any known generation so far
-if (memberData.generation.isPresent() &&
memberData.generation.get() >= maxGeneration
-|| !memberData.generation.isPresent() && maxGeneration ==
DEFAULT_GENERATION) {
-
-// If the current member's generation is higher, all the
previously owned partitions are invalid
-if (memberData.generation.isPresent() &&
memberData.generation.get() > maxGeneration) {
-allPreviousPartitionsToOwner.clear();
-partitionsWithMultiplePreviousOwners.clear();
-for (String droppedOutConsumer :
membersOfCurrentHighestGeneration) {
-
consumerToOwnedPartitions.get(droppedOutConsumer).clear();
-}
-
-membersOfCurrentHighestGeneration.clear();
-maxGeneration = memberData.generation.get();
-}
+// the member has a valid generation, so we can consider its owned
partitions if it has the highest
+// generation amongst
+for (final TopicPartition tp : memberData.partitions) {
+if (allTopics.contains(tp.topic())) {
+String otherConsumer =
allPreviousPartitionsToOwner.put(tp, consumer);
+if (otherConsumer == null) {
+// this partition is not owned by other consumer in
the same generation
+ownedPartitions.add(tp);
+} else {
+final int memberGeneration =
memberData.generation.orElse(DEFAULT_GENERATION);
+final int otherMemberGeneration =
subscriptions.get(otherConsumer).generationId().orElse(DEFAULT_GENERATION);
+
+if (memberGeneration == otherMemberGeneration) {
+if
(subscriptions.get(otherConsumer).generationId().orElse(DEFAULT_GENERATION) ==
memberData.generation.orElse(DEFAULT_GENERATION)) {
+log.error("Found multiple consumers {} and {}
claiming the same TopicPartition {} in the "
++ "same generation {}, this will be
invalidated and removed from their previous assignment.",
+consumer, otherConsumer, tp,
memberGeneration);
+partitionsWithMultiplePreviousOwners.add(tp);
+}
+
consumerToOwnedPartitions.get(otherConsumer).remove(tp);
+allPreviousPartitionsToOwner.put(tp, consumer);
+continue;
Review Comment:
It could be. I got into the habit of returning early, I thought it makes it
easier to read.
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639: A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1177934535
##
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignorTest.java:
##
@@ -1038,6 +1038,96 @@ public void
testPartitionsTransferringOwnershipIncludeThePartitionClaimedByMulti
assertTrue(isFullyBalanced(assignment));
}
+@ParameterizedTest(name = TEST_NAME_WITH_RACK_CONFIG)
+@EnumSource(RackConfig.class)
+public void
testOwnedPartitionsAreStableForConsumerWithMultipleGeneration(RackConfig
rackConfig) {
+initializeRacks(rackConfig);
+Map> partitionsPerTopic = new HashMap<>();
+partitionsPerTopic.put(topic, partitionInfos(topic, 3));
+partitionsPerTopic.put(topic2, partitionInfos(topic2, 3));
+partitionsPerTopic.put(topic3, partitionInfos(topic3, 3));
+
+int currentGeneration = 10;
+
+subscriptions.put(consumer1, buildSubscriptionV2Above(topics(topic,
topic2, topic3),
+partitions(tp(topic, 0), tp(topic2, 0), tp(topic3, 0)),
currentGeneration, 0));
+subscriptions.put(consumer2, buildSubscriptionV2Above(topics(topic,
topic2, topic3),
+partitions(tp(topic, 1), tp(topic2, 1), tp(topic3, 1)),
currentGeneration - 1, 1));
+subscriptions.put(consumer3, buildSubscriptionV2Above(topics(topic,
topic2, topic3),
+partitions(tp(topic2, 2), tp(topic3, 2), tp(topic, 1)),
currentGeneration - 2, 1));
+
+Map> assignment =
assignor.assignPartitions(partitionsPerTopic, subscriptions);
+assertEquals(new HashSet<>(partitions(tp(topic, 0), tp(topic2, 0),
tp(topic3, 0))),
+new HashSet<>(assignment.get(consumer1)));
+assertEquals(new HashSet<>(partitions(tp(topic, 1), tp(topic2, 1),
tp(topic3, 1))),
+new HashSet<>(assignment.get(consumer2)));
+assertEquals(new HashSet<>(partitions(tp(topic, 2), tp(topic2, 2),
tp(topic3, 2))),
+new HashSet<>(assignment.get(consumer3)));
+assertTrue(assignor.partitionsTransferringOwnership.isEmpty());
+
+verifyValidityAndBalance(subscriptions, assignment,
partitionsPerTopic);
+assertTrue(isFullyBalanced(assignment));
+}
+
+@ParameterizedTest(name = TEST_NAME_WITH_RACK_CONFIG)
+@EnumSource(RackConfig.class)
+public void testNoReassignmentOnCurrentMembers(RackConfig rackConfig) {
+initializeRacks(rackConfig);
+Map> partitionsPerTopic = new HashMap<>();
+partitionsPerTopic.put(topic, partitionInfos(topic, 3));
+partitionsPerTopic.put(topic2, partitionInfos(topic2, 3));
+partitionsPerTopic.put(topic3, partitionInfos(topic3, 3));
+partitionsPerTopic.put(topic1, partitionInfos(topic1, 3));
+
+int currentGeneration = 10;
+
+subscriptions.put(consumer1, buildSubscriptionV2Above(topics(topic,
topic2, topic3, topic1), partitions(),
+DEFAULT_GENERATION, 0));
+subscriptions.put(consumer2, buildSubscriptionV2Above(topics(topic,
topic2, topic3, topic1),
+partitions(tp(topic, 0), tp(topic2, 0), tp(topic1, 0)),
currentGeneration - 1, 1));
+subscriptions.put(consumer3, buildSubscriptionV2Above(topics(topic,
topic2, topic3, topic1),
+partitions(tp(topic3, 2), tp(topic2, 2), tp(topic1, 1)),
currentGeneration - 2, 2));
+subscriptions.put(consumer4, buildSubscriptionV2Above(topics(topic,
topic2, topic3, topic1),
+partitions(tp(topic3, 1), tp(topic, 1), tp(topic, 2)),
currentGeneration - 3, 3));
+
+Map> assignment =
assignor.assignPartitions(partitionsPerTopic, subscriptions);
+// ensure assigned partitions don't get reassigned
+assertTrue(assignment.get(consumer1).containsAll(
+Arrays.asList(tp(topic2, 1),
+tp(topic3, 0),
+tp(topic1, 2;
+assertTrue(assignor.partitionsTransferringOwnership.isEmpty());
+
+verifyValidityAndBalance(subscriptions, assignment,
partitionsPerTopic);
+assertTrue(isFullyBalanced(assignment));
+}
+
+@ParameterizedTest(name = TEST_NAME_WITH_RACK_CONFIG)
+@EnumSource(RackConfig.class)
+public void
testOwnedPartitionsAreInvalidatedForConsumerWithMultipleGeneration(RackConfig
rackConfig) {
+initializeRacks(rackConfig);
+Map> partitionsPerTopic = new HashMap<>();
+partitionsPerTopic.put(topic, partitionInfos(topic, 3));
+partitionsPerTopic.put(topic2, partitionInfos(topic2, 3));
+
+int currentGeneration = 10;
+
+subscriptions.put(consumer1, buildSubscriptionV2Above(topics(topic,
topic2),
+partitions(tp(topic, 0), tp(topic2, 1), tp(topic, 1)),
currentGeneration, 0));
+subscriptions.put(consumer2, buildSubscriptionV2Above(topics(topic,
topic2),
+partitions(tp(topic, 0), tp(topic2, 1), tp(topic2, 2)),
currentGeneration - 2, 1));
+
+Map> assignment =
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639: A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1177936654
##
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignorTest.java:
##
@@ -1038,6 +1038,96 @@ public void
testPartitionsTransferringOwnershipIncludeThePartitionClaimedByMulti
assertTrue(isFullyBalanced(assignment));
}
+@ParameterizedTest(name = TEST_NAME_WITH_RACK_CONFIG)
+@EnumSource(RackConfig.class)
+public void
testOwnedPartitionsAreStableForConsumerWithMultipleGeneration(RackConfig
rackConfig) {
+initializeRacks(rackConfig);
+Map> partitionsPerTopic = new HashMap<>();
+partitionsPerTopic.put(topic, partitionInfos(topic, 3));
+partitionsPerTopic.put(topic2, partitionInfos(topic2, 3));
+partitionsPerTopic.put(topic3, partitionInfos(topic3, 3));
+
+int currentGeneration = 10;
+
+subscriptions.put(consumer1, buildSubscriptionV2Above(topics(topic,
topic2, topic3),
+partitions(tp(topic, 0), tp(topic2, 0), tp(topic3, 0)),
currentGeneration, 0));
+subscriptions.put(consumer2, buildSubscriptionV2Above(topics(topic,
topic2, topic3),
+partitions(tp(topic, 1), tp(topic2, 1), tp(topic3, 1)),
currentGeneration - 1, 1));
+subscriptions.put(consumer3, buildSubscriptionV2Above(topics(topic,
topic2, topic3),
+partitions(tp(topic2, 2), tp(topic3, 2), tp(topic, 1)),
currentGeneration - 2, 1));
+
+Map> assignment =
assignor.assignPartitions(partitionsPerTopic, subscriptions);
+assertEquals(new HashSet<>(partitions(tp(topic, 0), tp(topic2, 0),
tp(topic3, 0))),
+new HashSet<>(assignment.get(consumer1)));
+assertEquals(new HashSet<>(partitions(tp(topic, 1), tp(topic2, 1),
tp(topic3, 1))),
+new HashSet<>(assignment.get(consumer2)));
+assertEquals(new HashSet<>(partitions(tp(topic, 2), tp(topic2, 2),
tp(topic3, 2))),
+new HashSet<>(assignment.get(consumer3)));
+assertTrue(assignor.partitionsTransferringOwnership.isEmpty());
+
+verifyValidityAndBalance(subscriptions, assignment,
partitionsPerTopic);
+assertTrue(isFullyBalanced(assignment));
+}
+
+@ParameterizedTest(name = TEST_NAME_WITH_RACK_CONFIG)
+@EnumSource(RackConfig.class)
+public void testNoReassignmentOnCurrentMembers(RackConfig rackConfig) {
+initializeRacks(rackConfig);
+Map> partitionsPerTopic = new HashMap<>();
+partitionsPerTopic.put(topic, partitionInfos(topic, 3));
+partitionsPerTopic.put(topic2, partitionInfos(topic2, 3));
+partitionsPerTopic.put(topic3, partitionInfos(topic3, 3));
+partitionsPerTopic.put(topic1, partitionInfos(topic1, 3));
+
+int currentGeneration = 10;
+
+subscriptions.put(consumer1, buildSubscriptionV2Above(topics(topic,
topic2, topic3, topic1), partitions(),
+DEFAULT_GENERATION, 0));
+subscriptions.put(consumer2, buildSubscriptionV2Above(topics(topic,
topic2, topic3, topic1),
+partitions(tp(topic, 0), tp(topic2, 0), tp(topic1, 0)),
currentGeneration - 1, 1));
+subscriptions.put(consumer3, buildSubscriptionV2Above(topics(topic,
topic2, topic3, topic1),
+partitions(tp(topic3, 2), tp(topic2, 2), tp(topic1, 1)),
currentGeneration - 2, 2));
+subscriptions.put(consumer4, buildSubscriptionV2Above(topics(topic,
topic2, topic3, topic1),
+partitions(tp(topic3, 1), tp(topic, 1), tp(topic, 2)),
currentGeneration - 3, 3));
+
+Map> assignment =
assignor.assignPartitions(partitionsPerTopic, subscriptions);
+// ensure assigned partitions don't get reassigned
+assertTrue(assignment.get(consumer1).containsAll(
+Arrays.asList(tp(topic2, 1),
+tp(topic3, 0),
+tp(topic1, 2;
+assertTrue(assignor.partitionsTransferringOwnership.isEmpty());
+
+verifyValidityAndBalance(subscriptions, assignment,
partitionsPerTopic);
+assertTrue(isFullyBalanced(assignment));
+}
+
+@ParameterizedTest(name = TEST_NAME_WITH_RACK_CONFIG)
+@EnumSource(RackConfig.class)
+public void
testOwnedPartitionsAreInvalidatedForConsumerWithMultipleGeneration(RackConfig
rackConfig) {
+initializeRacks(rackConfig);
+Map> partitionsPerTopic = new HashMap<>();
+partitionsPerTopic.put(topic, partitionInfos(topic, 3));
+partitionsPerTopic.put(topic2, partitionInfos(topic2, 3));
+
+int currentGeneration = 10;
+
+subscriptions.put(consumer1, buildSubscriptionV2Above(topics(topic,
topic2),
+partitions(tp(topic, 0), tp(topic2, 1), tp(topic, 1)),
currentGeneration, 0));
+subscriptions.put(consumer2, buildSubscriptionV2Above(topics(topic,
topic2),
+partitions(tp(topic, 0), tp(topic2, 1), tp(topic2, 2)),
currentGeneration - 2, 1));
+
+Map> assignment =
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639: A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550: URL: https://github.com/apache/kafka/pull/13550#discussion_r1177937393 ## clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignorTest.java: ## @@ -42,10 +30,22 @@ import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.ValueSource; +import java.nio.ByteBuffer; Review Comment: oh yes, will do it thanks for my IDE's import optimization. -- 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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639: A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1178034681
##
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignorTest.java:
##
@@ -1038,6 +1038,96 @@ public void
testPartitionsTransferringOwnershipIncludeThePartitionClaimedByMulti
assertTrue(isFullyBalanced(assignment));
}
+@ParameterizedTest(name = TEST_NAME_WITH_RACK_CONFIG)
+@EnumSource(RackConfig.class)
+public void
testOwnedPartitionsAreStableForConsumerWithMultipleGeneration(RackConfig
rackConfig) {
Review Comment:
Ok I renamed the test to `testEnsurePartitionsAssignedToHighestGeneration`
as the goal of this test is to make sure partitions are always assigned to the
member with the highest generation.
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639: A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1178121430
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java:
##
@@ -149,47 +148,51 @@ private boolean allSubscriptionsEqual(Set
allTopics,
}
MemberData memberData = memberData(subscription);
+maxGeneration = Math.max(maxGeneration,
memberData.generation.orElse(DEFAULT_GENERATION));
List ownedPartitions = new ArrayList<>();
consumerToOwnedPartitions.put(consumer, ownedPartitions);
-// Only consider this consumer's owned partitions as valid if it
is a member of the current highest
-// generation, or it's generation is not present but we have not
seen any known generation so far
-if (memberData.generation.isPresent() &&
memberData.generation.get() >= maxGeneration
-|| !memberData.generation.isPresent() && maxGeneration ==
DEFAULT_GENERATION) {
-
-// If the current member's generation is higher, all the
previously owned partitions are invalid
-if (memberData.generation.isPresent() &&
memberData.generation.get() > maxGeneration) {
-allPreviousPartitionsToOwner.clear();
-partitionsWithMultiplePreviousOwners.clear();
-for (String droppedOutConsumer :
membersOfCurrentHighestGeneration) {
-
consumerToOwnedPartitions.get(droppedOutConsumer).clear();
-}
-
-membersOfCurrentHighestGeneration.clear();
-maxGeneration = memberData.generation.get();
-}
+// the member has a valid generation, so we can consider its owned
partitions if it has the highest
+// generation amongst
+for (final TopicPartition tp : memberData.partitions) {
+if (allTopics.contains(tp.topic())) {
+String otherConsumer =
allPreviousPartitionsToOwner.put(tp, consumer);
+if (otherConsumer == null) {
+// this partition is not owned by other consumer in
the same generation
+ownedPartitions.add(tp);
+} else {
+final int memberGeneration =
memberData.generation.orElse(DEFAULT_GENERATION);
+final int otherMemberGeneration =
subscriptions.get(otherConsumer).generationId().orElse(DEFAULT_GENERATION);
+
+if (memberGeneration == otherMemberGeneration) {
+if
(subscriptions.get(otherConsumer).generationId().orElse(DEFAULT_GENERATION) ==
memberData.generation.orElse(DEFAULT_GENERATION)) {
+log.error("Found multiple consumers {} and {}
claiming the same TopicPartition {} in the "
++ "same generation {}, this will be
invalidated and removed from their previous assignment.",
+consumer, otherConsumer, tp,
memberGeneration);
+partitionsWithMultiplePreviousOwners.add(tp);
Review Comment:
From KAFKA-12984:
```
...the assignor will now explicitly look out for partitions that are being
claimed by multiple consumers ... we have to invalidate this partition from the
ownedPartitions of both consumers, since we can't tell who, if anyone, has the
valid claim to this partition.
```
--
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
[GitHub] [kafka] philipnee commented on a diff in pull request #13550: KAFKA-14639: A single partition may be revoked and assign during a single round of rebalance
philipnee commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1179185376
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java:
##
@@ -149,47 +148,57 @@ private boolean allSubscriptionsEqual(Set
allTopics,
}
MemberData memberData = memberData(subscription);
+final int memberGeneration =
memberData.generation.orElse(DEFAULT_GENERATION);
+maxGeneration = Math.max(maxGeneration, memberGeneration);
List ownedPartitions = new ArrayList<>();
consumerToOwnedPartitions.put(consumer, ownedPartitions);
-// Only consider this consumer's owned partitions as valid if it
is a member of the current highest
-// generation, or it's generation is not present but we have not
seen any known generation so far
-if (memberData.generation.isPresent() &&
memberData.generation.get() >= maxGeneration
-|| !memberData.generation.isPresent() && maxGeneration ==
DEFAULT_GENERATION) {
-
-// If the current member's generation is higher, all the
previously owned partitions are invalid
-if (memberData.generation.isPresent() &&
memberData.generation.get() > maxGeneration) {
-allPreviousPartitionsToOwner.clear();
-partitionsWithMultiplePreviousOwners.clear();
-for (String droppedOutConsumer :
membersOfCurrentHighestGeneration) {
-
consumerToOwnedPartitions.get(droppedOutConsumer).clear();
-}
-
-membersOfCurrentHighestGeneration.clear();
-maxGeneration = memberData.generation.get();
-}
+// the member has a valid generation, so we can consider its owned
partitions if it has the highest
+// generation amongst
+for (final TopicPartition tp : memberData.partitions) {
+if (allTopics.contains(tp.topic())) {
+String otherConsumer =
allPreviousPartitionsToOwner.put(tp, consumer);
+if (otherConsumer == null) {
+// this partition is not owned by other consumer in
the same generation
+ownedPartitions.add(tp);
+} else {
+final int otherMemberGeneration =
subscriptions.get(otherConsumer).generationId().orElse(DEFAULT_GENERATION);
-membersOfCurrentHighestGeneration.add(consumer);
-for (final TopicPartition tp : memberData.partitions) {
-// filter out any topics that no longer exist or aren't
part of the current subscription
-if (allTopics.contains(tp.topic())) {
-String otherConsumer =
allPreviousPartitionsToOwner.put(tp, consumer);
-if (otherConsumer == null) {
-// this partition is not owned by other consumer
in the same generation
-ownedPartitions.add(tp);
-} else {
+if (memberGeneration == otherMemberGeneration) {
log.error("Found multiple consumers {} and {}
claiming the same TopicPartition {} in the "
-+ "same generation {}, this will be
invalidated and removed from their previous assignment.",
- consumer, otherConsumer, tp,
maxGeneration);
-
consumerToOwnedPartitions.get(otherConsumer).remove(tp);
++ "same generation {}, this will
be invalidated and removed from their previous assignment.",
+consumer, otherConsumer, tp,
memberGeneration);
partitionsWithMultiplePreviousOwners.add(tp);
+
consumerToOwnedPartitions.get(otherConsumer).remove(tp);
+allPreviousPartitionsToOwner.put(tp, consumer);
+} else if (memberGeneration > otherMemberGeneration) {
+// move partition from the member with an older
generation to the member with the newer generation
+consumerToOwnedPartitions.get(consumer).add(tp);
+
consumerToOwnedPartitions.get(otherConsumer).remove(tp);
+allPreviousPartitionsToOwner.put(tp, consumer);
+// if memberGeneration > otherMemberGeneration,
the other member continue owns the generation
+log.warn("{} in generation {} and {} in generation
{} claiming the same TopicPartition {} in " +
+"different generations. The topic
partition wil be assigned to the member with " +
+
