@Override
  public TimelineEntity getEntity(TimelineReaderContext context,
      TimelineDataToRetrieve dataToRetrieve) throws IOException {
    String flowRunPathStr = getFlowRunPath(context.getUserId(),
        context.getClusterId(), context.getFlowName(), context.getFlowRunId(),
        context.getAppId());
    Path clusterIdPath = new Path(entitiesPath, context.getClusterId());
    Path flowRunPath = new Path(clusterIdPath, flowRunPathStr);
    Path appIdPath = new Path(flowRunPath, context.getAppId());
    Path entityTypePath = new Path(appIdPath, context.getEntityType());
    Path entityFilePath = getNormalPath(new Path(entityTypePath,
        context.getEntityId() + TIMELINE_SERVICE_STORAGE_EXTENSION));
    if (entityFilePath == null) {
      return null;
    }
    try (BufferedReader reader =
             new BufferedReader(new InputStreamReader(
                 fs.open(entityFilePath), StandardCharsets.UTF_8))) {
      TimelineEntity entity = readEntityFromFile(reader);
      return createEntityToBeReturned(
          entity, dataToRetrieve.getFieldsToRetrieve());
    } catch (FileNotFoundException e) {
      LOG.info("Cannot find entity {id:" + context.getEntityId() + " , type:" +
          context.getEntityType() + "}. Will send HTTP 404 in response.");
      return null;
    }
  }
@Test
  void testGetEntityCustomFields() throws Exception {
    // Specified fields in addition to default view will be returned.
    TimelineEntity result = reader.getEntity(
        new TimelineReaderContext("cluster1", "user1", "flow1", 1L, "app1",
            "app", "id_1"),
        new TimelineDataToRetrieve(null, null,
            EnumSet.of(Field.INFO, Field.CONFIGS, Field.METRICS), null, null,
            null));
    assertEquals(
        (new TimelineEntity.Identifier("app", "id_1")).toString(),
        result.getIdentifier().toString());
    assertEquals((Long) 1425016502000L, result.getCreatedTime());
    assertEquals(3, result.getConfigs().size());
    assertEquals(3, result.getMetrics().size());
    assertEquals(2, result.getInfo().size());
    // No events will be returned
    assertEquals(0, result.getEvents().size());
  }