### Question:
TimeEventQueueProcessor { public ProcessedEventMetadata<OFFSET> process(int partition, Iterator<ProcessorEvent<OFFSET>> eventIterator) { int totalEvents = 0; Map<Integer, Checkpoint<OFFSET>> checkpoints = new HashMap<>(); while (eventIterator.hasNext()) { if (eventQueue.getEventSize() >= maxBufferSize) { OUTAGE_LOG.info("Maximum queue size {} reached for log pipeline {}.", maxBufferSize, context.getName()); ProcessedEventMetadata<OFFSET> eventsMetadata = append(); if (eventsMetadata.getTotalEventsProcessed() <= 0) { break; } totalEvents += eventsMetadata.getTotalEventsProcessed(); checkpoints.putAll(eventsMetadata.getCheckpoints()); } ProcessorEvent<OFFSET> processorEvent = eventIterator.next(); eventQueue.add(processorEvent.getEvent(), processorEvent.getEvent().getTimeStamp(), processorEvent.getEventSize(), partition, processorEvent.getOffset()); } ProcessedEventMetadata<OFFSET> eventsMetadata = append(); if (eventsMetadata.getTotalEventsProcessed() > 0) { totalEvents += eventsMetadata.getTotalEventsProcessed(); checkpoints.putAll(eventsMetadata.getCheckpoints()); } return new ProcessedEventMetadata<>(totalEvents, checkpoints); } TimeEventQueueProcessor(LogProcessorPipelineContext context, long maxBufferSize, long eventDelayMillis,
                                 Iterable<Integer> partitions); ProcessedEventMetadata<OFFSET> process(int partition, Iterator<ProcessorEvent<OFFSET>> eventIterator); boolean isQueueEmpty(int partition);  }### Answer:
@Test public void test() throws Exception { LoggerContext loggerContext = LogPipelineTestUtil.createLoggerContext("WARN", ImmutableMap.of("test.logger", "INFO"), MockAppender.class.getName()); LogProcessorPipelineContext context = new LogProcessorPipelineContext(CConfiguration.create(), "test", loggerContext, NO_OP_METRICS_CONTEXT, 0); context.start(); TimeEventQueueProcessor<TestOffset> processor = new TimeEventQueueProcessor<>(context, 50, 1, ImmutableList.of(0)); long now = System.currentTimeMillis(); List<ILoggingEvent> events = ImmutableList.of( LogPipelineTestUtil.createLoggingEvent("test.logger", Level.INFO, "1", now - 1000), LogPipelineTestUtil.createLoggingEvent("test.logger", Level.INFO, "3", now - 700), LogPipelineTestUtil.createLoggingEvent("test.logger", Level.INFO, "5", now - 500), LogPipelineTestUtil.createLoggingEvent("test.logger", Level.INFO, "2", now - 900), LogPipelineTestUtil.createLoggingEvent("test.logger", Level.ERROR, "4", now - 600), LogPipelineTestUtil.createLoggingEvent("test.logger", Level.INFO, "6", now - 100)); ProcessedEventMetadata<TestOffset> metadata = processor.process(0, new TransformingIterator(events.iterator())); Assert.assertEquals(6, metadata.getTotalEventsProcessed()); for (Map.Entry<Integer, Checkpoint<TestOffset>> entry : metadata.getCheckpoints().entrySet()) { Checkpoint<TestOffset> value = entry.getValue(); Assert.assertEquals(6, value.getOffset().getOffset()); } }