Split test-binder from core spring-cloud-stream

* Remove all SI test-binder based components from core spring-cloud-stream-module
* Create a new module - spring-cloud-stream-test-binder that contains the test-binder
  and all it's related components
* Migrate tests from core module that use the test-binder into a separate module
  called spring-cloud-stream-integration-tests
* Remove the test-jar dependency using the classifier approach
* Update Spring Cloud Stream BOM with the new test-binder dependency
* Update Schema-Registry tests that use the old approach (using the test-jar with the classifier)
  with the new test-binder dependency

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2565

core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/AbstractDestination.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2017-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.integration.channel.AbstractSubscribableChannel;
+import org.springframework.messaging.SubscribableChannel;
+
+/**
+ * @author Oleg Zhurakousky
+ *
+ */
+abstract class AbstractDestination {
+
+	private final List<AbstractSubscribableChannel> channels = new ArrayList<>();
+
+	SubscribableChannel getChannel(int index) {
+		return this.channels.get(index);
+	}
+
+	void setChannel(SubscribableChannel channel) {
+		this.channels.add((AbstractSubscribableChannel) channel);
+		this.afterChannelIsSet(this.channels.size() - 1, ((AbstractSubscribableChannel) channel).getBeanName());
+	}
+
+	void afterChannelIsSet(int channelIndex, String name) {
+		// noop
+	}
+
+	SubscribableChannel getChannelByName(String name) {
+		name = name.endsWith(".destination") ? name : name + ".destination";
+		for (AbstractSubscribableChannel subscribableChannel : channels) {
+			if (subscribableChannel.getBeanName().equals(name)) {
+				return subscribableChannel;
+			}
+		}
+		return null;
+	}
+}

core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/FunctionBindingTestUtils.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright 2019-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.test;
+
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.cloud.function.context.FunctionCatalog;
+import org.springframework.cloud.function.context.FunctionRegistration;
+import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
+import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.binding.BindableProxyFactory;
+import org.springframework.cloud.stream.config.BindingProperties;
+import org.springframework.cloud.stream.config.BindingServiceProperties;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * A utility class to assist with just-in-time bindings.
+ * It is intended for internal framework testing and is NOT for public use. Breaking changes are likely!!!
+ *
+ * @author Oleg Zhurakousky
+ * @since 3.0.2
+ *
+ */
+public final class FunctionBindingTestUtils {
+
+	private FunctionBindingTestUtils() {
+
+	}
+
+	@SuppressWarnings("rawtypes")
+	public static void bind(ConfigurableApplicationContext applicationContext, Object function) {
+		try {
+			Object targetFunction = function;
+			if (function instanceof FunctionRegistration) {
+				targetFunction = ((FunctionRegistration) function).getTarget();
+			}
+			String functionName = targetFunction instanceof Function ? "function" : (targetFunction instanceof Consumer ? "consumer" : "supplier");
+
+			System.setProperty("spring.cloud.function.definition", functionName);
+			applicationContext.getBeanFactory().registerSingleton(functionName, function);
+
+			Object actualFunction =  ((FunctionInvocationWrapper) applicationContext
+				.getBean(FunctionCatalog.class).lookup(functionName)).getTarget();
+
+			InitializingBean functionBindingRegistrar = applicationContext.getBean("functionBindingRegistrar", InitializingBean.class);
+			functionBindingRegistrar.afterPropertiesSet();
+
+			BindableProxyFactory bindingProxy = applicationContext.getBean("&" + functionName + "_binding", BindableProxyFactory.class);
+			bindingProxy.afterPropertiesSet();
+
+			InitializingBean functionBinder = applicationContext.getBean("functionInitializer", InitializingBean.class);
+			functionBinder.afterPropertiesSet();
+
+			BindingServiceProperties bindingProperties = applicationContext.getBean(BindingServiceProperties.class);
+			String inputBindingName = functionName + "-in-0";
+			String outputBindingName = functionName + "-out-0";
+			Map<String, BindingProperties> bindings = bindingProperties.getBindings();
+			BindingProperties inputProperties = bindings.get(inputBindingName);
+			BindingProperties outputProperties = bindings.get(outputBindingName);
+			ConsumerProperties consumerProperties = inputProperties.getConsumer();
+			ProducerProperties producerProperties = outputProperties.getProducer();
+
+			TestChannelBinder binder = applicationContext.getBean(TestChannelBinder.class);
+			if (actualFunction instanceof Supplier || actualFunction instanceof Function) {
+				Binding<MessageChannel> bindProducer = binder.bindProducer(outputProperties.getDestination(),
+					applicationContext.getBean(outputBindingName, MessageChannel.class),
+					producerProperties == null ? new ProducerProperties() : producerProperties);
+				bindProducer.start();
+			}
+			if (actualFunction instanceof Consumer || actualFunction instanceof Function) {
+				Binding<MessageChannel> bindConsumer = binder.bindConsumer(inputProperties.getDestination(), null,
+					applicationContext.getBean(inputBindingName, MessageChannel.class),
+					consumerProperties == null ? new ConsumerProperties() : consumerProperties);
+				bindConsumer.start();
+			}
+		}
+		catch (Exception e) {
+			throw new IllegalStateException("Failed to bind function", e);
+		}
+		finally {
+			System.clearProperty("spring.cloud.function.definition");
+		}
+	}
+}

