[
https://issues.apache.org/jira/browse/TWILL-199?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Chengfeng Mao updated TWILL-199:
--------------------------------
Description:
Offsets of Kafka messages may not always be consistent. They can change when
switching to a new Kafka instance while the messages' content remains the same.
Therefore, we need a way to check the inconsistent offsets and correct them. An
instance of a new class {{KafkaOffsetProvider}} can be added to the parameter
of {{KafkaConsumer.MessageCallback}} for this purpose. For backward
compatibility, if this instance is not provided, its default value will be null
and no check for inconsistent offset will be done.
The new class {{KafkaOffsetProvider}} requires the timestamp of the message to
be processed. It provides a method to verify whether the message fetched with
the given offset matches the intended one by checking whether the fetched
message's timestamp is equal to the expected timestamp. If not, this method
returns the offset of the first message found with the expected timestamp.
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 call call the
method in {{KafkaOffsetProvider}} to get the correct offset to start fetching
messages with, when the original starting offset does not exist or the message
given doesn't match the timestamp stored in KafkaOffsetProvider . If neither of
the offset mismatches happen or the {{KafkaOffsetProvider}} instance is null,
there's no check for inconsistent offset and this method will just return null.
was:
Offsets of Kafka messages may not always be consistent. They can change when
switching to a new Kafka instance while the messages' content remains the same.
Therefore, we need a way to check the inconsistent offsets and correct them. An
instance of a new class {{KafkaOffsetProvider}} can be added to the parameter
of {{KafkaConsumer.MessageCallback}} for this purpose. For backward
compatibility, if this instance is not provided, its default value will be null
and no check for inconsistent offset will be done.
The new class {{KafkaOffsetProvider}} requires the timestamp of the message to
be processed. It provides a method to verify whether the message fetched with
the given offset matches the intended one by checking whether the fetched
message's timestamp is equal to the expected timestamp. If not, this method
returns the offset of the first message found with the expected timestamp.
Method {{void onReceived(Iterator<FetchedMessage> messages)}} in
KafkaConsumer.MessageCallback should be changed to {{Long
onReceived(Iterator<FetchedMessage> messages)}} so that it can call call the
method in {{KafkaOffsetProvider}} to get the correct offset to start fetching
messages with, when the original starting offset does not exist or the message
given doesn't match the timestamp stored in KafkaOffsetProvider . If neither of
the offset mismatches happen or the {{KafkaOffsetProvider}} instance is null,
there's no check for inconsistent offset and this method will just return null.
> Return 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
>
> Offsets of Kafka messages may not always be consistent. They can change when
> switching to a new Kafka instance while the messages' content remains the
> same. Therefore, we need a way to check the inconsistent offsets and correct
> them. An instance of a new class {{KafkaOffsetProvider}} can be added to the
> parameter of {{KafkaConsumer.MessageCallback}} for this purpose. For backward
> compatibility, if this instance is not provided, its default value will be
> null and no check for inconsistent offset will be done.
> The new class {{KafkaOffsetProvider}} requires the timestamp of the message
> to be processed. It provides a method to verify whether the message fetched
> with the given offset matches the intended one by checking whether the
> fetched message's timestamp is equal to the expected timestamp. If not, this
> method returns the offset of the first message found with the expected
> timestamp.
> 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 call call the
> method in {{KafkaOffsetProvider}} to get the correct offset to start fetching
> messages with, when the original starting offset does not exist or the
> message given doesn't match the timestamp stored in KafkaOffsetProvider . If
> neither of the offset mismatches happen or the {{KafkaOffsetProvider}}
> instance is null, there's no check for inconsistent offset and this method
> will just return null.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
