INT-3792: Fix Some SONAR Vulnerabilities

JIRA: https://jira.spring.io/browse/INT-3792

* Remove `Serializable` requirement for `ThreadStatePropagationChannelInterceptor` generic type
* Increase `PriorityChannelTests` timeout

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java

@@ -226,7 +226,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 	@Override
 	public ChannelInterceptor removeInterceptor(int index) {
 		ChannelInterceptor interceptor = super.removeInterceptor(index);
-		if (interceptor != null && interceptor instanceof ExecutorChannelInterceptor) {
+		if (interceptor instanceof ExecutorChannelInterceptor) {
 			this.executorInterceptorsSize--;
 		}
 		return interceptor;

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java

@@ -111,7 +111,7 @@ public abstract class AbstractExecutorChannel extends AbstractSubscribableChanne
 	@Override
 	public ChannelInterceptor removeInterceptor(int index) {
 		ChannelInterceptor interceptor = super.removeInterceptor(index);
-		if (interceptor != null && interceptor instanceof ExecutorChannelInterceptor) {
+		if (interceptor instanceof ExecutorChannelInterceptor) {
 			this.executorInterceptorsSize--;
 		}
 		return interceptor;
@@ -170,12 +170,13 @@ public abstract class AbstractExecutorChannel extends AbstractSubscribableChanne
 		}
 
 		private Message<?> applyBeforeHandle(Message<?> message, Deque<ExecutorChannelInterceptor> interceptorStack) {
+			Message<?> theMessage = message;
 			for (ChannelInterceptor interceptor : AbstractExecutorChannel.this.interceptors.interceptors) {
 				if (interceptor instanceof ExecutorChannelInterceptor) {
 					ExecutorChannelInterceptor executorInterceptor = (ExecutorChannelInterceptor) interceptor;
-					message = executorInterceptor.beforeHandle(message, AbstractExecutorChannel.this,
+					theMessage = executorInterceptor.beforeHandle(theMessage, AbstractExecutorChannel.this,
 							this.delegate.getMessageHandler());
-					if (message == null) {
+					if (theMessage == null) {
 						if (isLoggingEnabled() && logger.isDebugEnabled()) {
 							logger.debug(executorInterceptor.getClass().getSimpleName()
 									+ " returned null from beforeHandle, i.e. precluding the send.");
@@ -186,7 +187,7 @@ public abstract class AbstractExecutorChannel extends AbstractSubscribableChanne
 					interceptorStack.add(executorInterceptor);
 				}
 			}
-			return message;
+			return theMessage;
 		}
 
 		private void triggerAfterMessageHandled(Message<?> message, Exception ex,

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java

@@ -38,7 +38,7 @@ import org.springframework.util.CollectionUtils;
 public abstract class AbstractPollableChannel extends AbstractMessageChannel
 		implements PollableChannel, PollableChannelManagement, ExecutorChannelInterceptorAware {
 
-	protected volatile int executorInterceptorsSize;
+	private volatile int executorInterceptorsSize;
 
 	@Override
 	public int getReceiveCount() {
@@ -168,7 +168,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
 	@Override
 	public ChannelInterceptor removeInterceptor(int index) {
 		ChannelInterceptor interceptor = super.removeInterceptor(index);
-		if (interceptor != null && interceptor instanceof ExecutorChannelInterceptor) {
+		if (interceptor instanceof ExecutorChannelInterceptor) {
 			this.executorInterceptorsSize--;
 		}
 		return interceptor;

spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java

@@ -134,5 +134,4 @@ public class ExecutorChannel extends AbstractExecutorChannel {
 		this.dispatcher = unicastingDispatcher;
 	}
 
-
 }

spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ThreadStatePropagationChannelInterceptor.java

@@ -16,8 +16,6 @@
 
 package org.springframework.integration.channel.interceptor;
 
-import java.io.Serializable;
-
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
@@ -50,7 +48,7 @@ import org.springframework.messaging.support.ExecutorChannelInterceptor;
  * @author Artem Bilan
  * @since 4.2
  */
-public abstract class ThreadStatePropagationChannelInterceptor<S extends Serializable>
+public abstract class ThreadStatePropagationChannelInterceptor<S>
 		extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor {
 
 	@Override
@@ -67,7 +65,7 @@ public abstract class ThreadStatePropagationChannelInterceptor<S extends Seriali
 	@Override
 	@SuppressWarnings("unchecked")
 	public final Message<?> postReceive(Message<?> message, MessageChannel channel) {
-		if (message != null && message instanceof MessageWithThreadState) {
+		if (message instanceof MessageWithThreadState) {
 			MessageWithThreadState<S> messageWithThreadState = (MessageWithThreadState<S>) message;
 			Message<?> messageToHandle = messageWithThreadState.message;
 			populatePropagatedContext(messageWithThreadState.state, messageToHandle, channel);
@@ -92,13 +90,11 @@ public abstract class ThreadStatePropagationChannelInterceptor<S extends Seriali
 	protected abstract void populatePropagatedContext(S state, Message<?> message, MessageChannel channel);
 
 
-	private static class MessageWithThreadState<S> implements Message<Object>, Serializable {
+	private static class MessageWithThreadState<S> implements Message<Object> {
 
-		private static final long serialVersionUID = 1548216539234073073L;
+		private final Message<?> message;
 
-		final Message<?> message;
-
-		final S state;
+		private final S state;
 
 		public MessageWithThreadState(Message<?> message, S state) {
 			this.message = message;

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java

@@ -193,7 +193,7 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa
 	private Runnable createMessageHandlingTask(final MessageHandler handler, final Message<?> message) {
 		MessageHandlingRunnable task = new MessageHandlingRunnable() {
 
-			final MessageHandler delegate = new MessageHandler() {
+			private final MessageHandler delegate = new MessageHandler() {
 
 				@Override
 				public void handleMessage(Message<?> message) throws MessagingException {

spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java

@@ -89,36 +89,37 @@ public class PollingConsumer extends AbstractPollingEndpoint {
 
 	@Override
 	protected void handleMessage(Message<?> message) {
+		Message<?> theMessage = message;
 		Deque<ExecutorChannelInterceptor> interceptorStack = null;
 		try {
 			if (this.channelInterceptors != null
 					&& ((ExecutorChannelInterceptorAware) this.inputChannel).hasExecutorInterceptors()) {
 				interceptorStack = new ArrayDeque<ExecutorChannelInterceptor>();
-				message = applyBeforeHandle(message, interceptorStack);
-				if (message == null) {
+				theMessage = applyBeforeHandle(theMessage, interceptorStack);
+				if (theMessage == null) {
 					return;
 				}
 			}
-			this.handler.handleMessage(message);
+			this.handler.handleMessage(theMessage);
 			if (!CollectionUtils.isEmpty(interceptorStack)) {
-				triggerAfterMessageHandled(message, null, interceptorStack);
+				triggerAfterMessageHandled(theMessage, null, interceptorStack);
 			}
 		}
 		catch (Exception ex) {
 			if (!CollectionUtils.isEmpty(interceptorStack)) {
-				triggerAfterMessageHandled(message, ex, interceptorStack);
+				triggerAfterMessageHandled(theMessage, ex, interceptorStack);
 			}
 			if (ex instanceof MessagingException) {
 				throw (MessagingException) ex;
 			}
-			String description = "Failed to handle " + message + " to " + this + " in " + this.handler;
-			throw new MessageDeliveryException(message, description, ex);
+			String description = "Failed to handle " + theMessage + " to " + this + " in " + this.handler;
+			throw new MessageDeliveryException(theMessage, description, ex);
 		}
 		catch (Error ex) {//NOSONAR - ok, we re-throw below
 			if (!CollectionUtils.isEmpty(interceptorStack)) {
-				String description = "Failed to handle " + message + " to " + this + " in " + this.handler;
-				triggerAfterMessageHandled(message,
-						new MessageDeliveryException(message, description, ex),
+				String description = "Failed to handle " + theMessage + " to " + this + " in " + this.handler;
+				triggerAfterMessageHandled(theMessage,
+						new MessageDeliveryException(theMessage, description, ex),
 						interceptorStack);
 			}
 			throw ex;
@@ -126,10 +127,11 @@ public class PollingConsumer extends AbstractPollingEndpoint {
 	}
 
 	private Message<?> applyBeforeHandle(Message<?> message, Deque<ExecutorChannelInterceptor> interceptorStack) {
+		Message<?> theMessage = message;
 		for (ChannelInterceptor interceptor : this.channelInterceptors) {
 			if (interceptor instanceof ExecutorChannelInterceptor) {
 				ExecutorChannelInterceptor executorInterceptor = (ExecutorChannelInterceptor) interceptor;
-				message = executorInterceptor.beforeHandle(message, this.inputChannel, this.handler);
+				theMessage = executorInterceptor.beforeHandle(theMessage, this.inputChannel, this.handler);
 				if (message == null) {
 					if (logger.isDebugEnabled()) {
 						logger.debug(executorInterceptor.getClass().getSimpleName()
@@ -141,7 +143,7 @@ public class PollingConsumer extends AbstractPollingEndpoint {
 				interceptorStack.add(executorInterceptor);
 			}
 		}
-		return message;
+		return theMessage;
 	}
 
 	private void triggerAfterMessageHandled(Message<?> message, Exception ex,

spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -16,6 +16,12 @@
 
 package org.springframework.integration.channel;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
 import java.util.Comparator;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
@@ -25,15 +31,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.Test;
 
+import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
-import org.springframework.integration.support.MessageBuilder;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
 
 /**
  * @author Mark Fisher
@@ -50,7 +50,7 @@ public class PriorityChannelTests {
 		channel.receive(0);
 		assertTrue(channel.send(new GenericMessage<String>("test5")));
 	}
-	
+
 	@Test
 	public void testDefaultComparatorWithTimestampFallback() throws Exception{
 		PriorityChannel channel = new PriorityChannel();
@@ -91,11 +91,13 @@ public class PriorityChannelTests {
 		for (int i = 0; i < 1000; i++) {
 			channel.send(message);
 			new Thread(new Runnable() {
+				@Override
 				public void run() {
 					channel.receive();
 				}
 			}).start();
 			new Thread(new Runnable() {
+				@Override
 				public void run() {
 					message.getHeaders().toString();
 				}
@@ -120,9 +122,9 @@ public class PriorityChannelTests {
 		assertEquals("B", channel.receive(0).getPayload());
 		assertEquals("C", channel.receive(0).getPayload());
 		assertEquals("D", channel.receive(0).getPayload());
-		assertEquals("E", channel.receive(0).getPayload());		
+		assertEquals("E", channel.receive(0).getPayload());
 	}
-	
+
 	@Test
 	public void testWithCustomComparatorAndSequence() {
 		PriorityChannel channel = new PriorityChannel(10, new FooHeaderComparator());
@@ -199,7 +201,7 @@ public class PriorityChannelTests {
 		assertEquals(2, receivedFour);
 		assertEquals(3, receivedFive);
 		assertEquals(6, receivedSix);
-		assertEquals(7, receivedSeven);		
+		assertEquals(7, receivedSeven);
 	}
 
 	@Test
@@ -238,17 +240,18 @@ public class PriorityChannelTests {
 		Executor executor = Executors.newSingleThreadScheduledExecutor();
 		channel.send(new GenericMessage<String>("test-1"));
 		executor.execute(new Runnable() {
+			@Override
 			public void run() {
 				sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), 10));
 				latch.countDown();
 			}
 		});
 		assertFalse(sentSecondMessage.get());
-		Thread.sleep(500);
+		Thread.sleep(1000);
 		Message<?> message1 = channel.receive();
 		assertNotNull(message1);
 		assertEquals("test-1", message1.getPayload());
-		latch.await(1000, TimeUnit.MILLISECONDS);
+		latch.await(10000, TimeUnit.MILLISECONDS);
 		assertFalse(sentSecondMessage.get());
 		assertNull(channel.receive(0));
 	}
@@ -261,6 +264,7 @@ public class PriorityChannelTests {
 		Executor executor = Executors.newSingleThreadScheduledExecutor();
 		channel.send(new GenericMessage<String>("test-1"));
 		executor.execute(new Runnable() {
+			@Override
 			public void run() {
 				sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), 3000));
 				latch.countDown();
@@ -286,6 +290,7 @@ public class PriorityChannelTests {
 		Executor executor = Executors.newSingleThreadScheduledExecutor();
 		channel.send(new GenericMessage<String>("test-1"));
 		executor.execute(new Runnable() {
+			@Override
 			public void run() {
 				sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), -1));
 				latch.countDown();
@@ -305,27 +310,29 @@ public class PriorityChannelTests {
 
 
 	private static Message<String> createPriorityMessage(int priority) {
-		return MessageBuilder.withPayload("test:" + priority).setPriority(priority).build(); 
+		return MessageBuilder.withPayload("test:" + priority).setPriority(priority).build();
 	}
 
 
 	public static class StringPayloadComparator implements Comparator<Message<?>> {
 
+		@Override
 		public int compare(Message<?> message1, Message<?> message2) {
 			String s1 = (String) message1.getPayload();
 			String s2 = (String) message2.getPayload();
 			return s1.compareTo(s2);
-		}	
+		}
 	}
-	
+
 	public static class FooHeaderComparator implements Comparator<Message<?>> {
+		@Override
 		public int compare(Message<?> message1, Message<?> message2) {
 			Integer foo1 = (Integer) message1.getHeaders().get("foo");
 			Integer foo2 = (Integer) message2.getHeaders().get("foo");
 			foo1 = foo1 != null ? foo1 : 0;
 			foo2 = foo2 != null ? foo2 : 0;
 			return foo2.compareTo(foo1);
-		}	
+		}
 	}
 
 }

spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScanner.java

@@ -38,6 +38,9 @@ public class RecursiveLeafOnlyDirectoryScanner extends DefaultDirectoryScanner {
 	@Override
 	protected File[] listEligibleFiles(File directory) throws IllegalArgumentException {
 		File[] rootFiles = directory.listFiles();
+		if (rootFiles == null) {
+			return new File[0];
+		}
 		List<File> files = new ArrayList<File>(rootFiles.length);
 		for (File rootFile : rootFiles) {
 			if (rootFile.isDirectory()) {

spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java

@@ -22,6 +22,7 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.servlet.http.Cookie;
@@ -464,8 +465,9 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa
 				payload = this.payloadExpression.getValue(evaluationContext);
 			}
 			if (!CollectionUtils.isEmpty(this.headerExpressions)) {
-				for (String headerName : this.headerExpressions.keySet()) {
-					Expression headerExpression = this.headerExpressions.get(headerName);
+				for (Entry<String, Expression> entry : this.headerExpressions.entrySet()) {
+					String headerName = entry.getKey();
+					Expression headerExpression = entry.getValue();
 					Object headerValue = headerExpression.getValue(evaluationContext);
 					if (headerValue != null) {
 						headers.put(headerName, headerValue);

spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java

@@ -181,7 +181,7 @@ public class PollableJmsChannel extends AbstractJmsChannel
 	@Override
 	public ChannelInterceptor removeInterceptor(int index) {
 		ChannelInterceptor interceptor = super.removeInterceptor(index);
-		if (interceptor != null && interceptor instanceof ExecutorChannelInterceptor) {
+		if (interceptor instanceof ExecutorChannelInterceptor) {
 			this.executorInterceptorsSize--;
 		}
 		return interceptor;

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java

@@ -844,8 +844,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements Applicati
 		}
 		StringBuilder builder = new StringBuilder();
 
-		for (Object key : objectNameStaticProperties.keySet()) {
-			builder.append("," + key + "=" + objectNameStaticProperties.get(key));
+		for (Entry<Object, Object> entry : this.objectNameStaticProperties.entrySet()) {
+			builder.append("," + entry.getKey() + "=" + entry.getValue());
 		}
 		return builder.toString();
 	}

spring-integration-security/src/main/java/org/springframework/integration/security/channel/SecurityContextPropagationChannelInterceptor.java

@@ -36,7 +36,6 @@ import org.springframework.security.core.context.SecurityContextHolder;
  * in the containers Threads for channels like
  * {@link org.springframework.integration.channel.ExecutorChannel}
  * and {@link org.springframework.integration.channel.QueueChannel}.
- *
  * @author Artem Bilan
  * @see ThreadStatePropagationChannelInterceptor
  * @since 4.2
@@ -46,7 +45,7 @@ public class SecurityContextPropagationChannelInterceptor
 
 	private final static SecurityContext EMPTY_CONTEXT = SecurityContextHolder.createEmptyContext();
 
-	private static final ThreadLocal<SecurityContext> ORIGINAL_CONTEXT = new ThreadLocal<SecurityContext>();
+	private final static ThreadLocal<SecurityContext> ORIGINAL_CONTEXT = new ThreadLocal<SecurityContext>();
 
 	@Override
 	public void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler, Exception ex) {
@@ -63,7 +62,7 @@ public class SecurityContextPropagationChannelInterceptor
 
 	@Override
 	protected void populatePropagatedContext(Authentication authentication, Message<?> message,
-											 MessageChannel channel) {
+	                                         MessageChannel channel) {
 		if (authentication != null) {
 			SecurityContext currentContext = SecurityContextHolder.getContext();
 
@@ -75,9 +74,8 @@ public class SecurityContextPropagationChannelInterceptor
 		}
 	}
 
-	public static void cleanup() {
+	private void cleanup() {
 		SecurityContext originalContext = ORIGINAL_CONTEXT.get();
-
 		try {
 			if (originalContext == null || EMPTY_CONTEXT.equals(originalContext)) {
 				SecurityContextHolder.clearContext();
@@ -87,7 +85,7 @@ public class SecurityContextPropagationChannelInterceptor
 				SecurityContextHolder.setContext(originalContext);
 			}
 		}
-		catch (Throwable t) {
+		catch (Throwable t) {//NOSONAR
 			SecurityContextHolder.clearContext();
 		}
 	}

spring-integration-test/src/main/java/org/springframework/integration/test/util/OnlyOnceTrigger.java

@@ -23,16 +23,17 @@ import org.springframework.scheduling.TriggerContext;
 /**
  *
  * @author Gunnar Hillert
+ * @author Gary Russell
  * @since 2.2
  *
  */
 public class OnlyOnceTrigger implements Trigger {
 
-	private static final AtomicBoolean hasRun = new AtomicBoolean();
+	private final AtomicBoolean hasRun = new AtomicBoolean();
 
 	private final Date executionTime;
 
-	private static volatile CountDownLatch latch = new CountDownLatch(1);
+	private volatile CountDownLatch latch = new CountDownLatch(1);
 
 
 	public OnlyOnceTrigger() {
@@ -40,10 +41,11 @@ public class OnlyOnceTrigger implements Trigger {
 		executionTime = new Date();
 	}
 
+	@Override
 	public Date nextExecutionTime(TriggerContext triggerContext) {
 
-		if (OnlyOnceTrigger.hasRun.getAndSet(true)) {
-			OnlyOnceTrigger.latch.countDown();
+		if (this.hasRun.getAndSet(true)) {
+			this.latch.countDown();
 			return null;
 		}
 
@@ -61,29 +63,34 @@ public class OnlyOnceTrigger implements Trigger {
 
 	@Override
 	public boolean equals(Object obj) {
-		if (this == obj)
+		if (this == obj) {
 			return true;
-		if (obj == null)
+		}
+		if (obj == null) {
 			return false;
-		if (getClass() != obj.getClass())
+		}
+		if (getClass() != obj.getClass()) {
 			return false;
+		}
 		OnlyOnceTrigger other = (OnlyOnceTrigger) obj;
 		if (executionTime == null) {
-			if (other.executionTime != null)
+			if (other.executionTime != null) {
 				return false;
-		} else if (!executionTime.equals(other.executionTime))
+			}
+		} else if (!executionTime.equals(other.executionTime)) {
 			return false;
+		}
 		return true;
 	}
 
 	public void reset() {
-		OnlyOnceTrigger.latch = new CountDownLatch(1);
-		OnlyOnceTrigger.hasRun.set(false);
+		this.latch = new CountDownLatch(1);
+		this.hasRun.set(false);
 	}
 
 	public void await() {
 		try {
-			OnlyOnceTrigger.latch.await(5000, TimeUnit.MILLISECONDS);
+			this.latch.await(5000, TimeUnit.MILLISECONDS);
 			if (latch.getCount() != 0) {
 				throw new RuntimeException("test latch.await() did not count down");
 			}