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<String>
allTopics,
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
+ 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
