private boolean populateClientStatesMap(final Map<ProcessId, ClientState> clientStates,
                                            final Map<ProcessId, ClientMetadata> clientMetadataMap,
                                            final Map<TopicPartition, TaskId> taskForPartition,
                                            final ChangelogTopics changelogTopics) {
        boolean fetchEndOffsetsSuccessful;
        Map<TaskId, Long> allTaskEndOffsetSums;
        try {
            // Make the listOffsets request first so it can  fetch the offsets for non-source changelogs
            // asynchronously while we use the blocking Consumer#committed call to fetch source-changelog offsets;
            // note that we would need to wrap all exceptions as Streams exception with partition-level fine-grained
            // error messages
            final ListOffsetsResult endOffsetsResult =
                fetchEndOffsetsResult(changelogTopics.preExistingNonSourceTopicBasedPartitions(), adminClient);

            final Map<TopicPartition, Long> sourceChangelogEndOffsets =
                fetchCommittedOffsets(changelogTopics.preExistingSourceTopicBasedPartitions(), mainConsumerSupplier.get());

            final Map<TopicPartition, ListOffsetsResultInfo> endOffsets = ClientUtils.getEndOffsets(
                endOffsetsResult, changelogTopics.preExistingNonSourceTopicBasedPartitions());

            allTaskEndOffsetSums = computeEndOffsetSumsByTask(
                endOffsets,
                sourceChangelogEndOffsets,
                changelogTopics
            );
            fetchEndOffsetsSuccessful = true;
        } catch (final StreamsException | TimeoutException e) {
            log.info("Failed to retrieve all end offsets for changelogs, and hence could not calculate the per-task lag; " +
                "this is not a fatal error but would cause the assignor to fallback to a naive algorithm", e);
            allTaskEndOffsetSums = changelogTopics.statefulTaskIds().stream().collect(Collectors.toMap(t -> t, t -> UNKNOWN_OFFSET_SUM));
            fetchEndOffsetsSuccessful = false;
        }

        for (final Map.Entry<ProcessId, ClientMetadata> entry : clientMetadataMap.entrySet()) {
            final ProcessId processId = entry.getKey();
            final ClientState state = entry.getValue().state;
            state.initializePrevTasks(taskForPartition, taskManager.topologyMetadata().hasNamedTopologies());

            state.computeTaskLags(processId, allTaskEndOffsetSums);
            clientStates.put(processId, state);
        }

        return fetchEndOffsetsSuccessful;
    }
Builds a map from client to state, and readies each ClientState for assignment by adding any missing prev tasks
and computing the per-task overall lag based on the fetched end offsets for each changelog.

@param clientStates a map from each client to its state, including offset lags. Populated by this method.
@param clientMetadataMap a map from each client to its full metadata
@param taskForPartition map from topic partition to its corresponding task
@param changelogTopics object that manages changelog topics

@return whether we were able to successfully fetch the changelog end offsets and compute each client's lag
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java