GH-3032: Add RecoveryCallback to RetryingDeserializer

Fixes: #3032

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/support/serializer/RetryingDeserializer.java
#	spring-kafka/src/test/java/org/springframework/kafka/support/serializer/RetryingDeserializerTests.java

Author: sobychacko

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/serdes.adoc

@@ -397,6 +397,8 @@ ConsumerFactory cf = new DefaultKafkaConsumerFactory(myConsumerConfigs,
     new RetryingDeserializer(myUnreliableValueDeserializer, retryTemplate));
 ----
 
+Starting with version `3.1.2`, a `RecoveryCallback` can be set on the `RetryingDeserializer` optionally.
+
 Refer to the https://github.com/spring-projects/spring-retry[spring-retry] project for configuration of the `RetryTemplate` with a retry policy, back off policy, etc.
 
 

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/RetryingDeserializer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,11 +16,14 @@
 
 package org.springframework.kafka.support.serializer;
 
+import java.nio.ByteBuffer;
 import java.util.Map;
 
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.serialization.Deserializer;
 
+import org.springframework.lang.Nullable;
+import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.RetryOperations;
 import org.springframework.util.Assert;
 
@@ -31,39 +34,54 @@
  * @param <T> Type to be deserialized into.
  *
  * @author Gary Russell
- * @since 2.3
+ * @author Wang Zhiyang
+ * @author Soby Chacko
  *
+ * @since 2.3
  */
 public class RetryingDeserializer<T> implements Deserializer<T> {
 
 	private final Deserializer<T> delegate;
 
 	private final RetryOperations retryOperations;
 
+	@Nullable
+	private RecoveryCallback<T> recoveryCallback;
+
 	public RetryingDeserializer(Deserializer<T> delegate, RetryOperations retryOperations) {
 		Assert.notNull(delegate, "the 'delegate' deserializer cannot be null");
 		Assert.notNull(retryOperations, "the 'retryOperations' deserializer cannot be null");
 		this.delegate = delegate;
 		this.retryOperations = retryOperations;
 	}
 
+	/**
+	 * Set a recovery callback to execute when the retries are exhausted.
+	 * @param recoveryCallback {@link RecoveryCallback} to execute
+	 * @since 3.1.2
+	 */
+	public void setRecoveryCallback(@Nullable RecoveryCallback<T> recoveryCallback) {
+		this.recoveryCallback = recoveryCallback;
+	}
+
 	@Override
 	public void configure(Map<String, ?> configs, boolean isKey) {
 		this.delegate.configure(configs, isKey);
 	}
 
 	@Override
 	public T deserialize(String topic, byte[] data) {
-		return this.retryOperations.execute(context -> {
-			return this.delegate.deserialize(topic, data);
-		});
+		return this.retryOperations.execute(context -> this.delegate.deserialize(topic, data), this.recoveryCallback);
 	}
 
 	@Override
 	public T deserialize(String topic, Headers headers, byte[] data) {
-		return this.retryOperations.execute(context -> {
-			return this.delegate.deserialize(topic, headers, data);
-		});
+		return this.retryOperations.execute(context -> this.delegate.deserialize(topic, headers, data), this.recoveryCallback);
+	}
+
+	@Override
+	public T deserialize(String topic, Headers headers, ByteBuffer data) {
+		return this.retryOperations.execute(context -> this.delegate.deserialize(topic, headers, data), this.recoveryCallback);
 	}
 
 	@Override

spring-kafka/src/test/java/org/springframework/kafka/support/serializer/RetryingDeserializerTests.java

@@ -16,65 +16,61 @@
 
 package org.springframework.kafka.support.serializer;
 
-import static org.assertj.core.api.Assertions.assertThat;
-
-import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 
-import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.retry.RecoveryCallback;
+import org.springframework.retry.RetryContext;
 import org.springframework.retry.support.RetryTemplate;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
 /**
  * @author Gary Russell
- * @since 2.3
+ * @author Wang Zhiyang
+ * @author Soby Chacko
+ * @author Artem Bilan
  *
+ * @since 2.3
  */
-public class RetryingDeserializerTests {
+class RetryingDeserializerTests {
 
 	@Test
-	void testRetry() {
-		Deser delegate = new Deser();
+	void basicRetryingDeserializer() {
+		AtomicInteger n = new AtomicInteger();
+		Deserializer<String> delegate =
+				(topic, data) -> {
+					if (n.incrementAndGet() < 2) {
+						throw new RuntimeException();
+					}
+					return new String(data);
+				};
+
 		RetryingDeserializer<String> rdes = new RetryingDeserializer<>(delegate, new RetryTemplate());
 		assertThat(rdes.deserialize("foo", "bar".getBytes())).isEqualTo("bar");
-		assertThat(delegate.n).isEqualTo(3);
-		delegate.n = 0;
+		assertThat(n.get()).isEqualTo(3);
+		n.set(0);
 		assertThat(rdes.deserialize("foo", new RecordHeaders(), "bar".getBytes())).isEqualTo("bar");
-		assertThat(delegate.n).isEqualTo(3);
-		rdes.close();
+		assertThat(n.get()).isEqualTo(3);
 	}
 
-	public static class Deser implements Deserializer<String> {
-
-		int n;
-
-		@Override
-		public void configure(Map<String, ?> configs, boolean isKey) {
-		}
-
-		@Override
-		public String deserialize(String topic, byte[] data) {
-			if (n++ < 2) {
-				throw new RuntimeException();
-			}
-			return new String(data);
-		}
-
-		@Override
-		public String deserialize(String topic, Headers headers, byte[] data) {
-			if (n++ < 2) {
-				throw new RuntimeException();
-			}
-			return new String(data);
-		}
-
-		@Override
-		public void close() {
-			// empty
-		}
-
+	@Test
+	void retryingDeserializerWithRecoveryCallback() throws Exception {
+		RetryingDeserializer<String> rdes =
+				new RetryingDeserializer<>(
+						(s, b) -> {
+							throw new RuntimeException();
+						}, new RetryTemplate());
+		RecoveryCallback<String> recoveryCallback = mock();
+		rdes.setRecoveryCallback(recoveryCallback);
+		rdes.deserialize("my-topic", "my-data".getBytes());
+		verify(recoveryCallback).recover(any(RetryContext.class));
 	}
 
 }