@Override
  public TimelineWriteResponse write(TimelineCollectorContext context,
      TimelineEntities entities, UserGroupInformation callerUgi)
      throws IOException {
    TimelineWriteResponse response = new TimelineWriteResponse();
    String clusterId = context.getClusterId();
    String userId = context.getUserId();
    String flowName = context.getFlowName();
    String flowVersion = context.getFlowVersion();
    long flowRunId = context.getFlowRunId();
    String appId = context.getAppId();

    for (TimelineEntity entity : entities.getEntities()) {
      writeInternal(clusterId, userId, flowName, flowVersion,
              flowRunId, appId, entity, response);
    }
    return response;
  }
@Test
  void testWriteEntitiesWithEmptyFlowName() throws Exception {
    String id = "appId";
    String type = "app";

    TimelineEntities te = new TimelineEntities();
    TimelineEntity entity = new TimelineEntity();
    entity.setId(id);
    entity.setType(type);
    entity.setCreatedTime(1425016501000L);
    te.addEntity(entity);

    FileSystemTimelineWriterImpl fsi = null;
    try {
      fsi = new FileSystemTimelineWriterImpl();
      Configuration conf = new YarnConfiguration();
      String outputRoot = tmpFolder.getAbsolutePath();
      conf.set(FileSystemTimelineWriterImpl.TIMELINE_SERVICE_STORAGE_DIR_ROOT,
          outputRoot);
      fsi.init(conf);
      fsi.start();
      fsi.write(
          new TimelineCollectorContext("cluster_id", "user_id", "",
              "flow_version", 12345678L, "app_id"),
          te, UserGroupInformation.createRemoteUser("user_id"));

      String fileName = outputRoot + File.separator + "entities" +
          File.separator + "cluster_id" + File.separator + "user_id" +
          File.separator + "" + File.separator + "flow_version" +
          File.separator + "12345678" + File.separator + "app_id" +
          File.separator + type + File.separator + id +
          FileSystemTimelineWriterImpl.TIMELINE_SERVICE_STORAGE_EXTENSION;
      Path path = new Path(fileName);
      FileSystem fs = FileSystem.get(conf);
      assertTrue(fs.exists(path),
          "Specified path(" + fileName + ") should exist: ");
      FileStatus fileStatus = fs.getFileStatus(path);
      assertFalse(fileStatus.isDirectory(), "specified path should be a file");
      List<String> data = readFromFile(fs, path);
      assertEquals(2, data.size(), "data size is:" + data.size());
      String d = data.get(0);
      // confirm the contents same as what was written
      assertEquals(d, TimelineUtils.dumpTimelineRecordtoJSON(entity));
    } finally {
      if (fsi != null) {
        fsi.close();
      }
    }
  }