Edit data of all messages (value) before submitting to DLT topic

[Watlas]

Hello, I was reading the documentation and could not identify a solution to the following problem:

I want to add new attributes to the message before sending to the DLT topic, injecting this at the configuration level, I want to add attributes to the message body coming from the exception itself.

I believe what I'm looking for is a type of interceptor that allows me to change data.

[artembilan]

By default the framework uses a DeadLetterPublishingRecoverer. This one has a method like:

	/**
	 * Subclasses can override this method to customize the producer record to send to the
	 * DLQ. The default implementation simply copies the key and value from the consumer
	 * record and adds the headers. The timestamp is not set (the original timestamp is in
	 * one of the headers). IMPORTANT: if the partition in the {@link TopicPartition} is
	 * less than 0, it must be set to null in the {@link ProducerRecord}.
	 * @param record the failed record
	 * @param topicPartition the {@link TopicPartition} returned by the destination
	 * resolver.
	 * @param headers the headers - original record headers plus DLT headers.
	 * @param key the key to use instead of the consumer record key.
	 * @param value the value to use instead of the consumer record value.
	 * @return the producer record to send.
	 * @see KafkaHeaders
	 */
	protected ProducerRecord<Object, Object> createProducerRecord(ConsumerRecord<?, ?> record,
			TopicPartition topicPartition, Headers headers, @Nullable byte[] key, @Nullable byte[] value) {

		return new ProducerRecord<>(topicPartition.topic(),
				topicPartition.partition() < 0 ? null : topicPartition.partition(),
				key != null ? key : record.key(),
				value != null ? value : record.value(), headers);
	}

so, you can extend this DeadLetterPublishingRecoverer and override that method whatever way you need.

You inject your custom recoverer into a DefaultErrorHandler: https://docs.spring.io/spring-kafka/docs/current/reference/html/#dead-letters.
Which you will provide for the listener container factory:

	/**
	 * Set the {@link CommonErrorHandler} which can handle errors for both record and
	 * batch listeners. Replaces the use of
	 * {@link org.springframework.kafka.listener.GenericErrorHandler}s.
	 * @param commonErrorHandler the handler.
	 * @since 2.8
	 */
	public void setCommonErrorHandler(CommonErrorHandler commonErrorHandler) {

[artembilan]

Not yet. Will look on Monday.
Perhaps @tomazfernandes can spot something immediately.
Thanks

[tomazfernandes]

I don't think the DeadLetterPublishingRecovererFactory currently offers any extension points that would enable replacing the record, or using a DeadLetterPublishingFactory subclass instead of the provided one.

You can open an issue so we can look into offering one or both of these.

Thanks.

[Watlas]

@tomazfernandes So it is currently not possible to use the non-blocking retry solution and the ProducerRecord edit solution together?

[garyrussell]

Correct. We need to refactor the DeadLetterPublishingRecovererFactory to allow creation of the subclass. Not too difficult, but it will require some other refactoring because the non-blocking retry mechanism uses different header names.

[tomazfernandes]

Agreed. Another option would be extracting the anonymous retry topic DLPR subclass into a proper class, so that users could subclass that instead, and then just offer an extensibility point for the actual instance creation.

For the matter at hand, we do offer extensibility in ListenerContainerFactoryConfigurer#createDefaultErrorHandlerInstance, so we could replace the DefaultErrorHandler with a subclass. If we can intercept the ConsumerRecord instead to change the payload, we might be able to achieve the expected result. Not an ideal solution of course.