core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/InputDestination.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2017-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.test;
+
+import org.springframework.messaging.Message;
+
+/**
+ * Implementation of binder endpoint that represents the source destination (e.g.,
+ * destination from which messages will be received by Processor.INPUT). <br>
+ * You can interact with it by calling {@link #send(Message)} operation.
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class InputDestination extends AbstractDestination {
+
+	/**
+	 * Allows the {@link Message} to be sent to a Binder to be delegated to a default binding
+	 * destination (e.g., "function-in-0" for cases where you only have a single function with the name 'function').
+	 * @param message message to send
+	 */
+	public void send(Message<?> message) {
+		this.getChannel(0).send(message);
+	}
+
+	/**
+	 * @param message message to send
+	 * @param inputIndex input index
+	 * @deprecated since 3.0.2 in favor of {@link #receive(long, String)} where you should use the actual binding name (e.g., "foo-in-0")
+	 */
+	@Deprecated
+	public void send(Message<?> message, int inputIndex) {
+		this.getChannel(inputIndex).send(message);
+	}
+
+	/**
+	 * Allows the {@link Message} to be sent to a Binder's destination.<br>
+	 * This needs a bit of clarification. Just like with any binder, 'destination'
+	 * name and 'binding' name are usually the same unless additional configuration
+	 * is provided. For example; Assume you have a function 'uppercase'. The
+	 * 'binding' names for this function would be 'uppercase-in-0' (for input) and
+	 * 'uppercase-out-0' (for output). The 'destination' names would match as well
+	 * unless you decide to provide something like
+	 * 'spring.cloud.stream.bindings.uppercase-in-0.destination=upper' at which
+	 * point the binding names and destination names are different. <br>
+	 * <br>
+	 * So, it is important to remember that since this binder's goal is to emulate
+	 * real binders and real messaging systems you are sending TO and receiving FROM
+	 * destination (as if it was real broker destination) and binder will map and
+	 * delegate what you send to the binder's destination to individual bindings as
+	 * would the real binder.
+	 *
+	 * *
+	 *
+	 * <pre>
+	 *
+	 * // assume the following properties
+	 * "--spring.cloud.function.definition=uppercase",
+	 * "--spring.cloud.stream.bindings.uppercase-in-0.destination=upper",
+	 * "--spring.cloud.stream.bindings.uppercase-out-0.destination=upperout"
+	 *
+	 * // send/receive
+	 * inputDestination.send(message, "upper");
+	 * Message&lt;byte[]&gt; resultMessage = outputDestination.receive(1000, "upperout");
+	 *
+	 * // if 'destination' property is not provided for both input and output
+	 * inputDestination.send(message, "uppercase-in-0");
+	 * Message&lt;byte[]&gt; resultMessage = outputDestination.receive(1000, "uppercase-out-0");
+	 * </pre>
+	 *
+	 * @param message    message to send
+	 * @param destinationName the name of the destination
+	 */
+	public void send(Message<?> message, String destinationName) {
+		this.getChannelByName(destinationName).send(message);
+	}
+
+}

core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/OutputDestination.java

