public long parseForStore(TimelineDataManager tdm, Path appDirPath,
      boolean appCompleted, JsonFactory jsonFactory, ObjectMapper objMapper,
      FileSystem fs) throws IOException {
    LOG.debug("Parsing for log dir {} on attempt {}", appDirPath,
        attemptDirName);
    Path logPath = getPath(appDirPath);
    FileStatus status = fs.getFileStatus(logPath);
    long numParsed = 0;
    if (status != null) {
      long curModificationTime = status.getModificationTime();
      if (curModificationTime > getLastProcessedTime()) {
        long startTime = Time.monotonicNow();
        try {
          LOG.info("Parsing {} at offset {}", logPath, offset);
          long count =
              parsePath(tdm, logPath, appCompleted, jsonFactory, objMapper, fs);
          setLastProcessedTime(curModificationTime);
          LOG.info("Parsed {} entities from {} in {} msec", count, logPath,
              Time.monotonicNow() - startTime);
          numParsed += count;
        } catch (RuntimeException e) {
          // If AppLogs cannot parse this log, it may be corrupted or just empty
          if (e.getCause() instanceof JsonParseException
              && (status.getLen() > 0 || offset > 0)) {
            // log on parse problems if the file as been read in the past or
            // is visibly non-empty
            LOG.info("Log {} appears to be corrupted. Skip. ", logPath);
          } else {
            LOG.error("Failed to parse " + logPath + " from offset " + offset,
                e);
          }
        }
      } else {
        LOG.info("Skip Parsing {} as there is no change", logPath);
      }
    } else {
      LOG.warn("{} no longer exists. Skip for scanning. ", logPath);
    }
    return numParsed;
  }
@Test
  void testParseBrokenEntity() throws Exception {
    // Load test data
    TimelineDataManager tdm = PluginStoreTestUtils.getTdmWithMemStore(config);
    EntityLogInfo testLogInfo = new EntityLogInfo(TEST_ATTEMPT_DIR_NAME,
        TEST_BROKEN_FILE_NAME,
        UserGroupInformation.getLoginUser().getUserName());
    DomainLogInfo domainLogInfo = new DomainLogInfo(TEST_ATTEMPT_DIR_NAME,
        TEST_BROKEN_FILE_NAME,
        UserGroupInformation.getLoginUser().getUserName());
    // Try parse, should not fail
    testLogInfo.parseForStore(tdm, getTestRootPath(), true, jsonFactory, objMapper,
        fs);
    domainLogInfo.parseForStore(tdm, getTestRootPath(), true, jsonFactory, objMapper,
        fs);
    tdm.close();
  }