Class DeadLetterRecordManager

java.lang.Object
org.springframework.kafka.listener.DeadLetterRecordManager
public class DeadLetterRecordManager extends Object
Manage the creation of headers for records published to a DLT by a DeadLetterPublishingRecoverer or a AbstractRecoveringExceptionHandler implementation.
Since:
4.1.0

  • Field Details

    • logger

      protected final org.springframework.core.log.LogAccessor logger

  • Constructor Details

    • DeadLetterRecordManager

      public DeadLetterRecordManager()

  • Method Details

    • enrichHeadersAndCreateProducerRecord

      public org.apache.kafka.clients.producer.ProducerRecord<byte[],byte[]> enrichHeadersAndCreateProducerRecord(org.apache.kafka.clients.consumer.ConsumerRecord<byte[],byte[]> record, Exception exception, org.apache.kafka.common.TopicPartition topicPartition)
      Enriches headers with deserialization exception information and metadata, then creates a producer record for dead letter publishing.
      Parameters:
      record - the source consumer record
      exception - the exception that triggered the dead letter publishing
      topicPartition - the target topic and partition
      Returns:
      a producer record
      Since:
      4.1.0

    • addAndEnhanceHeaders

      public void addAndEnhanceHeaders(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record, Exception exception, @Nullable DeserializationException vDeserEx, @Nullable DeserializationException kDeserEx, org.apache.kafka.common.header.Headers headers)
      Adds and enriches headers with deserialization exception information, original record metadata, and custom headers.
      Parameters:
      record - the consumer record
      exception - the exception that triggered the dead letter publishing
      vDeserEx - the value deserialization exception, if any occurred during deserialization
      kDeserEx - the key deserialization exception, if any occurred during deserialization
      headers - the headers to be enriched

    • setHeaderNamesSupplier

      public void setHeaderNamesSupplier(Supplier<DeadLetterPublishingRecoverer.HeaderNames> supplier)
      Set a Supplier for DeadLetterPublishingRecoverer.HeaderNames.
      Parameters:
      supplier - the supplier.
      Since:
      4.1.0

    • setRetainExceptionHeader

      public void setRetainExceptionHeader(boolean retainExceptionHeader)
      Set to true to retain a Java serialized DeserializationException header. By default, such headers are removed from the published record, unless both key and value deserialization exceptions occur, in which case, the DLT_* headers are created from the value exception and the key exception header is retained.
      Parameters:
      retainExceptionHeader - true to retain the exception header.
      Since:
      4.1.0

    • setExceptionHeadersCreator

      public void setExceptionHeadersCreator(DeadLetterPublishingRecoverer.ExceptionHeadersCreator headersCreator)
      Set a DeadLetterPublishingRecoverer.ExceptionHeadersCreator implementation to completely take over setting the exception headers in the output record. Disables all headers that are set by default.
      Parameters:
      headersCreator - the creator.
      Since:
      4.1.0

    • setHeadersFunction

      public void setHeadersFunction(BiFunction<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>, Exception, @Nullable org.apache.kafka.common.header.Headers> headersFunction)
      Set a function which will be called to obtain additional headers to add to the published record. If a Header returned is an instance of DeadLetterPublishingRecoverer.SingleRecordHeader, then that header will replace any existing header of that name, rather than being appended as a new value.
      Parameters:
      headersFunction - the headers function.
      Since:
      4.1.0

    • setAppendOriginalHeaders

      public void setAppendOriginalHeaders(boolean appendOriginalHeaders)
      Set to false if you don't want to append the current "original" headers (topic, partition, etc.) if they are already present. When false, only the first "original" headers are retained.
      Parameters:
      appendOriginalHeaders - set to false not to replace.
      Since:
      4.1.0

    • setStripPreviousExceptionHeaders

      public void setStripPreviousExceptionHeaders(boolean stripPreviousExceptionHeaders)
      Set to false to retain previous exception headers as well as headers for the current exception. Default is true, which means only the current headers are retained; setting it to false this can cause a growth in record size when a record is republished many times.
      Parameters:
      stripPreviousExceptionHeaders - false to retain all.
      Since:
      4.1.0

    • isDefaultHeadersFunction

      public boolean isDefaultHeadersFunction()
      Is the headers function the default one.
      Returns:
      true if it is, false otherwise.
      Since:
      4.1.0

    • getHeadersFunction

      public BiFunction<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>, Exception, @Nullable org.apache.kafka.common.header.Headers> getHeadersFunction()
      Return the headers function.
      Returns:
      the headers function
      Since:
      4.1.0

    • removeHeaderToAdd

      public void removeHeaderToAdd(DeadLetterPublishingRecoverer.HeaderNames.HeadersToAdd header)
      Remove a header from the set of headers to be added to the published record.
      Parameters:
      header - the header to exclude
      Since:
      4.1.0

    • addAllToHeaders

      public void addAllToHeaders(DeadLetterPublishingRecoverer.HeaderNames.HeadersToAdd... headers)
      Add all headers to be added to the published record.
      Parameters:
      headers - the headers to include
      Since:
      4.1.0