Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix IntegrationFlowContext concurrency issue

Browse Source 
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.
This commit is contained in:
Artem Bilan
2018-05-17 17:28:31 -04:00
committed by Gary Russell
parent a750a7847c
commit b21dc0dff3
6 changed files with 160 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java
View File

@@ -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)) {

56
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java
View File

@@ -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<String, IntegrationFlowRegistration> registry = new HashMap<>();
private final Map<String, IntegrationFlowRegistration> 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.");
}
}

107
spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java
View File

@@ -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
.<String, String>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<Message<?>> 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<IntegrationFlowRegistration> 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() {