private CompletableFuture<Void> assignPartitions(
            SortedSet<TopicIdPartition> assignedPartitions,
            SortedSet<TopicPartition> addedPartitions) {

        // Update assignment in the subscription state, and ensure that no fetching or positions
        // initialization happens for the newly added partitions while the callback runs.
        updateSubscriptionAwaitingCallback(assignedPartitions, addedPartitions);

        // Invoke user call back.
        CompletableFuture<Void> result = invokeOnPartitionsAssignedCallback(addedPartitions);
        result.whenComplete((__, exception) -> {
            if (exception == null) {
                // Enable newly added partitions to start fetching and updating positions for them.
                subscriptions.enablePartitionsAwaitingCallback(addedPartitions);
            } else {
                // Keeping newly added partitions as non-fetchable after the callback failure.
                // They will be retried on the next reconciliation loop, until it succeeds or the
                // broker removes them from the assignment.
                if (!addedPartitions.isEmpty()) {
                    log.warn("Leaving newly assigned partitions {} marked as non-fetchable and not " +
                            "requiring initializing positions after onPartitionsAssigned callback failed.",
                        addedPartitions, exception);
                }
            }
        });

        // Clear topic names cache, removing topics that are not assigned to the member anymore.
        Set<String> assignedTopics = assignedPartitions.stream().map(TopicIdPartition::topic).collect(Collectors.toSet());
        assignedTopicNamesCache.values().retainAll(assignedTopics);

        return result;
    }
Make new assignment effective and trigger onPartitionsAssigned callback for the partitions
added. This will also update the local topic names cache, removing from it all topics that
are not assigned to the member anymore. This also ensures that records are not fetched and
positions are not initialized for the newly added partitions until the callback completes.

@param assignedPartitions Full assignment that will be updated in the member subscription
                          state. This includes previously owned and newly added partitions.
@param addedPartitions    Partitions contained in the new assignment that were not owned by
                          the member before. These will be provided to the
                          onPartitionsAssigned callback.
@return Future that will complete when the callback execution completes.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java