@Override
    public boolean isAllowedTaskMovement(final ClientState source, final ClientState destination) {
        final Map<String, String> sourceClientTags = clientTagFunction.apply(source.processId(), source);
        final Map<String, String> destinationClientTags = clientTagFunction.apply(destination.processId(), destination);

        for (final Entry<String, String> sourceClientTagEntry : sourceClientTags.entrySet()) {
            if (!sourceClientTagEntry.getValue().equals(destinationClientTags.get(sourceClientTagEntry.getKey()))) {
                return false;
            }
        }

        return true;
    }
The algorithm distributes standby tasks for the {@param statefulTaskIds} over different tag dimensions.
For each stateful task, the number of standby tasks will be assigned based on configured {@link AssignmentConfigs#numStandbyReplicas()}.
Rack aware standby tasks distribution only takes into account tags specified via {@link AssignmentConfigs#rackAwareAssignmentTags()}.
Ideally, all standby tasks for any given stateful task will be located on different tag dimensions to have the best possible distribution.
However, if the ideal (or partially ideal) distribution is impossible, the algorithm will fall back to the least-loaded clients without taking rack awareness constraints into consideration.
The least-loaded clients are determined based on the total number of tasks (active and standby tasks) assigned to the client.
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientTagAwareStandbyTaskAssignor.java