[GitHub] storm pull request #2181: [STORM-2607] Offset consumer + 1

[srdo <gi...@git.apache.org>]

Github user srdo commented on a diff in the pull request:

    https://github.com/apache/storm/pull/2181#discussion_r125522989
  
    --- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java ---
    @@ -96,31 +98,34 @@ public OffsetAndMetadata findNextCommitOffset() {
                             the next logical point in the topic. Next logical offset should be the
                             first element after committedOffset in the ascending ordered emitted set.
                          */
    -                    LOG.debug("Processed non contiguous offset."
    -                        + " (committedOffset+1) is no longer part of the topic."
    -                        + " Committed: [{}], Processed: [{}]", committedOffset, currOffset);
    -                    final Long nextEmittedOffset = emittedOffsets.ceiling(nextCommitOffset);
    +                    LOG.debug("Processed non-contiguous offset."
    +                        + " The earliest uncommitted offset is no longer part of the topic."
    +                        + " Missing uncommitted offset: [{}], Processed: [{}]", nextEarliestUncommittedOffset, currOffset);
    +                    final Long nextEmittedOffset = emittedOffsets.ceiling(nextEarliestUncommittedOffset);
                         if (nextEmittedOffset != null && currOffset == nextEmittedOffset) {
    +                        LOG.debug("Found committable offset: [{}] after missing offset: [{}], skipping to the committable offset",
    +                            currOffset, nextEarliestUncommittedOffset);
                             nextCommitMsg = currAckedMsg;
    -                        nextCommitOffset = currOffset;
    +                        nextEarliestUncommittedOffset = currOffset + 1;
                         } else {
    -                        LOG.debug("topic-partition [{}] has non-continuous offset [{}]."
    -                            + " Next Offset to commit should be [{}]", tp, currOffset, nextEmittedOffset);
    +                        LOG.debug("topic-partition [{}] has non-contiguous offset [{}]."
    +                            + " Next Offset to commit should be [{}]", tp, currOffset, nextEarliestUncommittedOffset - 1);
                             break;
                         }
                     }
                 } else {
                     //Received a redundant ack. Ignore and continue processing.
    -                LOG.warn("topic-partition [{}] has unexpected offset [{}]. Current committed Offset [{}]",
    -                    tp, currOffset, committedOffset);
    +                LOG.warn("topic-partition [{}] has unexpected offset [{}]. Current earliest uncommitted offset [{}]",
    +                    tp, currOffset, earliestUncommittedOffset);
                 }
             }
     
             OffsetAndMetadata nextCommitOffsetAndMetadata = null;
             if (nextCommitMsg != null) {
    -            nextCommitOffsetAndMetadata = new OffsetAndMetadata(nextCommitOffset, nextCommitMsg.getMetadata(Thread.currentThread()));
    +            nextCommitOffsetAndMetadata = new OffsetAndMetadata(nextEarliestUncommittedOffset,
    +                nextCommitMsg.getMetadata(Thread.currentThread()));
                 LOG.debug("topic-partition [{}] has offsets [{}-{}] ready to be committed",
    -                tp, committedOffset + 1, nextCommitOffsetAndMetadata.offset());
    +                tp, earliestUncommittedOffset, nextCommitOffsetAndMetadata.offset());
    --- End diff --
    
    Whoops, missed a -1 here. It should be nextCommitOffsetAndMetadata.offset() - 1


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---