GH-3234 Fix TestChannelBinder subscriber leak on context pause/resume

Author: olegz

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

@@ -158,7 +158,7 @@ protected MessageProducer createConsumerEndpoint(ConsumerDestination destination
 			.getChannel();
 
 		IntegrationMessageListeningContainer messageListenerContainer = new IntegrationMessageListeningContainer();
-		IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter(
+		IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter(siBinderInputChannel,
 			messageListenerContainer);
 
 		String groupName = StringUtils.hasText(group) ? group : "anonymous";
@@ -228,9 +228,12 @@ private static class IntegrationBinderInboundChannelAdapter
 
 		private RecoveryCallback<? extends Object> recoveryCallback;
 
-		IntegrationBinderInboundChannelAdapter(
+		private final SubscribableChannel subscribableChannel;
+
+		IntegrationBinderInboundChannelAdapter(SubscribableChannel subscribableChannel,
 			IntegrationMessageListeningContainer listenerContainer) {
 			this.listenerContainer = listenerContainer;
+			this.subscribableChannel = subscribableChannel;
 		}
 
 		// Temporarily unused until DLQ strategy for this binder becomes a requirement
@@ -243,6 +246,12 @@ public void setRetryTemplate(RetryTemplate retryTemplate) {
 			this.retryTemplate = retryTemplate;
 		}
 
+		@Override
+		protected void doStop() {
+			super.doStop();
+			this.subscribableChannel.unsubscribe(listenerContainer);
+		}
+
 		@Override
 		protected void onInit() {
 			if (this.retryTemplate != null) {