Package 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
  Description
  `OffsetManager(org.apache.kafka.common.TopicPartition tp, long initialFetchOffset)`
  Creates a new OffsetManager.

- ## Method Summary

  Modifier and Type
  Method
  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.
  `final String`
  `toString()`

  ### Methods inherited from class java.lang.Object

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

- ## Constructor Details

  - ### 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 Details

  - ### 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`