[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 (via GitHub)" <gi...@apache.org>]

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<String> allTopics,
             }
 
             MemberData memberData = memberData(subscription);
+            maxGeneration = Math.max(maxGeneration, memberData.generation.orElse(DEFAULT_GENERATION));
 
             List<TopicPartition> 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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org