TimelineCollector extends CompositeService { public TimelineWriteResponse putEntities(TimelineEntities entities, UserGroupInformation callerUgi) throws IOException { LOG.debug("putEntities(entities={}, callerUgi={})", entities, callerUgi); TimelineWriteResponse response; synchronized (writer) { response = writeTimelineEntities(entities, callerUgi); flushBufferedTimelineEntities(); } return response; } 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 testPutEntity() throws IOException { TimelineWriter writer = mock(TimelineWriter.class); TimelineCollector collector = new TimelineCollectorForTest(writer); TimelineEntities entities = generateTestEntities(1, 1); collector.putEntities( entities, UserGroupInformation.createRemoteUser("test-user")); verify(writer, times(1)).write(any(TimelineCollectorContext.class), any(TimelineEntities.class), any(UserGroupInformation.class)); verify(writer, times(1)).flush(); }