diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 1f7e75193..f7bafa971 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/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 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java index 706a7932e..bb78b95e8 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java +++ b/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>> 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 binding = binder.bindConsumer(target, - bindingServiceProperties.getGroup(inputName), input, - consumerProperties); + Binding binding = doBindConsumer(input, inputName, binder, consumerProperties, target); bindings.add(binding); } bindings = Collections.unmodifiableCollection(bindings); @@ -105,6 +114,48 @@ public class BindingService { return bindings; } + public Binding doBindConsumer(T input, String inputName, Binder 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 late = new LateBinding(); + rescheduleConsumerBinding(input, inputName, binder, consumerProperties, target, late, e); + return late; + } + } + } + + public void rescheduleConsumerBinding(final T input, final String inputName, + final Binder binder, final ConsumerProperties consumerProperties, + final String target, final LateBinding 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 Binding bindProducer(T output, String outputName) { String bindingTarget = this.bindingServiceProperties @@ -122,8 +173,7 @@ public class BindingService { producerProperties = extendedProducerProperties; } validate(producerProperties); - Binding binding = binder.bindProducer(bindingTarget, output, - producerProperties); + Binding binding = doBindProducer(output, bindingTarget, binder, producerProperties); this.producerBindings.put(outputName, binding); return binding; } @@ -139,6 +189,42 @@ public class BindingService { } } + public Binding doBindProducer(T output, String bindingTarget, Binder 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 late = new LateBinding(); + rescheduleProducerBinding(output, bindingTarget, binder, producerProperties, late, e); + return late; + } + } + } + + public void rescheduleProducerBinding(final T output, final String bindingTarget, + final Binder binder, final ProducerProperties producerProperties, + final LateBinding 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> 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 implements Binding { + + private Binding delegate; + + private boolean unbound; + + LateBinding() { + super(); + } + + public synchronized void setDelegate(Binding 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 + "]"; + } + + } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index ceae60c79..a402fc338 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/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 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java index 2a34dad25..470a0f7c5 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java +++ b/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 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); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java index cd1f91857..d22e87a6f 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java +++ b/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 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> bindings = service.bindConsumer(inputChannel, inputChannelName); @@ -101,24 +106,13 @@ public class BindingServiceTests { Binding 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 mockBinding1 = Mockito.mock(Binding.class); - @SuppressWarnings("unchecked") Binding 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> 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 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 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 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 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 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> bindings = service.bindConsumer(inputChannel, inputChannelName); + assertThat(fail.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(bindings).hasSize(1); + Binding 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 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 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 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 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 { @Override @@ -427,4 +508,5 @@ public class BindingServiceTests { public static class SomeBindableType { } + }