DeadLetterPublishingRecoverer should respect onlyLogRecordMatadata listener property

[kdowbecki]

Expected Behavior

Currently the org.springframework.kafka.listener.KafkaMessageListenerContainer class has onlyLogRecordMatadata property that governs how ConsumerRecord is logged on processing failure. Same onlyLogRecordMatadata property should be respected by the org.springframework.kafka.listener.DeadLetterPublishingRecoverer class for ProducerRecord.

Current Behavior

DeadLetterPublishingRecoverer currently logs the whole ProducerRecord when publication fails with no option to switch to just the metadata.

Context

We already have set onlyLogRecordMatadata = true because we don't want the message payload to appear in the logs when something fails. Since DeadLetterPublishingRecoverer copies the payload from the failed message into a ProducerRecord we still see the original message payload in the logs when publication fails.

[garyrussell]

Which version are you using? And which log message specifically?

It looks like all logs in 2.8.x properly log the record; in 2.7.x, I see one debug and 2 exceptions that log the producer record.

[garyrussell]

That flag only controls logging of the consumer record.

The DLPR does log the producer record; in 2.8 we added a new feature to format the logs for producer record.

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/support/KafkaUtils.java

Lines 168 to 176 in f94620a

	/**
	* Set a formatter for logging {@link ProducerRecord}s.
	* @param formatter a function to format the record as a String
	* @since 2.7.12
	*/
	public static void setProducerRecordFormatter(Function<ProducerRecord<?, ?>, String> formatter) {
	Assert.notNull(formatter, "'formatter' cannot be null");
	prFormatter = formatter;
	}

The default formatter logs the full record.

[garyrussell]

Although it was back-ported to 2.7.x, the DLPR doesn't use it.

[kdowbecki]

Good point @garyrussell. I'm on 2.7.12. I don't see DeadLetterPublishingRecoverer using KafkaUtils formatting in these lines:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

Line 501 in 85cbbed

this.logger.error(ex, () -> "Dead-letter publication failed for: " + outRecord);

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

Line 525 in 85cbbed

throw new KafkaException("Publication failed for: " + outRecord, e);

[kdowbecki]

It might be useful if failIfSendResultIsError field and verifySendResult method are changed from private to protected just like publish method is already protected. That way when when someone wants to override publish they can still reuse failIfSendResultIsError + verifySendResult functionality.

[garyrussell]

Yes, I will do some polishing around that at the same time; we don't allow protected fields (except loggers), but I can add a getter for it.

[garyrussell]

Closed by 56b7756

(Not sure why this didn't auto-close).