Log error for class cast exception on lambda

**cherry-pick to 5.0.x**

* Fix test

* Polish log message.

* Polishing - docs and javadocs.

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

@@ -460,7 +460,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * @see GenericEndpointSpec
 	 */
 	public B controlBus(Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
-		return this.handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor(
+		return handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor(
 				new ControlBusMethodFilter())), endpointConfigurer);
 	}
 
@@ -537,7 +537,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	/**
-	 * Populate the {@link MessageTransformingHandler} instance for the provided {@link GenericTransformer}.
+	 * Populate the {@link MessageTransformingHandler} instance for the provided
+	 * {@link GenericTransformer}. Use {@link #transform(Class, GenericTransformer)} if
+	 * you need to access the entire message.
 	 * @param genericTransformer the {@link GenericTransformer} to populate.
 	 * @param <S> the source type - 'transform from'.
 	 * @param <T> the target type - 'transform to'.
@@ -599,16 +601,22 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * @see LambdaMessageProcessor
 	 */
 	public <P> B convert(Class<P> payloadType) {
+		Assert.isTrue(!payloadType.equals(Message.class), ".convert() does not support Message as an explicit type");
 		return transform(payloadType, p -> p);
 	}
 
 
 	/**
-	 * Populate the {@link MessageTransformingHandler} instance for the provided {@link GenericTransformer}
-	 * for the specific {@code payloadType} to convert at runtime.
-	 * @param payloadType the {@link Class} for expected payload type.
+	 * Populate the {@link MessageTransformingHandler} instance for the provided
+	 * {@link GenericTransformer} for the specific {@code payloadType} to convert at
+	 * runtime.
+	 * Use {@link #transform(Class, GenericTransformer)} if you need access to the
+	 * entire message.
+	 * @param payloadType the {@link Class} for expected payload type. It can also be
+	 * {@code Message.class} if you wish to access the entire message in the transformer.
+	 * Conversion to this type will be attempted, if necessary.
 	 * @param genericTransformer the {@link GenericTransformer} to populate.
-	 * @param <P> the payload type - 'transform from'.
+	 * @param <P> the payload type - 'transform from' or {@code Message.class}.
 	 * @param <T> the target type - 'transform to'.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 * @see MethodInvokingTransformer
@@ -619,10 +627,14 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	/**
-	 * Populate the {@link MessageTransformingHandler} instance for the provided {@link GenericTransformer}.
-	 * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.
+	 * Populate the {@link MessageTransformingHandler} instance for the provided
+	 * {@link GenericTransformer}. In addition accept options for the integration endpoint
+	 * using {@link GenericEndpointSpec}. Use
+	 * {@link #transform(Class, GenericTransformer, Consumer)} if you need to access the
+	 * entire message.
 	 * @param genericTransformer the {@link GenericTransformer} to populate.
-	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
+	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint
+	 * options.
 	 * @param <S> the source type - 'transform from'.
 	 * @param <T> the target type - 'transform to'.
 	 * @return the current {@link IntegrationFlowDefinition}.
@@ -632,7 +644,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 */
 	public <S, T> B transform(GenericTransformer<S, T> genericTransformer,
 			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
-		return this.transform(null, genericTransformer, endpointConfigurer);
+
+		return transform(null, genericTransformer, endpointConfigurer);
 	}
 
 	/**
@@ -651,6 +664,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	public <P> B convert(Class<P> payloadType,
 			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 
+		Assert.isTrue(!payloadType.equals(Message.class), ".convert() does not support Message");
 		return transform(payloadType, p -> p, endpointConfigurer);
 	}
 
@@ -658,10 +672,12 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * Populate the {@link MessageTransformingHandler} instance for the provided {@link GenericTransformer}
 	 * for the specific {@code payloadType} to convert at runtime.
 	 * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.
-	 * @param payloadType the {@link Class} for expected payload type.
+	 * @param payloadType the {@link Class} for expected payload type. It can also be
+	 * {@code Message.class} if you wish to access the entire message in the transformer.
+	 * Conversion to this type will be attempted, if necessary.
 	 * @param genericTransformer the {@link GenericTransformer} to populate.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
-	 * @param <P> the payload type - 'transform from'.
+	 * @param <P> the payload type - 'transform from', or {@code Message.class}.
 	 * @param <T> the target type - 'transform to'.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 * @see MethodInvokingTransformer
@@ -758,6 +774,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *  .filter("World"::equals)
 	 * }
 	 * </pre>
+	 * Use {@link #filter(Class, GenericSelector)} if you need to access the entire
+	 * message.
 	 * @param genericSelector the {@link GenericSelector} to use.
 	 * @param <P> the source payload type.
 	 * @return the current {@link IntegrationFlowDefinition}.
@@ -813,14 +831,16 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *  .filter(Date.class, p -> p.after(new Date()))
 	 * }
 	 * </pre>
-	 * @param payloadType the {@link Class} for desired {@code payload} type.
+	 * @param payloadType the {@link Class} for expected payload type. It can also be
+	 * {@code Message.class} if you wish to access the entire message in the selector.
+	 * Conversion to this type will be attempted, if necessary.
 	 * @param genericSelector the {@link GenericSelector} to use.
-	 * @param <P> the source payload type.
+	 * @param <P> the source payload type or {@code Message.class}.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 */
 	public <P> B filter(Class<P> payloadType, GenericSelector<P> genericSelector) {
-		return this.filter(payloadType, genericSelector, null);
+		return filter(payloadType, genericSelector, null);
 	}
 
 	/**
@@ -833,6 +853,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *  .filter("World"::equals, e -> e.autoStartup(false))
 	 * }
 	 * </pre>
+	 * Use {@link #filter(Class, GenericSelector, Consumer)} if you need to access the entire
+	 * message.
 	 * @param genericSelector the {@link GenericSelector} to use.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
 	 * @param <P> the source payload type.
@@ -853,10 +875,12 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *  .filter(Date.class, p -> p.after(new Date()), e -> e.autoStartup(false))
 	 * }
 	 * </pre>
-	 * @param payloadType the {@link Class} for desired {@code payload} type.
+	 * @param payloadType the {@link Class} for expected payload type. It can also be
+	 * {@code Message.class} if you wish to access the entire message in the selector.
+	 * Conversion to this type will be attempted, if necessary.
 	 * @param genericSelector the {@link GenericSelector} to use.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
-	 * @param <P> the source payload type.
+	 * @param <P> the source payload type or {@code Message.class}.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 * @see FilterEndpointSpec
@@ -914,7 +938,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 */
 	public B handle(String beanName, String methodName) {
-		return this.handle(beanName, methodName, null);
+		return handle(beanName, methodName, null);
 	}
 
 	/**
@@ -989,6 +1013,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *  .<Integer>handle((p, h) -> p / 2)
 	 * }
 	 * </pre>
+	 * Use {@link #handle(Class, GenericHandler)} if you need to access the entire
+	 * message.
 	 * @param handler the handler to invoke.
 	 * @param <P> the payload type to expect.
 	 * @return the current {@link IntegrationFlowDefinition}.
@@ -1009,6 +1035,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *  .<Integer>handle((p, h) -> p / 2, e -> e.autoStartup(false))
 	 * }
 	 * </pre>
+	 * Use {@link #handle(Class, GenericHandler, Consumer)} if you need to access the entire
+	 * message.
 	 * @param handler the handler to invoke.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
 	 * @param <P> the payload type to expect.
@@ -1018,7 +1046,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 */
 	public <P> B handle(GenericHandler<P> handler,
 			Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
-		return this.handle(null, handler, endpointConfigurer);
+		return handle(null, handler, endpointConfigurer);
 	}
 
 	/**
@@ -1031,15 +1059,16 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *  .handle(Integer.class, (p, h) -> p / 2)
 	 * }
 	 * </pre>
-	 * @param payloadType the expected payload type.
-	 * The accepted payload can be converted to this one at runtime
+	 * @param payloadType the {@link Class} for expected payload type. It can also be
+	 * {@code Message.class} if you wish to access the entire message in the handler.
+	 * Conversion to this type will be attempted, if necessary.
 	 * @param handler the handler to invoke.
-	 * @param <P> the payload type to expect.
+	 * @param <P> the payload type to expect, or {@code Message.class}.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 */
 	public <P> B handle(Class<P> payloadType, GenericHandler<P> handler) {
-		return this.handle(payloadType, handler, null);
+		return handle(payloadType, handler, null);
 	}
 
 	/**
@@ -1053,11 +1082,12 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *  .handle(Integer.class, (p, h) -> p / 2, e -> e.autoStartup(false))
 	 * }
 	 * </pre>
-	 * @param payloadType the expected payload type.
-	 * The accepted payload can be converted to this one at runtime
+	 * @param payloadType the {@link Class} for expected payload type. It can also be
+	 * {@code Message.class} if you wish to access the entire message in the handler.
+	 * Conversion to this type will be attempted, if necessary.
 	 * @param handler the handler to invoke.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
-	 * @param <P> the payload type to expect.
+	 * @param <P> the payload type to expect or {@code Message.class}.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 */
@@ -1070,7 +1100,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		else {
 			serviceActivatingHandler = new ServiceActivatingHandler(handler, ClassUtils.HANDLER_HANDLE_METHOD);
 		}
