kirktrue commented on code in PR #14357: URL: https://github.com/apache/kafka/pull/14357#discussion_r1361294748
########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/AssignmentReconciler.java: ########## @@ -0,0 +1,328 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.clients.consumer.internals; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.internals.events.BackgroundEvent; +import org.apache.kafka.clients.consumer.internals.events.RebalanceListenerInvokedEvent; +import org.apache.kafka.clients.consumer.internals.events.RebalanceStartedEvent; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData.Assignment; +import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData.TopicPartitions; +import org.apache.kafka.clients.consumer.internals.Utils.TopicPartitionComparator; +import org.apache.kafka.common.utils.LogContext; +import org.slf4j.Logger; + +import java.time.Duration; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.SortedSet; +import java.util.TreeSet; +import java.util.concurrent.BlockingQueue; + +import static org.apache.kafka.clients.consumer.internals.RebalanceStep.ASSIGN; +import static org.apache.kafka.clients.consumer.internals.RebalanceStep.LOSE; +import static org.apache.kafka.clients.consumer.internals.RebalanceStep.REVOKE; + +/** + * {@code AssignmentReconciler} performs the work of reconciling this consumer's partition assignment as directed + * by the consumer group coordinator. When the coordinator determines that a change to the partition ownership of + * the group is required, it will communicate with each consumer to relay its respective <em>target</em> + * assignment, that is, the set of partitions for which that consumer should now assume ownership. It is the then the + * responsibility of the consumer to work toward that target by performing the necessary internal modifications to + * satisfy the assignment from the coordinator. In practical terms, this means that it must first determine the set + * difference between the <em>{@link SubscriptionState#assignedPartitions() current assignment}</em> and the + * <em>{@link Assignment#assignedTopicPartitions() target assignment}</em>. + * + * <p/> + * + * Internally, reconciliation requires the following steps: + * + * <ol> + * <li> + * Calculate the partitions to revoke; these are the partitions in the current assignment that are + * <em>not in</em> the target assignment + * </li> + * <li> + * Send a {@link RebalanceListenerInvokedEvent} so that the application thread will execute the + * {@link ConsumerRebalanceListener#onPartitionsRevoked(Collection)} callback method + * </li> + * <li> + * On confirmation of the callback method execution, remove the revoked partitions from the + * {@link SubscriptionState#assignFromSubscribed(Collection) current assignment} + * </li> + * <li> + * Signal to the heartbeat request manager to perform a heartbeat acknowledgement with the group coordinator Review Comment: The code has been changed to perform callbacks for both revoke and assign in one shot, with no heartbeats between them, as discussed above. -- 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
