fix can't get latest offset in KafkaEightSimpleConsumerFirehoseFactory (

apache#3355)
mark1900 · Aug 12, 2016 · df89f25 · df89f25
1 parent 4545878
commit df89f25
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/...nsumer/src/main/java/io/druid/firehose/kafka/KafkaEightSimpleConsumerFirehoseFactory.java b/...nsumer/src/main/java/io/druid/firehose/kafka/KafkaEightSimpleConsumerFirehoseFactory.java
@@ -153,7 +153,7 @@ public FirehoseV2 connect(final ByteBufferInputRowParser firehoseParser, Object
       );
       Long startOffset = lastOffsets.get(partition);
       PartitionConsumerWorker worker = new PartitionConsumerWorker(
-          feed, kafkaSimpleConsumer, partition, startOffset == null ? 0 : startOffset
+          feed, kafkaSimpleConsumer, partition, startOffset == null ? -1 : startOffset
       );
       consumerWorkers.add(worker);
     }

diff --git a/...kafka-eight-simpleConsumer/src/main/java/io/druid/firehose/kafka/KafkaSimpleConsumer.java b/...kafka-eight-simpleConsumer/src/main/java/io/druid/firehose/kafka/KafkaSimpleConsumer.java
@@ -238,7 +238,7 @@ public Iterable<BytesMessageWithOffset> fetch(long offset, int timeoutMs) throws
       }
       catch (Exception e) {
         ensureNotInterrupted(e);
-        log.warn(e, "caughte exception in fetch {} - {}", topic, partitionId);
+        log.warn(e, "caught exception in fetch {} - {}", topic, partitionId);
         response = null;
       }