@Override
  public Map<String, List<ContainerLogFileInfo>> getLogMetaFilesOfNode(
      ExtendedLogMetaRequest logRequest, FileStatus currentNodeFile,
      ApplicationId appId) throws IOException {
    Map<String, List<ContainerLogFileInfo>> logMetaFiles = new HashMap<>();

    Long checkSumIndex = parseChecksum(currentNodeFile);
    long endIndex = -1;
    if (checkSumIndex != null) {
      endIndex = checkSumIndex;
    }
    IndexedLogsMeta current = loadIndexedLogsMeta(
        currentNodeFile.getPath(), endIndex, appId);
    if (current != null) {
      for (IndexedPerAggregationLogMeta logMeta :
          current.getLogMetas()) {
        for (Entry<String, List<IndexedFileLogMeta>> log : logMeta
            .getLogMetas().entrySet()) {
          String currentContainerId = log.getKey();
          if (!(logRequest.getContainerId() == null ||
              logRequest.getContainerId().equals(currentContainerId))) {
            continue;
          }
          logMetaFiles.put(currentContainerId, new ArrayList<>());
          for (IndexedFileLogMeta aMeta : log.getValue()) {
            ContainerLogFileInfo file = new ContainerLogFileInfo();
            file.setFileName(aMeta.getFileName());
            file.setFileSize(Long.toString(aMeta.getFileSize()));
            file.setLastModifiedTime(
                Long.toString(aMeta.getLastModifiedTime()));
            logMetaFiles.get(currentContainerId).add(file);
          }
        }
      }
    }

    return logMetaFiles;
  }
@Test
  void testGetLogMetaFilesOfNode() throws Exception {
    if (fs.exists(rootLocalLogDirPath)) {
      fs.delete(rootLocalLogDirPath, true);
    }
    assertTrue(fs.mkdirs(rootLocalLogDirPath));

    Path appLogsDir = new Path(rootLocalLogDirPath, appId.toString());
    if (fs.exists(appLogsDir)) {
      fs.delete(appLogsDir, true);
    }
    assertTrue(fs.mkdirs(appLogsDir));

    List<String> logTypes = new ArrayList<String>();
    logTypes.add("syslog");
    logTypes.add("stdout");
    logTypes.add("stderr");

    Set<File> files = new HashSet<>();

    LogKey key1 = new LogKey(containerId.toString());

    for (String logType : logTypes) {
      File file = createAndWriteLocalLogFile(containerId, appLogsDir,
          logType);
      files.add(file);
    }
    files.add(createZeroLocalLogFile(appLogsDir));

    LogValue value = mock(LogValue.class);
    when(value.getPendingLogFilesToUploadForThisContainer()).thenReturn(files);

    LogAggregationIndexedFileController fileFormat =
        new LogAggregationIndexedFileController();

    fileFormat.initialize(getConf(), "Indexed");

    Map<ApplicationAccessType, String> appAcls = new HashMap<>();
    Path appDir = fileFormat.getRemoteAppLogDir(appId,
        USER_UGI.getShortUserName());
    if (fs.exists(appDir)) {
      fs.delete(appDir, true);
    }
    assertTrue(fs.mkdirs(appDir));

    Path logPath = fileFormat.getRemoteNodeLogFileForApp(
        appId, USER_UGI.getShortUserName(), nodeId);
    LogAggregationFileControllerContext context =
        new LogAggregationFileControllerContext(
            logPath, logPath, true, 1000, appId, appAcls, nodeId, USER_UGI);
    // initialize the writer
    fileFormat.initializeWriter(context);

    fileFormat.write(key1, value);
    fileFormat.postWrite(context);
    fileFormat.closeWriter();

    ContainerLogsRequest logRequest = new ContainerLogsRequest();
    logRequest.setAppId(appId);
    logRequest.setNodeId(nodeId.toString());
    logRequest.setAppOwner(USER_UGI.getShortUserName());
    logRequest.setContainerId(containerId.toString());
    logRequest.setBytes(Long.MAX_VALUE);
    // create a checksum file
    final ControlledClock clock = new ControlledClock();
    clock.setTime(System.currentTimeMillis());
    Path checksumFile = new Path(fileFormat.getRemoteAppLogDir(
            appId, USER_UGI.getShortUserName()),
        LogAggregationUtils.getNodeString(nodeId)
            + LogAggregationIndexedFileController.CHECK_SUM_FILE_SUFFIX);
    FSDataOutputStream fInput = null;
    try {
      String nodeName = logPath.getName() + "_" + clock.getTime();
      fInput = FileSystem.create(fs, checksumFile, LOG_FILE_UMASK);
      fInput.writeInt(nodeName.length());
      fInput.write(nodeName.getBytes(
          StandardCharsets.UTF_8));
      fInput.writeLong(0);
    } finally {
      IOUtils.closeStream(fInput);
    }

    Path nodePath = LogAggregationUtils.getRemoteAppLogDir(
        fileFormat.getRemoteRootLogDir(), appId, USER_UGI.getShortUserName(),
        fileFormat.getRemoteRootLogDirSuffix());
    FileStatus[] nodes = fs.listStatus(nodePath);
    ExtendedLogMetaRequest req =
        new ExtendedLogMetaRequest.ExtendedLogMetaRequestBuilder().build();
    for (FileStatus node : nodes) {
      Map<String, List<ContainerLogFileInfo>> metas =
          fileFormat.getLogMetaFilesOfNode(req, node, appId);

      if (node.getPath().getName().contains(
          LogAggregationIndexedFileController.CHECK_SUM_FILE_SUFFIX)) {
        assertTrue(metas.isEmpty(),
            "Checksum node files should not contain any logs");
      } else {
        assertFalse(metas.isEmpty(),
            "Non-checksum node files should contain log files");
        assertEquals(4, metas.values().stream().findFirst().get().size());
      }
    }

  }