TimelineCollector extends CompositeService { 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); } } }); } TimelineCollector(String name); abstract TimelineCollectorContext getTimelineEntityContext(); TimelineWriteResponse putEntities(TimelineEntities entities,
      UserGroupInformation callerUgi); TimelineWriteResponse putDomain(TimelineDomain domain,
      UserGroupInformation callerUgi); void putEntitiesAsync(TimelineEntities entities,
      UserGroupInformation callerUgi); static TimelineEntity aggregateEntities(
      TimelineEntities entities, String resultEntityId,
      String resultEntityType, boolean needsGroupIdInResult); static final String SEPARATOR; }
@Test public void testPutEntityAsync() throws Exception { TimelineWriter writer = mock(TimelineWriter.class); TimelineCollector collector = new TimelineCollectorForTest(writer); collector.init(new Configuration()); collector.start(); TimelineEntities entities = generateTestEntities(1, 1); collector.putEntitiesAsync( entities, UserGroupInformation.createRemoteUser("test-user")); Thread.sleep(1000); verify(writer, times(1)).write(any(TimelineCollectorContext.class), any(TimelineEntities.class), any(UserGroupInformation.class)); verify(writer, never()).flush(); collector.stop(); }
@Test public 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(); }