-		return this.handle(serviceActivatingHandler, endpointConfigurer);
+		return handle(serviceActivatingHandler, endpointConfigurer);
 	}
 
 	/**
@@ -1275,9 +1305,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 
 	/**
 	 * Accept a {@link Map} of values to be used for the
-	 * {@link org.springframework.messaging.Message} header enrichment.
+	 * {@link Message} header enrichment.
 	 * {@code values} can apply an {@link org.springframework.expression.Expression}
-	 * to be evaluated against a request {@link org.springframework.messaging.Message}.
+	 * to be evaluated against a request {@link Message}.
 	 * @param headers the Map of headers to enrich.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 */
@@ -1287,9 +1317,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 
 	/**
 	 * Accept a {@link Map} of values to be used for the
-	 * {@link org.springframework.messaging.Message} header enrichment.
+	 * {@link Message} header enrichment.
 	 * {@code values} can apply an {@link org.springframework.expression.Expression}
-	 * to be evaluated against a request {@link org.springframework.messaging.Message}.
+	 * to be evaluated against a request {@link Message}.
 	 * @param headers the Map of headers to enrich.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
 	 * @return the current {@link IntegrationFlowDefinition}.
@@ -1330,7 +1360,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 */
 	public B split() {
-		return this.split((Consumer<SplitterEndpointSpec<DefaultMessageSplitter>>) null);
+		return split((Consumer<SplitterEndpointSpec<DefaultMessageSplitter>>) null);
 	}
 
 	/**
@@ -1348,7 +1378,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * @see SplitterEndpointSpec
 	 */
 	public B split(Consumer<SplitterEndpointSpec<DefaultMessageSplitter>> endpointConfigurer) {
-		return this.split(new DefaultMessageSplitter(), endpointConfigurer);
+		return split(new DefaultMessageSplitter(), endpointConfigurer);
 	}
 
 	/**
@@ -1432,7 +1462,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 */
 	public B split(String beanName, String methodName) {
-		return this.split(beanName, methodName, null);
+		return split(beanName, methodName, null);
 	}
 
 	/**
@@ -1508,9 +1538,11 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *                           new Foo(rs.getInt(1), rs.getString(2)))))
 	 * }
 	 * </pre>
-	 * @param payloadType the expected payload type. Used at runtime to convert received payload type to.
+	 * @param payloadType the {@link Class} for expected payload type. It can also be
+	 * {@code Message.class} if you wish to access the entire message in the splitter.
+	 * Conversion to this type will be attempted, if necessary.
 	 * @param splitter the splitter {@link Function}.
-	 * @param <P> the payload type.
+	 * @param <P> the payload type or {@code Message.class}.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 */
@@ -1562,20 +1594,23 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *       , e -> e.applySequence(false))
 	 * }
 	 * </pre>
