public Map<Partition, SystemStreamPartitionMetadata> getPartitionMetadataMap(String streamName,
    @Nullable Map<Partition, List<String>> existingPartitionDescriptorMap) {
    LOG.info("Trying to obtain metadata for " + streamName);
    LOG.info("Existing partition descriptor: " + (MapUtils.isEmpty(existingPartitionDescriptorMap) ? "empty"
      : existingPartitionDescriptorMap));
    Map<Partition, SystemStreamPartitionMetadata> partitionMetadataMap = new HashMap<>();
    partitionDescriptorMap.putIfAbsent(streamName, new HashMap<>());
    List<FileMetadata> filteredFiles = getFilteredFiles(streamName);
    if (!MapUtils.isEmpty(existingPartitionDescriptorMap)) {
      filteredFiles = validateAndGetOriginalFilteredFiles(filteredFiles, existingPartitionDescriptorMap);
    }
    List<List<FileMetadata>> groupedPartitions = generatePartitionGroups(filteredFiles);
    int partitionId = 0;
    for (List<FileMetadata> fileGroup : groupedPartitions) {
      Partition partition = new Partition(partitionId);
      List<String> pathList = new ArrayList<>();
      List<String> lengthList = new ArrayList<>();
      fileGroup.forEach(fileMetadata -> {
        pathList.add(fileMetadata.getPath());
        lengthList.add(String.valueOf(fileMetadata.getLen()));
      });
      String oldestOffset = MultiFileHdfsReader.generateOffset(0, "0");
      String newestOffset = MultiFileHdfsReader.generateOffset(lengthList.size() - 1, String.valueOf(lengthList.get(lengthList.size() - 1)));
      SystemStreamPartitionMetadata metadata =
        new SystemStreamPartitionMetadata(oldestOffset, newestOffset, null);
      partitionMetadataMap.put(partition, metadata);
      partitionDescriptorMap.get(streamName).put(partition, pathList);
      partitionId++;
    }
    LOG.info("Obtained metadata map as: " + partitionMetadataMap);
    LOG.info("Computed partition description as: " + partitionDescriptorMap);
    return partitionMetadataMap;
  }
Get partition metadata for a stream
@param streamName name of the stream; should contain the information about the path of the
                  root directory
@param existingPartitionDescriptorMap map of the existing partition descriptor
@return map of SSP metadata
samza-hdfs/src/main/java/org/apache/samza/system/hdfs/partitioner/DirectoryPartitioner.java