Fixed getting the position in PulsarKafkaConsumer when we don't have …

…an offset yet (apache#900)
willome · Nov 14, 2017 · 635b846 · 635b846
1 parent 6278855
commit 635b846
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/...sar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java b/...sar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java
@@ -456,7 +456,8 @@ public void seekToEnd(Collection<TopicPartition> partitions) {
 
     @Override
     public long position(TopicPartition partition) {
-        return lastReceivedOffset.get(partition);
+        Long offset = lastReceivedOffset.get(partition);
+        return offset != null ? offset : -1l;
     }
 
     @Override