INT-2931 Fix Router Race Condition

AbstractMessageRouter.getRequiredConversionService() and
IntegrationObjectSupport.getConversionService() try to initialize the
conversion service and may be called from multiple threads. There is
a race condition where this could end up in an uninitialized conversion service
causing NPEs in subsequent method calls.

To solve this issue, the conversion service is initialized with
double-checked locking.

For further details see https://jira.springsource.org/browse/INT-2931

INT-2931 Polishing - Add Test Case

Add a test case that reliably reproduces the issue and verifies
the fix.

spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java

@@ -18,7 +18,6 @@ package org.springframework.integration.context;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.BeanInitializationException;
@@ -37,10 +36,11 @@ import org.springframework.util.StringUtils;
  * components whereas code built upon the integration framework should not
  * require tight coupling with the context but rather rely on standard
  * dependency injection.
- * 
+ *
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Josh Long
+ * @author Stefan Ferstl
  */
 public abstract class IntegrationObjectSupport implements BeanNameAware, NamedComponent, BeanFactoryAware, InitializingBean {
 
@@ -65,16 +65,16 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 	}
 
 	/**
-	 * Will return the name of this component identified by {@link #componentName} field. 
+	 * Will return the name of this component identified by {@link #componentName} field.
 	 * If {@link #componentName} was not set this method will default to the 'beanName' of this component;
 	 */
-	public final String getComponentName() {	
+	public final String getComponentName() {
 		return StringUtils.hasText(this.componentName) ? this.componentName : this.beanName;
 	}
 
 	/**
-	 * Sets the name of this component. 
-	 * 
+	 * Sets the name of this component.
+	 *
 	 * @param componentName
 	 */
 	public void setComponentName(String componentName) {
@@ -129,9 +129,13 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 
 	protected final ConversionService getConversionService() {
 		if (this.conversionService == null && this.beanFactory != null) {
-			this.conversionService = IntegrationContextUtils.getConversionService(this.beanFactory);
-			if (this.conversionService == null && logger.isDebugEnabled()) {
-				logger.debug("Unable to attempt conversion of Message payload types. Component '" +
+			synchronized (this) {
+				if (this.conversionService == null) {
+					this.conversionService = IntegrationContextUtils.getConversionService(this.beanFactory);
+				}
+			}
+			if (this.conversionService == null && this.logger.isDebugEnabled()) {
+				this.logger.debug("Unable to attempt conversion of Message payload types. Component '" +
 						this.getComponentName() + "' has no explicit ConversionService reference, " +
 						"and there is no 'integrationConversionService' bean within the context.");
 			}

spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java

@@ -37,6 +37,7 @@ import org.springframework.jmx.export.annotation.ManagedResource;
  * @author Oleg Zhurakousky
  * @author Gunnar Hillert
  * @author Soby Chacko
+ * @author Stefan Ferstl
  */
 @ManagedResource
 public abstract class AbstractMessageRouter extends AbstractMessageHandler {
@@ -103,7 +104,11 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler {
 
 	protected ConversionService getRequiredConversionService() {
 		if (this.getConversionService() == null) {
-			this.setConversionService(new DefaultConversionService());
+			synchronized (this) {
+				if (this.getConversionService() == null) {
+					this.setConversionService(new DefaultConversionService());
+				}
+			}
 		}
 		return this.getConversionService();
 	}

spring-integration-core/src/test/java/org/springframework/integration/router/RouterConcurrencyTest.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.router;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.context.IntegrationContextUtils;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class RouterConcurrencyTest {
+
+	@Test
+	public void test() throws Exception {
+		final AtomicInteger count = new AtomicInteger();
+		final Semaphore semaphore = new Semaphore(1);
+		final AbstractMessageRouter router = new AbstractMessageRouter() {
+			@Override
+			protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
+				return null;
+			}
+
+			@Override
+			protected void setConversionService(ConversionService conversionService) {
+				try {
+					if (count.incrementAndGet() > 1) {
+						Thread.sleep(2000);
+					}
+					super.setConversionService(conversionService);
+					semaphore.release();
+					Thread.sleep(1000);
+				}
+				catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+				}
+			}
+
+		};
+
+		final AtomicInteger beanCounter = new AtomicInteger();
+		BeanFactory beanFactory = mock(BeanFactory.class);
+		doAnswer(new Answer<Boolean>() {
+
+			public Boolean answer(InvocationOnMock invocation) throws Throwable {
+				if (beanCounter.getAndIncrement() < 2) {
+					semaphore.tryAcquire(4, TimeUnit.SECONDS);
+				}
+				return false;
+			}
+		}).when(beanFactory).containsBean(IntegrationContextUtils.INTEGRATION_CONVERSION_SERVICE_BEAN_NAME);
+		router.setBeanFactory(beanFactory);
+
+		ExecutorService exec = Executors.newFixedThreadPool(2);
+		final List<ConversionService> returns = new ArrayList<ConversionService>();
+		Runnable runnable = new Runnable() {
+
+			public void run() {
+				ConversionService requiredConversionService = router.getRequiredConversionService();
+				System.out.println("Adding " + requiredConversionService);
+				returns.add(requiredConversionService);
+			}
+		};
+		exec.execute(runnable);
+		exec.execute(runnable);
+		exec.shutdown();
+		exec.awaitTermination(10, TimeUnit.SECONDS);
+		assertTrue(returns.size() == 2);
+		assertNotNull(returns.get(0));
+		assertNotNull(returns.get(1));
+	}
+
+}