KafkaUtils { public static long getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config) { long startOffsetTime = kafka.api.OffsetRequest.LatestTime(); if ( config.forceFromStart ) { startOffsetTime = config.startOffsetTime; } return getOffset(consumer, topic, partition, startOffsetTime); }  static IBrokerReader makeBrokerReader(Map stormConf, KafkaConfig conf); static long getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config); static long getOffset(SimpleConsumer consumer, String topic, int partition, long startOffsetTime); static ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset); static Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg); static List<Partition> calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex); static String taskId(int taskIndex, int totalTasks); static final Logger LOG; }
@Test public void getOffsetFromConfigAndDontForceFromStart() { config.forceFromStart = false; config.startOffsetTime = OffsetRequest.EarliestTime(); createTopicAndSendMessage(); long latestOffset = KafkaUtils.getOffset(simpleConsumer, config.topic, 0, OffsetRequest.LatestTime()); long offsetFromConfig = KafkaUtils.getOffset(simpleConsumer, config.topic, 0, config); assertThat(latestOffset, is(equalTo(offsetFromConfig))); }
@Test public void getOffsetFromConfigAndFroceFromStart() { config.forceFromStart = true; config.startOffsetTime = OffsetRequest.EarliestTime(); createTopicAndSendMessage(); long earliestOffset = KafkaUtils.getOffset(simpleConsumer, config.topic, 0, OffsetRequest.EarliestTime()); long offsetFromConfig = KafkaUtils.getOffset(simpleConsumer, config.topic, 0, config); assertThat(earliestOffset, is(equalTo(offsetFromConfig))); }