@@ -0,0 +1,125 @@
+/*
+ * Copyright 2017-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.test;
+
+import java.util.ArrayList;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedTransferQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.channel.AbstractSubscribableChannel;
+import org.springframework.messaging.Message;
+import org.springframework.util.StringUtils;
+
+/**
+ * Implementation of binder endpoint that represents the target destination (e.g.,
+ * destination which receives messages sent to Processor.OUTPUT) <br>
+ * You can interact with it by calling {@link #receive()} operation.
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class OutputDestination extends AbstractDestination {
+
+	private final Log log = LogFactory.getLog(OutputDestination.class);
+
+	private final ConcurrentHashMap<String, BlockingQueue<Message<byte[]>>> messageQueues = new ConcurrentHashMap<>();
+
+	public Message<byte[]> receive(long timeout, String bindingName) {
+		try {
+			bindingName = bindingName.endsWith(".destination") ? bindingName : bindingName + ".destination";
+			return this.outputQueue(bindingName).poll(timeout, TimeUnit.MILLISECONDS);
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+		}
+		return null;
+	}
+
+	/**
+	 * Will clear all output destinations.
+	 *
+	 * @since 3.0.6
+	 */
+	public void clear() {
+		this.messageQueues.values().forEach(v -> v.clear());
+	}
+
+	/**
+	 * Will clear output destination with specified name.
+	 *
+	 * @param destinationName the name of the output destination to be cleared.
+	 * @return true if attempt to clear specific destination is successful otherwise false.
+	 * @since 3.0.6
+	 */
+	public boolean clear(String destinationName) {
+		String queueName = destinationName.endsWith(".destination") ? destinationName : destinationName + ".destination";
+		if (StringUtils.hasText(destinationName) && this.messageQueues.containsKey(queueName)) {
+			this.messageQueues.get(queueName).clear();
+			return true;
+		}
+		return false;
+	}
+	/**
+	 * Allows to access {@link Message}s received by this {@link OutputDestination}.
+	 * @param timeout how long to wait before giving up
+	 * @return received message
+	 * @deprecated since 3.0.2 in favor of {@link #receive(long, String)} where you should use the actual binding name (e.g., "foo-in-0")
+	 */
+	@Deprecated
+	public Message<byte[]> receive(long timeout, int bindingIndex) {
+		log.warn("!!!While 'receive(long timeout, int bindingIndex)' method may still work it is deprecated no longer supported. "
+			+ "It will be removed after 3.1.3 release. Please use 'receive(long timeout, String bindingName)'");
+		try {
+			BlockingQueue<Message<byte[]>> destinationQueue = (new ArrayList<>(this.messageQueues.values())).get(bindingIndex);
+			return destinationQueue.poll(timeout, TimeUnit.MILLISECONDS);
+		}
+		catch (Exception e) {
+			Thread.currentThread().interrupt();
+		}
+		return null;
+	}
+
+	/**
+	 * Allows to access {@link Message}s received by this {@link OutputDestination}.
+	 * @return received message
+	 */
+	public Message<byte[]> receive() {
+		return this.receive(0, 0);
+	}
+
+	public Message<byte[]> receive(long timeout) {
+		return this.receive(timeout, 0);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	void afterChannelIsSet(int channelIndex, String bindingName) {
+		if (((AbstractSubscribableChannel) this.getChannelByName(bindingName)).getSubscriberCount() < 1) {
+			this.getChannelByName(bindingName).subscribe(message -> this.outputQueue(bindingName).offer((Message<byte[]>) message));
+		}
+	}
+
+	private BlockingQueue<Message<byte[]>> outputQueue(String bindingName) {
+		this.messageQueues.putIfAbsent(bindingName, new LinkedTransferQueue<>());
+		return this.messageQueues.get(bindingName);
+	}
+}

core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java

