[ https://issues.apache.org/jira/browse/STORM-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Ethan Li updated STORM-2720: ---------------------------- Fix Version/s: (was: 2.0.1) 2.1.0 > Add timestamp based FirstPollOffsetStrategy in KafkaTridentSpoutOpaque > ---------------------------------------------------------------------- > > Key: STORM-2720 > URL: https://issues.apache.org/jira/browse/STORM-2720 > Project: Apache Storm > Issue Type: Improvement > Components: storm-kafka-client, trident > Affects Versions: 1.1.1 > Reporter: Janith Kaiprath Valiyalappil > Assignee: Janith Kaiprath Valiyalappil > Priority: Minor > Labels: pull-request-available > Fix For: 2.1.0 > > Time Spent: 10h > Remaining Estimate: 0h > > Offsets for a given partition at a particular timestamp can now be found > using offsetsForTimes API. > https://kafka.apache.org/0110/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#offsetsForTimes(java.util.Map). > One way to make use of this api would be to : > Add a new option for FirstPollOffsetStrategy called TIMESTAMP > Add a new startTimeStamp option to KafkaSpoutConfig, which would be used only > when FirstPollOffsetStrategy is set to TIMESTAMP. > Later in the KafkaTridentSpoutEmitter, when we do the first seek, we can do > something like : > {code} > if(firstPollOffsetStrategy.equals(TIMESTAMP)) { > try { > startTimeStampOffset = > > kafkaConsumer.offsetsForTimes(Collections.singletonMap(tp, > startTimeStamp)).get(tp).offset(); > } catch (IllegalArgumentException e) { > LOG.error("Illegal timestamp {} provided for tp {} > ",startTimeStamp,tp.toString()); > } catch (UnsupportedVersionException e) { > LOG.error("Kafka Server do not support offsetsForTimes(), > probably < 0.10.1",e); > } > if(startTimeStampOffset!=null) { > LOG.info("Kafka consumer offset reset for TopicPartition > {}, TimeStamp {}, Offset {}",tp,startTimeStamp,startTimeStampOffset); > kafkaConsumer.seek(tp, startTimeStampOffset); > } else { > LOG.info("Kafka consumer offset reset by timestamp failed > for TopicPartition {}, TimeStamp {}, Offset {}. Restart with a different > Strategy ",tp,startTimeStamp,startTimeStampOffset); > } > } > {code} -- This message was sent by Atlassian JIRA (v7.6.14#76016)