[STORM-2607] Offset consumer + 1 #2181

tiodollar · 2017-06-29T15:07:43Z

When i put a message in a partition, the storm-kafka-client consume this message.
But storm-kafka-client commit the offset -1.
storm-kafka-client: 1.1.0
storm-core : 1.1.0
kafka: 0.10.2.0
Steps to bug
#1 - Insert message in kafka
#2 - Read with storm Spout this topic
#3 - Get the offset for the consumer group and the offset is always offset -1

https://issues.apache.org/jira/browse/STORM-2607

srdo · 2017-06-29T17:32:47Z

@tiodollar Nice find. The fix is not workable though. We need to update OffsetManager to handle that the commit offset should be the last processed offset +1, as per the Kafka docs for commitSync: "The committed offset should be the next message your application will consume, i.e. lastProcessedMessageOffset + 1".

Would you like to take a look at fixing this? I'd also be happy to take a stab at it.

tiodollar · 2017-06-29T17:41:52Z

@srdo I'll take look, and try fixing this!

srdo · 2017-06-29T20:37:35Z

...al/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java

                    preCommitCommittedOffsets + 1, this.committedOffset, numCommittedOffsets, tp);

-        return numCommittedOffsets;
+        return numCommittedOffsets - 1;// The committed offset should be the next message


Please fix this in L144 instead, so the log messages above print the right number as well

srdo · 2017-06-29T20:37:42Z

...al/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java

        OffsetAndMetadata nextCommitOffsetAndMetadata = null;
        if (nextCommitMsg != null) {
+            //The committed offset should be the next message
+            nextCommitOffset = nextCommitOffset + 1;


I would prefer that we fix the offsets in the code block above (L76-117) instead of adjusting the offset here. That way the log messages won't be printing the wrong offset.

… instead of last committed offset for compatibility with commitSync consumer API

@srdo

STORM-2607: Switch OffsetManager to track earliest uncommitted offset… by @srdo

srdo

@tiodollar Thanks. Tested this with an example topology, and the spout seems to commit all the messages now.
@hmcl Could I get you to review this?

srdo · 2017-07-04T19:33:35Z

...al/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java

+                nextCommitMsg.getMetadata(Thread.currentThread()));
            LOG.debug("topic-partition [{}] has offsets [{}-{}] ready to be committed",
-                tp, committedOffset + 1, nextCommitOffsetAndMetadata.offset());
+                tp, earliestUncommittedOffset, nextCommitOffsetAndMetadata.offset());


Whoops, missed a -1 here. It should be nextCommitOffsetAndMetadata.offset() - 1

@srdo I corrected that!

tiodollar · 2017-07-05T14:00:32Z

@hmcl Could I get you to review this?

hmcl · 2017-07-06T14:47:57Z

@tiodollar ok

srdo · 2017-07-19T10:42:14Z

@tiodollar Seems like there's a conflict with master. Could you fix it?

tiodollar · 2017-07-21T16:06:19Z

@srdo i'll fix this! Tks!

HeartSaVioR · 2017-08-07T23:46:14Z

@tiodollar Any updates?

srdo · 2017-08-14T13:21:08Z

We should also remove the +1 here I think https://github.com/tiodollar/storm/blob/da7f49b94502914af769d7f40929bd54d3fae865/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java#L201

tiodollar · 2017-08-14T14:15:00Z

@HeartSaVioR i'll fix this merge today!

hmcl · 2017-08-15T23:16:01Z

@tiodollar once you do the merge, can you please squash all the commits. Thanks.

srdo · 2017-09-05T15:35:18Z

@tiodollar Please let me know if you're having trouble resolving the conflicts. I'd be happy to help.

Chandan83 · 2017-09-18T09:01:04Z

Is there any reason this merge is pending? Can I help?

HeartSaVioR · 2017-09-19T12:12:47Z

Anyone could take this over since this PR looks like inactive.
Please keep in mind that you are encouraged to preserve @tiodollar commits with keeping authorship. Also encouraged to squash commits into one, and add your commits if needed.

srdo · 2017-09-19T16:03:39Z

Yes, we should get this in soon if @tiodollar is busy. @Chandan83 if you'd like to work on this please create an account at https://issues.apache.org/jira and let us know, so we can assign the issue to you.

