private UserTaskAssignmentListener assignTasksToClients(final Cluster fullMetadata,
                                                            final GroupSubscription groupSubscription,
                                                            final Set<String> allSourceTopics,
                                                            final Map<Subtopology, TopicsInfo> topicGroups,
                                                            final Map<ProcessId, ClientMetadata> clientMetadataMap,
                                                            final Map<TaskId, Set<TopicPartition>> partitionsForTask,
                                                            final Map<ProcessId, Map<String, Optional<String>>> racksForProcessConsumer,
                                                            final Set<TaskId> statefulTasks) {
        if (!statefulTasks.isEmpty()) {
            throw new TaskAssignmentException("The stateful tasks should not be populated before assigning tasks to clients");
        }

        final Map<TopicPartition, TaskId> taskForPartition = new HashMap<>();
        final Map<Subtopology, Set<TaskId>> tasksForTopicGroup = new HashMap<>();
        populateTasksForMaps(taskForPartition, tasksForTopicGroup, allSourceTopics, partitionsForTask, fullMetadata);

        final ChangelogTopics changelogTopics = new ChangelogTopics(
            internalTopicManager,
            topicGroups,
            tasksForTopicGroup,
            logPrefix
        );
        changelogTopics.setup();

        final Map<ProcessId, ClientState> clientStates = new HashMap<>();
        final boolean lagComputationSuccessful =
            populateClientStatesMap(clientStates, clientMetadataMap, taskForPartition, changelogTopics);


        log.info("{} client nodes and {} consumers participating in this rebalance: \n{}.",
                 clientStates.size(),
                 clientStates.values().stream().map(ClientState::capacity).reduce(Integer::sum).orElse(0),
                 clientStates.entrySet().stream()
                     .sorted(comparingByKey())
                     .map(entry -> entry.getKey() + ": " + entry.getValue().consumers())
                     .collect(Collectors.joining(Utils.NL)));

        final Set<TaskId> allTasks = partitionsForTask.keySet();
        statefulTasks.addAll(changelogTopics.statefulTaskIds());

        log.info("Assigning stateful tasks: {}\n"
                     + "and stateless tasks: {}",
                 statefulTasks,
                 allTasks.stream().filter(t -> !statefulTasks.contains(t)).collect(Collectors.toSet()));
        log.debug("Assigning tasks and {} standby replicas to client nodes {}",
                  numStandbyReplicas(), clientStates);

        final Optional<org.apache.kafka.streams.processor.assignment.TaskAssignor> userTaskAssignor =
            customTaskAssignorSupplier.get();
        final UserTaskAssignmentListener customTaskAssignmentListener;
        if (userTaskAssignor.isPresent()) {
            final ApplicationState applicationState = buildApplicationState(
                taskManager.topologyMetadata(),
                clientMetadataMap,
                topicGroups,
                fullMetadata
            );
            final org.apache.kafka.streams.processor.assignment.TaskAssignor assignor = userTaskAssignor.get();
            final TaskAssignment taskAssignment = assignor.assign(applicationState);
            final AssignmentError assignmentError = TaskAssignmentUtils.validateTaskAssignment(applicationState, taskAssignment);
            processStreamsPartitionAssignment(assignor, taskAssignment, assignmentError, clientMetadataMap, groupSubscription);
            customTaskAssignmentListener = (assignment, subscription) -> {
                assignor.onAssignmentComputed(assignment, subscription, assignmentError);
                if (assignmentError != AssignmentError.NONE) {
                    log.error("Rebalance failed due to task assignor returning assignment with error {}", assignmentError);
                    throw new StreamsException("Task assignment with " + assignor.getClass().getName() +
                                               " returned an error: " + assignmentError);
                }
            };
        } else {
            customTaskAssignmentListener = (assignment, subscription) -> { };
            final LegacyTaskAssignor taskAssignor = createTaskAssignor(lagComputationSuccessful);
            final RackAwareTaskAssignor rackAwareTaskAssignor = new RackAwareTaskAssignor(
                fullMetadata,
                partitionsForTask,
                changelogTopics.changelogPartionsForTask(),
                tasksForTopicGroup,
                racksForProcessConsumer,
                internalTopicManager,
                assignmentConfigs,
                time
            );
            final boolean probingRebalanceNeeded = taskAssignor.assign(clientStates,
                allTasks,
                statefulTasks,
                rackAwareTaskAssignor,
                assignmentConfigs);
            if (probingRebalanceNeeded) {
                // Arbitrarily choose the leader's client to be responsible for triggering the probing rebalance,
                // note once we pick the first consumer within the process to trigger probing rebalance, other consumer
                // would not set to trigger any more.
                final ClientMetadata rebalanceClientMetadata = clientMetadataMap.get(taskManager.processId());
                if (rebalanceClientMetadata != null) {
                    final Instant rebalanceDeadline = Instant.ofEpochMilli(time.milliseconds() + probingRebalanceIntervalMs());
                    rebalanceClientMetadata.state.setFollowupRebalanceDeadline(rebalanceDeadline);
                }
            }
        }

        // Break this up into multiple logs to make sure the summary info gets through, which helps avoid
        // info loss for example due to long line truncation with large apps
        log.info("Assigned {} total tasks including {} stateful tasks to {} client nodes.",
                 allTasks.size(),
                 statefulTasks.size(),
                 clientStates.size());
        log.info("Assignment of tasks to nodes: {}",
                 clientStates.entrySet().stream()
                     .sorted(comparingByKey())
                     .map(entry -> entry.getKey() + "=" + entry.getValue().currentAssignment())
                     .collect(Collectors.joining(Utils.NL)));
        return customTaskAssignmentListener;
    }
Assigns a set of tasks to each client (Streams instance) using the configured task assignor, and also
populate the stateful tasks that have been assigned to the clients
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java