Package org.apache.pulsar.client.api

Interface ConsumerInterceptor<T>

All Superinterfaces:
AutoCloseable
@Public @Stable public interface ConsumerInterceptor<T> extends AutoCloseable
A plugin interface that allows you to intercept (and possibly mutate) messages received by the consumer.

A primary use case is to hook into consumer applications for custom monitoring, logging, etc.

Exceptions thrown by interceptor methods will be caught, logged, but not propagated further.

  • Method Details

    • close

      void close()
      Close the interceptor.
      Specified by:
      close in interface AutoCloseable

    • onArrival

      default Message<T> onArrival(Consumer<T> consumer, Message<T> message)
      This method is called when a message arrives in the consumer.

      This method provides visibility into the messages that have been received by the consumer but have not yet been processed. This can be useful for monitoring the state of the consumer's receiver queue and understanding the consumer's processing rate.

      The method is allowed to modify the message, in which case the modified message will be returned.

      Any exception thrown by this method will be caught by the caller, logged, but not propagated to the client.

      Since the consumer may run multiple interceptors, a particular interceptor's onArrival callback will be called in the order specified by ConsumerBuilder.intercept(ConsumerInterceptor[]). The first interceptor in the list gets the consumed message, the following interceptor will be passed the message returned by the previous interceptor, and so on. Since interceptors are allowed to modify the message, interceptors may potentially get the messages already modified by other interceptors. However, building a pipeline of mutable interceptors that depend on the output of the previous interceptor is discouraged, because of potential side-effects caused by interceptors potentially failing to modify the message and throwing an exception. If one of the interceptors in the list throws an exception from onArrival, the exception is caught, logged, and the next interceptor is called with the message returned by the last successful interceptor in the list, or otherwise the original consumed message.

      Parameters:
      consumer - the consumer which contains the interceptor
      message - the message that has arrived in the receiver queue
      Returns:
      the message that is either modified by the interceptor or the same message passed into the method

    • beforeConsume

      Message<T> beforeConsume(Consumer<T> consumer, Message<T> message)
      This is called just before the message is returned by Consumer.receive(), MessageListener.received(Consumer, Message) or the CompletableFuture returned by Consumer.receiveAsync() completes.

      This method is allowed to modify message, in which case the new message will be returned.

      Any exception thrown by this method will be caught by the caller, logged, but not propagated to client.

      Since the consumer may run multiple interceptors, a particular interceptor's beforeConsume callback will be called in the order specified by ConsumerBuilder.intercept(ConsumerInterceptor[]). The first interceptor in the list gets the consumed message, the following interceptor will be passed the message returned by the previous interceptor, and so on. Since interceptors are allowed to modify message, interceptors may potentially get the messages already modified by other interceptors. However building a pipeline of mutable interceptors that depend on the output of the previous interceptor is discouraged, because of potential side-effects caused by interceptors potentially failing to modify the message and throwing an exception. if one of interceptors in the list throws an exception from beforeConsume, the exception is caught, logged, and the next interceptor is called with the message returned by the last successful interceptor in the list, or otherwise the original consumed message.

      Parameters:
      consumer - the consumer which contains the interceptor
      message - the message to be consumed by the client.
      Returns:
      message that is either modified by the interceptor or same message passed into the method.

    • onAcknowledge

      void onAcknowledge(Consumer<T> consumer, MessageId messageId, Throwable exception)
      This is called consumer sends the acknowledgment to the broker.

      Any exception thrown by this method will be ignored by the caller.

      Parameters:
      consumer - the consumer which contains the interceptor
      messageId - message to ack, null if acknowledge fail.
      exception - the exception on acknowledge.

    • onAcknowledgeCumulative

      void onAcknowledgeCumulative(Consumer<T> consumer, MessageId messageId, Throwable exception)
      This is called consumer send the cumulative acknowledgment to the broker.

      Any exception thrown by this method will be ignored by the caller.

      Parameters:
      consumer - the consumer which contains the interceptor
      messageId - message to ack, null if acknowledge fail.
      exception - the exception on acknowledge.

    • onNegativeAcksSend

      void onNegativeAcksSend(Consumer<T> consumer, Set<MessageId> messageIds)
      This method will be called when a redelivery from a negative acknowledge occurs.

      Any exception thrown by this method will be ignored by the caller.

      Parameters:
      consumer - the consumer which contains the interceptor
      messageIds - the set of message ids to negatively ack

    • onAckTimeoutSend

      void onAckTimeoutSend(Consumer<T> consumer, Set<MessageId> messageIds)
      This method will be called when a redelivery from an acknowledge timeout occurs.

      Any exception thrown by this method will be ignored by the caller.

      Parameters:
      consumer - the consumer which contains the interceptor
      messageIds - message to ack, null if acknowledge fail.

    • onPartitionsChange

      default void onPartitionsChange(String topicName, int partitions)
      This method is called when partitions of the topic (partitioned-topic) changes.
      Parameters:
      topicName - topic name
      partitions - new updated number of partitions