From 8763c956223cb076b5ab6aab40e7842b11d44b9b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 21 Jan 2021 14:55:14 -0500 Subject: [PATCH] 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 --- .../amqp/dsl/AmqpBaseInboundGatewaySpec.java | 5 +- .../integration/handler/DelayHandler.java | 2 +- .../JsonNodeWrapperToJsonNodeConverter.java | 15 ++++-- .../json/JsonPropertyAccessor.java | 16 +++---- .../integration/store/MessageGroupQueue.java | 2 +- .../leader/LockRegistryLeaderInitiator.java | 24 +++++----- .../file/FileWritingMessageHandler.java | 6 +-- .../file/RecursiveDirectoryScanner.java | 7 ++- .../AbstractRemoteFileOutboundGateway.java | 47 +++++++++---------- .../AbstractFilePayloadTransformer.java | 30 ++++++------ .../ftp/gateway/FtpOutboundGateway.java | 30 +++++++----- .../ip/tcp/connection/TcpNioConnection.java | 34 ++++++-------- .../ReactiveRedisStreamMessageProducer.java | 3 +- ...RedisLockRegistryLeaderInitiatorTests.java | 4 +- .../integration/test/util/TestUtils.java | 14 +++--- .../ws/AbstractWebServiceInboundGateway.java | 4 +- .../integration/zeromq/ZeroMqProxy.java | 10 ++-- 17 files changed, 121 insertions(+), 132 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.java index 4edfee473b..25b1c2b111 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.java +++ b/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> * @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); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 0a5273a823..8c0006a0bd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/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> 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. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/json/JsonNodeWrapperToJsonNodeConverter.java b/spring-integration-core/src/main/java/org/springframework/integration/json/JsonNodeWrapperToJsonNodeConverter.java index 75daf74c4f..07ec545692 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/json/JsonNodeWrapperToJsonNodeConverter.java +++ b/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; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/json/JsonPropertyAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/json/JsonPropertyAccessor.java index 3064ddd4f5..a8ff685da5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/json/JsonPropertyAccessor.java +++ b/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); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java index 9b1a4e5499..4205b42700 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java @@ -169,7 +169,7 @@ public class MessageGroupQueue extends AbstractQueue> implements Bloc try { this.storeLock.lockInterruptibly(); try (Stream> messageStream = stream()) { - return messageStream.findFirst().orElse(null); + return messageStream.findFirst().orElse(null); // NOSONAR } finally { this.storeLock.unlock(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java b/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java index ef98dd73b1..1b59060102 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java +++ b/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. diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index 712f6e8c09..841893cdbc 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/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) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveDirectoryScanner.java b/spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveDirectoryScanner.java index b3754c8837..9fcefcc8c8 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveDirectoryScanner.java +++ b/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 pathStream = Files.walk(directory.toPath(), this.maxDepth, this.fileVisitOptions);) { Stream fileStream = pathStream - .skip(1) + .skip(1) // NOSONAR .map(Path::toFile) - .filter(file -> !supportAcceptFilter - || ((AbstractFileListFilter) filter).accept(file)); + .filter(file -> !supportAcceptFilter || filter.accept(file)); if (supportAcceptFilter) { return fileStream.collect(Collectors.toList()); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index 6af89e7e52..5ca454ec3b 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/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 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 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 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 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 { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/transformer/AbstractFilePayloadTransformer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/transformer/AbstractFilePayloadTransformer.java index c98f616431..bd27cda6b7 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/transformer/AbstractFilePayloadTransformer.java +++ b/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 implements Transformer, /** * Specify whether to delete the File after transformation. - * Default is false. - * + * Default is {@code false}. * @param deleteFiles true to delete the file. */ public void setDeleteFiles(boolean deleteFiles) { @@ -85,27 +84,26 @@ public abstract class AbstractFilePayloadTransformer 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. diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/gateway/FtpOutboundGateway.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/gateway/FtpOutboundGateway.java index 9ea8305183..18d55ef6f5 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/gateway/FtpOutboundGateway.java +++ b/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 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; } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java index a45968e10b..3d925a4ec0 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java +++ b/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. diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java index 717ca886ba..e82127ec9c 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java +++ b/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); diff --git a/spring-integration-test-support/src/main/java/org/springframework/integration/test/util/TestUtils.java b/spring-integration-test-support/src/main/java/org/springframework/integration/test/util/TestUtils.java index 851a1ba611..58e43a9e52 100644 --- a/spring-integration-test-support/src/main/java/org/springframework/integration/test/util/TestUtils.java +++ b/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); } } } diff --git a/spring-integration-ws/src/main/java/org/springframework/integration/ws/AbstractWebServiceInboundGateway.java b/spring-integration-ws/src/main/java/org/springframework/integration/ws/AbstractWebServiceInboundGateway.java index b042e1782b..64fbef299e 100644 --- a/spring-integration-ws/src/main/java/org/springframework/integration/ws/AbstractWebServiceInboundGateway.java +++ b/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(); } diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java index 727c8f9650..f38f48eeb5 100644 --- a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java +++ b/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); } }