[
https://issues.apache.org/jira/browse/STORM-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16691001#comment-16691001
]
Janith Kaiprath Valiyalappil edited comment on STORM-2720 at 11/18/18 7:10 PM:
-------------------------------------------------------------------------------
[[~Srdo]]I have the code done. at https://github.com/apache/storm/pull/2911. I
think I should be able to add unit test and fix merge conflicts after your code
at https://github.com/apache/storm/pull/2907 goes in.
was (Author: janithkv):
[[~Srdo]]I have the code done. Working on unit test
> 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
>
> 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.3#76005)