@@ -0,0 +1,326 @@
+/*
+ * Copyright 2015-2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.test;
+
+import java.util.function.Consumer;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderProvisioner.SpringIntegrationConsumerDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderProvisioner.SpringIntegrationProducerDestination;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.core.AttributeAccessor;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.acks.AcknowledgmentCallback;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.handler.BridgeHandler;
+import org.springframework.integration.support.DefaultErrorMessageStrategy;
+import org.springframework.integration.support.ErrorMessageStrategy;
+import org.springframework.integration.support.MapBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.retry.RecoveryCallback;
+import org.springframework.retry.RetryCallback;
+import org.springframework.retry.RetryContext;
+import org.springframework.retry.RetryListener;
+import org.springframework.retry.support.RetryTemplate;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * Implementation of {@link Binder} backed by Spring Integration framework. It is useful
+ * for localized demos and testing.
+ * <p>
+ * This binder extends from the same base class ({@link AbstractMessageChannelBinder}) as
+ * other binders (i.e., Rabbit, Kafka etc). Interaction with this binder is done via
+ * source and target destination which emulate real binder's destinations (i.e., Kafka
+ * topic) <br>
+ * The destination classes are
+ * <ul>
+ * <li>{@link InputDestination}</li>
+ * <li>{@link OutputDestination}</li>
+ * </ul>
+ * Simply autowire them in your your application and send/receive messages.
+ * </p>
+ * You must also add {@link TestChannelBinderConfiguration} to your configuration. Below
+ * is the example using Spring Boot test. <pre class="code">
+ *
+ * &#064;RunWith(SpringJUnit4ClassRunner.class)
+ * &#064;SpringBootTest(classes = {SpringIntegrationBinderConfiguration.class, TestWithSIBinder.MyProcessor.class})
+ * public class TestWithSIBinder {
+ *     &#064;Autowired
+ *     private SourceDestination sourceDestination;
+ *
+ *     &#064;Autowired
+ *     private TargetDestination targetDestination;
+ *
+ *     &#064;Test
+ *     public void testWiring() {
+ *         sourceDestination.send(new GenericMessage&lt;String&gt;("Hello"));
+ *         assertEquals("Hello world",
+ *             new String((byte[])targetDestination.receive().getPayload(), StandardCharsets.UTF_8));
+ *     }
+ *
+ *     &#064;SpringBootApplication
+ *     &#064;EnableBinding(Processor.class)
+ *     public static class MyProcessor {
+ *         &#064;StreamListener(Processor.INPUT)
+ *         &#064;SendTo(Processor.OUTPUT)
+ *         public String transform(String in) {
+ *             return in + " world";
+ *         }
+ *     }
+ * }
+ * </pre>
+ *
+ * @author Oleg Zhurakousky
+ * @author Gary Russell
+ *
+ */
+public class TestChannelBinder extends
+	AbstractMessageChannelBinder<ConsumerProperties, ProducerProperties, TestChannelBinderProvisioner> {
+
+	@Autowired
+	private BeanFactory beanFactory;
+
+	private Message<?> lastError;
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	private MessageSource<?> messageSourceDelegate = () -> new GenericMessage<>(
+		"polled data", new MapBuilder()
+		.put(MessageHeaders.CONTENT_TYPE, "text/plain")
+		.put(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, (AcknowledgmentCallback) status -> {
+		}).get());
+
+	public TestChannelBinder(TestChannelBinderProvisioner provisioningProvider) {
+		super(new String[] {}, provisioningProvider);
+	}
+
+	/**
+	 * Set a delegate {@link MessageSource} for pollable consumers.
+	 * @param messageSourceDelegate the delegate.
+	 */
+	@Autowired(required = false)
+	public void setMessageSourceDelegate(MessageSource<byte[]> messageSourceDelegate) {
+		this.messageSourceDelegate = messageSourceDelegate;
+	}
+
+	public Message<?> getLastError() {
+		return this.lastError;
+	}
+
+	public void resetLastError() {
+		this.lastError = null;
+	}
+
+	@Override
+	protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
+														ProducerProperties producerProperties, MessageChannel errorChannel)
+		throws Exception {
+		BridgeHandler handler = new BridgeHandler();
+		handler.setBeanFactory(this.beanFactory);
+		handler.setOutputChannel(
+			((SpringIntegrationProducerDestination) destination).getChannel());
+		return handler;
+	}
+
+	@Override
+	protected MessageProducer createConsumerEndpoint(ConsumerDestination destination,
+													String group, ConsumerProperties properties) throws Exception {
+		ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
+		SubscribableChannel siBinderInputChannel = ((SpringIntegrationConsumerDestination) destination)
+			.getChannel();
+
+		IntegrationMessageListeningContainer messageListenerContainer = new IntegrationMessageListeningContainer();
+		IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter(
+			messageListenerContainer);
+
+		String groupName = StringUtils.hasText(group) ? group : "anonymous";
+		ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination,
+			groupName, properties);
+		if (properties.getMaxAttempts() > 1) {
+			adapter.setRetryTemplate(buildRetryTemplate(properties));
+			adapter.setRecoveryCallback(errorInfrastructure.getRecoverer());
+		}
+		else {
+			adapter.setErrorMessageStrategy(errorMessageStrategy);
+			adapter.setErrorChannel(errorInfrastructure.getErrorChannel());
+		}
+
+		siBinderInputChannel.subscribe(messageListenerContainer);
+
+		return adapter;
+	}
+
+	@Override
+	protected PolledConsumerResources createPolledConsumerResources(String name,
+																	String group, ConsumerDestination destination,
+																	ConsumerProperties consumerProperties) {
+		return new PolledConsumerResources(this.messageSourceDelegate,
+			registerErrorInfrastructure(destination, group, consumerProperties));
+	}
+
+	@Override
+	protected MessageHandler getErrorMessageHandler(ConsumerDestination destination,
+													String group, ConsumerProperties consumerProperties) {
+		return m -> {
+			this.logger.debug("Error handled: " + m);
+			this.lastError = m;
+		};
+	}
+
+	/**
+	 * Implementation of simple message listener container modeled after AMQP
+	 * SimpleMessageListenerContainer.
+	 */
+	private static class IntegrationMessageListeningContainer implements MessageHandler {
+
+		private Consumer<Message<?>> listener;
+
+		@Override
+		public void handleMessage(Message<?> message) throws MessagingException {
+			this.listener.accept(message);
+		}
+
+		public void setMessageListener(Consumer<Message<?>> listener) {
+			this.listener = listener;
+		}
+
+	}
+
+	/**
+	 * Implementation of inbound channel adapter modeled after AmqpInboundChannelAdapter.
+	 */
+	private static class IntegrationBinderInboundChannelAdapter
+		extends MessageProducerSupport {
+
+		private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<AttributeAccessor>();
+
+		private final IntegrationMessageListeningContainer listenerContainer;
+
+		private RetryTemplate retryTemplate;
+
+		private RecoveryCallback<? extends Object> recoveryCallback;
+
+		IntegrationBinderInboundChannelAdapter(
+			IntegrationMessageListeningContainer listenerContainer) {
+			this.listenerContainer = listenerContainer;
+		}
+
+		// Temporarily unused until DLQ strategy for this binder becomes a requirement
+		public void setRecoveryCallback(
+			RecoveryCallback<? extends Object> recoveryCallback) {
+			this.recoveryCallback = recoveryCallback;
+		}
+
+		public void setRetryTemplate(RetryTemplate retryTemplate) {
+			this.retryTemplate = retryTemplate;
+		}
+
+		@Override
+		protected void onInit() {
+			if (this.retryTemplate != null) {
+				Assert.state(getErrorChannel() == null,
+					"Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
+						+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
+						+ "send an error message when retries are exhausted");
+			}
+			Listener messageListener = new Listener();
+			if (this.retryTemplate != null) {
+				this.retryTemplate.registerListener(messageListener);
+			}
+			this.listenerContainer.setMessageListener(messageListener);
+		}
+
+		protected class Listener implements RetryListener, Consumer<Message<?>> {
+
+			@Override
+			@SuppressWarnings("unchecked")
+			public void accept(Message<?> message) {
+				try {
+					if (IntegrationBinderInboundChannelAdapter.this.retryTemplate == null) {
+						try {
+							processMessage(message);
+						}
+						finally {
+							attributesHolder.remove();
+						}
+					}
+					else {
+						IntegrationBinderInboundChannelAdapter.this.retryTemplate
+							.execute(context -> {
+								processMessage(message);
+								return null;
+							}, (RecoveryCallback<Object>) IntegrationBinderInboundChannelAdapter.this.recoveryCallback);
+					}
+				}
+				catch (RuntimeException e) {
+					if (getErrorChannel() != null) {
+						getMessagingTemplate()
+							.send(getErrorChannel(),
+								buildErrorMessage(null, new IllegalStateException(
+									"Message conversion failed: " + message,
+									e)));
+					}
+					else {
+						throw e;
+					}
+				}
+			}
+
+			private void processMessage(Message<?> message) {
+				sendMessage(message);
+			}
+
+			@Override
+			public <T, E extends Throwable> boolean open(RetryContext context,
+														RetryCallback<T, E> callback) {
+				if (IntegrationBinderInboundChannelAdapter.this.recoveryCallback != null) {
+					attributesHolder.set(context);
+				}
+				return true;
+			}
+
+			@Override
+			public <T, E extends Throwable> void close(RetryContext context,
+													RetryCallback<T, E> callback, Throwable throwable) {
+				attributesHolder.remove();
+			}
+
+			@Override
+			public <T, E extends Throwable> void onError(RetryContext context,
+														RetryCallback<T, E> callback, Throwable throwable) {
+				// Empty
+			}
+
+		}
+
+	}
+
+}

