Fixes: #3066

Issue link: https://github.com/spring-cloud/spring-cloud-stream/issues/3066

Signed-off-by: DidierLoiseau <didierloiseau+github@gmail.com>

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -165,6 +165,7 @@ import org.springframework.util.backoff.FixedBackOff;
  * @author Byungjun You
  * @author Oliver Führer
  * @author Omer Celik
+ * @author Didier Loiseau
  */
 public class KafkaMessageChannelBinder extends
 		// @checkstyle:off
@@ -566,10 +567,16 @@ public class KafkaMessageChannelBinder extends
 	}
 
 	@Override
-	@SuppressWarnings("unchecked")
 	protected MessageProducer createConsumerEndpoint(
 			final ConsumerDestination destination, final String group,
 			final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
+		return createConsumerEndpointCaptureHelper(destination, group, extendedConsumerProperties);
+	}
+
+	@SuppressWarnings("unchecked")
+	private <K, V> MessageProducer createConsumerEndpointCaptureHelper(
+			final ConsumerDestination destination, final String group,
+			final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
 
 		boolean anonymous = !StringUtils.hasText(group);
 		Assert.isTrue(
@@ -577,7 +584,7 @@ public class KafkaMessageChannelBinder extends
 				"DLQ support is not available for anonymous subscriptions");
 		String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString()
 				: group;
-		final ConsumerFactory<?, ?> consumerFactory = createKafkaConsumerFactory(
+		final ConsumerFactory<K, V> consumerFactory = (ConsumerFactory<K, V>) createKafkaConsumerFactory(
 				anonymous, consumerGroup, extendedConsumerProperties, destination.getName() + ".consumer", destination.getName());
 		int partitionCount = extendedConsumerProperties.getInstanceCount()
 				* extendedConsumerProperties.getConcurrency();
@@ -647,9 +654,8 @@ public class KafkaMessageChannelBinder extends
 		}
 		resetOffsetsForAutoRebalance(extendedConsumerProperties, consumerFactory, containerProperties);
 		containerProperties.setAuthExceptionRetryInterval(this.configurationProperties.getAuthorizationExceptionRetryInterval());
-		@SuppressWarnings("rawtypes")
-		final ConcurrentMessageListenerContainer<?, ?> messageListenerContainer = new ConcurrentMessageListenerContainer(
-				consumerFactory, containerProperties) {
+		final ConcurrentMessageListenerContainer<K, V> messageListenerContainer =
+			new ConcurrentMessageListenerContainer<>(consumerFactory, containerProperties) {
 
 			@Override
 			public void stop(Runnable callback) {
@@ -677,8 +683,7 @@ public class KafkaMessageChannelBinder extends
 		ContainerProperties.AckMode ackMode = extendedConsumerProperties.getExtension().getAckMode();
 
 		if (ackMode != null) {
-			if ((extendedConsumerProperties.isBatchMode() && ackMode != ContainerProperties.AckMode.RECORD) ||
-					!extendedConsumerProperties.isBatchMode()) {
+			if (!extendedConsumerProperties.isBatchMode() || ackMode != ContainerProperties.AckMode.RECORD) {
 				messageListenerContainer.getContainerProperties()
 						.setAckMode(ackMode);
 			}
@@ -715,7 +720,7 @@ public class KafkaMessageChannelBinder extends
 			}
 		}
 		else if (!extendedConsumerProperties.isBatchMode() && transMan != null) {
-			messageListenerContainer.setAfterRollbackProcessor(new DefaultAfterRollbackProcessor<>(
+			var afterRollbackProcessor = new DefaultAfterRollbackProcessor<K, V>(
 					(record, exception) -> {
 						MessagingException payload =
 								new MessagingException(((RecordMessageConverter) messageConverter)
@@ -740,7 +745,31 @@ public class KafkaMessageChannelBinder extends
 						}
 					}, createBackOff(extendedConsumerProperties),
 					new KafkaTemplate<>(transMan.getProducerFactory()),
-					extendedConsumerProperties.getExtension().isTxCommitRecovered()));
+					extendedConsumerProperties.getExtension().isTxCommitRecovered());
+			if (!CollectionUtils.isEmpty(extendedConsumerProperties.getRetryableExceptions())) {
+				// mimic AbstractBinder.buildRetryTemplate(properties)’s retryPolicy
+				if (!extendedConsumerProperties.isDefaultRetryable()) {
+					afterRollbackProcessor.defaultFalse(true);
+				}
+				extendedConsumerProperties.getRetryableExceptions()
+						.forEach((t, retry) -> {
+							if (Exception.class.isAssignableFrom(t)) {
+								var ex = t.asSubclass(Exception.class);
+								if (retry) {
+									afterRollbackProcessor.addRetryableExceptions(ex);
+								}
+								else {
+									afterRollbackProcessor.addNotRetryableExceptions(ex);
+								}
+							}
+							else {
+								throw new IllegalArgumentException(
+										"Only Exception types can be configured as retryable-exceptions together with transactions. "
+												+ "Unsupported type: " + t.getName());
+							}
+						});
+			}
+			messageListenerContainer.setAfterRollbackProcessor(afterRollbackProcessor);
 		}
 		else {
 			kafkaMessageDrivenChannelAdapter.setErrorChannel(errorInfrastructure.getErrorChannel());

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java

@@ -36,6 +36,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
 import java.util.stream.IntStream;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -172,7 +173,8 @@ import static org.mockito.Mockito.verify;
  * @author Henryk Konsek
  * @author Gary Russell
  * @author Chris Bono
- * @Author Oliver Führer
+ * @author Oliver Führer
+ * @author Didier Loiseau
  */
 @EmbeddedKafka(count = 1, controlledShutdown = true, topics = "error.pollableDlq.group-pcWithDlq", brokerProperties = {"transaction.state.log.replication.factor=1",
 		"transaction.state.log.min.isr=1"})
@@ -1052,11 +1054,41 @@ class KafkaBinderTests extends
 		testDlqGuts(true, null, null, false, false);
 	}
 
+	@Test
+	void dlqAndRetryWithNonRetryableException() throws Exception {
+		testDlqGuts(true, null, null, false, false, true, true);
+	}
+
+	@Test
+	void dlqAndRetryDefaultFalse() throws Exception {
+		testDlqGuts(true, null, null, false, false, false, false);
+	}
+
+	@Test
+	void dlqAndRetryDefaultFalseWithRetryableException() throws Exception {
+		testDlqGuts(true, null, null, false, false, false, true);
+	}
+
 	@Test
 	void dlqAndRetryTransactional() throws Exception {
 		testDlqGuts(true, null, null, true, false);
 	}
 
+	@Test
+	void dlqAndRetryWithNonRetryableExceptionTransactional() throws Exception {
+		testDlqGuts(true, null, null, true, false, true, true);
+	}
+
+	@Test
+	void dlqAndRetryDefaultFalseTransactional() throws Exception {
+		testDlqGuts(true, null, null, true, false, false, false);
+	}
+
+	@Test
+	void dlqAndRetryDefaultFalseWithRetryableExceptionTransactional() throws Exception {
+		testDlqGuts(true, null, null, true, false, false, true);
+	}
+
 	@Test
 	void dlq() throws Exception {
 		testDlqGuts(false, null, 3, false, false);
@@ -1084,6 +1116,14 @@ class KafkaBinderTests extends
 
 	private void testDlqGuts(boolean withRetry, HeaderMode headerMode, Integer dlqPartitions,
 			boolean transactional, boolean useDlqDestResolver) throws Exception {
+		testDlqGuts(withRetry, headerMode, dlqPartitions, transactional,
+				useDlqDestResolver, true, false);
+	}
+
+	private void testDlqGuts(boolean withRetry, HeaderMode headerMode,
+			Integer dlqPartitions, boolean transactional, boolean useDlqDestResolver,
+			boolean defaultRetryable, boolean useConfiguredRetryableException)
+			throws Exception {
 
 		int expectedDlqPartition = dlqPartitions == null ? 0 : dlqPartitions - 1;
 		KafkaBinderConfigurationProperties binderConfig = createConfigurationProperties();
@@ -1128,12 +1168,18 @@ class KafkaBinderTests extends
 		consumerProperties.getExtension().setDlqPartitions(dlqPartitions);
 		consumerProperties.setConcurrency(2);
 		consumerProperties.populateBindingName("foobar");
+		consumerProperties.setDefaultRetryable(defaultRetryable);
+		consumerProperties.getRetryableExceptions().put(NumberFormatException.class,
+				!defaultRetryable);
 
 		DirectChannel moduleInputChannel = createBindableChannel("input",
 				createConsumerBindingProperties(consumerProperties));
 
 		var dlqChannel = new QueueChannel();
-		var handler = new FailingInvocationCountingMessageHandler();
+		var handler = new FailingInvocationCountingMessageHandler(
+				() -> useConfiguredRetryableException
+						? new NumberFormatException("fail")
+						: new RuntimeException("fail"));
 		moduleInputChannel.subscribe(handler);
 
 		long uniqueBindingId = System.currentTimeMillis();
@@ -1253,8 +1299,10 @@ class KafkaBinderTests extends
 					.get(KafkaHeaders.RECEIVED_PARTITION)).isEqualTo(expectedDlqPartition);
 		}
 		else if (!HeaderMode.none.equals(headerMode)) {
+			boolean shouldHaveRetried = defaultRetryable != useConfiguredRetryableException;
 			assertThat(handler.getInvocationCount())
-					.isEqualTo(consumerProperties.getMaxAttempts());
+					.isEqualTo(
+							shouldHaveRetried ? consumerProperties.getMaxAttempts() : 1);
 
 			assertThat(receivedMessage.getHeaders()
 					.get(KafkaMessageChannelBinder.X_ORIGINAL_TOPIC))
@@ -4090,14 +4138,27 @@ class KafkaBinderTests extends
 	private final class FailingInvocationCountingMessageHandler
 			implements MessageHandler {
 
+		private final Supplier<? extends RuntimeException> exceptionProvider;
+
 		private volatile int invocationCount;
 
 		private final LinkedHashMap<Long, Message<?>> receivedMessages = new LinkedHashMap<>();
 
 		private final CountDownLatch latch;
 
-		private FailingInvocationCountingMessageHandler(int latchSize) {
+		private FailingInvocationCountingMessageHandler(int latchSize,
+				Supplier<? extends RuntimeException> exceptionProvider) {
 			latch = new CountDownLatch(latchSize);
+			this.exceptionProvider = exceptionProvider;
+		}
+
+		private FailingInvocationCountingMessageHandler(
+				Supplier<? extends RuntimeException> exceptionProvider) {
+			this(1, exceptionProvider);
+		}
+
+		private FailingInvocationCountingMessageHandler(int latchSize) {
+			this(latchSize, () -> new RuntimeException("fail"));
 		}
 
 		private FailingInvocationCountingMessageHandler() {
@@ -4115,7 +4176,7 @@ class KafkaBinderTests extends
 				receivedMessages.put(offset, message);
 				latch.countDown();
 			}
-			throw new RuntimeException("fail");
+			throw exceptionProvider.get();
 		}
 
 		public LinkedHashMap<Long, Message<?>> getReceivedMessages() {