void assignStandbyTasksToClientsWithDifferentTags(final int numberOfStandbyClients,
                                                      final ConstrainedPrioritySet standbyTaskClientsByTaskLoad,
                                                      final TaskId activeTaskId,
                                                      final ProcessId activeTaskClient,
                                                      final Set<String> rackAwareAssignmentTags,
                                                      final Map<ProcessId, ClientState> clientStates,
                                                      final Map<TaskId, Integer> tasksToRemainingStandbys,
                                                      final Map<String, Set<String>> tagKeyToValues,
                                                      final Map<TagEntry, Set<ProcessId>> tagEntryToClients,
                                                      final Map<TaskId, ProcessId> pendingStandbyTasksToClientId) {
        standbyTaskClientsByTaskLoad.offerAll(clientStates.keySet());

        // We set countOfUsedClients as 1 because client where active task is located has to be considered as used.
        int countOfUsedClients = 1;
        int numRemainingStandbys = tasksToRemainingStandbys.get(activeTaskId);

        final Map<TagEntry, Set<ProcessId>> tagEntryToUsedClients = new HashMap<>();

        ProcessId lastUsedClient = activeTaskClient;
        do {
            updateClientsOnAlreadyUsedTagEntries(
                lastUsedClient,
                countOfUsedClients,
                rackAwareAssignmentTags,
                clientStates,
                tagEntryToClients,
                tagKeyToValues,
                tagEntryToUsedClients
            );

            final ProcessId clientOnUnusedTagDimensions = standbyTaskClientsByTaskLoad.poll(
                activeTaskId, uuid -> !isClientUsedOnAnyOfTheTagEntries(uuid, tagEntryToUsedClients)
            );

            if (clientOnUnusedTagDimensions == null) {
                break;
            }

            final ClientState clientStateOnUsedTagDimensions = clientStates.get(clientOnUnusedTagDimensions);
            countOfUsedClients++;
            numRemainingStandbys--;

            log.debug("Assigning {} out of {} standby tasks for an active task [{}] with client tags {}. " +
                    "Standby task client tags are {}.",
                    numberOfStandbyClients - numRemainingStandbys, numberOfStandbyClients, activeTaskId,
                    clientTagFunction.apply(activeTaskClient, clientStates.get(activeTaskClient)),
                    clientTagFunction.apply(clientStateOnUsedTagDimensions.processId(), clientStateOnUsedTagDimensions));

            clientStateOnUsedTagDimensions.assignStandby(activeTaskId);
            lastUsedClient = clientOnUnusedTagDimensions;
        } while (numRemainingStandbys > 0);

        if (numRemainingStandbys > 0) {
            pendingStandbyTasksToClientId.put(activeTaskId, activeTaskClient);
            tasksToRemainingStandbys.put(activeTaskId, numRemainingStandbys);
            log.warn("Rack aware standby task assignment was not able to assign {} of {} standby tasks for the " +
                     "active task [{}] with the rack aware assignment tags {}. " +
                     "This may happen when there aren't enough application instances on different tag " +
                     "dimensions compared to an active and corresponding standby task. " +
                     "Consider launching application instances on different tag dimensions than [{}]. " +
                     "Standby task assignment will fall back to assigning standby tasks to the least loaded clients.",
                     numRemainingStandbys, numberOfStandbyClients,
                     activeTaskId, rackAwareAssignmentTags,
                     clientTagFunction.apply(activeTaskClient, clientStates.get(activeTaskClient)));

        } else {
            tasksToRemainingStandbys.remove(activeTaskId);
        }
    }
Whether one task can be moved from source to destination. If the number of distinct tags including active
and standby after the movement isn't decreased, then we can move the task. Otherwise, we can not move
the task.
@param source Source client
@param destination Destination client
@param sourceTask Task to move
@param clientStateMap All client metadata
@return If the task can be moved
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientTagAwareStandbyTaskAssignor.java