GH-1893 Fix SpEL partition support

Consolidated code around PartitionHandler and interceptor

Consolidated ContentTypeConfiguration and the returning CompositeMessageConverter

Added support (reluctantly) to allow SpEL expressions for determining partition for function programming model. In other words something like  would work

That said, it is highly recommended to never use payload as a source for any kind of expressions (input or output) as payload constitutes privileged information that should only be accessible to he producer and consumer.

Resolves #1893
Resolves #1877

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PartitionHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2020 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,9 +16,18 @@
 
 package org.springframework.cloud.stream.binder;
 
+import java.lang.reflect.Field;
+import java.util.Map;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.context.expression.BeanFactoryResolver;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.ReflectionUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Utility class to determine if a binding is configured for partitioning (based on the
@@ -31,6 +40,7 @@ import org.springframework.util.Assert;
  * @author Ilayaperumal Gopinathan
  * @author Mark Fisher
  * @author Marius Bogoevici
+ * @author Oleg Zhurakousky
  */
 public class PartitionHandler {
 
@@ -42,6 +52,8 @@ public class PartitionHandler {
 
 	private final PartitionSelectorStrategy partitionSelectorStrategy;
 
+	private final ConfigurableListableBeanFactory beanFactory;
+
 	private volatile int partitionCount;
 
 	/**
@@ -50,16 +62,35 @@ public class PartitionHandler {
 	 * @param properties binder properties
 	 * @param partitionKeyExtractorStrategy PartitionKeyExtractor strategy
 	 * @param partitionSelectorStrategy PartitionSelector strategy
+	 *
+	 * @deprecated since 3.0.2. Please use another constructor which allows you to pass an instance of beanFactory
 	 */
+	@Deprecated
 	public PartitionHandler(EvaluationContext evaluationContext,
 			ProducerProperties properties,
 			PartitionKeyExtractorStrategy partitionKeyExtractorStrategy,
 			PartitionSelectorStrategy partitionSelectorStrategy) {
+
+		this(evaluationContext, properties, (ConfigurableListableBeanFactory) extractBeanFactoryFromEvaluationContext(evaluationContext));
+	}
+
+	/**
+	 * Construct a {@code PartitionHandler}.
+	 * @param evaluationContext evaluation context for binder
+	 * @param properties binder properties
+	 * @param beanFactory instance of ConfigurableListableBeanFactory
+	 *
+	 * @since 3.0.2
+	 */
+	public PartitionHandler(EvaluationContext evaluationContext,
+			ProducerProperties properties, ConfigurableListableBeanFactory beanFactory) {
+		this.beanFactory = beanFactory;
 		this.evaluationContext = evaluationContext;
 		this.producerProperties = properties;
-		this.partitionKeyExtractorStrategy = partitionKeyExtractorStrategy;
-		this.partitionSelectorStrategy = partitionSelectorStrategy;
+		this.partitionKeyExtractorStrategy = this.getPartitionKeyExtractorStrategy(properties);
+		this.partitionSelectorStrategy = this.getPartitionSelectorStrategy(properties);
 		this.partitionCount = this.producerProperties.getPartitionCount();
+
 	}
 
 	/**
@@ -120,4 +151,87 @@ public class PartitionHandler {
 		return null;
 	}
 
+	private PartitionKeyExtractorStrategy getPartitionKeyExtractorStrategy(
+			ProducerProperties producerProperties) {
+		PartitionKeyExtractorStrategy partitionKeyExtractor;
+		if (StringUtils.hasText(producerProperties.getPartitionKeyExtractorName())) {
+			partitionKeyExtractor = (PartitionKeyExtractorStrategy) this.beanFactory.getBean(
+					producerProperties.getPartitionKeyExtractorName(),
+					PartitionKeyExtractorStrategy.class);
+			Assert.notNull(partitionKeyExtractor,
+					"PartitionKeyExtractorStrategy bean with the name '"
+							+ producerProperties.getPartitionKeyExtractorName()
+							+ "' can not be found. Has it been configured (e.g., @Bean)?");
+		}
+		else {
+			Map<String, PartitionKeyExtractorStrategy> extractors = this.beanFactory
+					.getBeansOfType(PartitionKeyExtractorStrategy.class);
+			Assert.isTrue(extractors.size() <= 1,
+					"Multiple  beans of type 'PartitionKeyExtractorStrategy' found. "
+							+ extractors + ". Please "
+							+ "use 'spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName' property to specify "
+							+ "the name of the bean to be used.");
+			partitionKeyExtractor = CollectionUtils.isEmpty(extractors) ? null
+					: extractors.values().iterator().next();
+		}
+		return partitionKeyExtractor;
+	}
+
+	private PartitionSelectorStrategy getPartitionSelectorStrategy(
+			ProducerProperties producerProperties) {
+		PartitionSelectorStrategy partitionSelector;
+		if (StringUtils.hasText(producerProperties.getPartitionSelectorName())) {
+			partitionSelector = this.beanFactory.getBean(
+					producerProperties.getPartitionSelectorName(),
+					PartitionSelectorStrategy.class);
+			Assert.notNull(partitionSelector,
+					"PartitionSelectorStrategy bean with the name '"
+							+ producerProperties.getPartitionSelectorName()
+							+ "' can not be found. Has it been configured (e.g., @Bean)?");
+		}
+		else {
+			Map<String, PartitionSelectorStrategy> selectors = this.beanFactory
+					.getBeansOfType(PartitionSelectorStrategy.class);
+			Assert.isTrue(selectors.size() <= 1,
+					"Multiple  beans of type 'PartitionSelectorStrategy' found. "
+							+ selectors + ". Please "
+							+ "use 'spring.cloud.stream.bindings.output.producer.partitionSelectorName' property to specify "
+							+ "the name of the bean to be used.");
+			partitionSelector = CollectionUtils.isEmpty(selectors)
+					? new DefaultPartitionSelector()
+					: selectors.values().iterator().next();
+		}
+		return partitionSelector;
+	}
+
+	private static BeanFactory extractBeanFactoryFromEvaluationContext(EvaluationContext evaluationContext) {
+		try {
+			Field field = ReflectionUtils.findField(BeanFactoryResolver.class, "beanFactory");
+			field.setAccessible(true);
+			return (BeanFactory) field.get(evaluationContext);
+		}
+		catch (Exception e) {
+			throw new RuntimeException("Failed to extract beanFactory from EvaluationContext. Please use different constructor"
+					+ " which allows you to pass the instance of the beanFactory.");
+		}
+	}
+
+	/**
+	 * Default partition strategy; only works on keys with "real" hash codes, such as
+	 * String. Caller now always applies modulo so no need to do so here.
+	 */
+	private static class DefaultPartitionSelector implements PartitionSelectorStrategy {
+
+		@Override
+		public int selectPartition(Object key, int partitionCount) {
+			int hashCode = key.hashCode();
+			if (hashCode == Integer.MIN_VALUE) {
+				hashCode = 0;
+			}
+			return Math.abs(hashCode);
+		}
+	}
+
+
+
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2020 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.
@@ -28,8 +28,6 @@ import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.DefaultPollableMessageSource;
 import org.springframework.cloud.stream.binder.JavaClassMimeTypeUtils;
 import org.springframework.cloud.stream.binder.PartitionHandler;
-import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
-import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
 import org.springframework.cloud.stream.binder.PollableMessageSource;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.config.BindingProperties;
@@ -49,11 +47,11 @@ import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.util.Assert;
-import org.springframework.util.CollectionUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
+//import org.springframework.util.StringUtils;
 
 /**
  * A {@link MessageChannelConfigurer} that sets data types and message converters based on
@@ -147,23 +145,21 @@ public class MessageConverterConfigurer
 		boolean partitioned = !inbound && producerProperties != null && producerProperties.isPartitioned();
 		boolean functional = streamFunctionProperties != null && StringUtils.hasText(streamFunctionProperties.getDefinition());
 		if (partitioned) {
-			messageChannel.addInterceptor(new PartitioningInterceptor(bindingProperties,
-					getPartitionKeyExtractorStrategy(producerProperties),
-					getPartitionSelectorStrategy(producerProperties)));
+			if (inbound || !functional) {
+				messageChannel.addInterceptor(new PartitioningInterceptor(bindingProperties));
+			}
 		}
 
 		ConsumerProperties consumerProperties = bindingProperties.getConsumer();
 		if (this.isNativeEncodingNotSet(producerProperties, consumerProperties, inbound)) {
-			if (partitioned || !functional) {
-				if (inbound) {
-					messageChannel.addInterceptor(
-							new InboundContentTypeEnhancingInterceptor(contentType));
-				}
-				else {
-					messageChannel.addInterceptor(
-							new OutboundContentTypeConvertingInterceptor(contentType,
-									this.compositeMessageConverter));
-				}
+			if (inbound) {
+				messageChannel.addInterceptor(
+						new InboundContentTypeEnhancingInterceptor(contentType));
+			}
+			else {
+				messageChannel.addInterceptor(
+						new OutboundContentTypeConvertingInterceptor(contentType,
+								this.compositeMessageConverter));
 			}
 		}
 	}
@@ -180,76 +176,6 @@ public class MessageConverterConfigurer
 		}
 	}
 
-	private PartitionKeyExtractorStrategy getPartitionKeyExtractorStrategy(
-			ProducerProperties producerProperties) {
-		PartitionKeyExtractorStrategy partitionKeyExtractor;
-		if (StringUtils.hasText(producerProperties.getPartitionKeyExtractorName())) {
-			partitionKeyExtractor = this.beanFactory.getBean(
-					producerProperties.getPartitionKeyExtractorName(),
-					PartitionKeyExtractorStrategy.class);
-			Assert.notNull(partitionKeyExtractor,
-					"PartitionKeyExtractorStrategy bean with the name '"
-							+ producerProperties.getPartitionKeyExtractorName()
-							+ "' can not be found. Has it been configured (e.g., @Bean)?");
-		}
-		else {
-			Map<String, PartitionKeyExtractorStrategy> extractors = this.beanFactory
-					.getBeansOfType(PartitionKeyExtractorStrategy.class);
-			Assert.isTrue(extractors.size() <= 1,
-					"Multiple  beans of type 'PartitionKeyExtractorStrategy' found. "
-							+ extractors + ". Please "
-							+ "use 'spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName' property to specify "
-							+ "the name of the bean to be used.");
-			partitionKeyExtractor = CollectionUtils.isEmpty(extractors) ? null
-					: extractors.values().iterator().next();
-		}
-		return partitionKeyExtractor;
-	}
-
-	private PartitionSelectorStrategy getPartitionSelectorStrategy(
-			ProducerProperties producerProperties) {
-		PartitionSelectorStrategy partitionSelector;
-		if (StringUtils.hasText(producerProperties.getPartitionSelectorName())) {
-			partitionSelector = this.beanFactory.getBean(
-					producerProperties.getPartitionSelectorName(),
-					PartitionSelectorStrategy.class);
-			Assert.notNull(partitionSelector,
-					"PartitionSelectorStrategy bean with the name '"
-							+ producerProperties.getPartitionSelectorName()
-							+ "' can not be found. Has it been configured (e.g., @Bean)?");
-		}
-		else {
-			Map<String, PartitionSelectorStrategy> selectors = this.beanFactory
-					.getBeansOfType(PartitionSelectorStrategy.class);
-			Assert.isTrue(selectors.size() <= 1,
-					"Multiple  beans of type 'PartitionSelectorStrategy' found. "
-							+ selectors + ". Please "
-							+ "use 'spring.cloud.stream.bindings.output.producer.partitionSelectorName' property to specify "
-							+ "the name of the bean to be used.");
-			partitionSelector = CollectionUtils.isEmpty(selectors)
-					? new DefaultPartitionSelector()
-					: selectors.values().iterator().next();
-		}
-		return partitionSelector;
-	}
-
-	/**
-	 * Default partition strategy; only works on keys with "real" hash codes, such as
-	 * String. Caller now always applies modulo so no need to do so here.
-	 */
-	private static class DefaultPartitionSelector implements PartitionSelectorStrategy {
-
-		@Override
-		public int selectPartition(Object key, int partitionCount) {
-			int hashCode = key.hashCode();
-			if (hashCode == Integer.MIN_VALUE) {
-				hashCode = 0;
-			}
-			return Math.abs(hashCode);
-		}
-
-	}
-
 	/**
 	 * Primary purpose of this interceptor is to enhance/enrich Message that sent to the
 	 * *inbound* channel with 'contentType' header for cases where 'contentType' is not
@@ -381,15 +307,12 @@ public class MessageConverterConfigurer
 
 		private final PartitionHandler partitionHandler;
 
-		PartitioningInterceptor(BindingProperties bindingProperties,
-				PartitionKeyExtractorStrategy partitionKeyExtractorStrategy,
-				PartitionSelectorStrategy partitionSelectorStrategy) {
+		PartitioningInterceptor(BindingProperties bindingProperties) {
 			this.bindingProperties = bindingProperties;
 			this.partitionHandler = new PartitionHandler(
 					ExpressionUtils.createStandardEvaluationContext(
 							MessageConverterConfigurer.this.beanFactory),
-					this.bindingProperties.getProducer(), partitionKeyExtractorStrategy,
-					partitionSelectorStrategy);
+					this.bindingProperties.getProducer(), MessageConverterConfigurer.this.beanFactory);
 		}
 
 		public void setPartitionCount(int partitionCount) {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ContentTypeConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 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,7 +16,6 @@
 
 package org.springframework.cloud.stream.config;
 
-import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -29,20 +28,21 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Role;
 import org.springframework.integration.context.IntegrationContextUtils;
-import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter;
 import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter;
+import org.springframework.messaging.converter.CompositeMessageConverter;
 import org.springframework.messaging.converter.MessageConverter;
 
 /**
  * @author Vinicius Carvalho
  * @author Artem Bilan
+ * @author Oleg Zhurakousky
  */
 @Configuration
 @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
 class ContentTypeConfiguration {
 
 	@Bean(name = IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME)
-	public ConfigurableCompositeMessageConverter configurableCompositeMessageConverter(
+	public CompositeMessageConverter configurableCompositeMessageConverter(
 			ObjectProvider<ObjectMapper> objectMapperObjectProvider,
 			List<MessageConverter> customMessageConverters) {
 
@@ -50,12 +50,9 @@ class ContentTypeConfiguration {
 				.filter(c -> !(c instanceof DefaultDatatypeChannelMessageConverter)).collect(Collectors.toList());
 
 		CompositeMessageConverterFactory factory =
-				new CompositeMessageConverterFactory(new ArrayList<>(), objectMapperObjectProvider.getIfAvailable(ObjectMapper::new));
+				new CompositeMessageConverterFactory(customMessageConverters, objectMapperObjectProvider.getIfAvailable(ObjectMapper::new));
 
-		ArrayList<MessageConverter> messageConverters = new ArrayList<>(customMessageConverters);
-		messageConverters.addAll(factory.getMessageConverterForAllRegistered().getConverters());
-
-		return new ConfigurableCompositeMessageConverter(messageConverters);
+		return factory.getMessageConverterForAllRegistered();
 	}
 
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java

@@ -58,9 +58,11 @@ import org.springframework.cloud.function.context.config.ContextFunctionCatalogA
 import org.springframework.cloud.function.context.config.FunctionContextUtils;
 import org.springframework.cloud.function.context.config.RoutingFunction;
 import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.BinderTypeRegistry;
 import org.springframework.cloud.stream.binder.BindingCreatedEvent;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.PartitionHandler;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.binding.BindableProxyFactory;
 import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
@@ -81,11 +83,13 @@ import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.core.env.Environment;
 import org.springframework.core.type.MethodMetadata;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.MessageChannelReactiveUtils;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
 import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.handler.ServiceActivatingHandler;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.lang.Nullable;
@@ -127,10 +131,6 @@ public class FunctionConfiguration {
 			BindingServiceProperties serviceProperties, ConfigurableApplicationContext applicationContext,
 			FunctionBindingRegistrar bindingHolder, BinderAwareChannelResolver dynamicDestinationResolver) {
 
-//		boolean shouldCreateInitializer = bindableProxyFactories != null
-//				&& (applicationContext.containsBean("output") // need this to compose to existing legacy message source
-//				|| ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class)));
-
 		boolean shouldCreateInitializer = applicationContext.containsBean("output")
 				|| ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class));
 
@@ -417,7 +417,8 @@ public class FunctionConfiguration {
 			ProducerProperties producerProperties = StringUtils.hasText(outputChannelName)
 					? this.serviceProperties.getBindingProperties(outputChannelName).getProducer()
 							: null;
-			ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function, consumerProperties, producerProperties)) {
+			ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function, consumerProperties,
+					producerProperties, applicationContext)) {
 				@Override
 				protected void sendOutputs(Object result, Message<?> requestMessage) {
 					if (result instanceof Message && ((Message<?>) result).getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
@@ -512,7 +513,10 @@ public class FunctionConfiguration {
 
 		private final Field headersField;
 
-		FunctionWrapper(Function function, ConsumerProperties consumerProperties, ProducerProperties producerProperties) {
+		private final ConfigurableApplicationContext applicationContext;
+
+		FunctionWrapper(Function function, ConsumerProperties consumerProperties,
+				ProducerProperties producerProperties, ConfigurableApplicationContext applicationContext) {
 			this.function = function;
 			Type type =  ((FunctionInvocationWrapper) function).getFunctionType();
 			if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getOutputType(type, 0))) {
@@ -522,7 +526,9 @@ public class FunctionConfiguration {
 			this.producerProperties = producerProperties;
 			this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
 			this.headersField.setAccessible(true);
+			this.applicationContext = applicationContext;
 		}
+
 		@SuppressWarnings("unchecked")
 		@Override
 		public Object apply(Message<byte[]> message) {
@@ -531,7 +537,21 @@ public class FunctionConfiguration {
 						.getField(this.headersField, message.getHeaders());
 				headersMap.put(FunctionProperties.SKIP_CONVERSION_HEADER, consumerProperties.isUseNativeDecoding());
 			}
-			Object result = function.apply(message);
+
+			Function<Message, Message> outputMessageEnricher = null;
+			if (producerProperties != null && producerProperties.isPartitioned()) {
+				StandardEvaluationContext evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.applicationContext.getBeanFactory());
+				PartitionHandler partitionHandler = new PartitionHandler(evaluationContext, producerProperties, this.applicationContext.getBeanFactory());
+
+				outputMessageEnricher = outputMessage -> {
+					int partitionId = partitionHandler.determinePartition(outputMessage);
+					return MessageBuilder
+						.fromMessage(outputMessage)
+						.setHeader(BinderHeaders.PARTITION_HEADER, partitionId).build();
+				};
+			}
+
+			Object result = ((FunctionInvocationWrapper) function).apply(message, outputMessageEnricher);
 			if (result instanceof Publisher && ((FunctionInvocationWrapper) this.function).getTarget() instanceof RoutingFunction) {
 				throw new IllegalStateException("Routing to functions that return Publisher "
 						+ "is not supported in the context of Spring Cloud Stream.");

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java

@@ -552,6 +552,29 @@ public class ImplicitFunctionBindingTests {
 		}
 	}
 
+	@Test
+	public void partitionOnOutputPayloadTest() {
+		System.clearProperty("spring.cloud.function.definition");
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+				.getCompleteConfiguration(PojoFunctionConfiguration.class))
+						.web(WebApplicationType.NONE).run("--spring.cloud.function.definition=func",
+								"--spring.cloud.stream.bindings.func-out-0.producer.partitionKeyExpression=payload.id",
+								"--spring.cloud.stream.bindings.func-out-0.producer.partitionCount=5",
+								"--spring.jmx.enabled=false")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder.withPayload("Jim Lahey".getBytes()).build();
+
+			inputDestination.send(inputMessage, "func-in-0");
+
+			assertThat(outputDestination.receive(100, "func-out-0").getHeaders().get("scst_partition")).isEqualTo(3);
+
+			assertThat(outputDestination.receive(100)).isNull();
+		}
+	}
+
 	@EnableAutoConfiguration
 	public static class NoEnableBindingConfiguration {
 
@@ -737,4 +760,39 @@ public class ImplicitFunctionBindingTests {
 		}
 	}
 
+	@EnableAutoConfiguration
+	public static class PojoFunctionConfiguration {
+
+		@Bean
+		public Function<String, Person> func() {
+			return x -> {
+				Person person = new Person();
+				person.setName(x);
+				person.setId(3);
+				return person;
+			};
+		}
+	}
+
+	public static class Person {
+		private String name;
+		private int id;
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		public int getId() {
+			return id;
+		}
+
+		public void setId(int id) {
+			this.id = id;
+		}
+	}
+
 }