You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by srdo <gi...@git.apache.org> on 2017/10/10 19:46:35 UTC
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
GitHub user srdo opened a pull request:
https://github.com/apache/storm/pull/2367
STORM-2607: Storm-kafka-client never commits the last message on a partition.
This finishes up https://github.com/apache/storm/pull/2181, which has been blocked on a rebase for a while. It was suggested on the mailing list that we finish this up quickly since the fix should ideally go in 1.1.2.
I've squashed the commits as much as possible without mixing up which changes were made by @tiodollar and which were made by me. Let me know if squashing is not okay in this case, and I'll rebase on the commits in the original PR.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/srdo/storm STORM-2607-fixup
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/2367.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 #2367
----
commit 4dca66c24b4ec40f0ab4badac9042fc3f2d9e01e
Author: Rodolfo Ribeiro <ro...@b2wdigital.com>
Date: 2017-06-29T15:06:20Z
[STORM-2607] Offset consumer + 1
commit 51e5f707623e33b877a21c4fe05ad7564aaaf702
Author: Stig Rohde Døssing <st...@gmail.com>
Date: 2017-07-01T15:29:56Z
STORM-2607: Switch OffsetManager to track earliest uncommitted offset instead of last committed offset for compatibility with commitSync consumer API
commit 99569b2b2318f0559a9705be53ed3e5646b468ad
Author: Rodolfo Ribeiro <ro...@b2wdigital.com>
Date: 2017-07-04T22:13:05Z
[STORM-2607] Fixed PR
commit fd0dc1fbd36508069e8f6a906a65120e8a6f60b4
Author: Stig Rohde Døssing <sr...@apache.org>
Date: 2017-10-10T17:11:35Z
Merge branch 'asfgit-master' into STORM-2607-fixup
commit fd50c487f8a07b9705b8172a401374c9784436b2
Author: Stig Rohde Døssing <sr...@apache.org>
Date: 2017-10-10T19:20:06Z
STORM-2607: Fix merge conflicts, fix pending review comment, rename variables in OffsetManager to be consistent with the KafkaConsumer API definition of 'committed offset', fix tests, replace uses of deprecated Mockito methods
----
---
[GitHub] storm issue #2367: STORM-2607: Storm-kafka-client never commits the last mes...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2367
@hmcl Just signed in on Gitter, feel free to message me.
---
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2367#discussion_r144044782
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java ---
@@ -53,8 +54,8 @@
public OffsetManager(TopicPartition tp, long initialFetchOffset) {
this.tp = tp;
this.initialFetchOffset = initialFetchOffset;
- this.committedOffset = initialFetchOffset - 1;
- LOG.debug("Instantiated {}", this);
+ this.committedOffset = initialFetchOffset;
+ LOG.debug("Instantiated {}", this.toString());
--- End diff --
toString() is redundant
---
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2367#discussion_r143914514
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java ---
@@ -77,57 +78,58 @@ public void addToEmitMsgs(long offset) {
*/
public OffsetAndMetadata findNextCommitOffset() {
long currOffset;
- long nextEarliestUncommittedOffset = earliestUncommittedOffset;
+ long nextCommitOffset = committedOffset;
KafkaSpoutMessageId nextCommitMsg = null; // this is a convenience variable to make it faster to create OffsetAndMetadata
for (KafkaSpoutMessageId currAckedMsg : ackedMsgs) { // complexity is that of a linear scan on a TreeMap
currOffset = currAckedMsg.offset();
- if (currOffset == nextEarliestUncommittedOffset) { // found the next offset to commit
+ if (currOffset == nextCommitOffset) { // found the next offset to commit
nextCommitMsg = currAckedMsg;
- nextEarliestUncommittedOffset = currOffset + 1;
- } else if (currOffset > nextEarliestUncommittedOffset) {
- if (emittedOffsets.contains(nextEarliestUncommittedOffset)) {
- LOG.debug("topic-partition [{}] has non-contiguous offset [{}]."
+ nextCommitOffset = currOffset + 1;
+ } else if (currOffset > nextCommitOffset) {
+ if (emittedOffsets.contains(nextCommitOffset)) {
+ LOG.debug("topic-partition [{}] has non-sequential offset [{}]."
+ " It will be processed in a subsequent batch.", tp, currOffset);
break;
} else {
/*
- This case will arise in case of non contiguous offset being processed.
- So, if the topic doesn't contain offset = committedOffset + 1 (possible
+ This case will arise in case of non-sequential offset being processed.
+ So, if the topic doesn't contain offset = nextCommitOffset (possible
if the topic is compacted or deleted), the consumer should jump to
the next logical point in the topic. Next logical offset should be the
- first element after committedOffset in the ascending ordered emitted set.
+ first element after nextEarliestUncommittedOffset in the ascending ordered emitted set.
--- End diff --
Nice catch, fixed.
---
[GitHub] storm issue #2367: STORM-2607: Storm-kafka-client never commits the last mes...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on the issue:
https://github.com/apache/storm/pull/2367
@srdo are you on Gitter or some sort of chat room where I could ask you a few questions? Thanks.
---
[GitHub] storm issue #2367: STORM-2607: Storm-kafka-client never commits the last mes...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2367
Merged, thanks for the reviews guys.
---
[GitHub] storm issue #2367: STORM-2607: Storm-kafka-client never commits the last mes...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2367
@hmcl It's not that I feel strongly about keeping the contributions separate, but I think it's a rule for Apache projects that it must be possible to tell who contributed what, even historically. I can't recall where it says so though.
Squashed to two commits.
---
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2367#discussion_r144078200
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java ---
@@ -53,8 +54,8 @@
public OffsetManager(TopicPartition tp, long initialFetchOffset) {
this.tp = tp;
this.initialFetchOffset = initialFetchOffset;
- this.committedOffset = initialFetchOffset - 1;
- LOG.debug("Instantiated {}", this);
+ this.committedOffset = initialFetchOffset;
+ LOG.debug("Instantiated {}", this.toString());
--- End diff --
Thanks for the explanation. I just noticed that `this` is called within its own constructor, and therefore makes sense. However, IntelliJ does not give me this warning.
---
[GitHub] storm issue #2367: STORM-2607: Storm-kafka-client never commits the last mes...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2367
@hmcl I agree that this many commits is not ideal. The issue is with being able to track who contributed what code. If I squash we can't tell which parts of this PR was contributed by me, and which were contributed by @tiodollar. It's my impression that we need to keep the history of who contributed what intact.
---
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/2367
---
[GitHub] storm issue #2367: STORM-2607: Storm-kafka-client never commits the last mes...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on the issue:
https://github.com/apache/storm/pull/2367
I read a mail at dev@ from @hmcl and the talk was not relevant to this patch. +1 again.
---
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2367#discussion_r144080187
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java ---
@@ -163,8 +167,9 @@ public long commit(OffsetAndMetadata committedOffset) {
LOG.trace("{}", this);
- LOG.debug("Committed [{}] offsets in the range [{}-{}] for topic-partition [{}].",
- numCommittedOffsets, preCommitCommittedOffsets + 1, this.committedOffset, tp);
+ LOG.debug("Committed [{}] offsets in the range [{}-{}] for topic-partition [{}]."
+ + " Processing will resume at [{}] if the spout restarts.",
--- End diff --
Good catch, will fix
---
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2367#discussion_r144062571
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java ---
@@ -53,8 +54,8 @@
public OffsetManager(TopicPartition tp, long initialFetchOffset) {
this.tp = tp;
this.initialFetchOffset = initialFetchOffset;
- this.committedOffset = initialFetchOffset - 1;
- LOG.debug("Instantiated {}", this);
+ this.committedOffset = initialFetchOffset;
+ LOG.debug("Instantiated {}", this.toString());
--- End diff --
Yes, but I get a warning about leaking this from the constructor, which this fixes.
---
[GitHub] storm issue #2367: STORM-2607: Storm-kafka-client never commits the last mes...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on the issue:
https://github.com/apache/storm/pull/2367
+1
Would this be a clean merge to 1.x version lines?
---
[GitHub] storm issue #2367: STORM-2607: Storm-kafka-client never commits the last mes...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on the issue:
https://github.com/apache/storm/pull/2367
@srdo in my opinion the most important thing is to keep the project, and in particular the git log clean. I often have to cherry-pick changes from Apache branch into my own branch, and multiple commits for a JIRA turn branch maintenance nightmare. Several Apache projects have a strict rule of one JIRA one Git commit, which I believe is very beneficial.
If you feel really strongly about differentiating yours and his contributions, this patch must have at most two commits. One has all his contributions, and another yours. You can squash the commits and put @tiodollar as author as described in [here](https://stackoverflow.com/questions/3042437/change-commit-author-at-one-specific-commit).
---
[GitHub] storm issue #2367: STORM-2607: Storm-kafka-client never commits the last mes...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2367
@HeartSaVioR No, it doesn't look like it. I'll put up a 1.x and 1.1.x version.
---
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2367#discussion_r144079454
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java ---
@@ -163,8 +167,9 @@ public long commit(OffsetAndMetadata committedOffset) {
LOG.trace("{}", this);
- LOG.debug("Committed [{}] offsets in the range [{}-{}] for topic-partition [{}].",
- numCommittedOffsets, preCommitCommittedOffsets + 1, this.committedOffset, tp);
+ LOG.debug("Committed [{}] offsets in the range [{}-{}] for topic-partition [{}]."
+ + " Processing will resume at [{}] if the spout restarts.",
--- End diff --
at offset [{}] in case of a spout restart.
---
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2367#discussion_r144069466
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java ---
@@ -53,8 +54,8 @@
public OffsetManager(TopicPartition tp, long initialFetchOffset) {
this.tp = tp;
this.initialFetchOffset = initialFetchOffset;
- this.committedOffset = initialFetchOffset - 1;
- LOG.debug("Instantiated {}", this);
+ this.committedOffset = initialFetchOffset;
+ LOG.debug("Instantiated {}", this.toString());
--- End diff --
The warning I get is just "leaking this in constructor", Netbeans has the following description: "Using this as parameter can be dangerous in the contructor because the object is not fully initialized.". IntelliJ and Eclipse have similar warnings I believe.
It's basically saying not to use `this` as a parameter from the constructor, because the called method may use fields that aren't initialized. I agree that leaking `this` from the constructor is not dangerous here, because sfl4j just calls toString, but I'd still like to get rid of the warning. The reason this is flagged is because `this` is used as a parameter to LOG.debug, and not as part of a string concatenation. The IDE can't tell whether `this` is used unsafely in LOG.debug, so it's flagged as a warning.
---
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/storm/pull/2367#discussion_r143863556
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java ---
@@ -77,57 +78,58 @@ public void addToEmitMsgs(long offset) {
*/
public OffsetAndMetadata findNextCommitOffset() {
long currOffset;
- long nextEarliestUncommittedOffset = earliestUncommittedOffset;
+ long nextCommitOffset = committedOffset;
KafkaSpoutMessageId nextCommitMsg = null; // this is a convenience variable to make it faster to create OffsetAndMetadata
for (KafkaSpoutMessageId currAckedMsg : ackedMsgs) { // complexity is that of a linear scan on a TreeMap
currOffset = currAckedMsg.offset();
- if (currOffset == nextEarliestUncommittedOffset) { // found the next offset to commit
+ if (currOffset == nextCommitOffset) { // found the next offset to commit
nextCommitMsg = currAckedMsg;
- nextEarliestUncommittedOffset = currOffset + 1;
- } else if (currOffset > nextEarliestUncommittedOffset) {
- if (emittedOffsets.contains(nextEarliestUncommittedOffset)) {
- LOG.debug("topic-partition [{}] has non-contiguous offset [{}]."
+ nextCommitOffset = currOffset + 1;
+ } else if (currOffset > nextCommitOffset) {
+ if (emittedOffsets.contains(nextCommitOffset)) {
+ LOG.debug("topic-partition [{}] has non-sequential offset [{}]."
+ " It will be processed in a subsequent batch.", tp, currOffset);
break;
} else {
/*
- This case will arise in case of non contiguous offset being processed.
- So, if the topic doesn't contain offset = committedOffset + 1 (possible
+ This case will arise in case of non-sequential offset being processed.
+ So, if the topic doesn't contain offset = nextCommitOffset (possible
if the topic is compacted or deleted), the consumer should jump to
the next logical point in the topic. Next logical offset should be the
- first element after committedOffset in the ascending ordered emitted set.
+ first element after nextEarliestUncommittedOffset in the ascending ordered emitted set.
--- End diff --
The field `nextEarliestUncommittedOffset` looks removed in the patch and this comment refers it.
---
[GitHub] storm issue #2367: STORM-2607: Storm-kafka-client never commits the last mes...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2367
I'll merge this weekend if I no one brings up more concerns.
---
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/2367#discussion_r144066568
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java ---
@@ -53,8 +54,8 @@
public OffsetManager(TopicPartition tp, long initialFetchOffset) {
this.tp = tp;
this.initialFetchOffset = initialFetchOffset;
- this.committedOffset = initialFetchOffset - 1;
- LOG.debug("Instantiated {}", this);
+ this.committedOffset = initialFetchOffset;
+ LOG.debug("Instantiated {}", this.toString());
--- End diff --
This is not a big deal. But can you tell me what kinda warning you are talking about? I never heard of such a thing. When you concatenate a string with an object reference, the toString() method of that object reference is called. I am not sure why both approaches are any different.
---
[GitHub] storm issue #2367: STORM-2607: Storm-kafka-client never commits the last mes...
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on the issue:
https://github.com/apache/storm/pull/2367
@srdo I understand that you want to keep the commit history for @tiodollar but I don't think we should have 4 commits for such a simple change. This patch should be one commit only, either yours or @tiodollar 's . Can either you or @tiodollar please squash the commits. Thanks.
---
[GitHub] storm pull request #2367: STORM-2607: Storm-kafka-client never commits the l...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2367#discussion_r144083046
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java ---
@@ -53,8 +54,8 @@
public OffsetManager(TopicPartition tp, long initialFetchOffset) {
this.tp = tp;
this.initialFetchOffset = initialFetchOffset;
- this.committedOffset = initialFetchOffset - 1;
- LOG.debug("Instantiated {}", this);
+ this.committedOffset = initialFetchOffset;
+ LOG.debug("Instantiated {}", this.toString());
--- End diff --
It's not enabled by default in IntelliJ. If you go to settings -> editor -> inspections and search for "this", there's one called "'this' reference escaped in object construction"
---