public void putEntitiesAsync(TimelineEntities entities,
      UserGroupInformation callerUgi) throws IOException {
    LOG.debug("putEntitiesAsync(entities={}, callerUgi={})", entities,
        callerUgi);

    pool.execute(new Runnable() {
      @Override public void run() {
        try {
          writeTimelineEntities(entities, callerUgi);
        } catch (IOException ie) {
          LOG.error("Got an exception while writing entity", ie);
        }
      }
    });
  }
@Test
  void testAsyncEntityDiscard() throws Exception {
    TimelineWriter writer = mock(TimelineWriter.class);

    when(writer.write(any(), any(), any())).thenAnswer(
        new AnswersWithDelay(500, new Returns(new TimelineWriteResponse())));
    TimelineCollector collector = new TimelineCollectorForTest(writer);
    Configuration config = new Configuration();
    config
        .setInt(YarnConfiguration.TIMELINE_SERVICE_WRITER_ASYNC_QUEUE_CAPACITY,
            3);
    collector.init(config);
    collector.start();
    for (int i = 0; i < 10; ++i) {
      TimelineEntities entities = generateTestEntities(i + 1, 1);
      collector.putEntitiesAsync(entities,
          UserGroupInformation.createRemoteUser("test-user"));
    }
    Thread.sleep(3000);
    verify(writer, times(4))
        .write(any(TimelineCollectorContext.class), any(TimelineEntities.class),
            any(UserGroupInformation.class));
    verify(writer, never()).flush();
    collector.stop();
  }