### Question:
KafkaOffsetResolver { long getStartOffset(final long nextOffset, final long nextEventTime, final int partition) { Preconditions.checkArgument(nextOffset > 0, "Invalid checkpoint offset"); String topic = config.getTopic(); BrokerInfo brokerInfo = brokerService.getLeader(topic, partition); if (brokerInfo == null) { throw new LeaderNotAvailableException( String.format("BrokerInfo from BrokerService is null for topic %s partition %d. Will retry in next run.", topic, partition)); } SimpleConsumer consumer = new SimpleConsumer(brokerInfo.getHost(), brokerInfo.getPort(), SO_TIMEOUT_MILLIS, BUFFER_SIZE, "offset-finder-" + topic + "-" + partition); long offset = nextOffset - 1; try { long timestamp = getEventTimeByOffset(consumer, partition, offset); if (timestamp == nextEventTime) { return nextOffset; } LOG.debug("Event timestamp in {}:{} at offset {} is {}. It doesn't match with checkpoint timestamp {}", topic, partition, offset, timestamp, nextEventTime); } catch (NotFoundException | OffsetOutOfRangeException e) { LOG.debug("Cannot get valid log event in {}:{} at offset {}", topic, partition, offset); } long startOffset = findStartOffset(consumer, partition, nextEventTime); LOG.debug("Found new nextOffset {} for topic {} partition {} with existing offset {} and event time {}.", startOffset, topic, partition, nextOffset, nextEventTime); return startOffset; } KafkaOffsetResolver(BrokerService brokerService, KafkaPipelineConfig config);   }### Answer:
@Test public void testInOrderEvents() throws InterruptedException, IOException { String topic = "testInOrderEvents"; KafkaPipelineConfig config = new KafkaPipelineConfig(topic, Collections.singleton(0), 1024L, EVENT_DELAY_MILLIS, 1048576, 200L); KAFKA_TESTER.createTopic(topic, 1); long baseTime = System.currentTimeMillis() - EVENT_DELAY_MILLIS; List<ILoggingEvent> inOrderEvents = new ArrayList<>(); for (int i = 0; i < 20; i++) { inOrderEvents.add(createLoggingEvent("test.logger", Level.INFO, Integer.toString(i), baseTime + i)); } publishLog(topic, inOrderEvents); waitForAllLogsPublished(topic, inOrderEvents.size()); KafkaOffsetResolver offsetResolver = new KafkaOffsetResolver(KAFKA_TESTER.getBrokerService(), config); for (int i = 0; i < inOrderEvents.size(); i++) { long targetTime = inOrderEvents.get(i).getTimeStamp(); long offset = offsetResolver.getStartOffset(Long.MAX_VALUE, targetTime, 0); Assert.assertEquals("Failed to find the expected event with the target time: " + targetTime, i + 1, offset); } }