-	 * @param payloadType the expected payload type. Used at runtime to convert received payload type to.
+	 * @param payloadType the {@link Class} for expected payload type. It can also be
+	 * {@code Message.class} if you wish to access the entire message in the splitter.
+	 * Conversion to this type will be attempted, if necessary.
 	 * @param splitter the splitter {@link Function}.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
-	 * @param <P> the payload type.
+	 * @param <P> the payload type or {@code Message.class}.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 * @see SplitterEndpointSpec
 	 */
 	public <P> B split(Class<P> payloadType, Function<P, ?> splitter,
 			Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
+
 		MethodInvokingSplitter split = isLambda(splitter)
 				? new MethodInvokingSplitter(new LambdaMessageProcessor(splitter, payloadType))
 				: new MethodInvokingSplitter(splitter, ClassUtils.FUNCTION_APPLY_METHOD);
-		return this.split(split, endpointConfigurer);
+		return split(split, endpointConfigurer);
 	}
 
 	/**
@@ -1665,7 +1700,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 */
 	public B headerFilter(HeaderFilter headerFilter,
 			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
-		return this.transform(headerFilter, endpointConfigurer);
+		return transform(headerFilter, endpointConfigurer);
 	}
 
 	/**
@@ -1689,7 +1724,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 */
 	public B claimCheckIn(MessageStore messageStore,
 			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
-		return this.transform(new ClaimCheckInTransformer(messageStore), endpointConfigurer);
+		return transform(new ClaimCheckInTransformer(messageStore), endpointConfigurer);
 	}
 
 	/**
@@ -1730,7 +1765,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		ClaimCheckOutTransformer claimCheckOutTransformer = new ClaimCheckOutTransformer(messageStore);
 		claimCheckOutTransformer.setRemoveMessage(removeMessage);
-		return this.transform(claimCheckOutTransformer, endpointConfigurer);
+		return transform(claimCheckOutTransformer, endpointConfigurer);
 	}
 
 	/**
@@ -1895,6 +1930,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *  .route(p -> p.equals("foo") || p.equals("bar") ? new String[] {"foo", "bar"} : null)
 	 * }
 	 * </pre>
+	 * Use {@link #route(Class, Function)} if you need to access the entire message.
 	 * @param router the {@link Function} to use.
 	 * @param <S> the source payload type.
 	 * @param <T> the target result type.
@@ -1913,9 +1949,11 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *  .route(Integer.class, p -> p % 2 == 0)
 	 * }
 	 * </pre>
-	 * @param payloadType the expected payload type.
+	 * @param payloadType the {@link Class} for expected payload type. It can also be
+	 * {@code Message.class} if you wish to access the entire message in the splitter.
+	 * Conversion to this type will be attempted, if necessary.
 	 * @param router  the {@link Function} to use.
-	 * @param <S> the source payload type.
+	 * @param <S> the source payload type or {@code Message.class}.
 	 * @param <T> the target result type.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
@@ -1938,6 +1976,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *                       .applySequence(false))
 	 * }
 	 * </pre>
+	 * Use {@link #route(Class, Function, Consumer)} if you need to access the entire message.
 	 * @param router the {@link Function} to use.
 	 * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options.
 	 * @param <S> the source payload type.
@@ -1962,16 +2001,19 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 *                       .applySequence(false))
 	 * }
 	 * </pre>
-	 * @param payloadType the expected payload type.
+	 * @param payloadType the {@link Class} for expected payload type. It can also be
+	 * {@code Message.class} if you wish to access the entire message in the splitter.
+	 * Conversion to this type will be attempted, if necessary.
 	 * @param router the {@link Function} to use.
 	 * @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options.
-	 * @param <P> the source payload type.
+	 * @param <P> the source payload type or {@code Message.class}.
 	 * @param <T> the target result type.
 	 * @return the current {@link IntegrationFlowDefinition}.
 	 * @see LambdaMessageProcessor
 	 */
 	public <P, T> B route(Class<P> payloadType, Function<P, T> router,
 			Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer) {
+
 		MethodInvokingRouter methodInvokingRouter = isLambda(router)
 				? new MethodInvokingRouter(new LambdaMessageProcessor(router, payloadType))
 				: new MethodInvokingRouter(router, ClassUtils.FUNCTION_APPLY_METHOD);

spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java

@@ -22,6 +22,9 @@ import java.lang.reflect.Modifier;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
@@ -38,11 +41,14 @@ import org.springframework.util.ReflectionUtils;
  * - functional interface implementations.
  *
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 5.0
  */
 public class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFactoryAware {
 
+	private static final Log logger = LogFactory.getLog(LambdaMessageProcessor.class);
+
 	private final Object target;
 
 	private final Method method;
@@ -124,6 +130,12 @@ public class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFac
 			return this.method.invoke(this.target, args);
 		}
 		catch (InvocationTargetException e) {
+			if (e.getTargetException() instanceof ClassCastException) {
+				logger.error("Could not invoke the method due to a class cast exception, if using a lambda in the DSL, "
+						+ "consider using an overloaded EIP method that takes a Class<?> argument to explicitly "
+						+ "specify the type. An example of when this often occurs is if the lambda is configured to "
+						+ "receive a Message<?> argument.", e.getCause());
+			}
 			throw new MessageHandlingException(message, e.getCause());
 		}
 		catch (Exception e) {

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.dsl;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
@@ -72,6 +73,15 @@ public class LambdaMessageProcessorTests {
 		assertSame(testMessage, result);
 	}
 
+	@Test
+	public void testMessageAsArgumentLambda() {
+		LambdaMessageProcessor lmp = new LambdaMessageProcessor(
+				(GenericTransformer<Message<?>, Message<?>>) source -> messageTransformer(source), null);
+		lmp.setBeanFactory(mock(BeanFactory.class));
+		GenericMessage<String> testMessage = new GenericMessage<>("foo");
+		assertThatThrownBy(() -> lmp.processMessage(testMessage)).hasCauseExactlyInstanceOf(ClassCastException.class);
+	}
+
 	private void handle(GenericHandler<?> h) {
 		LambdaMessageProcessor lmp = new LambdaMessageProcessor(h, String.class);
 		lmp.setBeanFactory(getBeanFactory());

src/reference/asciidoc/dsl.adoc

@@ -113,6 +113,30 @@ The flow is "'one way'".
 That is, it does not provide a reply message but only prints the payload to STDOUT.
 The endpoints are automatically wired together by using direct channels.
 
+[[java-dsl-class-cast]]
+.Lambdas And `Message<?>` Arguments
+IMPORTANT: When using lambdas in EIP methods, the "input" argument is generally the message payload.
+If you wish to access the entire message, use one of the overloaded methods that take a `Class<?>` as the first parameter.
+For example, this won't work:
+
+====
+[source, java]
+----
+.<Message<?>, Foo>transform(m -> newFooFromMessage(m))
+----
+====
+
+This will fail at runtime with a `ClassCastException` because the lambda doesn't retain the argument type and the framework will attempt to cast the payload to a `Message<?>`.
+
+Instead, use:
+
+====
+[source, java]
+----
+.(Message.class, m -> newFooFromMessage(m))
+----
+====
+
 [[java-dsl-channels]]
 === Message Channels
 
@@ -294,6 +318,8 @@ Nevertheless, the DSL parser takes care of bean declarations for inline objects,
 
 See [https://docs.spring.io/spring-integration/api/org/springframework/integration/dsl/Transformers.html] in the Javadoc for more information and supported factory methods.
 
+Also see <<java-dsl-class-cast>>.
+
 [[java-dsl-inbound-adapters]]
 === Inbound Channel Adapters
 
@@ -403,6 +429,8 @@ public IntegrationFlow recipientListFlow() {
 
 The `.defaultOutputToParentFlow()` of the `.routeToRecipients()` definition lets you set the router's `defaultOutput` as a gateway to continue a process for the unmatched messages in the main flow.
 
+Also see <<java-dsl-class-cast>>.
+
 [[java-dsl-splitters]]
 === Splitters
 
@@ -427,6 +455,8 @@ public IntegrationFlow splitFlow() {
 The preceding example creates a splitter that splits a message containing a comma-delimited `String`.
 Note: The `getT2()` method comes from a `Tuple` `Collection`, which is the result of `EndpointSpec.get()`, and represents a pair of `ConsumerEndpointFactoryBean` and `DefaultMessageSplitter` for the preceding example.
 
+Also see <<java-dsl-class-cast>>.
+
 [[java-dsl-aggregators]]
 === Aggregators and Resequencers
 
@@ -531,6 +561,8 @@ public IntegrationFlow integerFlow() {
 ----
 ====
 
+Also see <<java-dsl-class-cast>>.
+
 [[java-dsl-log]]
 === Operator log()