core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/TestChannelBinderConfiguration.java

@@ -0,0 +1,111 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration;
+import org.springframework.cloud.stream.config.BindingServiceConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.config.EnableIntegration;
+
+/**
+ * {@link Binder} configuration backed by Spring Integration.
+ *
+ * Please see {@link TestChannelBinder} for more details.
+ *
+ * @param <T> binding type
+ * @author Oleg Zhurakousky
+ * @author David Turanski
+ * @see TestChannelBinder
+ */
+@Configuration
+@ConditionalOnMissingBean(Binder.class)
+@Import(BinderFactoryAutoConfiguration.class)
+@EnableIntegration
+public class TestChannelBinderConfiguration<T> {
+
+	/**
+	 * The name of the test binder.
+	 */
+	public static final String NAME = "integration";
+
+	/**
+	 * Utility operation to return an array of configuration classes defined in
+	 * {@link EnableBinding} annotation. Typically used for tests that do not rely on
+	 * creating an SCSt boot application annotated with {@link EnableBinding}, yet require
+	 * full {@link Binder} configuration.
+	 * @param additionalConfigurationClasses config classes to be added to the default
+	 * config
+	 * @return an array of configuration classes defined in {@link EnableBinding}
+	 * annotation
+	 */
+	public static Class<?>[] getCompleteConfiguration(
+		Class<?>... additionalConfigurationClasses) {
+		List<Class<?>> configClasses = new ArrayList<>();
+		configClasses.add(TestChannelBinderConfiguration.class);
+		configClasses.add(BindingServiceConfiguration.class);
+		if (additionalConfigurationClasses != null) {
+			configClasses.addAll(Arrays.asList(additionalConfigurationClasses));
+		}
+		return configClasses.toArray(new Class<?>[] {});
+	}
+
+	/**
+	 * Create an {@link ApplicationContextRunner} with user configuration using {@link #getCompleteConfiguration}.
+	 * @param additionalConfigurationClasses config classes to be added to the default
+	 * config
+	 * @return the ApplicationContextRunner
+	 */
+	public static ApplicationContextRunner applicationContextRunner(Class<?>... additionalConfigurationClasses) {
+		return new ApplicationContextRunner()
+			.withUserConfiguration(getCompleteConfiguration(additionalConfigurationClasses));
+	}
+
+	@Bean
+	public InputDestination sourceDestination() {
+		return new InputDestination();
+	}
+
+	@Bean
+	public OutputDestination targetDestination() {
+		return new OutputDestination();
+	}
+
+	@SuppressWarnings("unchecked")
+	@Bean
+	public Binder<T, ? extends ConsumerProperties, ? extends ProducerProperties> springIntegrationChannelBinder(
+		TestChannelBinderProvisioner provisioner) {
+		return (Binder<T, ? extends ConsumerProperties, ? extends ProducerProperties>) new TestChannelBinder(
+			provisioner);
+	}
+
+	@Bean
+	public TestChannelBinderProvisioner springIntegrationProvisioner() {
+		return new TestChannelBinderProvisioner();
+	}
+
+}

