public OffsetAndMetadata findNextCommitOffset(final String commitMetadata) {
        boolean found = false;
        long currOffset;
        long nextCommitOffset = committedOffset;

        for (KafkaSpoutMessageId currAckedMsg : ackedMsgs) {  // complexity is that of a linear scan on a TreeMap
            currOffset = currAckedMsg.offset();
            if (currOffset == nextCommitOffset) {
                // found the next offset to commit
                found = true;
                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-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 nextCommitOffset in the ascending ordered emitted set.
                     */
                    LOG.debug("Processed non-sequential offset."
                        + " The earliest uncommitted offset is no longer part of the topic."
                        + " Missing offset: [{}], Processed: [{}]", nextCommitOffset, currOffset);
                    final Long nextEmittedOffset = emittedOffsets.ceiling(nextCommitOffset);
                    if (nextEmittedOffset != null && currOffset == nextEmittedOffset) {
                        LOG.debug("Found committable offset: [{}] after missing offset: [{}], skipping to the committable offset",
                            currOffset, nextCommitOffset);
                        found = true;
                        nextCommitOffset = currOffset + 1;
                    } else {
                        LOG.debug("Topic-partition [{}] has non-sequential offset [{}]."
                            + " Next offset to commit should be [{}]", tp, currOffset, nextCommitOffset);
                        break;
                    }
                }
            } else {
                throw new IllegalStateException("The offset [" + currOffset + "] is below the current nextCommitOffset "
                    + "[" + nextCommitOffset + "] for [" + tp + "]."
                    + " This should not be possible, and likely indicates a bug in the spout's acking or emit logic.");
            }
        }

        OffsetAndMetadata nextCommitOffsetAndMetadata = null;
        if (found) {
            nextCommitOffsetAndMetadata = new OffsetAndMetadata(nextCommitOffset, commitMetadata);

            LOG.debug("Topic-partition [{}] has offsets [{}-{}] ready to be committed."
                + " Processing will resume at offset [{}] upon spout restart",
                tp, committedOffset, nextCommitOffsetAndMetadata.offset() - 1, nextCommitOffsetAndMetadata.offset());
        } else {
            LOG.debug("Topic-partition [{}] has no offsets ready to be committed", tp);
        }
        LOG.trace("{}", this);
        return nextCommitOffsetAndMetadata;
    }
@Test
    public void testFindNextCommitOffsetWithMultipleOutOfOrderAcks() {
        emitAndAckMessage(getMessageId(initialFetchOffset + 1));
        emitAndAckMessage(getMessageId(initialFetchOffset));
        OffsetAndMetadata nextCommitOffset = manager.findNextCommitOffset(COMMIT_METADATA);
        assertThat("The next commit offset should be one past the processed message offset", nextCommitOffset.offset(), is(initialFetchOffset + 2));
    }