KGH-12: Late binding for kafka etc

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/12

Support late binding for binders that don't inherently support it.

Schedule attempts at 30 second intervals (default, configurable).

`IllegalState` and `IllegalArgument` exceptions are fatal.

Add test cases (mocks) to verify rebind after failure

Docs

Polishing - always use the setter on LateBinding

Remove volatile from the fields since they are only accessed from synchronized methods.

spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc

@@ -1123,6 +1123,12 @@ The typical usage of this property is to be nested in a customized environment <
 +
 Default: false.
 
+spring.cloud.stream.bindingRetryInterval::
+  The interval (seconds) between retrying binding creation when, for example, the binder doesn't support late binding and the broker is down (e.g. Apache Kafka).
+Set to zero to treat such conditions as fatal, preventing the application from starting.
++
+Default: 30
+
 [[binding-properties]]
 === Binding Properties
 

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

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.stream.binding;
 
+import java.sql.Date;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -36,6 +37,7 @@ import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 import org.springframework.validation.DataBinder;
@@ -63,15 +65,24 @@ public class BindingService {
 
 	private final Map<String, List<Binding<?>>> consumerBindings = new HashMap<>();
 
+	private final TaskScheduler taskScheduler;
+
 	private final BinderFactory binderFactory;
 
 	public BindingService(
 			BindingServiceProperties bindingServiceProperties,
 			BinderFactory binderFactory) {
+		this(bindingServiceProperties, binderFactory, null);
+	}
+
+	public BindingService(
+			BindingServiceProperties bindingServiceProperties,
+			BinderFactory binderFactory, TaskScheduler taskScheduler) {
 		this.bindingServiceProperties = bindingServiceProperties;
 		this.binderFactory = binderFactory;
 		this.validator = new CustomValidatorBean();
 		this.validator.afterPropertiesSet();
+		this.taskScheduler = taskScheduler;
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
@@ -95,9 +106,7 @@ public class BindingService {
 		}
 		validate(consumerProperties);
 		for (String target : bindingTargets) {
-			Binding<T> binding = binder.bindConsumer(target,
-					bindingServiceProperties.getGroup(inputName), input,
-					consumerProperties);
+			Binding<T> binding = doBindConsumer(input, inputName, binder, consumerProperties, target);
 			bindings.add(binding);
 		}
 		bindings = Collections.unmodifiableCollection(bindings);
@@ -105,6 +114,48 @@ public class BindingService {
 		return bindings;
 	}
 
+	public <T> Binding<T> doBindConsumer(T input, String inputName, Binder<T, ConsumerProperties, ?> binder,
+			ConsumerProperties consumerProperties, String target) {
+		if (this.taskScheduler == null || this.bindingServiceProperties.getBindingRetryInterval() <= 0) {
+			return binder.bindConsumer(target,
+					this.bindingServiceProperties.getGroup(inputName), input,
+					consumerProperties);
+		}
+		else {
+			try {
+				return binder.bindConsumer(target,
+						this.bindingServiceProperties.getGroup(inputName), input,
+						consumerProperties);
+			}
+			catch (RuntimeException e) {
+				LateBinding<T> late = new LateBinding<T>();
+				rescheduleConsumerBinding(input, inputName, binder, consumerProperties, target, late, e);
+				return late;
+			}
+		}
+	}
+
+	public <T> void rescheduleConsumerBinding(final T input, final String inputName,
+			final Binder<T, ConsumerProperties, ?> binder, final ConsumerProperties consumerProperties,
+			final String target, final LateBinding<T> late, RuntimeException exception) {
+		if (exception instanceof IllegalStateException || exception instanceof IllegalArgumentException) {
+			throw exception;
+		}
+		this.log.error("Failed to create consumer binding; retrying in " +
+			this.bindingServiceProperties.getBindingRetryInterval() + " seconds", exception);
+		this.taskScheduler.schedule(() -> {
+			try {
+				late.setDelegate(binder.bindConsumer(target,
+					this.bindingServiceProperties.getGroup(inputName), input,
+					consumerProperties));
+			}
+			catch (RuntimeException e) {
+				rescheduleConsumerBinding(input, inputName, binder, consumerProperties, target, late, e);
+			}
+		}, new Date(System.currentTimeMillis() +
+				this.bindingServiceProperties.getBindingRetryInterval() * 1_000));
+	}
+
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	public <T> Binding<T> bindProducer(T output, String outputName) {
 		String bindingTarget = this.bindingServiceProperties
@@ -122,8 +173,7 @@ public class BindingService {
 			producerProperties = extendedProducerProperties;
 		}
 		validate(producerProperties);
-		Binding<T> binding = binder.bindProducer(bindingTarget, output,
-				producerProperties);
+		Binding<T> binding = doBindProducer(output, bindingTarget, binder, producerProperties);
 		this.producerBindings.put(outputName, binding);
 		return binding;
 	}
@@ -139,6 +189,42 @@ public class BindingService {
 		}
 	}
 
+	public <T> Binding<T> doBindProducer(T output, String bindingTarget, Binder<T, ?, ProducerProperties> binder,
+			ProducerProperties producerProperties) {
+		if (this.taskScheduler == null || this.bindingServiceProperties.getBindingRetryInterval() <= 0) {
+			return binder.bindProducer(bindingTarget, output, producerProperties);
+		}
+		else {
+			try {
+				return binder.bindProducer(bindingTarget, output, producerProperties);
+			}
+			catch (RuntimeException e) {
+				LateBinding<T> late = new LateBinding<T>();
+				rescheduleProducerBinding(output, bindingTarget, binder, producerProperties, late, e);
+				return late;
+			}
+		}
+	}
+
+	public <T> void rescheduleProducerBinding(final T output, final String bindingTarget,
+			final Binder<T, ?, ProducerProperties> binder, final ProducerProperties producerProperties,
+			final LateBinding<T> late, final RuntimeException exception) {
+		if (exception instanceof IllegalStateException || exception instanceof IllegalArgumentException) {
+			throw exception;
+		}
+		this.log.error("Failed to create producer binding; retrying in " +
+				this.bindingServiceProperties.getBindingRetryInterval() + " seconds", exception);
+		this.taskScheduler.schedule(() -> {
+			try {
+				late.setDelegate(binder.bindProducer(bindingTarget, output, producerProperties));
+			}
+			catch (RuntimeException e) {
+				rescheduleProducerBinding(output, bindingTarget, binder, producerProperties, late, e);
+			}
+		}, new Date(System.currentTimeMillis() +
+				this.bindingServiceProperties.getBindingRetryInterval() * 1_000));
+	}
+
 	public void unbindConsumers(String inputName) {
 		List<Binding<?>> bindings = this.consumerBindings.remove(inputName);
 		if (bindings != null && !CollectionUtils.isEmpty(bindings)) {
@@ -188,4 +274,39 @@ public class BindingService {
 			throw new IllegalStateException(dataBinder.getBindingResult().toString());
 		}
 	}
+
+	private static class LateBinding<T> implements Binding<T> {
+
+		private Binding<T> delegate;
+
+		private boolean unbound;
+
+		LateBinding() {
+			super();
+		}
+
+		public synchronized void setDelegate(Binding<T> delegate) {
+			if (this.unbound) {
+				delegate.unbind();
+			}
+			else {
+				this.delegate = delegate;
+			}
+		}
+
+		@Override
+		public synchronized void unbind() {
+			this.unbound = true;
+			if (this.delegate != null) {
+				this.delegate.unbind();
+			}
+		}
+
+		@Override
+		public String toString() {
+			return "LateBinding [delegate=" + this.delegate + "]";
+		}
+
+	}
+
 }

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

@@ -65,6 +65,7 @@ import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.messaging.core.DestinationResolver;
 import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
 import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.tuple.spel.TuplePropertyAccessor;
 
 /**
@@ -82,7 +83,8 @@ import org.springframework.tuple.spel.TuplePropertyAccessor;
 @Import(ContentTypeConfiguration.class)
 public class BindingServiceConfiguration {
 
-	public static final String STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME = "streamListenerAnnotationBeanPostProcessor";
+	public static final String STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME =
+			"streamListenerAnnotationBeanPostProcessor";
 
 	public static final String ERROR_BRIDGE_CHANNEL = "errorBridgeChannel";
 
@@ -108,8 +110,8 @@ public class BindingServiceConfiguration {
 	// already exists).
 	@ConditionalOnMissingBean(BindingService.class)
 	public BindingService bindingService(BindingServiceProperties bindingServiceProperties,
-			BinderFactory binderFactory) {
-		return new BindingService(bindingServiceProperties, binderFactory);
+			BinderFactory binderFactory, TaskScheduler taskScheduler) {
+		return new BindingService(bindingServiceProperties, binderFactory, taskScheduler);
 	}
 
 	@Bean

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

@@ -48,6 +48,8 @@ import org.springframework.util.Assert;
 @JsonInclude(Include.NON_DEFAULT)
 public class BindingServiceProperties implements ApplicationContextAware, InitializingBean {
 
+	private static final int DEFAULT_BINDING_RETRY_INTERVAL = 30;
+
 	private ConversionService conversionService;
 
 	@Value("${INSTANCE_INDEX:${CF_INSTANCE_INDEX:0}}")
@@ -66,6 +68,8 @@ public class BindingServiceProperties implements ApplicationContextAware, Initia
 
 	private ConfigurableApplicationContext applicationContext;
 
+	private int bindingRetryInterval = DEFAULT_BINDING_RETRY_INTERVAL;
+
 	public Map<String, BindingProperties> getBindings() {
 		return this.bindings;
 	}
@@ -214,6 +218,14 @@ public class BindingServiceProperties implements ApplicationContextAware, Initia
 		return getBindingProperties(bindingName).getDestination();
 	}
 
+	public int getBindingRetryInterval() {
+		return this.bindingRetryInterval;
+	}
+
+	public void setBindingRetryInterval(int bindingRetryInterval) {
+		this.bindingRetryInterval = bindingRetryInterval;
+	}
+
 	public void updateProducerProperties(String bindingName, ProducerProperties producerProperties) {
 		if (this.bindings.containsKey(bindingName)) {
 			this.bindings.get(bindingName).setProducer(producerProperties);

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java

@@ -22,6 +22,8 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -52,8 +54,10 @@ import org.springframework.cloud.stream.converter.CompositeMessageConverterFacto
 import org.springframework.cloud.stream.reflection.GenericsUtils;
 import org.springframework.cloud.stream.utils.MockBinderConfiguration;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.core.DestinationResolutionException;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.fail;
@@ -66,6 +70,7 @@ import static org.mockito.ArgumentMatchers.same;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -78,6 +83,7 @@ import static org.mockito.Mockito.when;
  */
 public class BindingServiceTests {
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	public void testDefaultGroup() throws Exception {
 		BindingServiceProperties properties = new BindingServiceProperties();
@@ -91,9 +97,8 @@ public class BindingServiceTests {
 		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
 		BindingService service = new BindingService(properties, binderFactory);
 		MessageChannel inputChannel = new DirectChannel();
-		@SuppressWarnings("unchecked")
 		Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
-		when(binder.bindConsumer(eq("foo"), isNull(String.class), same(inputChannel),
+		when(binder.bindConsumer(eq("foo"), isNull(), same(inputChannel),
 				any(ConsumerProperties.class))).thenReturn(mockBinding);
 		Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel,
 				inputChannelName);
@@ -101,24 +106,13 @@ public class BindingServiceTests {
 		Binding<MessageChannel> binding = bindings.iterator().next();
 		assertThat(binding).isSameAs(mockBinding);
 		service.unbindConsumers(inputChannelName);
-		verify(binder).bindConsumer(eq("foo"), isNull(String.class), same(inputChannel),
+		verify(binder).bindConsumer(eq("foo"), isNull(), same(inputChannel),
 				any(ConsumerProperties.class));
 		verify(binding).unbind();
 		binderFactory.destroy();
 	}
 
-	private DefaultBinderFactory createMockBinderFactory() {
-		BinderTypeRegistry binderTypeRegistry = createMockBinderTypeRegistry();
-		return new DefaultBinderFactory(
-				Collections.singletonMap("mock", new BinderConfiguration("mock", new Properties(), true, true)),
-				binderTypeRegistry);
-	}
-
-	private DefaultBinderTypeRegistry createMockBinderTypeRegistry() {
-		return new DefaultBinderTypeRegistry(Collections.singletonMap("mock",
-				new BinderType("mock", new Class[] { MockBinderConfiguration.class })));
-	}
-
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	public void testMultipleConsumerBindings() throws Exception {
 		BindingServiceProperties properties = new BindingServiceProperties();
@@ -137,14 +131,12 @@ public class BindingServiceTests {
 				binderFactory);
 		MessageChannel inputChannel = new DirectChannel();
 
-		@SuppressWarnings("unchecked")
 		Binding<MessageChannel> mockBinding1 = Mockito.mock(Binding.class);
-		@SuppressWarnings("unchecked")
 		Binding<MessageChannel> mockBinding2 = Mockito.mock(Binding.class);
 
-		when(binder.bindConsumer(eq("foo"), isNull(String.class), same(inputChannel),
+		when(binder.bindConsumer(eq("foo"), isNull(), same(inputChannel),
 				any(ConsumerProperties.class))).thenReturn(mockBinding1);
-		when(binder.bindConsumer(eq("bar"), isNull(String.class), same(inputChannel),
+		when(binder.bindConsumer(eq("bar"), isNull(), same(inputChannel),
 				any(ConsumerProperties.class))).thenReturn(mockBinding2);
 
 		Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel,
@@ -160,9 +152,9 @@ public class BindingServiceTests {
 
 		service.unbindConsumers("input");
 
-		verify(binder).bindConsumer(eq("foo"), isNull(String.class), same(inputChannel),
+		verify(binder).bindConsumer(eq("foo"), isNull(), same(inputChannel),
 				any(ConsumerProperties.class));
-		verify(binder).bindConsumer(eq("bar"), isNull(String.class), same(inputChannel),
+		verify(binder).bindConsumer(eq("bar"), isNull(), same(inputChannel),
 				any(ConsumerProperties.class));
 		verify(binding1).unbind();
 		verify(binding2).unbind();
@@ -170,6 +162,7 @@ public class BindingServiceTests {
 		binderFactory.destroy();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	public void testExplicitGroup() throws Exception {
 		BindingServiceProperties properties = new BindingServiceProperties();
@@ -185,7 +178,6 @@ public class BindingServiceTests {
 		BindingService service = new BindingService(properties,
 				binderFactory);
 		MessageChannel inputChannel = new DirectChannel();
-		@SuppressWarnings("unchecked")
 		Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
 		when(binder.bindConsumer(eq("foo"), eq("fooGroup"), same(inputChannel),
 				any(ConsumerProperties.class))).thenReturn(mockBinding);
@@ -371,13 +363,6 @@ public class BindingServiceTests {
 		bindingService.bindProducer(new DirectChannel(), "output");
 	}
 
-	private BindingServiceProperties createBindingServiceProperties(HashMap<String, String> properties) {
-		BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
-		org.springframework.boot.context.properties.bind.Binder propertiesBinder = new org.springframework.boot.context.properties.bind.Binder(new MapConfigurationPropertySource(properties));
-		propertiesBinder.bind("spring.cloud.stream", org.springframework.boot.context.properties.bind.Bindable.ofInstance(bindingServiceProperties));
-		return bindingServiceProperties;
-	}
-
 	@Test
 	public void testUnrecognizedBinderDisallowedIfUsed() {
 		HashMap<String, String> properties = new HashMap<>();
@@ -408,6 +393,102 @@ public class BindingServiceTests {
 		assertThat(bindableType).isSameAs(SomeBindableType.class);
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	public void testLateBindingConsumer() throws Exception {
+		BindingServiceProperties properties = new BindingServiceProperties();
+		properties.setBindingRetryInterval(1);
+		Map<String, BindingProperties> bindingProperties = new HashMap<>();
+		BindingProperties props = new BindingProperties();
+		props.setDestination("foo");
+		final String inputChannelName = "input";
+		bindingProperties.put(inputChannelName, props);
+		properties.setBindings(bindingProperties);
+		DefaultBinderFactory binderFactory = createMockBinderFactory();
+		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
+		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+		scheduler.initialize();
+		BindingService service = new BindingService(properties, binderFactory, scheduler);
+		MessageChannel inputChannel = new DirectChannel();
+		final Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
+		final CountDownLatch fail = new CountDownLatch(2);
+		doAnswer(i -> {
+			fail.countDown();
+			if (fail.getCount() == 1) {
+				throw new RuntimeException("fail");
+			}
+			return mockBinding;
+		}).when(binder).bindConsumer(eq("foo"), isNull(), same(inputChannel), any(ConsumerProperties.class));
+		Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel, inputChannelName);
+		assertThat(fail.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(bindings).hasSize(1);
+		Binding<MessageChannel> binding = TestUtils.getPropertyValue(bindings.iterator().next(), "delegate",
+				Binding.class);
+		assertThat(binding).isSameAs(mockBinding);
+		service.unbindConsumers(inputChannelName);
+		verify(binder, times(2)).bindConsumer(eq("foo"), isNull(), same(inputChannel),
+				any(ConsumerProperties.class));
+		verify(binding).unbind();
+		binderFactory.destroy();
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	public void testLateBindingProducer() throws Exception {
+		BindingServiceProperties properties = new BindingServiceProperties();
+		properties.setBindingRetryInterval(1);
+		Map<String, BindingProperties> bindingProperties = new HashMap<>();
+		BindingProperties props = new BindingProperties();
+		props.setDestination("foo");
+		final String outputChannelName = "output";
+		bindingProperties.put(outputChannelName, props);
+		properties.setBindings(bindingProperties);
+		DefaultBinderFactory binderFactory = createMockBinderFactory();
+		Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
+		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+		scheduler.initialize();
+		BindingService service = new BindingService(properties, binderFactory, scheduler);
+		MessageChannel outputChannel = new DirectChannel();
+		final Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
+		final CountDownLatch fail = new CountDownLatch(2);
+		doAnswer(i -> {
+			fail.countDown();
+			if (fail.getCount() == 1) {
+				throw new RuntimeException("fail");
+			}
+			return mockBinding;
+		}).when(binder).bindProducer(eq("foo"), same(outputChannel), any(ProducerProperties.class));
+		Binding<MessageChannel> binding = service.bindProducer(outputChannel, outputChannelName);
+		assertThat(fail.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(binding).isNotNull();
+		Binding delegate = TestUtils.getPropertyValue(binding, "delegate", Binding.class);
+		assertThat(delegate).isSameAs(mockBinding);
+		service.unbindProducers(outputChannelName);
+		verify(binder, times(2)).bindProducer(eq("foo"), same(outputChannel), any(ProducerProperties.class));
+		verify(delegate).unbind();
+		binderFactory.destroy();
+		scheduler.destroy();
+	}
+
+	private DefaultBinderFactory createMockBinderFactory() {
+		BinderTypeRegistry binderTypeRegistry = createMockBinderTypeRegistry();
+		return new DefaultBinderFactory(
+				Collections.singletonMap("mock", new BinderConfiguration("mock", new Properties(), true, true)),
+				binderTypeRegistry);
+	}
+
+	private DefaultBinderTypeRegistry createMockBinderTypeRegistry() {
+		return new DefaultBinderTypeRegistry(Collections.singletonMap("mock",
+				new BinderType("mock", new Class[] { MockBinderConfiguration.class })));
+	}
+
+	private BindingServiceProperties createBindingServiceProperties(HashMap<String, String> properties) {
+		BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
+		org.springframework.boot.context.properties.bind.Binder propertiesBinder = new org.springframework.boot.context.properties.bind.Binder(new MapConfigurationPropertySource(properties));
+		propertiesBinder.bind("spring.cloud.stream", org.springframework.boot.context.properties.bind.Bindable.ofInstance(bindingServiceProperties));
+		return bindingServiceProperties;
+	}
+
 	public static class FooBinder
 			implements Binder<SomeBindableType, ConsumerProperties, ProducerProperties> {
 		@Override
@@ -427,4 +508,5 @@ public class BindingServiceTests {
 
 	public static class SomeBindableType {
 	}
+
 }