You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dajac (via GitHub)" <gi...@apache.org> on 2023/04/26 12:40:51 UTC
[GitHub] [kafka] dajac commented on a diff in pull request #13550: KAFKA-14639: A single partition may be revoked and assign during a single round of rebalance
dajac commented on code in PR #13550:
URL: https://github.com/apache/kafka/pull/13550#discussion_r1177783905
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java:
##########
@@ -137,7 +136,7 @@ private boolean allSubscriptionsEqual(Set<String> allTopics,
Map<TopicPartition, String> allPreviousPartitionsToOwner = new HashMap<>();
for (Map.Entry<String, Subscription> subscriptionEntry : subscriptions.entrySet()) {
- String consumer = subscriptionEntry.getKey();
+ final String consumer = subscriptionEntry.getKey();
Subscription subscription = subscriptionEntry.getValue();
Review Comment:
nit: While here, should this one be final as well?
##########
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:
So my understanding is that partitions put in `partitionsWithMultiplePreviousOwners` will be unassigned from all consumers claiming them. Is my understanding correct?
##########
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));
Review Comment:
nit: Should we define a variable for `memberData.generation.orElse(DEFAULT_GENERATION)`? The same code is reused later.
##########
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:
nit: Could we revert this as it is not related to the fix?
##########
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:
I actually wonder if this was expected or if it is just a typo in the test case because there is also `testOwnedPartitionsAreInvalidatedForConsumerWithMultipleGeneration`.
##########
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)) {
Review Comment:
Aren't those two if statements the same?
##########
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<String, List<PartitionInfo>> 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<String, List<TopicPartition>> 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<String, List<PartitionInfo>> 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<String, List<TopicPartition>> 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<String, List<PartitionInfo>> 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<String, List<TopicPartition>> assignment = assignor.assignPartitions(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));
+ }
+
Review Comment:
Should we also add a test for the case where a partition is claimed by two consumers with the same generation?
##########
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:
nit: It seems that we also test the case where a partition is claimed by two consumers with different generation in this test. Should we try to reflect this in its name?
##########
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<String, List<PartitionInfo>> 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<String, List<TopicPartition>> 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<String, List<PartitionInfo>> 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<String, List<TopicPartition>> 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))));
Review Comment:
nit: It may be better to verify the full assignment for completeness.
##########
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);
+ }
+ consumerToOwnedPartitions.get(otherConsumer).remove(tp);
+ allPreviousPartitionsToOwner.put(tp, consumer);
+ continue;
+ }
- 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);
+ 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);
- partitionsWithMultiplePreviousOwners.add(tp);
+ allPreviousPartitionsToOwner.put(tp, consumer);
}
+
+ // if memberGeneration < otherMemberGeneration, the other member continue owns the generation
+ log.warn("Found multiple members {} and {} claiming the same TopicPartition {} in " +
+ "different generations. The topic partition wil be assigned to the member with " +
+ "the higher generation {}.",
Review Comment:
Should we add the generation for each members? This will also tell us which member got it in the end. I would also use `consumers` instead of `members` in order to be consistent with the other logs.
##########
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);
+ }
+ consumerToOwnedPartitions.get(otherConsumer).remove(tp);
+ allPreviousPartitionsToOwner.put(tp, consumer);
+ continue;
Review Comment:
This does not look nice. Should we remove it and use `else if`?
--
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