With regard to squashing, I think it should be possible to squash the existing commits into one for @tiodollar's commits and one for mine and preserve authorship information that way.

Chandan83 · 2017-09-21T08:19:41Z

@srdo I have created an apache jira account with id ChandanKrSingh. Please assign the issue to me.

tiodollar · 2017-09-21T11:10:46Z

@srdo i'll resolve this issue today! I have been busy, but i reseved some hour to do this.

Chandan83 · 2017-09-21T11:40:23Z

@tiodollar Great. Thanks.

srdo · 2017-10-10T19:46:48Z

@tiodollar Since there is still a conflict here, and we'd like this to go in 1.1.2 which we hope to release before too long, I've resolved the conflicts and addressed the review comments. There's a PR with the full set of changes here #2367. Note that your commits are still present, so you should still be credited with the fix.

Let me know if this is not okay with you.

tiodollar · 2017-10-11T13:43:36Z

@srdo tks for this! I'm very busy, and is better that! For me it's ok!

srdo · 2017-10-11T16:57:55Z

@tiodollar Happy to hear it :)

[STORM-2607] Offset consumer + 1

34ffb95

[STORM-2607] Fixed committed offset should be the next message.

c3721f5

srdo reviewed Jun 29, 2017

View reviewed changes

Rodolfo Ribeiro and others added 2 commits June 29, 2017 19:30

[STORM-2607] Fixed PR

5c72981

STORM-2607: Switch OffsetManager to track earliest uncommitted offset…

f192ad2

… instead of last committed offset for compatibility with commitSync consumer API

srdo mentioned this pull request Jul 1, 2017

STORM-2607: Switch OffsetManager to track earliest uncommitted offset… tiodollar/storm#1

Merged

Merge pull request #1 from srdo/STORM-2607

c02a1e8

STORM-2607: Switch OffsetManager to track earliest uncommitted offset… by @srdo

srdo reviewed Jul 4, 2017

View reviewed changes

[STORM-2607] Fixed PR

da7f49b

srdo mentioned this pull request Oct 10, 2017

STORM-2607: Storm-kafka-client never commits the last message on a partition. #2367

Merged

tiodollar closed this Oct 14, 2017

[STORM-2607] Offset consumer + 1 #2181

[STORM-2607] Offset consumer + 1 #2181

Uh oh!

Conversation

tiodollar commented Jun 29, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

srdo commented Jun 29, 2017

Uh oh!

tiodollar commented Jun 29, 2017

Uh oh!

srdo Jun 29, 2017

Choose a reason for hiding this comment

Uh oh!

tiodollar Jun 29, 2017

Choose a reason for hiding this comment

Uh oh!

srdo Jun 29, 2017

Choose a reason for hiding this comment

Uh oh!

tiodollar Jun 29, 2017

Choose a reason for hiding this comment

Uh oh!

srdo left a comment

Choose a reason for hiding this comment

Uh oh!

srdo Jul 4, 2017

Choose a reason for hiding this comment

Uh oh!

tiodollar Jul 4, 2017

Choose a reason for hiding this comment

Uh oh!

srdo Jul 4, 2017

Choose a reason for hiding this comment

Uh oh!

tiodollar commented Jul 5, 2017

Uh oh!

hmcl commented Jul 6, 2017

Uh oh!

srdo commented Jul 19, 2017

Uh oh!

tiodollar commented Jul 21, 2017

Uh oh!

HeartSaVioR commented Aug 7, 2017

Uh oh!

srdo commented Aug 14, 2017

Uh oh!

tiodollar commented Aug 14, 2017

Uh oh!

hmcl commented Aug 15, 2017

Uh oh!

srdo commented Sep 5, 2017

Uh oh!

Chandan83 commented Sep 18, 2017

Uh oh!

HeartSaVioR commented Sep 19, 2017

Uh oh!

srdo commented Sep 19, 2017

Uh oh!

Chandan83 commented Sep 21, 2017

Uh oh!

tiodollar commented Sep 21, 2017

Uh oh!

Chandan83 commented Sep 21, 2017

Uh oh!

srdo commented Oct 10, 2017

Uh oh!

tiodollar commented Oct 11, 2017

Uh oh!

srdo commented Oct 11, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

tiodollar commented Jun 29, 2017 •

edited

Loading