public long commit(OffsetAndMetadata committedOffsetAndMeta) {
        committed = true;
        final long preCommitCommittedOffset = this.committedOffset;
        long numCommittedOffsets = 0;
        this.committedOffset = committedOffsetAndMeta.offset();
        for (Iterator<KafkaSpoutMessageId> iterator = ackedMsgs.iterator(); iterator.hasNext();) {
            if (iterator.next().offset() < committedOffsetAndMeta.offset()) {
                iterator.remove();
                numCommittedOffsets++;
            } else {
                break;
            }
        }

        for (Iterator<Long> iterator = emittedOffsets.iterator(); iterator.hasNext();) {
            if (iterator.next() < committedOffsetAndMeta.offset()) {
                iterator.remove();
            } else {
                break;
            }
        }

        LOG.trace("{}", this);
        
        LOG.debug("Committed [{}] offsets in the range [{}-{}] for topic-partition [{}]."
            + " Processing will resume at [{}] upon spout restart",
                numCommittedOffsets, preCommitCommittedOffset, this.committedOffset - 1, tp, this.committedOffset);
        
        return numCommittedOffsets;
    }
@Test
    public void testCommit() {
        emitAndAckMessage(getMessageId(initialFetchOffset));
        emitAndAckMessage(getMessageId(initialFetchOffset + 1));
        emitAndAckMessage(getMessageId(initialFetchOffset + 2));
        
        long committedMessages = manager.commit(new OffsetAndMetadata(initialFetchOffset + 2));
        
        assertThat("Should have committed all messages to the left of the earliest uncommitted offset", committedMessages, is(2L));
        assertThat("The committed messages should not be in the acked list anymore", manager.contains(getMessageId(initialFetchOffset)), is(false));
        assertThat("The committed messages should not be in the emitted list anymore", manager.containsEmitted(initialFetchOffset), is(false));
        assertThat("The committed messages should not be in the acked list anymore", manager.contains(getMessageId(initialFetchOffset + 1)), is(false));
        assertThat("The committed messages should not be in the emitted list anymore", manager.containsEmitted(initialFetchOffset + 1), is(false));
        assertThat("The uncommitted message should still be in the acked list", manager.contains(getMessageId(initialFetchOffset + 2)), is(true));
        assertThat("The uncommitted message should still be in the emitted list", manager.containsEmitted(initialFetchOffset + 2), is(true));
    }