core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/TestChannelBinderProvisioner.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright 2017-2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.cloud.stream.provisioning.ProvisioningException;
+import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
+import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.PublishSubscribeChannel;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.SubscribableChannel;
+
+/**
+ * {@link ProvisioningProvider} to support {@link TestChannelBinder}. It exists primarily
+ * to support {@link AbstractMessageChannel} semantics for creating
+ * {@link ConsumerDestination} and {@link ProducerDestination}, to interact with this
+ * {@link Binder}.
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class TestChannelBinderProvisioner
+	implements ProvisioningProvider<ConsumerProperties, ProducerProperties> {
+
+	private final Map<String, SubscribableChannel> provisionedDestinations = new HashMap<>();
+
+	@Autowired
+	private InputDestination source;
+
+	@Autowired
+	private OutputDestination target;
+
+	/**
+	 * Will provision producer destination as an SI {@link PublishSubscribeChannel}. <br>
+	 * This provides convenience of registering additional subscriber (handler in the test
+	 * method) along side of being able to call {@link OutputDestination#receive()} to get
+	 * a {@link Message} for additional assertions.
+	 */
+	@Override
+	public ProducerDestination provisionProducerDestination(String name,
+															ProducerProperties properties) throws ProvisioningException {
+		SubscribableChannel destination = this.provisionDestination(name, true);
+		this.target.setChannel(destination);
+		return new SpringIntegrationProducerDestination(name, destination);
+	}
+
+	/**
+	 * Will provision consumer destination as SI {@link DirectChannel}.
+	 */
+	@Override
+	public ConsumerDestination provisionConsumerDestination(String name, String group,
+															ConsumerProperties properties) throws ProvisioningException {
+		SubscribableChannel destination = this.provisionDestination(name, false);
+		if (this.source != null) {
+			this.source.setChannel(destination);
+		}
+		return new SpringIntegrationConsumerDestination(name, destination);
+	}
+
+	private SubscribableChannel provisionDestination(String name, boolean pubSub) {
+		String destinationName = name + ".destination";
+		SubscribableChannel destination = this.provisionedDestinations
+			.get(destinationName);
+		if (destination == null) {
+			destination = new PublishSubscribeChannel();
+			((AbstractMessageChannel) destination).setBeanName(destinationName);
+			((AbstractMessageChannel) destination).setComponentName(destinationName);
+			this.provisionedDestinations.put(destinationName, destination);
+		}
+		return destination;
+	}
+
+	class SpringIntegrationConsumerDestination implements ConsumerDestination {
+
+		private final String name;
+
+		private final SubscribableChannel channel;
+
+		SpringIntegrationConsumerDestination(String name, SubscribableChannel channel) {
+			this.name = name;
+			this.channel = channel;
+		}
+
+		public SubscribableChannel getChannel() {
+			return this.channel;
+		}
+
+		@Override
+		public String getName() {
+			return this.name;
+		}
+
+	}
+
+	class SpringIntegrationProducerDestination implements ProducerDestination {
+
+		private final String name;
+
+		private final SubscribableChannel channel;
+
+		SpringIntegrationProducerDestination(String name, SubscribableChannel channel) {
+			this.name = name;
+			this.channel = channel;
+		}
+
+		@Override
+		public String getNameForPartition(int partition) {
+			return this.getName() + partition;
+		}
+
+		public SubscribableChannel getChannel() {
+			return this.channel;
+		}
+
+		@Override
+		public String getName() {
+			return this.name;
+		}
+
+	}
+
+}

