private LegacyTaskAssignor createTaskAssignor(final boolean lagComputationSuccessful) {
        final LegacyTaskAssignor taskAssignor = legacyTaskAssignorSupplier.get();
        if (taskAssignor instanceof LegacyStickyTaskAssignor) {
            // special case: to preserve pre-existing behavior, we invoke the LegacyStickyTaskAssignor
            // whether or not lag computation failed.
            return taskAssignor;
        } else if (lagComputationSuccessful) {
            return taskAssignor;
        } else {
            log.info("Failed to fetch end offsets for changelogs, will return previous assignment to clients and "
                         + "trigger another rebalance to retry.");
            return new FallbackPriorTaskAssignor();
        }
    }
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