Fix new Sonar smells

build.gradle

@@ -610,6 +610,7 @@ project('spring-integration-jpa') {
 		}
 		testImplementation "com.h2database:h2:$h2Version"
 		testImplementation "org.hibernate:hibernate-entitymanager:$hibernateVersion"
+		testImplementation "org.hamcrest:hamcrest-core:$hamcrestVersion"
 	}
 }
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2020 the original author or authors.
+ * Copyright 2014-2021 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.
@@ -89,7 +89,7 @@ public final class FixedSubscriberChannel implements SubscribableChannel, BeanNa
 
 	@Override
 	public boolean subscribe(MessageHandler handler) {
-		if (handler != this.handler && LOGGER.isDebugEnabled()) {
+		if (!this.handler.equals(handler) && LOGGER.isDebugEnabled()) {
 			LOGGER.debug(getComponentName() + ": cannot be subscribed to (it has a fixed single subscriber).");
 		}
 		return false;

spring-integration-core/src/main/java/org/springframework/integration/config/PublisherRegistrar.java

@@ -103,7 +103,9 @@ public class PublisherRegistrar implements ImportBeanDefinitionRegistrar {
 			if (beanFactory != null) {
 				order = beanFactory.resolveEmbeddedValue(order);
 			}
-			postProcessor.setOrder(Integer.parseInt(order));
+			if (StringUtils.hasText(order)) {
+				postProcessor.setOrder(Integer.parseInt(order));
+			}
 		}
 		return postProcessor;
 	}

spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -46,9 +46,9 @@ public class PointToPointChannelParser extends AbstractChannelParser {
 	@Override
 	protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder builder = null;
-		Element queueElement = null;
+		Element queueElement;
 		String fixedSubscriberChannel = element.getAttribute("fixed-subscriber");
-		boolean isFixedSubscriber = "true".equals(fixedSubscriberChannel.trim().toLowerCase());
+		boolean isFixedSubscriber = "true".equalsIgnoreCase(fixedSubscriberChannel.trim());
 
 		// configure a queue-based channel if any queue sub-element is defined
 		String channel = element.getAttribute(ID_ATTRIBUTE);

spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -277,7 +277,7 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
 					if (shouldMapHeader(headerName, headerMatcher)) {
 						Object value = entry.getValue();
 						target.put(headerName, value);
-						if (this.replyHeaderMatcher == headerMatcher &&
+						if (this.replyHeaderMatcher.equals(headerMatcher) &&
 								JsonHeaders.TYPE_ID.equals(headerName) && value != null) {
 
 							ResolvableType resolvableType =

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -107,7 +107,7 @@ public class PayloadTypeRouter extends AbstractMappingMessageRouter {
 			}
 			// no match at this level, continue up the hierarchy
 			for (Class<?> superInterface : iface.getInterfaces()) {
-				int weight = this.determineTypeDifferenceWeight(candidate, superInterface, level + 3);
+				int weight = this.determineTypeDifferenceWeight(candidate, superInterface, level + 3); // NOSONAR
 				if (weight < Integer.MAX_VALUE) {
 					return weight;
 				}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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,7 @@
 
 package org.springframework.integration.router;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Iterator;
@@ -98,13 +99,8 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi
 	public void setRecipients(List<Recipient> recipients) {
 		Assert.notEmpty(recipients, "'recipients' must not be empty");
 		Queue<Recipient> newRecipients = new ConcurrentLinkedQueue<>(recipients);
-
 		newRecipients.forEach(this::setupRecipient);
-
-		if (logger.isDebugEnabled()) {
-			logger.debug("Channel Recipients: " + this.recipients + " replaced with: " + newRecipients);
-		}
-
+		logger.debug(() -> "Channel Recipients: " + this.recipients + " replaced with: " + newRecipients);
 		this.recipients = newRecipients;
 	}
 
@@ -126,10 +122,7 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi
 				addRecipient(next.getKey(), (MessageSelector) null, newRecipients);
 			}
 		}
-		if (logger.isDebugEnabled()) {
-			logger.debug("Channel Recipients: " + this.recipients + " replaced with: " + newRecipients);
-		}
-
+		logger.debug(() -> "Channel Recipients: " + this.recipients + " replaced with: " + newRecipients);
 		this.recipients = newRecipients;
 	}
 
@@ -193,7 +186,7 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi
 		int counter = 0;
 		MessageChannel channel = getChannelResolver().resolveDestination(channelName);
 		for (Iterator<Recipient> it = this.recipients.iterator(); it.hasNext(); ) {
-			if (it.next().getChannel() == channel) {
+			if (channel.equals(it.next().getChannel())) {
 				it.remove();
 				counter++;
 			}
@@ -236,9 +229,7 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi
 				addRecipient(key);
 			}
 		}
-		if (logger.isDebugEnabled()) {
-			logger.debug("Channel Recipients:" + originalRecipients + " replaced with:" + this.recipients);
-		}
+		logger.debug(() -> "Channel Recipients: " + originalRecipients + " replaced with: " + this.recipients);
 	}
 
 	@Override
@@ -259,10 +250,13 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi
 
 	@Override
 	protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
-		return this.recipients.stream()
-				.filter(recipient -> recipient.accept(message))
-				.map(Recipient::getChannel)
-				.collect(Collectors.toList());
+		List<MessageChannel> result = new ArrayList<>();
+		for (Recipient recipient : this.recipients) {
+			if (recipient.accept(message)) {
+				result.add(recipient.getChannel());
+			}
+		}
+		return result;
 	}
 
 

spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -31,11 +31,23 @@ import org.springframework.util.ErrorHandler;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class PollerMetadata {
 
+	/**
+	 * The constant for unlimited number of message to poll in one cycle.
+	 */
 	public static final int MAX_MESSAGES_UNBOUNDED = Integer.MIN_VALUE;
 
+	/**
+	 * The default receive timeout as one second.
+	 */
+	public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
+
+	/**
+	 * The bean name for global default poller.
+	 */
 	public static final String DEFAULT_POLLER_METADATA_BEAN_NAME =
 			"org.springframework.integration.context.defaultPollerMetadata";
 
@@ -44,21 +56,21 @@ public class PollerMetadata {
 	 */
 	public static final String DEFAULT_POLLER = DEFAULT_POLLER_METADATA_BEAN_NAME;
 
-	private volatile Trigger trigger;
+	private Trigger trigger;
 
-	private volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
+	private long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
 
-	private volatile long receiveTimeout = 1000;
+	private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
 
-	private volatile ErrorHandler errorHandler;
+	private ErrorHandler errorHandler;
 
-	private volatile List<Advice> adviceChain;
+	private List<Advice> adviceChain;
 
-	private volatile Executor taskExecutor;
+	private Executor taskExecutor;
 
-	private volatile long sendTimeout;
+	private long sendTimeout;
 
-	private volatile TransactionSynchronizationFactory transactionSynchronizationFactory;
+	private TransactionSynchronizationFactory transactionSynchronizationFactory;
 
 
 	public void setTransactionSynchronizationFactory(
@@ -91,11 +103,8 @@ public class PollerMetadata {
 	 * Set the maximum number of messages to receive for each poll.
 	 * A non-positive value indicates that polling should repeat as long
 	 * as non-null messages are being received and successfully sent.
-	 *
 	 * <p>The default is unbounded.
-	 *
 	 * @param maxMessagesPerPoll The maxMessagesPerPoll to set.
-	 *
 	 * @see #MAX_MESSAGES_UNBOUNDED
 	 */
 	public void setMaxMessagesPerPoll(long maxMessagesPerPoll) {

spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java

@@ -198,44 +198,46 @@ public class Jackson2JsonObjectMapper extends AbstractJacksonJsonObjectMapper<Js
 
 	private void registerWellKnownModulesIfAvailable() {
 		if (JDK8_MODULE_PRESENT) {
-			this.objectMapper.registerModule(Jdk8ModuleProvider.module);
+			this.objectMapper.registerModule(Jdk8ModuleProvider.MODULE);
 		}
 
 		if (JAVA_TIME_MODULE_PRESENT) {
-			this.objectMapper.registerModule(JavaTimeModuleProvider.module);
+			this.objectMapper.registerModule(JavaTimeModuleProvider.MODULE);
 		}
 
 		if (JODA_MODULE_PRESENT) {
-			this.objectMapper.registerModule(JodaModuleProvider.module);
+			this.objectMapper.registerModule(JodaModuleProvider.MODULE);
 		}
 
 		if (KOTLIN_MODULE_PRESENT) {
-			this.objectMapper.registerModule(KotlinModuleProvider.module);
+			this.objectMapper.registerModule(KotlinModuleProvider.MODULE);
 		}
 	}
 
 	private static final class Jdk8ModuleProvider {
 
-		static final com.fasterxml.jackson.databind.Module module = new com.fasterxml.jackson.datatype.jdk8.Jdk8Module();
+		static final com.fasterxml.jackson.databind.Module MODULE =
+				new com.fasterxml.jackson.datatype.jdk8.Jdk8Module();
 
 	}
 
 	private static final class JavaTimeModuleProvider {
 
-		static final com.fasterxml.jackson.databind.Module module =
+		static final com.fasterxml.jackson.databind.Module MODULE =
 				new com.fasterxml.jackson.datatype.jsr310.JavaTimeModule();
 
 	}
 
 	private static final class JodaModuleProvider {
 
-		static final com.fasterxml.jackson.databind.Module module = new com.fasterxml.jackson.datatype.joda.JodaModule();
+		static final com.fasterxml.jackson.databind.Module MODULE =
+				new com.fasterxml.jackson.datatype.joda.JodaModule();
 
 	}
 
 	private static final class KotlinModuleProvider {
 
-		static final com.fasterxml.jackson.databind.Module module =
+		static final com.fasterxml.jackson.databind.Module MODULE =
 				new com.fasterxml.jackson.module.kotlin.KotlinModule();
 
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -91,8 +91,7 @@ import org.springframework.util.Assert;
  * @author Artem Bilan
  * @author Steven Pearce
  */
-public class FileReadingMessageSource extends AbstractMessageSource<File>
-		implements ManageableLifecycle {
+public class FileReadingMessageSource extends AbstractMessageSource<File> implements ManageableLifecycle {
 
 	private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5;
 
@@ -124,14 +123,14 @@ public class FileReadingMessageSource extends AbstractMessageSource<File>
 	private WatchEventType[] watchEvents = { WatchEventType.CREATE };
 
 	/**
-	 * Creates a FileReadingMessageSource with a naturally ordered queue of unbounded capacity.
+	 * Create a FileReadingMessageSource with a naturally ordered queue of unbounded capacity.
 	 */
 	public FileReadingMessageSource() {
 		this(null);
 	}
 
 	/**
-	 * Creates a FileReadingMessageSource with a bounded queue of the given
+	 * Create a FileReadingMessageSource with a bounded queue of the given
 	 * capacity. This can be used to reduce the memory footprint of this
 	 * component when reading from a large directory.
 	 * @param internalQueueCapacity
@@ -150,15 +149,13 @@ public class FileReadingMessageSource extends AbstractMessageSource<File>
 	}
 
 	/**
-	 * Creates a FileReadingMessageSource with a {@link PriorityBlockingQueue}
+	 * Create a FileReadingMessageSource with a {@link PriorityBlockingQueue}
 	 * ordered with the passed in {@link Comparator}.
 	 * <p> The size of the queue used should be large enough to hold all the files
 	 * in the input directory in order to sort all of them, so restricting the
 	 * size of the queue is mutually exclusive with ordering. No guarantees
 	 * about file delivery order can be made under concurrent access.
-	 * @param receptionOrderComparator
-	 *            the comparator to be used to order the files in the internal
-	 *            queue
+	 * @param receptionOrderComparator the comparator to be used to order the files in the internal queue
 	 */
 	public FileReadingMessageSource(@Nullable Comparator<File> receptionOrderComparator) {
 		this.toBeReceived = new PriorityBlockingQueue<>(DEFAULT_INTERNAL_QUEUE_CAPACITY, receptionOrderComparator);
@@ -211,7 +208,7 @@ public class FileReadingMessageSource extends AbstractMessageSource<File>
 	}
 
 	/**
-	 * Sets a {@link FileListFilter}.
+	 * Set a {@link FileListFilter}.
 	 * By default a {@link org.springframework.integration.file.filters.AcceptOnceFileListFilter}
 	 * with no bounds is used. In most cases a customized {@link FileListFilter} will
 	 * be needed to deal with modification and duplication concerns.
@@ -227,10 +224,8 @@ public class FileReadingMessageSource extends AbstractMessageSource<File>
 	}
 
 	/**
-	 * Optional. Sets a {@link FileLocker} to be used to guard files against
-	 * duplicate processing.
-	 * <p>
-	 * <b>The supplied FileLocker must be thread safe</b>
+	 * Set a {@link FileLocker} to be used to guard files against duplicate processing.
+	 * <p> <b>The supplied FileLocker must be thread safe</b>
 	 * @param locker a locker
 	 */
 	public void setLocker(FileLocker locker) {
@@ -239,7 +234,7 @@ public class FileReadingMessageSource extends AbstractMessageSource<File>
 	}
 
 	/**
-	 * Optional. Set this flag if you want to make sure the internal queue is
+	 * Set this flag if you want to make sure the internal queue is
 	 * refreshed with the latest content of the input directory on each poll.
 	 * <p>
 	 * By default this implementation will empty its queue before looking at the
@@ -247,8 +242,7 @@ public class FileReadingMessageSource extends AbstractMessageSource<File>
 	 * consider the effects of setting this flag. The internal
 	 * {@link java.util.concurrent.BlockingQueue} that this class is keeping
 	 * will more likely be out of sync with the file system if this flag is set
-	 * to <code>false</code>, but it will change more often (causing expensive
-	 * reordering) if it is set to <code>true</code>.
+	 * to false, but it will change more often (causing expensive reordering) if it is set to true.
 	 * @param scanEachPoll
 	 *            whether or not the component should re-scan (as opposed to not
 	 *            rescanning until the entire backlog has been delivered)
@@ -489,13 +483,13 @@ public class FileReadingMessageSource extends AbstractMessageSource<File>
 			while (key != null) {
 				File parentDir = ((Path) key.watchable()).toAbsolutePath().toFile();
 				for (WatchEvent<?> event : key.pollEvents()) {
-					if (event.kind() == StandardWatchEventKinds.ENTRY_CREATE ||
-							event.kind() == StandardWatchEventKinds.ENTRY_MODIFY ||
-							event.kind() == StandardWatchEventKinds.ENTRY_DELETE) {
+					if (StandardWatchEventKinds.ENTRY_CREATE.equals(event.kind()) ||
+							StandardWatchEventKinds.ENTRY_MODIFY.equals(event.kind()) ||
+							StandardWatchEventKinds.ENTRY_DELETE.equals(event.kind())) {
 
 						processFilesFromNormalEvent(files, parentDir, event);
 					}
-					else if (event.kind() == StandardWatchEventKinds.OVERFLOW) {
+					else if (StandardWatchEventKinds.OVERFLOW.equals(event.kind())) {
 						processFilesFromOverflowEvent(files, event);
 					}
 				}
@@ -510,7 +504,7 @@ public class FileReadingMessageSource extends AbstractMessageSource<File>
 			File file = new File(parentDir, item.toFile().getName());
 			logger.debug(() -> "Watch event [" + event.kind() + "] for file [" + file + "]");
 
-			if (event.kind() == StandardWatchEventKinds.ENTRY_DELETE) {
+			if (StandardWatchEventKinds.ENTRY_DELETE.equals(event.kind())) {
 				if (getFilter() instanceof ResettableFileListFilter) {
 					((ResettableFileListFilter<File>) getFilter()).remove(file);
 				}

spring-integration-groovy/src/main/java/org/springframework/integration/groovy/config/GroovyAwareScriptExecutingProcessorFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -27,6 +27,7 @@ import org.springframework.scripting.ScriptSource;
  * if provided {@code language == "groovy"}, otherwise delegates to the super class.
  *
  * @author Artem Bilan
+ *
  * @since 5.0
  */
 public class GroovyAwareScriptExecutingProcessorFactory extends ScriptExecutingProcessorFactory {
@@ -34,7 +35,8 @@ public class GroovyAwareScriptExecutingProcessorFactory extends ScriptExecutingP
 	@Override
 	public AbstractScriptExecutingMessageProcessor<?> createMessageProcessor(String language,
 			ScriptSource scriptSource, ScriptVariableGenerator scriptVariableGenerator) {
-		if ("groovy".equals(language.toLowerCase())) {
+
+		if ("groovy".equalsIgnoreCase(language)) {
 			return new GroovyScriptExecutingMessageProcessor(scriptSource, scriptVariableGenerator);
 		}
 		else {

spring-integration-http/src/main/java/org/springframework/integration/http/support/DefaultHttpHeaderMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -249,10 +249,10 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
 	 * @param outboundHeaderNames The outbound header names.
 	 */
 	public void setOutboundHeaderNames(String... outboundHeaderNames) {
-		if (HTTP_REQUEST_HEADER_NAMES == outboundHeaderNames) {
+		if (Arrays.equals(HTTP_REQUEST_HEADER_NAMES, outboundHeaderNames)) {
 			this.isDefaultOutboundMapper = true;
 		}
-		else if (HTTP_RESPONSE_HEADER_NAMES == outboundHeaderNames) {
+		else if (Arrays.equals(HTTP_RESPONSE_HEADER_NAMES, outboundHeaderNames)) {
 			this.isDefaultInboundMapper = true;
 		}
 		this.outboundHeaderNames =

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java

@@ -20,6 +20,7 @@ import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.net.Socket;
 import java.time.Duration;
+import java.util.Objects;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Predicate;
@@ -53,7 +54,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 	private volatile TcpConnectionSupport theConnection;
 
 	/**
-	 * Constructs a factory that will established connections to the host and port.
+	 * Construct a factory that will established connections to the host and port.
 	 * @param host The host.
 	 * @param port The port.
 	 */
@@ -108,7 +109,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 	}
 
 	/**
-	 * Obtains a connection - if {@link #setSingleUse(boolean)} was called with
+	 * Obtain a connection - if {@link #setSingleUse(boolean)} was called with
 	 * true, a new connection is returned; otherwise a single connection is
 	 * reused for all requests while the connection remains open.
 	 * @throws InterruptedException if interrupted.
@@ -198,7 +199,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 	}
 
 	/**
-	 * Transfers attributes such as (de)serializers, singleUse etc to a new connection.
+	 * Transfer attributes such as (de)serializers, singleUse etc to a new connection.
 	 * When the connection factory has a reference to a TCPListener (to read
 	 * responses), or for single use connections, the connection is executed.
 	 * Single use connections need to read from the connection in order to
@@ -243,11 +244,10 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 	/**
 	 * Force close the connection and null the field if it's
 	 * a shared connection.
-	 *
 	 * @param connection The connection.
 	 */
 	public void forceClose(TcpConnection connection) {
-		if (this.theConnection == connection) {
+		if (Objects.equals(this.theConnection, connection)) {
 			this.theConnection = null;
 		}
 		connection.close();

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ClientModeConnectionManager.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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,8 @@
 
 package org.springframework.integration.ip.tcp.connection;
 
+import java.util.Objects;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -28,6 +30,8 @@ import org.springframework.util.Assert;
  * connection factory will create a new one (if possible).
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.1
  *
  */
@@ -53,7 +57,7 @@ public class ClientModeConnectionManager implements Runnable {
 		synchronized (this.clientConnectionFactory) {
 			try {
 				TcpConnection connection = this.clientConnectionFactory.getConnection();
-				if (connection != this.lastConnection) {
+				if (!Objects.equals(connection, this.lastConnection)) {
 					if (this.logger.isDebugEnabled()) {
 						this.logger.debug("Connection " + connection.getConnectionId() + " established");
 					}
@@ -65,8 +69,8 @@ public class ClientModeConnectionManager implements Runnable {
 					}
 				}
 			}
-			catch (Exception e) {
-				this.logger.error("Could not establish connection using " + this.clientConnectionFactory, e);
+			catch (Exception ex) {
+				this.logger.error("Could not establish connection using " + this.clientConnectionFactory, ex);
 			}
 		}
 	}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java

@@ -120,7 +120,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	}
 
 	/**
-	 * Creates a {@link TcpConnectionSupport} object and publishes a
+	 * Create a {@link TcpConnectionSupport} object and publishes a
 	 * {@link TcpConnectionOpenEvent}, if an event publisher is provided.
 	 * @param socket the underlying socket.
 	 * @param server true if this connection is a server connection
@@ -164,7 +164,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	}
 
 	/**
-	 * Closes this connection.
+	 * Close this connection.
 	 */
 	@Override
 	public void close() {
@@ -182,7 +182,6 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	/**
 	 * If we have been intercepted, propagate the close from the outermost interceptor;
 	 * otherwise, just call close().
-	 *
 	 * @param isException true when this call is the result of an Exception.
 	 */
 	protected void closeConnection(boolean isException) {
@@ -288,8 +287,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 
 	/**
 	 * Set a temporary listener to receive just the first incoming message.
-	 * Used in conjunction with a connectionTest in a client connection
-	 * factory.
+	 * Used in conjunction with a connectionTest in a client connection factory.
 	 * @param tListener the test listener.
 	 * @since 5.3
 	 */
@@ -309,7 +307,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	}
 
 	/**
-	 * Registers a sender. Used on server side connections so a
+	 * Register a sender. Used on server side connections so a
 	 * sender can determine which connection to send a reply
 	 * to.
 	 * @param senderToRegister the sender.
@@ -322,7 +320,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	}
 
 	/**
-	 * Registers the senders. Used on server side connections so a
+	 * Register the senders. Used on server side connections so a
 	 * sender can determine which connection to send a reply
 	 * to.
 	 * @param sendersToRegister the sender.
@@ -451,8 +449,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	protected final void sendExceptionToListener(Exception e) {
 		TcpListener listenerForException = getListener();
 		if (!this.exceptionSent.getAndSet(true) && listenerForException != null) {
-			Map<String, Object> headers = Collections.singletonMap(IpHeaders.CONNECTION_ID,
-					(Object) this.getConnectionId());
+			Map<String, Object> headers = Collections.singletonMap(IpHeaders.CONNECTION_ID, getConnectionId());
 			ErrorMessage errorMessage = new ErrorMessage(e, headers);
 			listenerForException.onMessage(errorMessage);
 		}
@@ -491,7 +488,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	 * @param event the event to publish.
 	 */
 	public void publishEvent(TcpConnectionEvent event) {
-		Assert.isTrue(event.getSource() == this, "Can only publish events with this as the source");
+		Assert.isTrue(equals(event.getSource()), "Can only publish events with this as the source");
 		this.doPublish(event);
 	}
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java

@@ -28,6 +28,7 @@ import java.nio.channels.ClosedChannelException;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
+import java.util.Arrays;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
@@ -65,6 +66,8 @@ public class TcpNioConnection extends TcpConnectionSupport {
 
 	private static final int SIXTY = 60;
 
+	private static final int MAX_MESSAGE_SIZE = 60 * 1024;
+
 	private static final long DEFAULT_PIPE_TIMEOUT = 60000;
 
 	private static final byte[] EOF = new byte[0]; // EOF marker buffer
@@ -87,8 +90,6 @@ public class TcpNioConnection extends TcpConnectionSupport {
 
 	private volatile ByteBuffer rawBuffer;
 
-	private volatile int maxMessageSize = 60 * 1024;
-
 	private volatile long lastRead;
 
 	private volatile long lastSend;
@@ -100,7 +101,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 	private volatile boolean timedOut;
 
 	/**
-	 * Constructs a TcpNetConnection for the SocketChannel.
+	 * Construct a TcpNetConnection for the SocketChannel.
 	 * @param socketChannel The socketChannel.
 	 * @param server If true, this connection was created as
 	 * a result of an incoming request.
@@ -211,7 +212,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 	}
 
 	/**
-	 * Allocates a ByteBuffer of the requested length using normal or
+	 * Allocate a ByteBuffer of the requested length using normal or
 	 * direct buffers, depending on the usingDirectBuffers field.
 	 *
 	 * @param length The buffer length.
@@ -229,8 +230,8 @@ public class TcpNioConnection extends TcpConnectionSupport {
 	}
 
 	/**
-	 * If there is no listener,
-	 * this method exits. When there is a listener, this method assembles
+	 * If there is no listener, this method exits.
+	 * When there is a listener, this method assembles
 	 * data into messages by invoking convertAndSend whenever there is
 	 * data in the input Stream. Method exits when a message is complete
 	 * and there is no more data; thus freeing the thread to work on other
@@ -409,7 +410,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 
 	private void doRead() throws IOException {
 		if (this.rawBuffer == null) {
-			this.rawBuffer = allocate(this.maxMessageSize);
+			this.rawBuffer = allocate(MAX_MESSAGE_SIZE);
 		}
 
 		this.writingLatch = new CountDownLatch(1);
@@ -722,7 +723,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 				}
 			}
 			int bite;
-			bite = this.currentBuffer[this.currentOffset++] & 0xff;
+			bite = this.currentBuffer[this.currentOffset++] & 0xff; // N0SONAR
 			this.available.decrementAndGet();
 			if (this.currentOffset >= this.currentBuffer.length) {
 				this.currentBuffer = null;
@@ -736,7 +737,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 			while (buffer == null) {
 				try {
 					buffer = this.buffers.poll(1, TimeUnit.SECONDS);
-					if (buffer == EOF || (buffer == null && this.isClosed)) {
+					if (Arrays.equals(EOF, buffer) || (buffer == null && this.isClosed)) {
 						return null;
 					}
 				}

spring-integration-jpa/src/test/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -17,7 +17,7 @@
 package org.springframework.integration.jpa.inbound;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
+import static org.awaitility.Awaitility.await;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -25,8 +25,7 @@ import java.util.List;
 
 import javax.persistence.EntityManager;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -44,8 +43,7 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.annotation.Rollback;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 import org.springframework.transaction.annotation.Transactional;
 
 /**
@@ -58,11 +56,10 @@ import org.springframework.transaction.annotation.Transactional;
  * @since 2.2
  *
  */
-@RunWith(SpringJUnit4ClassRunner.class)
-@ContextConfiguration
+@SpringJUnitConfig
 @Rollback
 @Transactional("transactionManager")
-@DirtiesContext
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
 public class JpaPollingChannelAdapterTests {
 
 	@Autowired
@@ -90,7 +87,6 @@ public class JpaPollingChannelAdapterTests {
 	 * to retrieve a list of records from the database.
 	 */
 	@Test
-	@DirtiesContext
 	public void testWithEntityClass() throws Exception {
 		testTrigger.reset();
 		//~~~~SETUP~~~~~
@@ -259,7 +255,6 @@ public class JpaPollingChannelAdapterTests {
 	 * will be deleted after the polling.
 	 */
 	@Test
-	@DirtiesContext
 	public void testWithJpaQueryAndDelete() throws Exception {
 		testTrigger.reset();
 
@@ -297,30 +292,11 @@ public class JpaPollingChannelAdapterTests {
 
 		assertThat(students.size()).isEqualTo(3);
 
-		Long studentCount = waitForDeletes(students);
-
-		assertThat(studentCount).isEqualTo(Long.valueOf(0));
-	}
-
-	private Long waitForDeletes(Collection<?> students) throws InterruptedException {
-		Long studentCount = (long) students.size();
-
-		int n = 0;
-
-		while (studentCount > 0) {
-			studentCount = entityManager.createQuery("select count(*) from Student", Long.class).getSingleResult();
-			if (studentCount > 0) {
-				Thread.sleep(100);
-				if (n++ > 100) {
-					fail("Failed to delete after poll");
-				}
-			}
-		}
-		return studentCount;
+		await().until(() -> entityManager.createQuery("select count(*) from Student", Long.class).getSingleResult(),
+				(count) -> count == 0);
 	}
 
 	@Test
-	@DirtiesContext
 	public void testWithJpaQueryButNoResultsAndDelete() throws Exception {
 		testTrigger.reset();
 
@@ -355,7 +331,6 @@ public class JpaPollingChannelAdapterTests {
 	 * will be deleted after the polling.
 	 */
 	@Test
-	@DirtiesContext
 	public void testWithJpaQueryAndDeletePerRow() throws Exception {
 		testTrigger.reset();
 
@@ -389,10 +364,8 @@ public class JpaPollingChannelAdapterTests {
 
 		assertThat(students.size()).isEqualTo(3);
 
-		Long studentCount = waitForDeletes(students);
-
-		assertThat(studentCount).isEqualTo(Long.valueOf(0));
-
+		await().until(() -> entityManager.createQuery("select count(*) from Student", Long.class).getSingleResult(),
+				(count) -> count == 0);
 	}
 
 	/**

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-2021 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.
@@ -440,7 +440,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 					this.consumerProperties.getClientId(), null, this.consumerProperties.getKafkaConsumerProperties());
 
 			ConsumerRebalanceListener rebalanceCallback =
-					new ItegrationConsumerRebalanceListener(this.consumerProperties.getConsumerRebalanceListener());
+					new IntegrationConsumerRebalanceListener(this.consumerProperties.getConsumerRebalanceListener());
 
 			Pattern topicPattern = this.consumerProperties.getTopicPattern();
 			TopicPartitionOffset[] partitions = this.consumerProperties.getTopicPartitions();
@@ -580,13 +580,13 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 		}
 	}
 
-	private class ItegrationConsumerRebalanceListener implements ConsumerRebalanceListener {
+	private class IntegrationConsumerRebalanceListener implements ConsumerRebalanceListener {
 
 		private final ConsumerRebalanceListener providedRebalanceListener;
 
 		private final boolean isConsumerAware;
 
-		ItegrationConsumerRebalanceListener(ConsumerRebalanceListener providedRebalanceListener) {
+		IntegrationConsumerRebalanceListener(ConsumerRebalanceListener providedRebalanceListener) {
 			this.providedRebalanceListener = providedRebalanceListener;
 			this.isConsumerAware = providedRebalanceListener instanceof ConsumerAwareRebalanceListener;
 		}
@@ -594,9 +594,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 		@Override
 		public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 			KafkaMessageSource.this.assignedPartitions.removeAll(partitions);
-			if (KafkaMessageSource.this.logger.isInfoEnabled()) {
-				KafkaMessageSource.this.logger.info("Partitions revoked: " + partitions);
-			}
+			KafkaMessageSource.this.logger.info(() -> "Partitions revoked: " + partitions);
 			if (this.providedRebalanceListener != null) {
 				if (this.isConsumerAware) {
 					((ConsumerAwareRebalanceListener) this.providedRebalanceListener)
@@ -625,9 +623,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 				KafkaMessageSource.this.logger.warn("Paused consumer resumed by Kafka due to rebalance; "
 						+ "consumer paused again, so the initial poll() will never return any records");
 			}
-			if (KafkaMessageSource.this.logger.isInfoEnabled()) {
-				KafkaMessageSource.this.logger.info("Partitions assigned: " + partitions);
-			}
+			KafkaMessageSource.this.logger.info(() -> "Partitions assigned: " + partitions);
 			if (this.providedRebalanceListener != null) {
 				if (this.isConsumerAware) {
 					((ConsumerAwareRebalanceListener) this.providedRebalanceListener)
@@ -783,7 +779,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 						// see if there are any pending acks for higher offsets
 						List<KafkaAckInfo<K, V>> toCommit = new ArrayList<>();
 						for (KafkaAckInfo<K, V> info : candidates) {
-							if (info != this.ackInfo) {
+							if (!this.ackInfo.equals(info)) {
 								if (info.isAckDeferred()) {
 									toCommit.add(info);
 								}
@@ -828,9 +824,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 						}
 					}
 					else {
-						if (this.logger.isDebugEnabled()) {
-							this.logger.debug("Deferring commit offset; earlier messages are in flight.");
-						}
+						this.logger.debug("Deferring commit offset; earlier messages are in flight.");
 					}
 				}
 			}

spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -35,7 +35,6 @@ import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.core.ResolvableType;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
-import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
@@ -47,6 +46,7 @@ import org.springframework.http.codec.ServerCodecConfigurer;
 import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpResponse;
 import org.springframework.integration.expression.ExpressionEvalMap;
+import org.springframework.integration.http.HttpHeaders;
 import org.springframework.integration.http.inbound.BaseHttpInboundEndpoint;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.messaging.Message;
@@ -281,9 +281,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 				prepareRequestMessageBuilder(request, payload, headers);
 
 		return exchange.getPrincipal()
-				.map(principal ->
-						messageBuilder
-								.setHeader(org.springframework.integration.http.HttpHeaders.USER_PRINCIPAL, principal))
+				.map(principal -> messageBuilder.setHeader(HttpHeaders.USER_PRINCIPAL, principal))
 				.defaultIfEmpty(messageBuilder)
 				.map(AbstractIntegrationMessageBuilder::build)
 				.zipWith(Mono.just(httpEntity));
@@ -308,19 +306,17 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 							.copyHeaders(headers);
 		}
 
-		messageBuilder.setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_URL,
-				request.getURI().toString());
+		messageBuilder.setHeader(HttpHeaders.REQUEST_URL, request.getURI().toString());
 		HttpMethod httpMethod = request.getMethod();
 		if (httpMethod != null) {
-			messageBuilder.setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_METHOD,
-					httpMethod.toString());
+			messageBuilder.setHeader(HttpHeaders.REQUEST_METHOD, httpMethod.toString());
 		}
 		return messageBuilder;
 	}
 
 	private EvaluationContext buildEvaluationContext(RequestEntity<?> httpEntity, ServerWebExchange exchange) {
 		ServerHttpRequest request = exchange.getRequest();
-		HttpHeaders requestHeaders = request.getHeaders();
+		org.springframework.http.HttpHeaders requestHeaders = request.getHeaders();
 		MultiValueMap<String, String> requestParams = request.getQueryParams();
 		Map<String, Object> exchangeAttributes = exchange.getAttributes();
 
@@ -375,8 +371,8 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 								exchange.getResponse().setStatusCode(e.getStatusCode());
 							}
 
-							HttpHeaders entityHeaders = e.getHeaders();
-							HttpHeaders responseHeaders = exchange.getResponse().getHeaders();
+							org.springframework.http.HttpHeaders entityHeaders = e.getHeaders();
+							org.springframework.http.HttpHeaders responseHeaders = exchange.getResponse().getHeaders();
 
 							if (!entityHeaders.isEmpty()) {
 								entityHeaders.entrySet().stream()
@@ -501,7 +497,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
 		if (adapter.isNoValue()) {
 			return ResolvableType.forClass(Void.class);
 		}
-		else if (genericType != ResolvableType.NONE) {
+		else if (!ResolvableType.NONE.equals(genericType)) {
 			return genericType;
 		}
 		else {

spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 the original author or authors.
+ * Copyright 2014-2021 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.
@@ -88,7 +88,7 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean {
 	}
 
 	public void setMessageListener(WebSocketListener messageListener) {
-		Assert.state(this.messageListener == null || this.messageListener == messageListener,
+		Assert.state(this.messageListener == null || this.messageListener.equals(messageListener),
 				"'messageListener' is already configured");
 		this.messageListener = messageListener;
 	}
@@ -119,7 +119,7 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean {
 
 	public WebSocketSession getSession(String sessionId) {
 		WebSocketSession session = this.sessions.get(sessionId);
-		Assert.notNull(session, "Session not found for id '" + sessionId + "'");
+		Assert.notNull(session, () -> "Session not found for id '" + sessionId + "'");
 		return session;
 	}
 
@@ -139,8 +139,8 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean {
 				try {
 					session.close(CloseStatus.GOING_AWAY);
 				}
-				catch (Exception e) {
-					this.logger.error("Failed to close session id '" + session.getId() + "': " + e.getMessage());
+				catch (Exception ex) {
+					this.logger.error("Failed to close session id '" + session.getId() + "': " + ex.getMessage());
 				}
 			}
 		}