TimelineCollector extends CompositeService { public TimelineWriteResponse putDomain(TimelineDomain domain, UserGroupInformation callerUgi) throws IOException { if (LOG.isDebugEnabled()) { LOG.debug( "putDomain(domain=" + domain + ", callerUgi=" + callerUgi + ")"); } TimelineWriteResponse response; synchronized (writer) { final TimelineCollectorContext context = getTimelineEntityContext(); response = writer.write(context, domain); 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 testPutDomain() throws IOException { TimelineWriter writer = mock(TimelineWriter.class); TimelineCollector collector = new TimelineCollectorForTest(writer); TimelineDomain domain = generateDomain("id", "desc", "owner", "reader1,reader2", "writer", 0L, 1L); collector.putDomain(domain, UserGroupInformation.createRemoteUser("owner")); verify(writer, times(1)) .write(any(TimelineCollectorContext.class), any(TimelineDomain.class)); verify(writer, times(1)).flush(); }