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.
---