core/spring-cloud-stream-test-binder/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderTests.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2021-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.test;
+
+import java.lang.reflect.Method;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class TestChannelBinderTests {
+
+	@Test
+	public void test() throws Exception {
+		// nothing to assert. no failure on this test signifies success
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+			TestChannelBinderConfiguration.getCompleteConfiguration(SampleConfiguration.class))
+			.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false",
+				"--spring.cloud.stream.bindings.function-in-0.destination=input")) {
+			TestChannelBinder binder = context.getBean(TestChannelBinder.class);
+			Method registerErrorInfrastructure = ReflectionUtils
+				.findMethod(TestChannelBinder.class, "registerErrorInfrastructure", ProducerDestination.class, String.class);
+			registerErrorInfrastructure.setAccessible(true);
+			ProducerDestination destination = new ProducerDestination() {
+				@Override
+				public String getNameForPartition(int partition) {
+					return "sample";
+				}
+
+				@Override
+				public String getName() {
+					return "sample";
+				}
+			};
+			registerErrorInfrastructure.invoke(binder, destination, "function-in-0");
+			destination = new ProducerDestination() {
+				@Override
+				public String getNameForPartition(int partition) {
+					return "sample";
+				}
+
+				@Override
+				public String getName() {
+					return "sample";
+				}
+			};
+			registerErrorInfrastructure.invoke(binder, destination, "function-in-0");
+		}
+	}
+
+	@EnableAutoConfiguration
+	public static class SampleConfiguration {
+
+	}
+}