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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
