Revise retry for IntegrationReactiveUtils (#3543)

Related to: https://github.com/spring-cloud/stream-applications/issues/156

The `IntegrationReactiveUtils` uses a general `Flux.retry()`
operator to always retry for all the errors.
On the other hand it has only a `.doOnError(MessagingException.class)`
which leads swallowing all the other exceptions from logs and handling.

* Replace with `retryWhen()` for the `MessagingException` predicate
failing for all other exceptions.
The end-user may add their own `retry` or error handling mechanism to
the returned `Flux` from the `IntegrationReactiveUtils`

**Cherry-pick to 5.4.x**

spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2021 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.
@@ -37,6 +37,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
 import reactor.core.scheduler.Schedulers;
+import reactor.util.retry.Retry;
 
 /**
  * Utilities for adapting integration components to/from reactive types.
@@ -100,7 +101,7 @@ public final class IntegrationReactiveUtils {
 										Mono.delay(ctx.getOrDefault(DELAY_WHEN_EMPTY_KEY,
 												DEFAULT_DELAY_WHEN_EMPTY)))))
 				.repeat()
-				.retry();
+				.retryWhen(Retry.indefinitely().filter(MessagingException.class::isInstance));
 	}
 
 	/**

spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java

@@ -22,11 +22,14 @@ import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.util.IntegrationReactiveUtils;
+import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
 
 import reactor.core.Disposable;
@@ -43,7 +46,7 @@ import reactor.util.concurrent.Queues;
  *
  * @since 5.1.9
  */
-class MessageChannelReactiveUtilsTests {
+class IntegrationReactiveUtilsTests {
 
 	private static final Scheduler SCHEDULER = Schedulers.boundedElastic();
 
@@ -125,4 +128,28 @@ class MessageChannelReactiveUtilsTests {
 		assertThat(publisherSubscribed.await(10, TimeUnit.SECONDS)).isTrue();
 	}
 
+
+	@Test
+	void testRetryOnMessagingExceptionOnly() {
+		AtomicInteger retryAttempts = new AtomicInteger(3);
+		AtomicReference<Throwable> finalException = new AtomicReference<>();
+		MessageSource<?> messageSource =
+				() -> {
+					if (retryAttempts.getAndDecrement() > 0) {
+						throw new MessagingException("retryable MessagingException");
+					}
+					else {
+						throw new RuntimeException("non-retryable RuntimeException");
+					}
+				};
+
+		StepVerifier.create(IntegrationReactiveUtils.messageSourceToFlux(messageSource).doOnError(finalException::set))
+				.expectSubscription()
+				.expectErrorMessage("non-retryable RuntimeException")
+				.verify(Duration.ofSeconds(1));
+
+		assertThat(retryAttempts.get()).isEqualTo(-1);
+		assertThat(finalException.get()).hasMessage("non-retryable RuntimeException");
+	}
+
 }