GH-3469 New Sonar smell & volatile busyWaitMillis

Fixes https://github.com/spring-projects/spring-integration/issues/3469

* Fix new Sonar smells
* Mark `LockRegistryLeaderInitiator.busyWaitMillis` as `volatile`
so runtime changes (e.g. `RedisLockRegistryLeaderInitiatorTests`)
will make an immediate effect

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.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.
@@ -19,7 +19,6 @@ package org.springframework.integration.amqp.dsl;
 import org.springframework.amqp.rabbit.batch.BatchingStrategy;
 import org.springframework.amqp.rabbit.retry.MessageRecoverer;
 import org.springframework.amqp.support.converter.MessageConverter;
-import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
 import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
@@ -183,7 +182,7 @@ public class AmqpBaseInboundGatewaySpec<S extends AmqpBaseInboundGatewaySpec<S>>
 	 * @param messageRecoverer the callback.
 	 * @return the spec.
 	 * @since 5.5
-	 * @see AmqpInboundChannelAdapter#setMessageRecoverer(MessageRecoverer)
+	 * @see org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter#setMessageRecoverer(MessageRecoverer)
 	 */
 	public S messageRecoverer(MessageRecoverer messageRecoverer) {
 		this.target.setMessageRecoverer(messageRecoverer);

spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java

@@ -565,7 +565,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
 		MessageGroup messageGroup = this.messageStore.getMessageGroup(this.messageGroupId);
 		try (Stream<Message<?>> messageStream = messageGroup.streamMessages()) {
 			TaskScheduler taskScheduler = getTaskScheduler();
-			messageStream.forEach((message) ->
+			messageStream.forEach((message) -> // NOSONAR
 					taskScheduler.schedule(() -> {
 						// This is fine to keep the reference to the message,
 						// because the scheduled task is performed immediately.

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

@@ -20,15 +20,16 @@ import java.util.Collections;
 import java.util.Set;
 
 import org.springframework.core.convert.TypeDescriptor;
-import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.convert.converter.GenericConverter;
 import org.springframework.integration.json.JsonPropertyAccessor.JsonNodeWrapper;
+import org.springframework.lang.Nullable;
 
 import com.fasterxml.jackson.databind.JsonNode;
 
 /**
- * The {@link Converter} implementation for the conversion of {@link JsonPropertyAccessor.JsonNodeWrapper} to
- * {@link JsonNode}, when the {@link JsonPropertyAccessor.JsonNodeWrapper} can be a result of the expression
+ * The {@link org.springframework.core.convert.converter.Converter} implementation for the conversion
+ * of {@link JsonPropertyAccessor.JsonNodeWrapper} to {@link JsonNode},
+ * when the {@link JsonPropertyAccessor.JsonNodeWrapper} can be a result of the expression
  * for JSON in case of the {@link JsonPropertyAccessor} usage.
  *
  * @author Pierre Lakreb
@@ -44,8 +45,12 @@ class JsonNodeWrapperToJsonNodeConverter implements GenericConverter {
 	}
 
 	@Override
-	public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {
-		return targetType.getObjectType().cast(((JsonNodeWrapper<?>) source).getRealNode());
+	@Nullable
+	public Object convert(@Nullable Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {
+		if (source != null) {
+			return targetType.getObjectType().cast(((JsonNodeWrapper<?>) source).getRealNode());
+		}
+		return null;
 	}
 
 }

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

@@ -161,7 +161,7 @@ public class JsonPropertyAccessor implements PropertyAccessor {
 			}
 			catch (IOException e) {
 				throw new AccessException(
-						"Can not get content of binary value : " + json);
+						"Can not get content of binary value: " + json, e);
 			}
 		}
 		throw new IllegalArgumentException("Json is not ValueNode.");
@@ -210,7 +210,7 @@ public class JsonPropertyAccessor implements PropertyAccessor {
 
 		@Override
 		public int compareTo(ComparableJsonNode o) {
-			return this.delegate.equals(o.delegate) ? 0 : 1;
+			return this.delegate.equals(o.delegate) ? 0 : 1; // NOSONAR
 		}
 
 	}
@@ -239,15 +239,13 @@ public class JsonPropertyAccessor implements PropertyAccessor {
 
 		@Override
 		public Object get(int index) {
-			if (index < 0) {
-				// negative index can be handled with that conversion
-				index = this.delegate.size() + index;
-			}
+			// negative index - get from the end of list
+			int i = index < 0 ? this.delegate.size() + index : index;
 			try {
-				return wrap(this.delegate.get(index));
+				return wrap(this.delegate.get(i));
 			}
-			catch (AccessException e) {
-				throw new IllegalArgumentException(e);
+			catch (AccessException ex) {
+				throw new IllegalArgumentException(ex);
 			}
 		}
 

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java

@@ -169,7 +169,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 		try {
 			this.storeLock.lockInterruptibly();
 			try (Stream<Message<?>> messageStream = stream()) {
-				return messageStream.findFirst().orElse(null);
+				return messageStream.findFirst().orElse(null); // NOSONAR
 			}
 			finally {
 				this.storeLock.unlock();

spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 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.
@@ -121,17 +121,6 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 	 */
 	private long heartBeatMillis = DEFAULT_HEART_BEAT_TIME;
 
-	/**
-	 * Time in milliseconds to wait in between attempts to acquire the lock, if it is not
-	 * held. The longer this is, the longer the system can be leaderless, if the leader
-	 * dies. If a leader dies without releasing its lock, the system might still have to
-	 * wait for the old lock to expire, but after that it should not have to wait longer
-	 * than the busy wait time to get a new leader. If the remote lock does not expire, or
-	 * if you know it interrupts the current thread when it expires or is broken, then you
-	 * can reduce the busy wait to zero.
-	 */
-	private long busyWaitMillis = DEFAULT_BUSY_WAIT_TIME;
-
 	private boolean publishFailedEvents = false;
 
 	private LeaderSelector leaderSelector;
@@ -153,6 +142,17 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 	 */
 	private int phase = Integer.MAX_VALUE - 1000;
 
+	/**
+	 * Time in milliseconds to wait in between attempts to acquire the lock, if it is not
+	 * held. The longer this is, the longer the system can be leaderless, if the leader
+	 * dies. If a leader dies without releasing its lock, the system might still have to
+	 * wait for the old lock to expire, but after that it should not have to wait longer
+	 * than the busy wait time to get a new leader. If the remote lock does not expire, or
+	 * if you know it interrupts the current thread when it expires or is broken, then you
+	 * can reduce the busy wait to zero.
+	 */
+	private volatile long busyWaitMillis = DEFAULT_BUSY_WAIT_TIME;
+
 	/**
 	 * Flag that indicates whether the leadership election for this {@link #candidate} is
 	 * running.

spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.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.
@@ -660,7 +660,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 
 				byte[] buffer = new byte[StreamUtils.BUFFER_SIZE];
 				int bytesRead;
-				while ((bytesRead = inputStream.read(buffer)) != -1) {
+				while ((bytesRead = inputStream.read(buffer)) != -1) { // NOSONAR
 					outputStream.write(buffer, 0, bytesRead);
 				}
 				if (this.appendNewLine) {
@@ -680,7 +680,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 			bos = state != null ? state.stream : createOutputStream(fileToWriteTo, true);
 			byte[] buffer = new byte[StreamUtils.BUFFER_SIZE];
 			int bytesRead = -1;
-			while ((bytesRead = inputStream.read(buffer)) != -1) {
+			while ((bytesRead = inputStream.read(buffer)) != -1) { // NOSONAR
 				bos.write(buffer, 0, bytesRead);
 			}
 			if (FileWritingMessageHandler.this.appendNewLine) {

spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveDirectoryScanner.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.
@@ -73,10 +73,9 @@ public class RecursiveDirectoryScanner extends DefaultDirectoryScanner {
 		try (Stream<Path> pathStream = Files.walk(directory.toPath(), this.maxDepth, this.fileVisitOptions);) {
 			Stream<File> fileStream =
 					pathStream
-							.skip(1)
+							.skip(1) // NOSONAR
 							.map(Path::toFile)
-							.filter(file -> !supportAcceptFilter
-									|| ((AbstractFileListFilter<File>) filter).accept(file));
+							.filter(file -> !supportAcceptFilter || filter.accept(file));
 
 			if (supportAcceptFilter) {
 				return fileStream.collect(Collectors.toList());

spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.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.
@@ -55,6 +55,7 @@ import org.springframework.integration.handler.AbstractReplyProducingMessageHand
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
 import org.springframework.integration.support.MutableMessage;
 import org.springframework.integration.support.PartialSuccessException;
+import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
@@ -586,38 +587,36 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 	}
 
 	private Object doLs(Message<?> requestMessage) {
-		String dir = this.fileNameProcessor != null
-				? this.fileNameProcessor.processMessage(requestMessage)
-				: null;
-		if (dir != null && !dir.endsWith(this.remoteFileTemplate.getRemoteFileSeparator())) {
-			dir += this.remoteFileTemplate.getRemoteFileSeparator();
-		}
-		final String fullDir = dir;
+		String dir = obtainRemoteDir(requestMessage);
 		return this.remoteFileTemplate.execute(session -> {
-			List<?> payload = ls(requestMessage, session, fullDir);
+			List<?> payload = ls(requestMessage, session, dir);
 			return getMessageBuilderFactory()
 					.withPayload(payload)
-					.setHeader(FileHeaders.REMOTE_DIRECTORY, fullDir)
+					.setHeader(FileHeaders.REMOTE_DIRECTORY, dir)
 					.setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort());
 		});
 
 	}
 
 	private Object doNlst(Message<?> requestMessage) {
+		String dir = obtainRemoteDir(requestMessage);
+		return this.remoteFileTemplate.execute(session -> {
+			List<?> payload = nlst(requestMessage, session, dir);
+			return getMessageBuilderFactory()
+					.withPayload(payload)
+					.setHeader(FileHeaders.REMOTE_DIRECTORY, dir)
+					.setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort());
+		});
+	}
+
+	private String obtainRemoteDir(Message<?> requestMessage) {
 		String dir = this.fileNameProcessor != null
 				? this.fileNameProcessor.processMessage(requestMessage)
 				: null;
 		if (dir != null && !dir.endsWith(this.remoteFileTemplate.getRemoteFileSeparator())) {
 			dir += this.remoteFileTemplate.getRemoteFileSeparator();
 		}
-		final String fullDir = dir;
-		return this.remoteFileTemplate.execute(session -> {
-			List<?> payload = nlst(requestMessage, session, fullDir);
-			return getMessageBuilderFactory()
-					.withPayload(payload)
-					.setHeader(FileHeaders.REMOTE_DIRECTORY, fullDir)
-					.setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort());
-		});
+		return dir;
 	}
 
 	/**
@@ -963,7 +962,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 			remotePath = (parent + child);
 		}
 		else if (StringUtils.hasText(child)) {
-			remotePath = "." + this.remoteFileTemplate.getRemoteFileSeparator() + child;
+			remotePath = '.' + this.remoteFileTemplate.getRemoteFileSeparator() + child;
 		}
 		return remotePath;
 	}
@@ -1055,7 +1054,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 			try {
 				session.read(remoteFilePath, outputStream);
 			}
-			catch (Exception e) {
+			catch (Exception ex) {
 				/* Some operation systems acquire exclusive file-lock during file processing
 				   and the file can't be deleted without closing streams before.
 				*/
@@ -1064,12 +1063,8 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 					this.logger.warn(() -> "Failed to delete tempFile " + tempFile);
 				}
 
-				if (e instanceof RuntimeException) {
-					throw (RuntimeException) e;
-				}
-				else {
-					throw new MessagingException("Failure occurred while copying from remote to local directory", e);
-				}
+				throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message,
+						() -> "Failure occurred while copying from remote to local directory", ex);
 			}
 			finally {
 				try {

spring-integration-file/src/main/java/org/springframework/integration/file/transformer/AbstractFilePayloadTransformer.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.
@@ -54,8 +54,7 @@ public abstract class AbstractFilePayloadTransformer<T> implements Transformer,
 
 	/**
 	 * Specify whether to delete the File after transformation.
-	 * Default is <em>false</em>.
-	 *
+	 * Default is {@code false}.
 	 * @param deleteFiles true to delete the file.
 	 */
 	public void setDeleteFiles(boolean deleteFiles) {
@@ -85,27 +84,26 @@ public abstract class AbstractFilePayloadTransformer<T> implements Transformer,
 			Assert.notNull(payload, "Message payload must not be null");
 			Assert.isInstanceOf(File.class, payload, "Message payload must be of type [java.io.File]");
 			File file = (File) payload;
-			T result = this.transformFile(file);
-			Message<?> transformedMessage = getMessageBuilderFactory().withPayload(result)
-					.copyHeaders(message.getHeaders())
-					.setHeaderIfAbsent(FileHeaders.ORIGINAL_FILE, file)
-					.setHeaderIfAbsent(FileHeaders.FILENAME, file.getName())
-					.build();
-			if (this.deleteFiles) {
-				if (!file.delete() && this.logger.isWarnEnabled()) {
-					this.logger.warn("failed to delete File '" + file + "'");
-				}
+			T result = transformFile(file);
+			Message<?> transformedMessage =
+					getMessageBuilderFactory()
+							.withPayload(result)
+							.copyHeaders(message.getHeaders())
+							.setHeaderIfAbsent(FileHeaders.ORIGINAL_FILE, file)
+							.setHeaderIfAbsent(FileHeaders.FILENAME, file.getName())
+							.build();
+			if (this.deleteFiles && !file.delete() && this.logger.isWarnEnabled()) {
+				this.logger.warn("failed to delete File '" + file + "'");
 			}
 			return transformedMessage;
 		}
-		catch (Exception e) {
-			throw new MessagingException(message, "failed to transform File Message", e);
+		catch (Exception ex) {
+			throw new MessagingException(message, "failed to transform File Message", ex);
 		}
 	}
 
 	/**
 	 * Subclasses must implement this method to transform the File contents.
-	 *
 	 * @param file The file.
 	 * @return The result of the transformation.
 	 * @throws IOException Any IOException.

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/gateway/FtpOutboundGateway.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.
@@ -284,17 +284,8 @@ public class FtpOutboundGateway extends AbstractRemoteFileOutboundGateway<FTPFil
 			}
 			return task.call();
 		}
-		catch (Exception e) {
-			if (e instanceof IOException) {
-				throw (IOException) e;
-
-			}
-			else if (e instanceof RuntimeException) {
-				throw (RuntimeException) e;
-			}
-			else {
-				throw new IOException("Uncategorised IO exception", e);
-			}
+		catch (Exception ex) {
+			throw rethrowAsIoExceptionIfAny(ex);
 		}
 		finally {
 			if (restoreWorkingDirectory) {
@@ -303,6 +294,19 @@ public class FtpOutboundGateway extends AbstractRemoteFileOutboundGateway<FTPFil
 		}
 	}
 
+	private IOException rethrowAsIoExceptionIfAny(Exception ex) {
+		if (ex instanceof IOException) {
+			return (IOException) ex;
+
+		}
+		else if (ex instanceof RuntimeException) {
+			throw (RuntimeException) ex;
+		}
+		else {
+			return new IOException("Uncategorized IO exception", ex);
+		}
+	}
+
 	@Override
 	public boolean isChmodCapable() {
 		return true;
@@ -316,7 +320,7 @@ public class FtpOutboundGateway extends AbstractRemoteFileOutboundGateway<FTPFil
 				client.sendSiteCommand(chModCommand);
 			}
 			catch (IOException e) {
-				throw new UncheckedIOException("Failed to execute '" + chModCommand  + "'", e);
+				throw new UncheckedIOException("Failed to execute '" + chModCommand + "'", e);
 			}
 		});
 	}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.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.
@@ -153,7 +153,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 				if (this.bufferedOutputStream == null) {
 					int writeBufferSize = this.socketChannel.socket().getSendBufferSize();
 					this.bufferedOutputStream = new BufferedOutputStream(getChannelOutputStream(),
-							writeBufferSize > 0 ? writeBufferSize : 8192);
+							writeBufferSize > 0 ? writeBufferSize : 8192); // NOSONAR
 				}
 				Object object = getMapper().fromMessage(message);
 				Assert.state(object != null, "Mapper mapped the message to 'null'.");
@@ -393,19 +393,17 @@ public class TcpNioConnection extends TcpConnectionSupport {
 			}
 			listener.onMessage(message);
 		}
-		catch (Exception e) {
-			if (e instanceof NoListenerException) { // could also be thrown by an interceptor
-				if (logger.isWarnEnabled()) {
-					logger.warn("Unexpected message - no endpoint registered with connection: "
-							+ getConnectionId()
-							+ " - "
-							+ message);
-				}
-			}
-			else {
-				logger.error("Exception sending message: " + message, e);
+		catch (NoListenerException ex) { // could also be thrown by an interceptor
+			if (logger.isWarnEnabled()) {
+				logger.warn("Unexpected message - no endpoint registered with connection: "
+						+ getConnectionId()
+						+ " - "
+						+ message);
 			}
 		}
+		catch (Exception ex) {
+			logger.error("Exception sending message: " + message, ex);
+		}
 	}
 
 	private void doRead() throws IOException {
@@ -424,7 +422,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 			checkForAssembler();
 
 			if (logger.isTraceEnabled()) {
-				logger.trace("Before read: " + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
+				logger.trace("Before read: " + this.rawBuffer.position() + '/' + this.rawBuffer.limit());
 			}
 			int len = this.socketChannel.read(this.rawBuffer);
 			if (len < 0) {
@@ -432,11 +430,11 @@ public class TcpNioConnection extends TcpConnectionSupport {
 				closeConnection(true);
 			}
 			if (logger.isTraceEnabled()) {
-				logger.trace("After read: " + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
+				logger.trace("After read: " + this.rawBuffer.position() + '/' + this.rawBuffer.limit());
 			}
 			this.rawBuffer.flip();
 			if (logger.isTraceEnabled()) {
-				logger.trace("After flip: " + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
+				logger.trace("After flip: " + this.rawBuffer.position() + '/' + this.rawBuffer.limit());
 			}
 			if (logger.isDebugEnabled()) {
 				logger.debug("Read " + this.rawBuffer.limit() + " into raw buffer");
@@ -475,10 +473,8 @@ public class TcpNioConnection extends TcpConnectionSupport {
 				}
 				catch (RejectedExecutionException e) {
 					this.executionControl.decrementAndGet();
-					if (logger.isInfoEnabled()) {
-						logger.info("Insufficient threads in the assembler fixed thread pool; consider increasing " +
+					logger.info("Insufficient threads in the assembler fixed thread pool; consider increasing " +
 								"this task executor pool size");
-					}
 					throw e;
 				}
 			}

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java

@@ -39,7 +39,6 @@ import org.springframework.integration.redis.support.RedisHeaders;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.converter.MessageConversionException;
 import org.springframework.util.Assert;
@@ -209,7 +208,7 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport {
 	 * Configure a resume Function to resume the main sequence when polling the stream fails.
 	 * Mutually exclusive with {@link #setStreamReceiverOptions(StreamReceiver.StreamReceiverOptions)}.
 	 * By default this function extract the failed {@link Record} and sends an
-	 * {@link org.springframework.messaging.support.ErrorMessage} to the provided {@link #setErrorChannel(MessageChannel)}.
+	 * {@link org.springframework.messaging.support.ErrorMessage} to the provided {@link #setErrorChannel}.
 	 * The failed message for this record may have a {@link IntegrationMessageHeaderAccessor#ACKNOWLEDGMENT_CALLBACK}
 	 * header when manual acknowledgment is configured for this message producer.
 	 * @param resumeFunction must not be null.

spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 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.
@@ -24,7 +24,6 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 
@@ -54,7 +53,6 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
 
 	@Test
 	@RedisAvailable
-	@Ignore("Intermittent failures")
 	public void testDistributedLeaderElection() throws Exception {
 		RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), "LeaderInitiator");
 		registry.expireUnusedOlderThan(-1);

spring-integration-test-support/src/main/java/org/springframework/integration/test/util/TestUtils.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.
@@ -119,7 +119,7 @@ public abstract class TestUtils {
 	public static TestApplicationContext createTestApplicationContext() {
 		TestApplicationContext context = new TestApplicationContext();
 		ErrorHandler errorHandler = new MessagePublishingErrorHandler(context);
-		ThreadPoolTaskScheduler scheduler = createTaskScheduler(10);
+		ThreadPoolTaskScheduler scheduler = createTaskScheduler(10); // NOSONAR
 		scheduler.setErrorHandler(errorHandler);
 		registerBean("taskScheduler", scheduler, context);
 		registerBean("integrationConversionService", new DefaultFormattingConversionService(), context);
@@ -255,15 +255,13 @@ public abstract class TestUtils {
 			boolean sent = false;
 			if (errorChannel != null) {
 				try {
-					sent = errorChannel.send(new ErrorMessage(throwable), 10000);
+					sent = errorChannel.send(new ErrorMessage(throwable), 10000); // NOSONAR
 				}
 				catch (Throwable errorDeliveryError) { //NOSONAR
 					// message will be logged only
-					if (logger.isWarnEnabled()) {
-						logger.warn("Error message was not delivered.", errorDeliveryError);
-					}
-					if (errorDeliveryError instanceof Error) {
-						throw ((Error) errorDeliveryError); // NOSONAR
+					logger.warn("Error message was not delivered.", errorDeliveryError);
+					if (errorDeliveryError instanceof Error) {  // NOSONAR
+						throw ((Error) errorDeliveryError);
 					}
 				}
 			}

spring-integration-ws/src/main/java/org/springframework/integration/ws/AbstractWebServiceInboundGateway.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.
@@ -72,7 +72,7 @@ public abstract class AbstractWebServiceInboundGateway extends MessagingGatewayS
 			this.doInvoke(messageContext);
 		}
 		catch (Exception e) {
-			while ((e instanceof MessagingException || e instanceof ExpressionException) &&
+			while ((e instanceof MessagingException || e instanceof ExpressionException) && // NOSONAR
 					e.getCause() instanceof Exception) {
 				e = (Exception) e.getCause();
 			}

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-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.
@@ -273,9 +273,9 @@ public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAw
 								this.backendSocketConfigurer.accept(backendSocket);
 							}
 
-							this.frontendPort.set(bindSocket(frontendSocket, this.frontendPort.get()));
-							this.backendPort.set(bindSocket(backendSocket, this.backendPort.get()));
-							boolean bound = controlSocket.bind(this.controlAddress);
+							this.frontendPort.set(bindSocket(frontendSocket, this.frontendPort.get())); // NOSONAR
+							this.backendPort.set(bindSocket(backendSocket, this.backendPort.get())); // NOSONAR
+							boolean bound = controlSocket.bind(this.controlAddress); // NOSONAR
 							if (!bound) {
 								throw new IllegalArgumentException("Cannot bind ZeroMQ socket to address: "
 										+ this.controlAddress);
@@ -306,7 +306,7 @@ public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAw
 	public synchronized void stop() {
 		if (this.running.getAndSet(false)) {
 			try (ZMQ.Socket commandSocket = this.context.createSocket(SocketType.PAIR)) {
-				commandSocket.connect(this.controlAddress);
+				commandSocket.connect(this.controlAddress); // NOSONAR
 				commandSocket.send(zmq.ZMQ.PROXY_TERMINATE);
 			}
 		}