Add Option to Log Recovered Record to DeadLetterPublishingRecoverer

[garyrussell]

See https://stackoverflow.com/questions/77368555/spring-boot-with-kafka-not-logging-exceptions-on-deserialization-when-using-dl/77368770#77368770

[Wzy19930507]

Hi @garyrussell , may i pick it up?

If my understanding is correct, it can be add code after DeadLetterPublishingRecoverer.accept()#L505

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

Lines 498 to 506 in 6876694

	if (this.skipSameTopicFatalExceptions
	&& tp.topic().equals(record.topic())
	&& !getClassifier().classify(exception)) {
	this.logger.error("Recovery of " + KafkaUtils.format(record)
	+ " skipped because not retryable exception " + exception.toString()
	+ " and the destination resolver routed back to the same topic");
	return;
	}
	if (consumer != null && this.verifyPartition) {

if (logRecordEnable) {
    logger.info("Recovered Record " + KafkaUtils.format(record) + " , exception " + exception);
}

[garyrussell]

Sure, but I think it actually belongs I the the DefaultErrorHandler after successful recovery.

[Wzy19930507]

thank you for your reminder.

Here are 4 methods need to Log in DefaultErrorHandler:

• handleOne
• handleRemaining
• handleBatchAndReturnRemaining
• handleBatch

note

I have a few questions that hope you can give me a hint

1. edge case at handleRemaining when SeekUtils.commitRecovered is true and ack mode is MANUAL_IMMEDIATE, successful recovery but consumer.sync() fail throw CommitFailedException, unable to log.
   

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

   Lines 210 to 218 in 6876694

   	if (commitRecovered) {
   	if (container.getContainerProperties().getAckMode().equals(AckMode.MANUAL_IMMEDIATE)) {
   	ConsumerRecord<?, ?> record = records.get(0);
   	Map<TopicPartition, OffsetAndMetadata> offsetToCommit = Collections.singletonMap(
   	new TopicPartition(record.topic(), record.partition()),
   	ListenerUtils.createOffsetAndMetadata(container, record.offset() + 1));
   	if (container.getContainerProperties().isSyncCommits()) {
   	consumer.commitSync(offsetToCommit, container.getContainerProperties().getSyncCommitTimeout());
   	}

Should i try catch CommitFailedException and Log recovered record at handleRemaining?

---

2. at handleBatchAndReturnRemaining and handleBatch
   

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

   Lines 180 to 182 in 6876694

   	public <K, V> ConsumerRecords<K, V> handleBatchAndReturnRemaining(Exception thrownException,
   	ConsumerRecords<?, ?> data, Consumer<?, ?> consumer, MessageListenerContainer container,
   	Runnable invokeListener) {

   
   only Log recovery record or Log ConsumerRecords data?
   if only Log recovery record, at handleBatch need change doHandle to handle

---

3. same as Q2, at handleRemaining
   

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

   Lines 165 to 166 in 6876694

   	public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records,
   	Consumer<?, ?> consumer, MessageListenerContainer container) {

   
   Log records.get(0) or Log records?

[garyrussell]

I see, yes, complicated; let's go back to logging in the DLPR only.