From ef06583388910bfcd925af106af22b2c124c38ed Mon Sep 17 00:00:00 2001 From: Philip Nee Date: Fri, 28 Apr 2023 02:08:32 -0700 Subject: [PATCH] KAFKA-14639: A single partition may be revoked and assign during a single round of rebalance (#13550) KAFKA-14639: A single partition may be revoked and assign during a single round of rebalance (#13550) This is a really long story, but the incident started in KAFKA-13419 when we observed a member sending out a topic partition owned from the previous generation when a member missed a rebalance cycle due to REBALANCE_IN_PROGRESS. This patch changes the AbstractStickyAssignor.AllSubscriptionsEqual method. In short, it should no long check and validate only the highest generation. Instead, we consider 3 cases: 1. Member will continue to hold on to its partition if there are no other owners 2. If there are 1+ owners to the same partition. One with the highest generation will win. 3. If two members of the same generation hold on to the same partition. We will log an error but remove both from the assignment. (Same with the current logic) Here are some important notes that lead to the patch: - If a member is kicked out of the group, and `UNKNOWN_MEMBER_ID` will be thrown. - It seems to be a common situation that members are late to joinGroup and therefore get `REBALANCE_IN_PROGRESS` error. This is why we don't want to reset generation because it might cause lots of revocations and can be disruptive To summarize the current behavior of different errors: `REBALANCE_IN_PROGRESS` - heartbeat: requestRejoin if member state is stable - joinGroup: rejoin immediately - syncGroup: rejoin immediately - commit: requestRejoin and fail the commit. Raise this exception if the generation is staled, i.e. another rebalance is already in progress. `UNKNOWN_MEMBER_ID` - heartbeat: resetStateAndRejoinif generation hasn't changed. otherwise, ignore - joinGroup: resetStateAndRejoin if generation unchanged, otherwise rejoin immediately - syncGroup: resetStateAndRejoin if generation unchanged, otherwise rejoin immediately `ILLEGAL_GENERATION` - heartbeat: resetStateAndRejoinif generation hasn't changed. otherwise, ignore - syncGroup: raised the exception if generation has been resetted or the member hasn't completed rebalancing. then resetStateAndRejoin if generation unchanged, otherwise rejoin immediately Reviewers: David Jacot --- .../internals/AbstractStickyAssignor.java | 73 +++++++++-------- .../internals/AbstractStickyAssignorTest.java | 81 +++++++++++++++++++ 2 files changed, 122 insertions(+), 32 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java index 66d0c2ded0756..467920a423569 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java @@ -96,7 +96,6 @@ private boolean allSubscriptionsEqual(Set allTopics, Map subscriptions, Map> consumerToOwnedPartitions, Set partitionsWithMultiplePreviousOwners) { - Set membersOfCurrentHighestGeneration = new HashSet<>(); boolean isAllSubscriptionsEqual = true; Set subscribedTopics = new HashSet<>(); @@ -106,8 +105,8 @@ private boolean allSubscriptionsEqual(Set allTopics, Map allPreviousPartitionsToOwner = new HashMap<>(); for (Map.Entry subscriptionEntry : subscriptions.entrySet()) { - String consumer = subscriptionEntry.getKey(); - Subscription subscription = subscriptionEntry.getValue(); + final String consumer = subscriptionEntry.getKey(); + final Subscription subscription = subscriptionEntry.getValue(); // initialize the subscribed topics set if this is the first subscription if (subscribedTopics.isEmpty()) { @@ -118,47 +117,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) { + // if two members of the same generation own the same partition, revoke the partition 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 + ownedPartitions.add(tp); + consumerToOwnedPartitions.get(otherConsumer).remove(tp); + allPreviousPartitionsToOwner.put(tp, consumer); + log.warn("Consumer {} in generation {} and consumer {} in generation {} claiming the same " + + "TopicPartition {} in different generations. The topic partition wil be " + + "assigned to the member with the higher generation {}.", + consumer, memberGeneration, + otherConsumer, otherMemberGeneration, + tp, + memberGeneration); + } else { + // let the other member continue to own the topic partition + log.warn("Consumer {} in generation {} and consumer {} in generation {} claiming the same " + + "TopicPartition {} in different generations. The topic partition wil be " + + "assigned to the member with the higher generation {}.", + consumer, memberGeneration, + otherConsumer, otherMemberGeneration, + tp, + otherMemberGeneration); } } } } } - return isAllSubscriptionsEqual; } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignorTest.java index 8a71ca0de5e47..b1147dccda0ae 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignorTest.java @@ -942,6 +942,87 @@ public void testPartitionsTransferringOwnershipIncludeThePartitionClaimedByMulti assertTrue(isFullyBalanced(assignment)); } + public void testEnsurePartitionsAssignedToHighestGeneration() { + Map partitionsPerTopic = new HashMap<>(); + partitionsPerTopic.put(topic, 3); + partitionsPerTopic.put(topic2, 3); + partitionsPerTopic.put(topic3, 3); + + int currentGeneration = 10; + + // ensure partitions are always assigned to the member with the highest generation + subscriptions.put(consumer1, buildSubscriptionV2Above(topics(topic, topic2, topic3), + partitions(tp(topic, 0), tp(topic2, 0), tp(topic3, 0)), currentGeneration)); + subscriptions.put(consumer2, buildSubscriptionV2Above(topics(topic, topic2, topic3), + partitions(tp(topic, 1), tp(topic2, 1), tp(topic3, 1)), currentGeneration - 1)); + subscriptions.put(consumer3, buildSubscriptionV2Above(topics(topic, topic2, topic3), + partitions(tp(topic2, 1), tp(topic3, 0), tp(topic3, 2)), currentGeneration - 2)); + + Map> assignment = assignor.assign(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)); + } + + public void testNoReassignmentOnCurrentMembers() { + Map partitionsPerTopic = new HashMap<>(); + partitionsPerTopic.put(topic, 3); + partitionsPerTopic.put(topic1, 3); + partitionsPerTopic.put(topic2, 3); + partitionsPerTopic.put(topic3, 3); + + int currentGeneration = 10; + + subscriptions.put(consumer1, buildSubscriptionV2Above(topics(topic, topic2, topic3, topic1), + partitions(), DEFAULT_GENERATION)); + subscriptions.put(consumer2, buildSubscriptionV2Above(topics(topic, topic2, topic3, topic1), + partitions(tp(topic, 0), tp(topic2, 0), tp(topic1, 0)), currentGeneration - 1)); + subscriptions.put(consumer3, buildSubscriptionV2Above(topics(topic, topic2, topic3, topic1), + partitions(tp(topic3, 2), tp(topic2, 2), tp(topic1, 1)), currentGeneration - 2)); + subscriptions.put(consumer4, buildSubscriptionV2Above(topics(topic, topic2, topic3, topic1), + partitions(tp(topic3, 1), tp(topic, 1), tp(topic, 2)), currentGeneration - 3)); + + Map> assignment = assignor.assign(partitionsPerTopic, subscriptions); + // ensure assigned partitions don't get reassigned + assertEquals(new HashSet<>(partitions(tp(topic1, 2), tp(topic2, 1), tp(topic3, 0))), + new HashSet<>(assignment.get(consumer1))); + assertTrue(assignor.partitionsTransferringOwnership.isEmpty()); + + verifyValidityAndBalance(subscriptions, assignment, partitionsPerTopic); + assertTrue(isFullyBalanced(assignment)); + } + + @Test + public void testOwnedPartitionsAreInvalidatedForConsumerWithMultipleGeneration() { + Map partitionsPerTopic = new HashMap<>(); + partitionsPerTopic.put(topic, 3); + partitionsPerTopic.put(topic2, 3); + + int currentGeneration = 10; + + subscriptions.put(consumer1, buildSubscriptionV2Above(topics(topic, topic2), + partitions(tp(topic, 0), tp(topic2, 1), tp(topic, 1)), currentGeneration)); + subscriptions.put(consumer2, buildSubscriptionV2Above(topics(topic, topic2), + partitions(tp(topic, 0), tp(topic2, 1), tp(topic2, 2)), currentGeneration - 2)); + + Map> assignment = assignor.assign(partitionsPerTopic, subscriptions); + assertEquals(new HashSet<>(partitions(tp(topic, 0), tp(topic2, 1), tp(topic, 1))), + new HashSet<>(assignment.get(consumer1))); + assertEquals(new HashSet<>(partitions(tp(topic, 2), tp(topic2, 2), tp(topic2, 0))), + new HashSet<>(assignment.get(consumer2))); + assertTrue(assignor.partitionsTransferringOwnership.isEmpty()); + + verifyValidityAndBalance(subscriptions, assignment, partitionsPerTopic); + assertTrue(isFullyBalanced(assignment)); + } + private String getTopicName(int i, int maxNum) { return getCanonicalName("t", i, maxNum); }