showuon commented on a change in pull request #10985:
URL: https://github.com/apache/kafka/pull/10985#discussion_r665386121
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java
##########
@@ -130,19 +146,26 @@ private boolean allSubscriptionsEqual(Set<String>
allTopics,
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())) {
- ownedPartitions.add(tp);
+
+ if (!allPreviousPartitionsToOwner.containsKey(tp)) {
+ allPreviousPartitionsToOwner.put(tp, consumer);
+ ownedPartitions.add(tp);
+ } else {
+ String otherConsumer =
allPreviousPartitionsToOwner.get(tp);
+ log.warn("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);
Review comment:
Should we log the generation number `maxGeneration` ?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java
##########
@@ -238,32 +275,51 @@ private boolean allSubscriptionsEqual(Set<String>
allTopics,
Iterator<String> unfilledConsumerIter = unfilledMembers.iterator();
// Round-Robin filling remaining members up to the expected numbers of
maxQuota, otherwise, to minQuota
for (TopicPartition unassignedPartition : unassignedPartitions) {
- if (!unfilledConsumerIter.hasNext()) {
- if (unfilledMembers.isEmpty()) {
- // Should not enter here since we have calculated the
exact number to assign to each consumer
- // There might be issues in the assigning algorithm, or
maybe assigning the same partition to two owners.
+ String consumer;
+ if (unfilledConsumerIter.hasNext()) {
+ consumer = unfilledConsumerIter.next();
+ } else {
+ if (unfilledMembers.isEmpty() &&
potentiallyUnfilledMembersAtMinQuota.isEmpty()) {
+ // Should not enter here since we have calculated the
exact number to assign to each consumer.
+ // This indicates issues in the assignment algorithm
int currentPartitionIndex =
unassignedPartitions.indexOf(unassignedPartition);
log.error("No more unfilled consumers to be assigned. The
remaining unassigned partitions are: {}",
- unassignedPartitions.subList(currentPartitionIndex,
unassignedPartitions.size()));
+
unassignedPartitions.subList(currentPartitionIndex,
unassignedPartitions.size()));
throw new IllegalStateException("No more unfilled
consumers to be assigned.");
+ } else if (unfilledMembers.isEmpty()) {
+ consumer = potentiallyUnfilledMembersAtMinQuota.poll();
+ } else {
+ unfilledConsumerIter = unfilledMembers.iterator();
+ consumer = unfilledConsumerIter.next();
}
- unfilledConsumerIter = unfilledMembers.iterator();
}
- String consumer = unfilledConsumerIter.next();
+
List<TopicPartition> consumerAssignment = assignment.get(consumer);
consumerAssignment.add(unassignedPartition);
// We already assigned all possible ownedPartitions, so we know
this must be newly assigned to this consumer
- if (allRevokedPartitions.contains(unassignedPartition))
+ // or else the partition was actually claimed by multiple previous
owners and had to be invalidated from all
+ // members claimed ownedPartitions
+ if (allRevokedPartitions.contains(unassignedPartition) ||
partitionsWithMultiplePreviousOwners.contains(unassignedPartition))
partitionsTransferringOwnership.put(unassignedPartition,
consumer);
int currentAssignedCount = consumerAssignment.size();
int expectedAssignedCount = numMembersAssignedOverMinQuota <
expectedNumMembersAssignedOverMinQuota ? maxQuota : minQuota;
Review comment:
`expectedAssignedCount` is not used any more.
--
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
