protected List<Task> readTasksFromCoordinatorStream(CoordinatorStreamSystemConsumer consumer) {
    CoordinatorStreamStore coordinatorStreamStore = new CoordinatorStreamStore(consumer.getConfig(), new MetricsRegistryMap());
    LocalityManager localityManager = new LocalityManager(coordinatorStreamStore);
    Map<String, ProcessorLocality> containerLocalities = localityManager.readLocality().getProcessorLocalities();
    TaskAssignmentManager taskAssignmentManager = new TaskAssignmentManager(new NamespaceAwareCoordinatorStreamStore(coordinatorStreamStore, SetTaskContainerMapping.TYPE), new NamespaceAwareCoordinatorStreamStore(coordinatorStreamStore, SetTaskModeMapping.TYPE));
    Map<String, String> taskNameToContainerIdMapping = taskAssignmentManager.readTaskAssignment();
    StorageConfig storageConfig = new StorageConfig(consumer.getConfig());
    List<String> storeNames = storageConfig.getStoreNames();
    return taskNameToContainerIdMapping.entrySet()
        .stream()
        .map(entry -> {
          String hostName = Optional.ofNullable(containerLocalities.get(entry.getValue()))
              .map(ProcessorLocality::host)
              .orElse(null);
          return new Task(hostName, entry.getKey(), entry.getValue(), new ArrayList<>(), storeNames);
        }).collect(Collectors.toList());
  }
Builds list of {@link Task} from job model in coordinator stream.
@param consumer system consumer associated with a job's coordinator stream.
@return list of {@link Task} constructed from job model in coordinator stream.
samza-rest/src/main/java/org/apache/samza/rest/proxy/task/SamzaTaskProxy.java