Interface KafkaConsumerProcessingObserver

@Internal public interface KafkaConsumerProcessingObserver
Internal integration point that lets an observability module observe the processing lifecycle of a single consumed ConsumerRecord, from the first delivery attempt through to the record's terminal outcome (successful processing, or terminal failure once retries are exhausted / the exception is not retryable).

The consumer processor invokes onStart(ConsumerRecord, String, String) when it starts handling a record and keeps the returned handle across blocking retries of that same Kafka record. Around every individual processing attempt (the first delivery and each blocking retry) it brackets the listener invocation and any handling of its failure with onAttemptStart(Object) and onAttemptEnd(Object), allowing the observation to be made the current context for that attempt only and to be cleared again before any retry delay. Each observation then ends with exactly one of onSuccess(Object) or onError(Object, Throwable), after the attempt has ended. This is deliberately different from ConsumerRecordInterceptor, which fires per delivery attempt before listener binding and therefore cannot observe the terminal outcome of a record.

A successful non-blocking retry-topic dispatch is treated as the terminal outcome of the source record's delivery (onError(Object, Throwable)): the failed record has been handed off to a separate retry topic and, when that retry record is later consumed, it begins its own observation. In other words a single observation spans all blocking attempts of one consumed Kafka record, not a distributed retry-topic pipeline.

Only listeners that consume single records are observed: batch listeners receive no callbacks. Failures to publish a listener result to its @SendTo destination are reported only if they are known when the listener invocation returns; a failure the producer reports later is not.

All callbacks are invoked fail-open by the processor: an exception or linkage error thrown by any method here is caught and logged, and never alters Kafka delivery semantics (listener invocation, retries, offset commits, or dead-letter publishing).

This type is marked Internal on purpose: it exists to support the framework's own Micrometer Observation integration and is not a stable, user-facing extension API. The signature may change without notice. It intentionally exposes only Kafka client and primitive types so that Micronaut Kafka does not gain a dependency on any observability library.

Since:
6.3.0
Author:
Sagar Kharab

  • Method Summary

    Modifier and Type
    Method
    Description
    void
    onAttemptEnd(@NonNull Object attempt)
    Invoked once per processing attempt for which onAttemptStart(Object) returned a token, after the listener invocation and the handling of any failure (exception handler, dead-letter or retry-topic publishing), so that any per-attempt context opened by onAttemptStart(Object) is cleared before a retry delay.
    @Nullable Object
    onAttemptStart(@NonNull Object handle)
    Invoked immediately before each individual processing attempt of the record (the first delivery and every blocking retry), after onStart(ConsumerRecord, String, String).
    void
    onError(@NonNull Object handle, @NonNull Throwable error)
    Invoked once when the observation of a record ends in failure: retries exhausted, the retry strategy stopped on the paused partition, the exception is not retryable, the failed record was handed off to a non-blocking retry topic, or the listener result could not be published to its destination.
    @Nullable Object
    onStart(@NonNull ConsumerRecord<?,?> record, @NonNull String clientId, @Nullable String groupId)
    Invoked when a new observation of a consumed record starts.
    void
    onSuccess(@NonNull Object handle)
    Invoked once when the record has been processed successfully and will not be retried.

  • Method Details

    • onStart

      @Nullable Object onStart(@NonNull ConsumerRecord<?,?> record, @NonNull String clientId, @Nullable String groupId)
      Invoked when a new observation of a consumed record starts. On subsequent blocking retries of the same record the processor reuses the handle returned here rather than starting again, so implementations should treat this as "the record's processing has begun". The opt-out decision (a null return) is remembered in the same way.

      This is called once per observation, which is not always once per record: a record that is delivered again after its observation ended, for example when a partition paused by stopOnExhaustedRetry is resumed or a revoked partition is assigned back, starts a new one.

      A record that failed to deserialize while polling is passed with a null key and value. It has the headers and timestamp of the failed record if the Kafka client reports them, and otherwise no headers and no timestamp.

      Parameters:
      record - the record about to be processed
      clientId - the Micronaut Kafka client id of the consumer
      groupId - the Kafka consumer group id, if any
      Returns:
      an opaque handle representing the started observation, or null to opt out of observing this record (in which case no further callbacks will be invoked for it)

    • onAttemptStart

      @Nullable Object onAttemptStart(@NonNull Object handle)
      Invoked immediately before each individual processing attempt of the record (the first delivery and every blocking retry), after onStart(ConsumerRecord, String, String). Implementations typically make the observation the current context here so that listener work and downstream instrumentation observe it. The returned token is passed back to onAttemptEnd(Object) once the attempt completes.
      Parameters:
      handle - the handle previously returned by onStart(ConsumerRecord, String, String)
      Returns:
      an opaque per-attempt token (for example a scope to close), or null if there is nothing to close at the end of the attempt

    • onAttemptEnd

      void onAttemptEnd(@NonNull Object attempt)
      Invoked once per processing attempt for which onAttemptStart(Object) returned a token, after the listener invocation and the handling of any failure (exception handler, dead-letter or retry-topic publishing), so that any per-attempt context opened by onAttemptStart(Object) is cleared before a retry delay. It does not signal the record's terminal outcome.
      Parameters:
      attempt - the token previously returned by onAttemptStart(Object)

    • onSuccess

      void onSuccess(@NonNull Object handle)
      Invoked once when the record has been processed successfully and will not be retried.
      Parameters:
      handle - the handle previously returned by onStart(ConsumerRecord, String, String)

    • onError

      void onError(@NonNull Object handle, @NonNull Throwable error)
      Invoked once when the observation of a record ends in failure: retries exhausted, the retry strategy stopped on the paused partition, the exception is not retryable, the failed record was handed off to a non-blocking retry topic, or the listener result could not be published to its destination. If the record is delivered again later, that starts a new observation (see onStart(ConsumerRecord, String, String)).

      It is also invoked, with a KafkaListenerException, when this consumer abandons a record before its terminal outcome: the partition was revoked, the consumer stopped, the partition moved to another offset, or an interceptor skipped the record on a retry.

      Parameters:
      handle - the handle previously returned by onStart(ConsumerRecord, String, String)
      error - the terminal error