Package com.linkedin.venice.pubsub.api

Interface PubSubConsumerAdapter

All Superinterfaces:
AutoCloseable, Closeable
All Known Implementing Classes:
ApacheKafkaConsumerAdapter, MockInMemoryConsumerAdapter
public interface PubSubConsumerAdapter extends AutoCloseable, Closeable
An adapter for consuming messages from a Pub-Sub topic. Implementations of this interface are not expected to be thread safe. However, they are expected to provide the following guarantees: 1) Honor the timeout parameter for all methods that have one. 2) Non-blocking behavior for methods that do not have an explicit timeout parameter. In other words, they should timeout after the default timeout period: PubSubConstants.PUBSUB_CONSUMER_API_DEFAULT_TIMEOUT_MS.

  • Method Details

    • subscribe

      void subscribe(PubSubTopicPartition pubSubTopicPartition, long lastReadOffset)
      Subscribes to a topic-partition if it is not already subscribed. If the topic-partition is already subscribed, this method is a no-op. The method assumes that the topic-partition exists.
      Parameters:
      pubSubTopicPartition - The topic-partition to subscribe to.
      lastReadOffset - The last read offset for the topic-partition. A poll call following a subscribe call will return messages from the offset (lastReadOffset + 1).
      Throws:
      IllegalArgumentException - If the topic-partition is null or if the partition number is negative.
      PubSubTopicDoesNotExistException - If the topic does not exist.

    • subscribe

      void subscribe(@Nonnull PubSubTopicPartition pubSubTopicPartition, @Nonnull PubSubPosition lastReadPubSubPosition)
      Subscribes to a specified topic-partition if it is not already subscribed. If the topic-partition is already subscribed, this method performs no action. The subscription uses the provided PubSubPosition to determine the starting offset for consumption. If the position is PubSubSymbolicPosition.EARLIEST, the consumer will seek to the earliest available message. If it is PubSubSymbolicPosition.LATEST, the consumer will seek to the latest available message. If a custom position is provided, implementations should resolve it to the corresponding offset or position in the underlying pub-sub system. Implementations of this interface should ensure proper validation of the topic-partition existence and manage consumer assignments. This method does not guarantee immediate subscription state changes and may defer them based on implementation details.
      Parameters:
      pubSubTopicPartition - the topic-partition to subscribe to
      lastReadPubSubPosition - the last known position for the topic-partition
      Throws:
      IllegalArgumentException - if lastReadPubSubPosition is null or of an unsupported type
      PubSubTopicDoesNotExistException - if the specified topic does not exist

    • subscribe

      void subscribe(@Nonnull PubSubTopicPartition pubSubTopicPartition, @Nonnull PubSubPosition position, boolean isInclusive)
      Subscribes to a specified topic-partition if it is not already subscribed. If the topic-partition is already subscribed, this method performs no action.

      The subscription uses the provided PubSubPosition to determine the starting position for consumption. If the position is PubSubSymbolicPosition.EARLIEST, the consumer will seek to the earliest available message. If it is PubSubSymbolicPosition.LATEST, the consumer will seek to the latest available message. If a concrete position is provided, implementations should resolve it to a specific offset or internal position based on the underlying pub-sub system.

      The inclusive flag determines whether the message at the specified position (if resolvable to an offset or equivalent) should be included in consumption:

      • If true, the consumer should begin from the exact position specified.
      • If false, consumption should begin immediately after the specified position.

      Implementations should validate the topic-partition's existence and handle any necessary assignment or internal state initialization. The method does not guarantee immediate subscription effect and may defer changes depending on the consumer's execution model.

      Parameters:
      pubSubTopicPartition - the topic-partition to subscribe to
      position - the position from which consumption should begin
      isInclusive - whether to include the message at the given position
      Throws:
      IllegalArgumentException - if position is null or of an unsupported type
      PubSubTopicDoesNotExistException - if the specified topic does not exist

    • unSubscribe

      void unSubscribe(PubSubTopicPartition pubSubTopicPartition)
      Unsubscribes the consumer from a specified topic-partition. If the consumer was previously subscribed to the given partition, it will be unsubscribed, and the associated partition assignments and tracked offsets will be updated accordingly.
      Parameters:
      pubSubTopicPartition - The PubSub topic-partition to unsubscribe from.

    • batchUnsubscribe

      void batchUnsubscribe(Set<PubSubTopicPartition> pubSubTopicPartitionSet)
      Unsubscribes the consumer from a batch of topic-partitions.
      Parameters:
      pubSubTopicPartitionSet - A set of topic-partitions to unsubscribe from.

    • resetOffset

      void resetOffset(PubSubTopicPartition pubSubTopicPartition) throws PubSubUnsubscribedTopicPartitionException
      Resets the offset for a specific topic-partition, seeking it to the beginning of the topic. If the topic partition is not currently subscribed, a PubSubUnsubscribedTopicPartitionException is thrown.
      Parameters:
      pubSubTopicPartition - The PubSub topic-partition for which to reset the offset.
      Throws:
      PubSubUnsubscribedTopicPartitionException - If the specified topic-partition is not currently subscribed.

    • close

      void close()
      Closes the PubSub consumer and releases any associated resources.
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable

    • poll

      Map<PubSubTopicPartition,List<DefaultPubSubMessage>> poll(long timeoutMs)
      Polls the Kafka consumer for messages from the subscribed topic partitions within the specified time duration.
      Parameters:
      timeoutMs - The maximum time, in milliseconds, to wait for messages to be polled.
      Returns:
      A mapping of PubSub topic partitions to lists of PubSub messages retrieved from Kafka.
      Throws:
      PubSubClientException - If there is an error during message retrieval from Kafka.
      PubSubClientRetriableException - If a retriable exception occurs during polling attempts, with retries as configured.

    • hasAnySubscription

      boolean hasAnySubscription()
      Checks if the consumer has any active topic-partition subscriptions.
      Returns:
      true if the consumer is subscribed to one or more topic partitions, false otherwise.

    • hasSubscription

      boolean hasSubscription(PubSubTopicPartition pubSubTopicPartition)
      Checks if the consumer is currently subscribed to the specified PubSub topic-partition.
      Parameters:
      pubSubTopicPartition - The PubSub topic-partition to check for subscription.
      Returns:
      true if the consumer is subscribed to the given topic-partition, false otherwise.

    • pause

      void pause(PubSubTopicPartition pubSubTopicPartition)
      Pauses message consumption for the specified PubSub topic-partition. If the partition was not previously subscribed, this method is a no-op.
      Parameters:
      pubSubTopicPartition - The PubSub topic-partition for which to pause message consumption.

    • resume

      void resume(PubSubTopicPartition pubSubTopicPartition)
      Resumes message consumption for the specified PubSub topic-partition. If the partition was not previously paused or if they were not subscribed at all, this method is a no-op.
      Parameters:
      pubSubTopicPartition - The PubSub topic partition for which to resume message consumption.

    • getAssignment

      Set<PubSubTopicPartition> getAssignment()
      Retrieves the set of PubSub topic-partitions currently assigned to the consumer.
      Returns:
      A set of PubSub topic-partitions representing the current assignment of the consumer.

    • getOffsetLag

      default long getOffsetLag(PubSubTopicPartition pubSubTopicPartition)
      Retrieves the consuming offset lag for a PubSub topic partition. The offset lag represents the difference between the last consumed message offset and the latest available message offset for the partition.
      Parameters:
      pubSubTopicPartition - The PubSub topic partition for which to fetch the offset lag.
      Returns:
      The offset lag, which is zero or a positive value if a valid lag was collected by the consumer, or -1 if the lag cannot be determined or is not applicable.

    • getLatestOffset

      default long getLatestOffset(PubSubTopicPartition pubSubTopicPartition)
      Retrieves the latest available offset for a PubSub topic partition.
      Parameters:
      pubSubTopicPartition - The PubSub topic partition for which to fetch the latest offset.
      Returns:
      The latest offset, which is zero or a positive value if an offset was collected by the consumer, or -1 if the offset cannot be determined or is not applicable.

    • offsetForTime

      Long offsetForTime(PubSubTopicPartition pubSubTopicPartition, long timestamp, Duration timeout)
      Retrieves the offset of the first message with a timestamp greater than or equal to the target timestamp for the specified PubSub topic-partition. If no such message is found, null will be returned for the partition.
      Parameters:
      pubSubTopicPartition - The PubSub topic-partition for which to fetch the offset.
      timestamp - The target timestamp to search for in milliseconds since the Unix epoch.
      timeout - The maximum duration to wait for the operation to complete.
      Returns:
      The offset of the first message with a timestamp greater than or equal to the target timestamp, or null if no such message is found for the partition.
      Throws:
      PubSubOpTimeoutException - If the operation times out while fetching the offset.
      PubSubClientException - If there is an error while attempting to fetch the offset.

    • getPositionByTimestamp

      PubSubPosition getPositionByTimestamp(PubSubTopicPartition pubSubTopicPartition, long timestamp, Duration timeout)
      Retrieves the offset of the first message with a timestamp greater than or equal to the target timestamp for the specified PubSub topic-partition. If no such message is found, null will be returned for the partition.
      Parameters:
      pubSubTopicPartition - The PubSub topic-partition for which to fetch the offset.
      timestamp - The target timestamp to search for in milliseconds since the Unix epoch.
      timeout - The maximum duration to wait for the operation to complete.
      Returns:
      The offset of the first message with a timestamp greater than or equal to the target timestamp, or null if no such message is found for the partition.
      Throws:
      PubSubOpTimeoutException - If the operation times out while fetching the offset.
      PubSubClientException - If there is an error while attempting to fetch the offset.

    • offsetForTime

      Long offsetForTime(PubSubTopicPartition pubSubTopicPartition, long timestamp)
      Retrieves the offset of the first message with a timestamp greater than or equal to the target timestamp for the specified PubSub topic-partition. If no such message is found, null will be returned for the partition.
      Parameters:
      pubSubTopicPartition - The PubSub topic-partition for which to fetch the offset.
      timestamp - The target timestamp to search for in milliseconds since the Unix epoch.
      Returns:
      The offset of the first message with a timestamp greater than or equal to the target timestamp, or null if no such message is found for the partition.
      Throws:
      PubSubOpTimeoutException - If the operation times out while fetching the offset.
      PubSubClientException - If there is an error while attempting to fetch the offset.

    • getPositionByTimestamp

      PubSubPosition getPositionByTimestamp(PubSubTopicPartition pubSubTopicPartition, long timestamp)
      Retrieves the offset of the first message with a timestamp greater than or equal to the target timestamp for the specified PubSub topic-partition. If no such message is found, null will be returned for the partition.
      Parameters:
      pubSubTopicPartition - The PubSub topic-partition for which to fetch the offset.
      timestamp - The target timestamp to search for in milliseconds since the Unix epoch.
      Returns:
      The offset of the first message with a timestamp greater than or equal to the target timestamp, or null if no such message is found for the partition.
      Throws:
      PubSubOpTimeoutException - If the operation times out while fetching the offset.
      PubSubClientException - If there is an error while attempting to fetch the offset.

    • beginningPosition

      PubSubPosition beginningPosition(PubSubTopicPartition pubSubTopicPartition, Duration timeout)
      Retrieves the beginning position for the specified PubSub topic-partition.
      Parameters:
      pubSubTopicPartition - The PubSub topic-partition for which to fetch the beginning offset.
      timeout - The maximum duration to wait for the operation to complete.
      Returns:
      The beginning offset of the specified topic-partition. If topic-partition exists but has no messages, the offset will be 0.
      Throws:
      PubSubOpTimeoutException - If the operation times out while fetching the beginning offset.
      PubSubClientException - If there is an error while attempting to fetch the beginning offset.

    • beginningPosition

      default PubSubPosition beginningPosition(PubSubTopicPartition pubSubTopicPartition)

    • beginningPositions

      Map<PubSubTopicPartition,PubSubPosition> beginningPositions(Collection<PubSubTopicPartition> partitions, Duration timeout)

    • endOffsets

      Map<PubSubTopicPartition,Long> endOffsets(Collection<PubSubTopicPartition> partitions, Duration timeout)
      Retrieves the end offsets for a collection of PubSub topic-partitions. The end offset represents the highest offset available in each specified partition, i.e., offset of the last message + 1. If there are no messages in a partition, the end offset will be 0.
      Parameters:
      partitions - A collection of PubSub topic-partitions for which to fetch the end offsets.
      timeout - The maximum duration to wait for the operation to complete.
      Returns:
      A mapping of PubSub topic partitions to their respective end offsets, or an empty map if the offsets cannot be determined.
      Throws:
      PubSubOpTimeoutException - If the operation times out while fetching the end offsets.
      PubSubClientException - If there is an error while attempting to fetch the end offsets.

    • endPositions

      Map<PubSubTopicPartition,PubSubPosition> endPositions(Collection<PubSubTopicPartition> partitions, Duration timeout)

    • endOffset

      Long endOffset(PubSubTopicPartition pubSubTopicPartition)
      Retrieves the end offset for the specified PubSub topic-partition. The end offset represents the highest offset available in each specified partition, i.e., offset of the last message + 1. If there are no messages in a partition, the end offset will be 0.
      Parameters:
      pubSubTopicPartition - The PubSub topic partition for which to fetch the end offset.
      Returns:
      The end offset of the specified topic partition.
      Throws:
      PubSubOpTimeoutException - If the operation times out while fetching the end offset.
      PubSubClientException - If there is an error while attempting to fetch the end offset.

    • endPosition

      PubSubPosition endPosition(PubSubTopicPartition pubSubTopicPartition)

    • partitionsFor

      List<PubSubTopicPartitionInfo> partitionsFor(PubSubTopic pubSubTopic)
      Retrieves the list of partitions associated with a given Pub-Sub topic.
      Parameters:
      pubSubTopic - The Pub-Sub topic for which partition information is requested.
      Returns:
      A list of PubSubTopicPartitionInfo representing the partitions of the topic, or null if the topic does not exist.

    • comparePositions

      long comparePositions(PubSubTopicPartition partition, PubSubPosition position1, PubSubPosition position2)
      Compares two PubSub positions within the specified topic partition.
      Parameters:
      partition - The topic partition where the comparison is being performed.
      position1 - The first PubSub position.
      position2 - The second PubSub position.
      Returns:
      A negative value if position1 is behind position2, zero if equal, or a positive value if position1 is ahead of position2.

    • positionDifference

      long positionDifference(PubSubTopicPartition partition, PubSubPosition position1, PubSubPosition position2)
      Computes the relative difference between two PubSubPosition instances for a given PubSubTopicPartition, as position1 - position2.

      Implementations must resolve symbolic positions such as PubSubSymbolicPosition.EARLIEST and PubSubSymbolicPosition.LATEST to concrete positions based on the partition's start and end positions. This ensures that symbolic references can be treated consistently during subtraction.

      For example:

      • If both positions are concrete, the result is the logical offset difference between them.
      • If position1 is symbolic (e.g., EARLIEST), it must be resolved to the concrete beginning position.
      • If position2 is symbolic (e.g., LATEST), it must be resolved to the concrete end position.
      Parameters:
      partition - The topic partition for which the difference is calculated.
      position1 - The first PubSub position (minuend).
      position2 - The second PubSub position (subtrahend).
      Returns:
      The signed offset difference between position1 and position2.
      Throws:
      IllegalArgumentException - if either position is null, or if symbolic positions cannot be resolved.

    • decodePosition

      default PubSubPosition decodePosition(PubSubTopicPartition partition, int positionTypeId, byte[] data)
      Decodes the given type-encoded byte array into a PubSubPosition for the specified topic partition.
      Parameters:
      partition - The topic partition this position belongs to.
      positionTypeId - The type ID of the position, which indicates how to decode the byte array.
      data - The byte array containing the encoded position.
      Returns:
      The decoded PubSubPosition.
      Throws:
      IllegalArgumentException - if the data cannot be decoded into a valid PubSubPosition.

    • decodePosition

      PubSubPosition decodePosition(PubSubTopicPartition partition, int positionTypeId, ByteBuffer buffer)
      Decodes the given ByteBuffer into a PubSubPosition for the specified topic partition.
      Parameters:
      partition - The topic partition this position belongs to.
      positionTypeId - The type ID of the position, which indicates how to decode the byte buffer.
      buffer - The ByteBuffer containing the encoded position.
      Returns:
      The decoded PubSubPosition.
      Throws:
      IllegalArgumentException - if the buffer cannot be decoded into a valid PubSubPosition.