Message offset adjust ticket #1010 #1017

breadpowder · 2019-10-18T15:49:33Z

This is for ticket id #1010 regarding offset.

The offset validation logic is refactored a bit to ensure working for both transactional and non-transactional mode. Since:

in transactional mode messages are not +1 between messages
in non-transaction mode, there might be very minor chances a message get delivered more than once due to retrying.

Now the offset logic in message writer works: If the current processing message offset is not greater than "last seen offset" (messages have written), it means there are rebalances and/or the broker is replaying message from last committed offset/failure point. Since the messages before offsets are written, we can safely trim those message to remove duplicates.

The existing commit offset validation logic in upload class remains intact . The existing logic ensures that after rebalances, if another consumer has made progress on a same topic/partition, we either trims those files or trim corresponding completed offset.

HenryCaiHaiying · 2019-10-18T17:32:24Z

There are too many files changed in this PR. Please check your coding style. We don't want to use wildcard imports, each individual package needs to be explicitly specified and imported.

breadpowder · 2019-10-18T19:20:43Z

Done. Please review the latest one. This one is rebased on most recent master and leaves my commits clear to read.

HenryCaiHaiying

I don't think the fix needs to be that complicated.

To distinguish the case between:

offset progression no longer go sequential because consumer rebalance happens
offset progression jumps because of transactional message

We can use ConsumerRebalanceListener to listen for consumer rebalance events, we already used SecurMessageIterator for those callback events, we can a bit more logic there to clear the lastSeenOffsets as well as remove underlying files if the partition assignment changes.

If we do that cleanup in the ConsumerRebalanceListener, we can simply change != to < in adjustOffset() method.

breadpowder · 2019-10-21T15:40:34Z

How about a bit enhancement over the above approach? Instead of removing the underlying files when rebalancing happens, let's upload the file and commit the offset in the partition revoked notification. Therefore, consumers don't need to rewrite the same data.

HenryCaiHaiying · 2019-10-21T17:27:28Z

If the rebalance happens, the files the current consumer was working on are no longer source of truth. The other consumer can already upload all or part of those offsets to S3. The other situation is the consumer was working on partition 0 for some time, then rebalance event 1 happens, he lost partition 0 but he still keep some unfinished partition 0 in local filesystem. After sometime another rebalance event 2 happens, he got back partition 0 but the offset jumps ahead already. If he continues appending the new messages from partition 0 to local files, those files will have a gap on message offsets. So it's best to clear all local files when offset inconsistency discovers.

The messages are not lost because some other consumers already uploaded those offsets to S3.

The transactional message is a bit different story, that's why we need to distinguish between this case and consumer rebalance.

breadpowder · 2019-10-21T18:12:32Z

Thank you! Here, I understand and agree with your points here regarding the other customer takes over and the possibility about offset jump when a previous partition assigned back .

