Fix NPE when there was no previous commit for the given partition.
@@ -144,7 +144,6 @@ public class SecorKafkaMessageIterator implements KafkaMessageIterator, Rebalanc
@Override
public void subscribe(RebalanceHandler handler, SecorConfig config) {
ConsumerRebalanceListener reBalanceListener = new SecorConsumerRebalanceListener(mKafkaConsumer, mZookeeperConnector, getSkipZookeeperOffsetSeek(config), config.getNewConsumerAutoOffsetReset(), handler);
- ;
String[] subscribeList = config.getKafkaTopicList();
if (subscribeList.length == 0 || Strings.isNullOrEmpty(subscribeList[0])) {
@@ -164,7 +163,10 @@ public class SecorKafkaMessageIterator implements KafkaMessageIterator, Rebalanc
public long getKafkaCommitedOffsetCount(final com.pinterest.secor.common.TopicPartition topicPartition) {
TopicPartition kafkaTopicPartition = new TopicPartition(topicPartition.getTopic(), topicPartition.getPartition());
OffsetAndMetadata offsetAndMetadata = mKafkaConsumer.committed(kafkaTopicPartition);
-
+ if (offsetAndMetadata != null) {
return offsetAndMetadata.offset();
+ } else {
+ return -1;
+ }
}
}
