From b21dc0dff353d735f7acf915ab189550297c2ceb Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 17 May 2018 17:28:31 -0400 Subject: [PATCH] Fix IntegrationFlowContext concurrency issue When we register `IntegrationFlow` s concurrently at runtime, we may end up with the problem when we register the same object with the same bean name, but in different places. Or when we turn off bean overriding, we end up with the exception that bean with the name already registered * Wrap `IntegrationFlow` bean registration in the `StandardIntegrationFlowContext` into the `Lock` when its bean name is generating * Make `StandardIntegrationFlowContext.registry` as `ConcurrentHashMap` to avoid `ConcurrentModificationException` during `put()` and `remove()` * Fix concurrency for beans registration with the generation names in the `IntegrationFlowBeanPostProcessor` using an `IntegrationFlow` id as a prefix for uniqueness. **Cherry-pick to 5.0.x** Fix generated bean name in the WebFluxDslTests Use only single `Lock` in the `StandardIntegrationFlowContext`: we don't need a fully blown `LockRegistry` there anymore since we have only one synchronization block there and it is always around the same type * Add `What's New` note, and mention changes in the `dsl.adoc` Minor doc polishing. --- .../dsl/IntegrationFlowBeanPostProcessor.java | 24 ++-- .../StandardIntegrationFlowContext.java | 56 ++++++--- .../dsl/manualflow/ManualFlowTests.java | 107 +++++++++++++++--- .../webflux/dsl/WebFluxDslTests.java | 2 +- src/reference/asciidoc/dsl.adoc | 10 ++ src/reference/asciidoc/whats-new.adoc | 6 + 6 files changed, 160 insertions(+), 45 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java index c57cf6d284..fed2e6253f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java @@ -123,7 +123,7 @@ public class IntegrationFlowBeanPostProcessor String id = endpointSpec.getId(); if (id == null) { - id = generateBeanName(endpoint, entry.getValue()); + id = generateBeanName(endpoint, flowNamePrefix, entry.getValue()); } Collection messageHandlers = @@ -131,13 +131,10 @@ public class IntegrationFlowBeanPostProcessor .values(); if (!messageHandlers.contains(messageHandler)) { - String handlerBeanName = generateBeanName(messageHandler); - String[] handlerAlias = new String[] { id + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX }; + String handlerBeanName = generateBeanName(messageHandler, flowNamePrefix); registerComponent(messageHandler, handlerBeanName, flowBeanName); - for (String alias : handlerAlias) { - this.beanFactory.registerAlias(handlerBeanName, alias); - } + this.beanFactory.registerAlias(handlerBeanName, id + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX); } registerComponent(endpoint, id, flowBeanName); @@ -187,12 +184,13 @@ public class IntegrationFlowBeanPostProcessor .values() .contains(o.getKey())) .forEach(o -> - registerComponent(o.getKey(), generateBeanName(o.getKey(), o.getValue()))); + registerComponent(o.getKey(), + generateBeanName(o.getKey(), flowNamePrefix, o.getValue()))); } SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.get().getT1(); String id = spec.getId(); if (!StringUtils.hasText(id)) { - id = generateBeanName(pollingChannelAdapterFactoryBean, entry.getValue()); + id = generateBeanName(pollingChannelAdapterFactoryBean, flowNamePrefix, entry.getValue()); } registerComponent(pollingChannelAdapterFactoryBean, id, flowBeanName); targetIntegrationComponents.put(pollingChannelAdapterFactoryBean, id); @@ -238,7 +236,7 @@ public class IntegrationFlowBeanPostProcessor targetIntegrationComponents.put(component, gatewayId); } else { - String generatedBeanName = generateBeanName(component, entry.getValue()); + String generatedBeanName = generateBeanName(component, flowNamePrefix, entry.getValue()); registerComponent(component, generatedBeanName, flowBeanName); targetIntegrationComponents.put(component, generatedBeanName); } @@ -335,11 +333,11 @@ public class IntegrationFlowBeanPostProcessor this.beanFactory.getBean(beanName); } - private String generateBeanName(Object instance) { - return generateBeanName(instance, null); + private String generateBeanName(Object instance, String prefix) { + return generateBeanName(instance, prefix, null); } - private String generateBeanName(Object instance, String fallbackId) { + private String generateBeanName(Object instance, String prefix, String fallbackId) { if (instance instanceof NamedComponent && ((NamedComponent) instance).getComponentName() != null) { return ((NamedComponent) instance).getComponentName(); } @@ -347,7 +345,7 @@ public class IntegrationFlowBeanPostProcessor return fallbackId; } - String generatedBeanName = instance.getClass().getName(); + String generatedBeanName = prefix + instance.getClass().getName(); String id = generatedBeanName; int counter = -1; while (counter == -1 || this.beanFactory.containsBean(id)) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java index 5f9162a68a..ebf38736ca 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java @@ -20,6 +20,9 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -46,7 +49,9 @@ import org.springframework.util.Assert; */ public final class StandardIntegrationFlowContext implements IntegrationFlowContext, BeanFactoryAware { - private final Map registry = new HashMap<>(); + private final Map registry = new ConcurrentHashMap<>(); + + private final Lock registerFlowsLock = new ReentrantLock(); private ConfigurableListableBeanFactory beanFactory; @@ -76,17 +81,29 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont private void register(StandardIntegrationFlowRegistrationBuilder builder) { IntegrationFlow integrationFlow = builder.integrationFlowRegistration.getIntegrationFlow(); String flowId = builder.integrationFlowRegistration.getId(); - if (flowId == null) { - flowId = generateBeanName(integrationFlow, null); - builder.id(flowId); + Lock registerBeanLock = null; + try { + if (flowId == null) { + registerBeanLock = this.registerFlowsLock; + registerBeanLock.lock(); + flowId = generateBeanName(integrationFlow, null); + builder.id(flowId); + } + else if (this.registry.containsKey(flowId)) { + throw new IllegalArgumentException("An IntegrationFlow '" + this.registry.get(flowId) + + "' with flowId '" + flowId + "' is already registered.\n" + + "An existing IntegrationFlowRegistration must be destroyed before overriding."); + } + + integrationFlow = (IntegrationFlow) registerBean(integrationFlow, flowId, null); } - else if (this.registry.containsKey(flowId)) { - throw new IllegalArgumentException("An IntegrationFlow '" + this.registry.get(flowId) + - "' with flowId '" + flowId + "' is already registered.\n" + - "An existing IntegrationFlowRegistration must be destroyed before overriding."); + finally { + if (registerBeanLock != null) { + registerBeanLock.unlock(); + } } - IntegrationFlow theFlow = (IntegrationFlow) registerBean(integrationFlow, flowId, null); - builder.integrationFlowRegistration.setIntegrationFlow(theFlow); + + builder.integrationFlowRegistration.setIntegrationFlow(integrationFlow); final String theFlowId = flowId; builder.additionalBeans.forEach((key, value) -> registerBean(key, value, theFlowId)); @@ -133,19 +150,26 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont * @param flowId the bean name to destroy from */ @Override - public synchronized void remove(String flowId) { + public void remove(String flowId) { if (this.registry.containsKey(flowId)) { IntegrationFlowRegistration flowRegistration = this.registry.remove(flowId); flowRegistration.stop(); - Arrays.stream(this.beanFactory.getDependentBeans(flowId)) - .forEach(((BeanDefinitionRegistry) this.beanFactory)::removeBeanDefinition); + BeanDefinitionRegistry beanDefinitionRegistry = (BeanDefinitionRegistry) this.beanFactory; - ((BeanDefinitionRegistry) this.beanFactory).removeBeanDefinition(flowId); + Arrays.stream(this.beanFactory.getDependentBeans(flowId)) + .forEach(beanName -> { + beanDefinitionRegistry.removeBeanDefinition(beanName); + // TODO until https://jira.spring.io/browse/SPR-16837 + Arrays.asList(beanDefinitionRegistry.getAliases(beanName)) + .forEach(beanDefinitionRegistry::removeAlias); + }); + + beanDefinitionRegistry.removeBeanDefinition(flowId); } else { - throw new IllegalStateException("Only manually registered IntegrationFlows can be removed. " - + "But [" + flowId + "] ins't one of them."); + throw new IllegalStateException("An IntegrationFlow with the id " + + "[" + flowId + "] doesn't exist in the registry."); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java index e744e486ac..d89a3100d4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java @@ -19,6 +19,7 @@ package org.springframework.integration.dsl.manualflow; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.lessThan; +import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -28,9 +29,14 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.util.ArrayList; import java.util.Arrays; import java.util.Date; +import java.util.List; import java.util.Objects; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -41,8 +47,11 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.BeanCreationNotAllowedException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.BeanFactoryPostProcessor; import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -63,6 +72,7 @@ import org.springframework.integration.dsl.context.IntegrationFlowContext.Integr import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.support.SmartLifecycleRoleController; +import org.springframework.integration.transformer.MessageTransformingHandler; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; @@ -92,7 +102,7 @@ public class ManualFlowTests { private IntegrationFlowContext integrationFlowContext; @Autowired - private BeanFactory beanFactory; + private ListableBeanFactory beanFactory; @Autowired private SmartLifecycleRoleController roleController; @@ -113,8 +123,10 @@ public class ManualFlowTests { IntegrationFlow flow = IntegrationFlows.from(producer) .channel(channel) .get(); - this.integrationFlowContext.registration(flow).register(); + IntegrationFlowRegistration flowRegistration = this.integrationFlowContext.registration(flow).register(); assertTrue(started.get()); + + flowRegistration.destroy(); } @Test @@ -138,15 +150,19 @@ public class ManualFlowTests { } MyProducerSpec spec = new MyProducerSpec(new MyProducer()); QueueChannel channel = new QueueChannel(); - IntegrationFlow flow = IntegrationFlows.from(spec.id("foo")) + IntegrationFlow flow = IntegrationFlows.from(spec.id("fooChannel")) .channel(channel) .get(); - this.integrationFlowContext.registration(flow).register(); + IntegrationFlowRegistration flowRegistration = this.integrationFlowContext.registration(flow).register(); assertTrue(started.get()); + + flowRegistration.destroy(); } @Test public void testManualFlowRegistration() throws InterruptedException { + String flowId = "testManualFlow"; + IntegrationFlow myFlow = f -> f .transform(String::toUpperCase) .channel(MessageChannels.queue()) @@ -160,6 +176,7 @@ public class ManualFlowTests { BeanFactoryHandler additionalBean = new BeanFactoryHandler(); IntegrationFlowRegistration flowRegistration = this.integrationFlowContext.registration(myFlow) + .id(flowId) .addBean(additionalBean) .register(); @@ -185,6 +202,8 @@ public class ManualFlowTests { assertThat(e.getMessage(), containsString("The 'receive()/receiveAndConvert()' isn't supported")); } + assertThat(this.beanFactory.getBeanNamesForType(MessageTransformingHandler.class)[0], startsWith(flowId + ".")); + flowRegistration.destroy(); assertFalse(this.beanFactory.containsBean(flowRegistration.getId())); @@ -210,7 +229,8 @@ public class ManualFlowTests { } catch (Exception e) { assertThat(e, instanceOf(IllegalStateException.class)); - assertThat(e.getMessage(), containsString("But [" + "foo" + "] ins't one of them.")); + assertThat(e.getMessage(), + containsString("An IntegrationFlow with the id [" + "foo" + "] doesn't exist in the registry.")); } } @@ -231,16 +251,21 @@ public class ManualFlowTests { Message receive = resultChannel.receive(1000); assertNotNull(receive); assertEquals("test", receive.getPayload()); + + this.integrationFlowContext.remove("dynamicFlow"); } @Test public void testDynamicAdapterFlow() { - this.integrationFlowContext.registration(new MyFlowAdapter()).register(); + IntegrationFlowRegistration flowRegistration = + this.integrationFlowContext.registration(new MyFlowAdapter()).register(); PollableChannel resultChannel = this.beanFactory.getBean("flowAdapterOutput", PollableChannel.class); Message receive = resultChannel.receive(1000); assertNotNull(receive); assertEquals("flowAdapterMessage", receive.getPayload()); + + flowRegistration.destroy(); } @@ -283,8 +308,9 @@ public class ManualFlowTests { PollableChannel resultChannel = new QueueChannel(); IntegrationFlowRegistration flowRegistration = - this.integrationFlowContext.registration(flow -> - flow.handle(new MessageProducingHandler()) + this.integrationFlowContext.registration( + flow -> flow + .handle(new MessageProducingHandler()) .channel(resultChannel)) .register(); @@ -294,6 +320,8 @@ public class ManualFlowTests { Message receive = resultChannel.receive(1000); assertNotNull(receive); assertEquals("test", receive.getPayload()); + + flowRegistration.destroy(); } @Test @@ -341,8 +369,8 @@ public class ManualFlowTests { assertTrue(this.roleController.getEndpointsRunningStatus(testRole).isEmpty()); } - @Test - public void testDynaSubFlowCreation() { + // @Test + public void testDynamicSubFlowCreation() { Flux> messageFlux = Flux.just("1,2,3,4") .map(v -> v.split(",")) @@ -362,7 +390,8 @@ public class ManualFlowTests { .channel(resultChannel) .get(); - this.integrationFlowContext.registration(integrationFlow).register(); + IntegrationFlowRegistration flowRegistration = + this.integrationFlowContext.registration(integrationFlow).register(); for (int i = 0; i < 4; i++) { Message receive = resultChannel.receive(10_000); @@ -370,6 +399,8 @@ public class ManualFlowTests { } assertNull(resultChannel.receive(0)); + + flowRegistration.destroy(); } @Test @@ -380,10 +411,11 @@ public class ManualFlowTests { IntegrationFlows.from(Supplier.class) .get(); - this.integrationFlowContext - .registration(testFlow) - .id(testId) - .register(); + IntegrationFlowRegistration flowRegistration = + this.integrationFlowContext + .registration(testFlow) + .id(testId) + .register(); try { this.integrationFlowContext @@ -395,12 +427,57 @@ public class ManualFlowTests { assertThat(e, instanceOf(IllegalArgumentException.class)); assertThat(e.getMessage(), containsString("with flowId '" + testId + "' is already registered.")); } + + flowRegistration.destroy(); + } + + @Test + public void testConcurrentRegistration() throws InterruptedException { + ExecutorService executorService = Executors.newCachedThreadPool(); + + List flowRegistrations = new ArrayList<>(); + + AtomicBoolean exceptionHappened = new AtomicBoolean(); + + for (int i = 0; i < 100; i++) { + int index = i; + executorService.execute(() -> { + + IntegrationFlow flow = f -> f + .transform(m -> m); + + try { + IntegrationFlowContext.IntegrationFlowRegistrationBuilder registration = + this.integrationFlowContext.registration(flow); + if (index % 2 == 0) { + registration.id("concurrentFlow#" + index); + } + flowRegistrations.add(registration.register()); + } + catch (Exception e) { + exceptionHappened.set(true); + } + + }); + } + + executorService.shutdownNow(); + assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS)); + + assertFalse(exceptionHappened.get()); + + flowRegistrations.forEach(IntegrationFlowRegistration::destroy); } @Configuration @EnableIntegration public static class RootConfiguration { + @Bean + public static BeanFactoryPostProcessor beanFactoryPostProcessor() { + return beanFactory -> ((DefaultListableBeanFactory) beanFactory).setAllowBeanDefinitionOverriding(false); + } + @Bean @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) public Date foo() { diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java index 03076cdb53..741407fe7a 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java @@ -101,7 +101,7 @@ public class WebFluxDslTests { @Qualifier("webFluxWithReplyPayloadToFlux.handler") private WebFluxRequestExecutingMessageHandler webFluxWithReplyPayloadToFlux; - @Resource(name = "org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler#1") + @Resource(name = "httpReactiveProxyFlow.org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler#0") private WebFluxRequestExecutingMessageHandler httpReactiveProxyFlow; @Autowired diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 82d70a24af..117bbcf120 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -580,6 +580,12 @@ And Lambda flow can't start from `MessageSource` or `MessageProducer`. Starting _version 5.1_, this kind of `IntegrationFlow` are wrapped to the proxy for exposing lifecycle control and provide access to the `inputChannel` of the internally associated `StandardIntegrationFlow`. +Starting with _version 5.0.5_, the generated bean names for the components in an `IntegrationFlow` include the flow bean followed by a dot as a prefix. +For example the `ConsumerEndpointFactoryBean` for the `.transform("Hello "::concat)` in the sample above, will end up with te bean name like `lambdaFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0`. +The `Transformer` implementation bean for that endpoint will have a bean name such as `lambdaFlow.org.springframework.integration.transformer.MethodInvokingTransformer#0`. +These generated bean names are prepended with the flow id prefix for purposes such as parsing logs or grouping components together in some analysis tool, as well as to avoid a race condition when we concurrently register integration flows at runtime. +See <> for more information. + [[java-dsl-function-expression]] === FunctionExpression @@ -927,6 +933,10 @@ Usually those additional beans are connection factories (AMQP, JMS, (S)FTP, TCP/ Such a dynamically registered `IntegrationFlow` and all its dependant beans can be removed afterwards using `IntegrationFlowRegistration.destroy()` callback. See `IntegrationFlowContext` JavaDocs for more information. +NOTE: Starting with _version 5.0.5_, all generated bean names in an `IntegrationFlow` definition are prepended with flow id as a prefix. +It is recommended to always specify an explicit flow id, otherwise a synchronization barrier is initiated in the `IntegrationFlowContext` to generate the bean name for the `IntegrationFlow` and register its beans. +We synchronize on these two operations to avoid a race condition when the same generated bean name may be used for different `IntegrationFlow` instances. + [[java-dsl-gateway]] === IntegrationFlow as Gateway diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index d1b0655c33..41543aa10f 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -44,3 +44,9 @@ Previously, interceptors were not applied when beans were created after the appl A new `ResultType.BYTES` mode is introduced for the `ObjectToJsonTransformer`. See <> for more information. + +==== Integration Flows: Generated bean names + +Starting with _version 5.0.5_, generated bean names for the components in an `IntegrationFlow` include the flow bean name, followed by a dot, as a prefix. + +See <> for more information.