Here, my thinking is we don't need to clear the local files to ensure consistency.Instead, we can do a upload and commit offset in the "partition revoke" callback given the kafka consumerRebalanceListener Api's promises (http://kafka.apache.org/20/javadoc/index.html?org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html). The api document says "this method will be called before a rebalance operation starts and after the consumer stops fetching data. It is recommended that offsets should be committed in this callback to either Kafka or a custom offset store to prevent duplicate data".

Therefore, we can safely upload, commit the offset and clear the lastseen offset after process existing messages buffer in the "partition revoke callback". It won't cause data inconsistency issue given the other consumer will start from the newly committed offset, as they work sequentially without any messages overlap.

By doing this, the benefit is that the other customer don't need to reprocess the same data. Let me know your thoughts. Thank you again for sharing your thoughts!

HenryCaiHaiying · 2019-10-21T22:36:31Z

If you add the uploading code in the ConsumerRebalanceListener, the uploading will clear the local files automatically so the old check and file clear might not be really needed. It's probably also a good idea to clear the lastSeen data structure on that rebalance listener so everything start fresh after rebalance. We don't have that consumer rebalance listener before, but this new listener we actually can do things more cleanly.

…

On Mon, Oct 21, 2019 at 11:12 AM breadpowder ***@***.***> wrote: Thank you! Here, I understand and agree with your points here regarding the other customer took over and the possibility about offset jump when a previous partition assigned back . Here, my thinking is we don't need to clear the local files to ensure consistency, instead, we can do a upload and commit offset in the "partition revoke" callback given the kafka consumerRebalanceListener Api's promises ( http://kafka.apache.org/20/javadoc/index.html?org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html). The api document says "this method will be called before a rebalance operation starts and after the consumer stops fetching data. It is recommended that offsets should be committed in this callback to either Kafka or a custom offset store to prevent duplicate data". Therefore, we can safely upload and commit offset after process existing messages buffer in the "partition revoke callback". It won't cause data inconsistency issue given the other consumer will start from the newly committed offset, as they work sequentially without any offset overlapping. By doing this, the benefit is that the other customer don't need to reprocess the same data. Let me know your thoughts. Thank you again for sharing your thoughts! — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1017?email_source=notifications&email_token=ABYJP7Z6WHSKVTWAHMPQGCLQPXWJJA5CNFSM4JCJGYLKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEB3IQSQ#issuecomment-544639050>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ABYJP7ZHNUZ3MD55HV3UTW3QPXWJJANCNFSM4JCJGYLA> .

breadpowder · 2019-10-21T22:48:29Z

Sure. will commit soon.

HenryCaiHaiying · 2019-10-21T23:39:21Z

Looking forward to your PR

…

On Mon, Oct 21, 2019 at 3:48 PM breadpowder ***@***.***> wrote: Sure. will commit soon. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1017?email_source=notifications&email_token=ABYJP753CG4IGF3VRQIS7EDQPYWUBA5CNFSM4JCJGYLKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEB4BH6Q#issuecomment-544740346>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ABYJP74YRR32GDQLPZRRR33QPYWUBANCNFSM4JCJGYLA> .

breadpowder · 2019-10-23T16:35:51Z

As discussed, rebalance logic is added into SecorConsumerRebalanceListener and RebalanceHandler. Please review.

HenryCaiHaiying

In general, the code change looks fine. But there are too many lines changed to make the diff review difficult. I wasn't quite sure why you need to restructure the class/inner-classes in SecorKafkaMessageIterator, it seems you can can just add the new reset logic in the ConsumerRebalancerListener inner class.

HenryCaiHaiying · 2019-10-23T18:25:20Z

src/main/java/com/pinterest/secor/common/OffsetTracker.java

-                        topicPartition.getTopic(),topicPartition.getPartition(),lastSeenOffset, offset);
-            } else {
+        if (offset < lastSeenOffset + 1) {
+            LOG.warn("offset for topic {} partition {} goes back from {} to {}",


How about we throw an exception here? I don't think we should see this happen with the code change in Rebalance listener?

For compatibility with LegacyKafkaMessageIterator, since no such listener exists, we can't throw exception here. Any idea?

Would prefer to use a separator class for RebalanceListener since rebalance and handler logic is separate concern from message iterator so better to be a complete seperate unit.

Let me know if the above is ok, will push shortly.

src/main/java/com/pinterest/secor/writer/MessageWriter.java

#fixed offset

breadpowder · 2019-10-23T23:26:54Z

Just happened to find that this issue also related another ticket #682

HenryCaiHaiying · 2019-10-25T17:36:30Z

lgtm, thanks for the effort.

breadpowder force-pushed the message-offset-adjust-1010 branch 2 times, most recently from 564b436 to 4a2f234 Compare October 18, 2019 19:16

HenryCaiHaiying reviewed Oct 19, 2019

View reviewed changes

breadpowder force-pushed the message-offset-adjust-1010 branch from 4a2f234 to dda16af Compare October 23, 2019 16:30

HenryCaiHaiying reviewed Oct 23, 2019

View reviewed changes

Fixed offset tracking in rebalance callback

2c7eec6

#fixed offset

breadpowder force-pushed the message-offset-adjust-1010 branch from dda16af to 2c7eec6 Compare October 23, 2019 19:34

breadpowder requested a review from HenryCaiHaiying October 24, 2019 20:19

HenryCaiHaiying merged commit 5c51d7e into pinterest:master Oct 25, 2019

HenryCaiHaiying mentioned this pull request May 25, 2020

Unable To Process Transaction API Events #682

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Message offset adjust ticket #1010 #1017

Message offset adjust ticket #1010 #1017

breadpowder commented Oct 18, 2019

HenryCaiHaiying commented Oct 18, 2019

breadpowder commented Oct 18, 2019 •

edited

Loading

HenryCaiHaiying left a comment

breadpowder commented Oct 21, 2019 •

edited

Loading

HenryCaiHaiying commented Oct 21, 2019

breadpowder commented Oct 21, 2019 •

edited

Loading

HenryCaiHaiying commented Oct 21, 2019 via email

breadpowder commented Oct 21, 2019

HenryCaiHaiying commented Oct 21, 2019 via email

breadpowder commented Oct 23, 2019

HenryCaiHaiying left a comment

HenryCaiHaiying Oct 23, 2019

breadpowder Oct 23, 2019 •

edited

Loading

breadpowder Oct 23, 2019 •

edited

Loading

breadpowder Oct 23, 2019

breadpowder commented Oct 23, 2019 •

edited

Loading

HenryCaiHaiying commented Oct 25, 2019

Message offset adjust ticket #1010 #1017

Message offset adjust ticket #1010 #1017

Conversation

breadpowder commented Oct 18, 2019

HenryCaiHaiying commented Oct 18, 2019

breadpowder commented Oct 18, 2019 • edited Loading

HenryCaiHaiying left a comment

Choose a reason for hiding this comment

breadpowder commented Oct 21, 2019 • edited Loading

HenryCaiHaiying commented Oct 21, 2019

breadpowder commented Oct 21, 2019 • edited Loading

HenryCaiHaiying commented Oct 21, 2019 via email

breadpowder commented Oct 21, 2019

HenryCaiHaiying commented Oct 21, 2019 via email

breadpowder commented Oct 23, 2019

HenryCaiHaiying left a comment

Choose a reason for hiding this comment

HenryCaiHaiying Oct 23, 2019

Choose a reason for hiding this comment

breadpowder Oct 23, 2019 • edited Loading

Choose a reason for hiding this comment

breadpowder Oct 23, 2019 • edited Loading

Choose a reason for hiding this comment

breadpowder Oct 23, 2019

Choose a reason for hiding this comment

breadpowder commented Oct 23, 2019 • edited Loading

HenryCaiHaiying commented Oct 25, 2019

breadpowder commented Oct 18, 2019 •

edited

Loading

breadpowder commented Oct 21, 2019 •

edited

Loading

breadpowder commented Oct 21, 2019 •

edited

Loading

breadpowder Oct 23, 2019 •

edited

Loading

breadpowder Oct 23, 2019 •

edited

Loading

breadpowder commented Oct 23, 2019 •

edited

Loading