Github user janithkv commented on a diff in the pull request:
https://github.com/apache/storm/pull/2911#discussion_r234878746
--- Diff:
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
---
@@ -225,7 +229,23 @@ private void emitTuple(TridentCollector collector,
ConsumerRecord<K, V> record)
*/
private long seek(TopicPartition tp, KafkaTridentSpoutBatchMetadata
lastBatchMeta) {
if (isFirstPoll(tp)) {
- if (firstPollOffsetStrategy == EARLIEST) {
+ if(firstPollOffsetStrategy.equals(TIMESTAMP)) {
+ Long startTimeStampOffset = null;
+ try {
+ startTimeStampOffset =
+
consumer.offsetsForTimes(Collections.singletonMap(tp,
startTimeStamp)).get(tp).offset();
+ } catch (IllegalArgumentException e) {
--- End diff --
ok will change
---
