org.apache.storm.kafka.spout.internal

Class OffsetManager

java.lang.Object

org.apache.storm.kafka.spout.internal.OffsetManager

public class OffsetManager
extends Object

Manages acked and committed offsets for a TopicPartition. This class is not thread safe

Constructor Summary

Constructors

Constructor and Description
OffsetManager(org.apache.kafka.common.TopicPartition tp, long initialFetchOffset) Creates a new OffsetManager.

Method Summary

Modifier and Type	Method and Description
void	addToAckMsgs(KafkaSpoutMessageId msgId)
void	addToEmitMsgs(long offset)
long	commit(org.apache.kafka.clients.consumer.OffsetAndMetadata committedOffsetAndMeta) Marks an offset as committed.
boolean	contains(KafkaSpoutMessageId msgId)
org.apache.kafka.clients.consumer.OffsetAndMetadata	findNextCommitOffset(String commitMetadata) An offset can only be committed when all emitted records with lower offset have been acked.
long	getCommittedOffset()
long	getLatestEmittedOffset()
long	getNthUncommittedOffsetAfterCommittedOffset(int index) Gets the offset of the nth emitted message after the committed offset.
int	getNumUncommittedOffsets()
boolean	hasCommitted() Checks if this OffsetManager has committed to Kafka.
String	toString()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Constructor Detail

OffsetManager

public OffsetManager(org.apache.kafka.common.TopicPartition tp,
                     long initialFetchOffset)

Creates a new OffsetManager.

Parameters:

tp - The TopicPartition

initialFetchOffset - The initial fetch offset for the given TopicPartition

Method Detail

addToAckMsgs

public void addToAckMsgs(KafkaSpoutMessageId msgId)

addToEmitMsgs

public void addToEmitMsgs(long offset)

getNumUncommittedOffsets

public int getNumUncommittedOffsets()

getNthUncommittedOffsetAfterCommittedOffset

public long getNthUncommittedOffsetAfterCommittedOffset(int index)

Gets the offset of the nth emitted message after the committed offset. Example: If the committed offset is 0 and offsets 1, 2, 8, 10 have been emitted, getNthUncommittedOffsetAfterCommittedOffset(3) returns 8.

Parameters:

index - The index of the message to get the offset for

Returns:

The offset

Throws:

NoSuchElementException - if the index is out of range

findNextCommitOffset

public org.apache.kafka.clients.consumer.OffsetAndMetadata findNextCommitOffset(String commitMetadata)

An offset can only be committed when all emitted records with lower offset have been acked. This guarantees that all offsets smaller than the committedOffset have been delivered, or that those offsets no longer exist in Kafka.

The returned offset points to the earliest uncommitted offset, and matches the semantics of the KafkaConsumer.commitSync API.

Parameters:

commitMetadata - Metadata information to commit to Kafka. It is constant per KafkaSpout instance per topology

Returns:

the next OffsetAndMetadata to commit, or null if no offset is ready to commit.

commit

public long commit(org.apache.kafka.clients.consumer.OffsetAndMetadata committedOffsetAndMeta)

Marks an offset as committed. This method has side effects - it sets the internal state in such a way that future calls to findNextCommitOffset(String) will return offsets greater than or equal to the offset specified, if any.

Parameters:

committedOffsetAndMeta - The committed offset. All lower offsets are expected to have been committed.

Returns:

Number of offsets committed in this commit

hasCommitted

public boolean hasCommitted()

Checks if this OffsetManager has committed to Kafka.

Returns:

true if this OffsetManager has made at least one commit to Kafka, false otherwise

contains

public boolean contains(KafkaSpoutMessageId msgId)

getLatestEmittedOffset

public long getLatestEmittedOffset()

getCommittedOffset

public long getCommittedOffset()

toString

public final String toString()

Overrides:

toString in class Object