private void updateClientsOnAlreadyUsedTagEntries(final ProcessId usedClient,
                                                             final int countOfUsedClients,
                                                             final Set<String> rackAwareAssignmentTags,
                                                             final Map<ProcessId, ClientState> clientStates,
                                                             final Map<TagEntry, Set<ProcessId>> tagEntryToClients,
                                                             final Map<String, Set<String>> tagKeyToValues,
                                                             final Map<TagEntry, Set<ProcessId>> tagEntryToUsedClients) {
        final Map<String, String> usedClientTags = clientTagFunction.apply(usedClient, clientStates.get(usedClient));

        for (final Entry<String, String> usedClientTagEntry : usedClientTags.entrySet()) {
            final String tagKey = usedClientTagEntry.getKey();

            if (!rackAwareAssignmentTags.contains(tagKey)) {
                log.warn("Client tag with key [{}] will be ignored when computing rack aware standby " +
                         "task assignment because it is not part of the configured rack awareness [{}].",
                         tagKey, rackAwareAssignmentTags);
                continue;
            }

            final Set<String> allTagValues = tagKeyToValues.get(tagKey);

            // Consider the following client setup where we need to distribute 2 standby tasks for each stateful task.
            //
            // # Kafka Streams Client 1
            // client.tag.zone: eu-central-1a
            // client.tag.cluster: k8s-cluster1
            // rack.aware.assignment.tags: zone,cluster
            //
            // # Kafka Streams Client 2
            // client.tag.zone: eu-central-1b
            // client.tag.cluster: k8s-cluster1
            // rack.aware.assignment.tags: zone,cluster
            //
            // # Kafka Streams Client 3
            // client.tag.zone: eu-central-1c
            // client.tag.cluster: k8s-cluster1
            // rack.aware.assignment.tags: zone,cluster
            //
            // # Kafka Streams Client 4
            // client.tag.zone: eu-central-1a
            // client.tag.cluster: k8s-cluster2
            // rack.aware.assignment.tags: zone,cluster
            //
            // # Kafka Streams Client 5
            // client.tag.zone: eu-central-1b
            // client.tag.cluster: k8s-cluster2
            // rack.aware.assignment.tags: zone,cluster
            //
            // # Kafka Streams Client 6
            // client.tag.zone: eu-central-1c
            // client.tag.cluster: k8s-cluster2
            // rack.aware.assignment.tags: zone,cluster
            //
            // Since we have only two unique `cluster` tag values,
            // we can only achieve "ideal" distribution on the 1st standby task assignment.
            // Ideal distribution for the 1st standby task can be achieved because we can assign standby task
            // to the client located on different cluster and zone compared to an active task.
            // We can't consider the `cluster` tag for the 2nd standby task assignment because the 1st standby
            // task would already be assigned on different cluster compared to the active one, which means
            // we have already used all the available cluster tag values. Taking the `cluster` tag into consideration
            // for the 2nd standby task assignment would effectively mean excluding all the clients.
            // Instead, for the 2nd standby task, we can only achieve partial rack awareness based on the `zone` tag.
            // As we don't consider the `cluster` tag for the 2nd standby task assignment, partial rack awareness
            // can be satisfied by placing the 2nd standby client on a different `zone` tag compared to active and corresponding standby tasks.
            // The `zone` on either `cluster` tags are valid candidates for the partial rack awareness, as our goal is to distribute clients on the different `zone` tags.

            // This statement checks if we have used more clients than the number of unique values for the given tag,
            // and if so, removes those tag entries from the tagEntryToUsedClients map.
            if (allTagValues.size() <= countOfUsedClients) {
                allTagValues.forEach(tagValue -> tagEntryToUsedClients.remove(new TagEntry(tagKey, tagValue)));
            } else {
                final String tagValue = usedClientTagEntry.getValue();
                final TagEntry tagEntry = new TagEntry(tagKey, tagValue);
                final Set<ProcessId> clientsOnUsedTagValue = tagEntryToClients.get(tagEntry);
                tagEntryToUsedClients.put(tagEntry, clientsOnUsedTagValue);
            }
        }
    }
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