public static Option<Pair<String, List<String>>> getFileSliceForFileGroupFromDeltaCommit(byte[] bytes, HoodieFileGroupId fileGroupId) {
    try {
      org.apache.hudi.avro.model.HoodieCommitMetadata commitMetadata = deserializeCommitMetadata(bytes);
      Map<String,List<org.apache.hudi.avro.model.HoodieWriteStat>> partitionToWriteStatsMap =
              commitMetadata.getPartitionToWriteStats();
      for (Map.Entry<String, List<org.apache.hudi.avro.model.HoodieWriteStat>> partitionToWriteStat: partitionToWriteStatsMap.entrySet()) {
        for (org.apache.hudi.avro.model.HoodieWriteStat writeStat: partitionToWriteStat.getValue()) {
          HoodieFileGroupId fgId = new HoodieFileGroupId(partitionToWriteStat.getKey(), writeStat.getFileId());
          if (fgId.equals(fileGroupId)) {
            return Option.of(Pair.of(writeStat.getBaseFile() == null ? "" : writeStat.getBaseFile(), writeStat.getLogFiles()));
          }
        }
      }
      return Option.empty();
    } catch (Exception e) {
      throw new HoodieException("Fail to parse the base file and log files from DeltaCommit", e);
    }
  }
@Test
  public void testGetFileSliceForFileGroupFromDeltaCommit() throws IOException {
    org.apache.hudi.avro.model.HoodieCommitMetadata commitMetadata = new org.apache.hudi.avro.model.HoodieCommitMetadata();
    org.apache.hudi.avro.model.HoodieWriteStat writeStat1 = createWriteStat("111", "111base", Arrays.asList("1.log", "2.log"));
    org.apache.hudi.avro.model.HoodieWriteStat writeStat2 = createWriteStat("111", "111base", Arrays.asList("3.log", "4.log"));
    org.apache.hudi.avro.model.HoodieWriteStat writeStat3 = createWriteStat("222", null, Collections.singletonList("5.log"));
    Map<String,List<org.apache.hudi.avro.model.HoodieWriteStat>> partitionToWriteStatsMap = new HashMap<>();
    partitionToWriteStatsMap.put("partition1", Arrays.asList(writeStat2, writeStat3));
    partitionToWriteStatsMap.put("partition2", Collections.singletonList(writeStat1));
    commitMetadata.setPartitionToWriteStats(partitionToWriteStatsMap);
    byte[] serializedCommitMetadata = TimelineMetadataUtils.serializeAvroMetadata(
            commitMetadata, org.apache.hudi.avro.model.HoodieCommitMetadata.class).get();

    Option<Pair<String, List<String>>> result = HoodieCommitMetadata.getFileSliceForFileGroupFromDeltaCommit(
            serializedCommitMetadata, new HoodieFileGroupId("partition1","111"));

    assertTrue(result.isPresent());
    assertEquals("111base", result.get().getKey());
    assertEquals(2, result.get().getValue().size());
    assertEquals("3.log", result.get().getValue().get(0));
    assertEquals("4.log", result.get().getValue().get(1));

    result = HoodieCommitMetadata.getFileSliceForFileGroupFromDeltaCommit(
            serializedCommitMetadata, new HoodieFileGroupId("partition1","222"));
    assertTrue(result.isPresent());
    assertTrue(result.get().getKey().isEmpty());
    assertEquals(1, result.get().getValue().size());
    assertEquals("5.log", result.get().getValue().get(0));
  }