public Map<TaskName, Integer> readPartitionMapping() {
    LOG.debug("Reading changelog partition information");
    final Map<TaskName, Integer> changelogMapping = new HashMap<>();
    metadataStore.all().forEach((taskName, partitionIdAsBytes) -> {
      String partitionId = valueSerde.fromBytes(partitionIdAsBytes);
      LOG.debug("TaskName: {} is mapped to {}", taskName, partitionId);
      if (StringUtils.isNotBlank(partitionId)) {
        changelogMapping.put(new TaskName(taskName), Integer.valueOf(partitionId));
      }
    });
    return changelogMapping;
  }
Reads the taskName to changelog partition assignments from the {@link MetadataStore}.

@return TaskName to change LOG partition mapping, or an empty map if there were no messages.
samza-core/src/main/java/org/apache/samza/storage/ChangelogStreamManager.java