lucasbru commented on code in PR #15271: URL: https://github.com/apache/kafka/pull/15271#discussion_r1472523826
########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java: ########## @@ -1350,21 +1321,33 @@ public Map<Uuid, SortedSet<Integer>> currentAssignment() { return this.currentAssignment; } - /** * @return Set of topic IDs received in a target assignment that have not been reconciled yet - * because topic names are not in metadata. Visible for testing. + * because topic names are not in metadata or reconciliation hasn't finished. Reconciliation + * hasn't finished for a topic if the currently active assignment has a different set of partitions + * for the topic than the target assignment. + * + * Visible for testing. */ - Set<Uuid> topicsWaitingForMetadata() { - return Collections.unmodifiableSet(assignmentUnresolved.keySet()); + Set<Uuid> topicsAwaitingReconciliation() { + return topicPartitionsAwaitingReconciliation().keySet(); } /** - * @return Topic partitions received in a target assignment that have been resolved in - * metadata and are ready to be reconciled. Visible for testing. + * @return Map of topics partitions received in a target assignment that have not been + * reconciled yet because topic names are not in metadata or reconciliation hasn't finished. + * The value will always contain all partitions in the target assignment. + * + * Visible for testing. */ - Set<TopicIdPartition> assignmentReadyToReconcile() { - return Collections.unmodifiableSet(assignmentReadyToReconcile); + Map<Uuid, SortedSet<Integer>> topicPartitionsAwaitingReconciliation() { + final Map<Uuid, SortedSet<Integer>> topicPartitionMap = new HashMap<>(); + currentTargetAssignment.forEach((x, y) -> { + if (!currentAssignment.containsKey(x) || !currentAssignment.get(x).equals(y)) { Review Comment: Yes, it's only used for testing. This method is a replacement for `assignmentReadyToReconcile` which was also used only in testing. You are right that semantics were a bit weird, I implemented so that it replicates the behavior of the (old) internal state. I changed it to have more reasonable semantics (the value contains only the partitions that are missing from the currently reconciled assignment). For the existing tests, there is no difference between the two semantics. ########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java: ########## @@ -1032,17 +1011,24 @@ private void resolveMetadataForUnresolvedAssignment() { Optional<String> nameFromMetadata = findTopicNameInGlobalOrLocalCache(topicId); nameFromMetadata.ifPresent(resolvedTopicName -> { // Name resolved, so assignment is ready for reconciliation. - addToAssignmentReadyToReconcile(topicId, resolvedTopicName, topicPartitions); + topicPartitions.forEach(tp -> { + TopicIdPartition topicIdPartition = new TopicIdPartition( + topicId, + new TopicPartition(resolvedTopicName, tp)); Review Comment: Done ########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java: ########## @@ -788,23 +767,28 @@ public void transitionToStale() { } /** - * Reconcile the assignment that has been received from the server and for which topic names - * are resolved, kept in the {@link #assignmentReadyToReconcile}. This will commit if needed, - * trigger the callbacks and update the subscription state. Note that only one reconciliation + * Reconcile the assignment that has been received from the server. If for some topics, the + * topic ID cannot be matched to a topic name, a metadata update will be triggered and only + * the subset of topics that are resolvable will be reconciled. Reconciliation will trigger the + * callbacks and update the subscription state. Note that only one reconciliation * can be in progress at a time. If there is already another one in progress when this is * triggered, it will be no-op, and the assignment will be reconciled on the next * reconciliation loop. */ - boolean reconcile() { + void reconcile() { + if (targetAssignmentReconciled()) { + log.debug("Ignoring reconciliation attempt. Target assignment is equal to the " + + "current assignment."); + return; + } if (reconciliationInProgress) { log.debug("Ignoring reconciliation attempt. Another reconciliation is already in progress. Assignment " + - assignmentReadyToReconcile + " will be handled in the next reconciliation loop."); - return false; + currentTargetAssignment + " will be handled in the next reconciliation loop."); + return; } - // Make copy of the assignment to reconcile as it could change as new assignments or metadata updates are received - SortedSet<TopicIdPartition> assignedTopicIdPartitions = new TreeSet<>(TOPIC_ID_PARTITION_COMPARATOR); - assignedTopicIdPartitions.addAll(assignmentReadyToReconcile); + // Resolve metadata for target assignment + SortedSet<TopicIdPartition> assignedTopicIdPartitions = resolveMetadataForTargetAssignment(); Review Comment: Done -- 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