GH-3946: Revise Router channelKeyFallback option (#3948)

* GH-3946: Revise Router channelKeyFallback option

Fixes https://github.com/spring-projects/spring-integration/issues/3946

The `AbstractMappingMessageRouter` has both `resolutionRequired` and `channelKeyFallback`
as `true` by default.
End-users expects them to back off when they set a `defaultOutputChannel`.
They really want something similar to Java `switch` statement

* Change the logic in the `AbstractMappingMessageRouter` to reset `channelKeyFallback`
to `false` when `defaultOutputChannel` to avoid attempts to resolve channel from name,
but rather fallback to `defaultOutputChannel` as it states from th mentioned
Java `switch` statement experience
* Deprecate `RouterSpec.noChannelKeyFallback()` in favor of newly introduced `channelKeyFallback(boolean)`
* Call `channelKeyFallback(false)` from an overloaded `defaultOutputToParentFlow()`
to reflect the mentioned expected behavior in Java DSL as well.
* Respectively, deprecate `KotlinRouterSpec.noChannelKeyFallback()` wrapper
in favor of newly introduced `channelKeyFallback(channelKeyFallback: Boolean)`
* Remove redundant already `noChannelKeyFallback()` option in the `NoFallbackAllowedTests`
* Document the change and new behavior

* Fix `IntegrationGraphServerTests` to `setChannelKeyFallback(true)` explicitly

* Remove not relevant `default-output-channel` from the `DynamicRouterTests-context.xml`

* Reject an `AbstractMappingMessageRouter` configuration where `defaultOutputChannel` is provided
and both `channelKeyFallback` & `resolutionRequired` are set to `true`.
Such a state makes `defaultOutputChannel` as not reachable and may cause some confusions in target
applications.

* Remove `&` symbol from JavaDocs

* Fix `boolean` expression for `AbstractMappingMessageRouter` configuration check

* Fix `IntegrationGraphServerTests` for new router behavior

* Improve language in docs

spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 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.
@@ -23,6 +23,7 @@ import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.router.AbstractMappingMessageRouter;
+import org.springframework.integration.router.AbstractMessageRouter;
 import org.springframework.integration.support.context.NamedComponent;
 import org.springframework.integration.support.management.MappingMessageRouterManagement;
 import org.springframework.messaging.MessageChannel;
@@ -118,9 +119,24 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
 	 * cause a stack overflow.
 	 * @return the router spec.
 	 * @since 5.2
+	 * @deprecated since 6.0 in favor of {@link  #channelKeyFallback(boolean)}
 	 */
+	@Deprecated(since = "6.0", forRemoval = true)
 	public RouterSpec<K, R> noChannelKeyFallback() {
-		this.handler.setChannelKeyFallback(false);
+		return channelKeyFallback(false);
+	}
+
+	/**
+	 * When true (default), if a resolved channel key does not exist in the channel map,
+	 * the key itself is used as the channel name, which we will attempt to resolve to a
+	 * channel. Set to {@code false} to disable this feature.
+	 * @param channelKeyFallback false to disable the fallback.
+	 * @return the router spec.
+	 * @since 6.0
+	 * @see AbstractMappingMessageRouter#setChannelKeyFallback(boolean)
+	 */
+	public RouterSpec<K, R> channelKeyFallback(boolean channelKeyFallback) {
+		this.handler.setChannelKeyFallback(channelKeyFallback);
 		return _this();
 	}
 
@@ -205,6 +221,20 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
 		return _this();
 	}
 
+	/**
+	 * Make a default output mapping of the router to the parent flow.
+	 * Use the next, after router, parent flow {@link MessageChannel} as a
+	 * {@link AbstractMessageRouter#setDefaultOutputChannel(MessageChannel)} of this router.
+	 * This option also disables {@link AbstractMappingMessageRouter#setChannelKeyFallback(boolean)},
+	 * if not called explicitly afterwards, to skip an attempt to resolve the channel name.
+	 * @return the router spec.
+	 * @since 6.0
+	 */
+	public RouterSpec<K, R> defaultOutputToParentFlow() {
+		return super.defaultOutputToParentFlow()
+				.channelKeyFallback(false);
+	}
+
 	@Override
 	public Map<Object, String> getComponentsToRegister() {
 		// The 'mappingProvider' must be added to the 'componentsToRegister' in the end to

spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -62,7 +62,7 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter
 	@SuppressWarnings("serial")
 	private final Map<String, MessageChannel> dynamicChannels =
 			Collections.synchronizedMap(
-					new LinkedHashMap<String, MessageChannel>(DEFAULT_DYNAMIC_CHANNEL_LIMIT, 0.75f, true) {
+					new LinkedHashMap<>(DEFAULT_DYNAMIC_CHANNEL_LIMIT, 0.75f, true) {
 
 						@Override
 						protected boolean removeEldestEntry(Entry<String, MessageChannel> eldest) {
@@ -79,6 +79,10 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter
 
 	private boolean channelKeyFallback = true;
 
+	private boolean defaultOutputChannelSet;
+
+	private boolean channelKeyFallbackSetExplicitly;
+
 	private volatile Map<String, String> channelMappings = new LinkedHashMap<>();
 
 
@@ -124,15 +128,72 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter
 	/**
 	 * When true (default), if a resolved channel key does not exist in the channel map,
 	 * the key itself is used as the channel name, which we will attempt to resolve to a
-	 * channel. Set to false to disable this feature. This could be useful to prevent
+	 * channel. Set to {@code false} to disable this feature. This could be useful to prevent
 	 * malicious actors from generating a message that could cause the message to be
 	 * routed to an unexpected channel, such as one upstream of the router, which would
 	 * cause a stack overflow.
-	 * @param channelKeyFallback false to disable the fall back.
+	 * @param channelKeyFallback false to disable the fallback.
 	 * @since 5.2
 	 */
 	public void setChannelKeyFallback(boolean channelKeyFallback) {
 		this.channelKeyFallback = channelKeyFallback;
+		this.channelKeyFallbackSetExplicitly = true;
+	}
+
+	/**
+	 * Set the default channel where Messages should be sent if channel resolution
+	 * fails to return any channels.
+	 * It also sets {@link #channelKeyFallback} to {@code false} to avoid
+	 * an attempt to resolve a channel from its key, but instead send the message
+	 * directly to this channel.
+	 * If {@link #channelKeyFallback} is set explicitly to {@code true},
+	 * the logic depends on the {@link #resolutionRequired} option
+	 * ({@code true} by default), and therefore a fallback to
+	 * this default output channel may never happen.
+	 * The configuration where a default output channel is present and
+	 * {@link #setResolutionRequired resolutionRequired} and
+	 * {@link #setChannelKeyFallback channelKeyFallback} are set to {@code true}
+	 * is rejected since it leads to ambiguity.
+	 * @param defaultOutputChannel The default output channel.
+	 * @since 6.0
+	 * @see #setChannelKeyFallback(boolean)
+	 * @see #setResolutionRequired(boolean)
+	 */
+	@Override
+	public void setDefaultOutputChannel(MessageChannel defaultOutputChannel) {
+		super.setDefaultOutputChannel(defaultOutputChannel);
+		if (!this.channelKeyFallbackSetExplicitly) {
+			this.channelKeyFallback = false;
+		}
+		this.defaultOutputChannelSet = true;
+	}
+
+	/**
+	 * Set the default channel where Messages should be sent if channel resolution
+	 * fails to return any channels.
+	 * It also sets {@link #channelKeyFallback} to {@code false} to avoid
+	 * an attempt to resolve a channel from its key, but instead send the message
+	 * directly to this channel.
+	 * If {@link #channelKeyFallback} is set explicitly to {@code true},
+	 * the logic depends on the {@link #resolutionRequired} option
+	 * ({@code true} by default), and therefore a fallback to
+	 * this default output channel may never happen.
+	 * The configuration where a default output channel is present and
+	 * {@link #setResolutionRequired resolutionRequired} and
+	 * {@link #setChannelKeyFallback channelKeyFallback} are set to {@code true}
+	 * is rejected since it leads to ambiguity.
+	 * @param defaultOutputChannelName the name of the channel bean for default output.
+	 * @since 6.0
+	 * @see #setChannelKeyFallback(boolean)
+	 * @see #setResolutionRequired(boolean)
+	 */
+	@Override
+	public void setDefaultOutputChannelName(String defaultOutputChannelName) {
+		super.setDefaultOutputChannelName(defaultOutputChannelName);
+		if (!this.channelKeyFallbackSetExplicitly) {
+			this.channelKeyFallback = false;
+		}
+		this.defaultOutputChannelSet = true;
 	}
 
 	/**
@@ -189,15 +250,14 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter
 		return Collections.unmodifiableSet(this.dynamicChannels.keySet());
 	}
 
-	/**
-	 * Subclasses must implement this method to return the channel keys.
-	 * A "key" might be present in this router's "channelMappings", or it
-	 * could be the channel's name or even the Message Channel instance itself.
-	 * @param message The message.
-	 * @return The channel keys.
-	 */
-	protected abstract List<Object> getChannelKeys(Message<?> message);
-
+	@Override
+	protected void onInit() {
+		super.onInit();
+		Assert.state(!this.channelKeyFallback || !this.resolutionRequired || !this.defaultOutputChannelSet,
+				"The 'defaultOutputChannel' cannot be reached " +
+						"when both 'channelKeyFallback' & 'resolutionRequired' are set to true. " +
+						"See their javadocs for more information.");
+	}
 
 	@Override
 	protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
@@ -207,6 +267,15 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter
 		return channels;
 	}
 
+	/**
+	 * Subclasses must implement this method to return the channel keys.
+	 * A "key" might be present in this router's "channelMappings", or it
+	 * could be the channel's name or even the Message Channel instance itself.
+	 * @param message The message.
+	 * @return The channel keys.
+	 */
+	protected abstract List<Object> getChannelKeys(Message<?> message);
+
 	/**
 	 * Convenience method allowing conversion of a list
 	 * of mappings in a control-bus message.

spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinRouterSpec.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2022 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.
@@ -47,8 +47,13 @@ class KotlinRouterSpec<K, R : AbstractMappingMessageRouter>(override val delegat
 		this.delegate.suffix(suffix)
 	}
 
+	@Deprecated(message = "Since 6.0", replaceWith = ReplaceWith("channelKeyFallback(false)"))
 	fun noChannelKeyFallback() {
-		this.delegate.noChannelKeyFallback()
+		channelKeyFallback(false)
+	}
+
+	fun channelKeyFallback(channelKeyFallback: Boolean) {
+		this.delegate.channelKeyFallback(channelKeyFallback)
 	}
 
 	fun channelMapping(key: K, channelName: String) {

spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/NoFallbackAllowedTests.java

@@ -33,6 +33,8 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 5.2
  *
  */
@@ -53,9 +55,7 @@ public class NoFallbackAllowedTests {
 
 		@Bean
 		public IntegrationFlow flow() {
-			return f -> f.route("headers.whereTo", r -> r
-					.noChannelKeyFallback()
-					.defaultOutputChannel(queue()));
+			return f -> f.route("headers.whereTo", r -> r.defaultOutputChannel(queue()));
 		}
 
 		@Bean

spring-integration-core/src/test/java/org/springframework/integration/graph/IntegrationGraphServerTests.java

@@ -135,7 +135,7 @@ public class IntegrationGraphServerTests {
 
 		List<Map<?, ?>> links = (List<Map<?, ?>>) map.get("links");
 		assertThat(links).isNotNull();
-		assertThat(links.size()).isEqualTo(34);
+		assertThat(links.size()).isEqualTo(32);
 
 		jsonArray =
 				JsonPathUtils.evaluate(baos.toByteArray(), "$..nodes[?(@.name == 'expressionRouter.router')]");
@@ -169,7 +169,7 @@ public class IntegrationGraphServerTests {
 		assertThat(nodes.size()).isEqualTo(34);
 		links = (List<Map<?, ?>>) map.get("links");
 		assertThat(links).isNotNull();
-		assertThat(links.size()).isEqualTo(37);
+		assertThat(links.size()).isEqualTo(35);
 
 		jsonArray = JsonPathUtils.evaluate(baos.toByteArray(), "$..nodes[?(@.name == 'router.router')]");
 		routerJson = jsonArray.toJSONString();
@@ -347,7 +347,7 @@ public class IntegrationGraphServerTests {
 			HeaderValueRouter router = new HeaderValueRouter("foo");
 			router.setChannelMapping("bar", "barChannel");
 			router.setChannelMapping("baz", "bazChannel");
-			router.setDefaultOutputChannel(discards());
+			router.setChannelKeyFallback(true);
 			return router;
 		}
 
@@ -365,7 +365,7 @@ public class IntegrationGraphServerTests {
 		public ExpressionEvaluatingRouter expressionRouter() {
 			ExpressionEvaluatingRouter router = new ExpressionEvaluatingRouter(
 					new SpelExpressionParser().parseExpression("headers['foo']"));
-			router.setDefaultOutputChannel(discards());
+			router.setChannelKeyFallback(true);
 			return router;
 		}
 
@@ -404,7 +404,7 @@ public class IntegrationGraphServerTests {
 		@Bean
 		@InboundChannelAdapter(channel = "fizChannel", autoStartup = "false")
 		public MessageSource<String> testSource() {
-			return new AbstractMessageSource<String>() {
+			return new AbstractMessageSource<>() {
 
 				@Override
 				public String getComponentType() {