[ https://issues.apache.org/jira/browse/TWILL-199?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15685552#comment-15685552 ]
ASF GitHub Bot commented on TWILL-199: -------------------------------------- GitHub user maochf opened a pull request: https://github.com/apache/twill/pull/16 [TWILL-199] Handle offset error and return next offset in KafkaConsumer.MessageCallback https://issues.apache.org/jira/browse/TWILL-199 You can merge this pull request into a Git repository by running: $ git pull https://github.com/maochf/twill feature/find-correct-offset Alternatively you can review and apply these changes as the patch at: https://github.com/apache/twill/pull/16.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #16 ---- commit 6e5dc84793a4eb35a1a0b10a36ed5a00b5275414 Author: Chengfeng <m...@cask.co> Date: 2016-11-22T03:04:52Z add KafkaOffsetProvider interface and return long in MessageCallback#onReceived ---- > Get next offset and handle offset error in KafkaConsumer.MessageCallback > ------------------------------------------------------------------------ > > Key: TWILL-199 > URL: https://issues.apache.org/jira/browse/TWILL-199 > Project: Apache Twill > Issue Type: Improvement > Reporter: Chengfeng Mao > > The method {{void onReceived(Iterator<FetchedMessage> messages)}} in > {{KafkaConsumer.MessageCallback}} can be more flexible with the change to > {{Long onReceived(Iterator<FetchedMessage> messages)}} so that it can provide > additional functionalities: > 1. To return the next offset to be fetched > 2. To handle offset non-existence or offset mismatch error and take action on > the error > This method will return null for backward compatibility when it doesn't need > to provide the next offset. > In concrete implementation, a class of a new interface > {{KafkaOffsetProvider}} can be added as a member in > {{KafkaConsumer.MessageCallback}} to perform the offset error handling and > provide the next offset. Besides, {{KafkaOffsetProvider}} also has methods to > provide the following functionalities: > 1. To fetch earliest/latest offset in Kafka > 2. To find the offset of a message with timestamp equal to the given > timestamp in Kafka > For backward compatibility, if {{KafkaOffsetProvider}} instance is not > provided, its default value will be null and none of its methods will be > called. -- This message was sent by Atlassian JIRA (v6.3.4#6332)