Github user srdo commented on a diff in the pull request: https://github.com/apache/storm/pull/2181#discussion_r125533526 --- 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 -- Thanks
--- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---