You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@storm.apache.org by "Ethan Li (JIRA)" <ji...@apache.org> on 2019/08/10 22:25:00 UTC
[jira] [Updated] (STORM-2720) Add timestamp based
FirstPollOffsetStrategy in KafkaTridentSpoutOpaque
[ 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)