From d55d8402d9dec98766332be05f4effdafd8ece12 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 8 Feb 2021 10:33:01 -0500 Subject: [PATCH] Fix race condition for FluxMessageChannelTests The logic in the `FluxMessageChannelTests.testFluxMessageChannelCleanUp()` relies on the `finishLatch.countDown()` which happens before we release the lock for a group in the aggregator. So, when we call a `destroy()` as the next statement, we interrupt the lock instead of unlocking. This cause a race condition in the Reactor's `Sink` to terminate properly * Expose `CorrelationHandlerSpec.releaseLockBeforeSend()` and set it to `true` in the test to properly unlock before we call `finishLatch.countDown()` * Deprecate a couple introduced before options in favor of their variants with `set` prefix * Fix new Sonar smells --- .../dsl/CorrelationHandlerSpec.java | 42 ++++++++++++++++++- .../reactive/FluxMessageChannelTests.java | 4 +- .../file/filters/CompositeFileListFilter.java | 7 +--- .../AbstractRemoteFileOutboundGateway.java | 10 ++--- 4 files changed, 49 insertions(+), 14 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java index 7a9295c206..0c4a259e21 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.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. @@ -336,8 +336,21 @@ public abstract class CorrelationHandlerSpec f .split(__ -> Flux.fromStream(IntStream.range(0, 100).boxed()), null) .channel(flux) - .aggregate(a -> a.releaseStrategy(m -> m.size() == 100)) + .aggregate(a -> a.releaseStrategy(m -> m.size() == 100).releaseLockBeforeSend(true)) .handle(__ -> finishLatch.countDown()); IntegrationFlowContext.IntegrationFlowRegistration flowRegistration = diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java index 7019884521..3eacbcc72a 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java @@ -100,14 +100,11 @@ public class CompositeFileListFilter } /** - * Not thread safe. Only a single thread may add filters at a time. - *

* Add the new filters to this CompositeFileListFilter while maintaining the existing filters. - * * @param filtersToAdd a list of filters to add * @return this CompositeFileListFilter instance with the added filters */ - public CompositeFileListFilter addFilters(Collection> filtersToAdd) { + public synchronized CompositeFileListFilter addFilters(Collection> filtersToAdd) { for (FileListFilter elf : filtersToAdd) { if (elf instanceof DiscardAwareFileListFilter) { ((DiscardAwareFileListFilter) elf).addDiscardCallback(this.discardCallback); @@ -120,7 +117,7 @@ public class CompositeFileListFilter throw new IllegalStateException(e); } } - this.allSupportAccept &= elf.supportsSingleFileFiltering(); + this.allSupportAccept = this.allSupportAccept && elf.supportsSingleFileFiltering(); this.oneIsForRecursion |= elf.isForRecursion(); } this.fileFilters.addAll(filtersToAdd); 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 638e8b7c62..862450fc47 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 @@ -1005,10 +1005,8 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply if (recursion && StringUtils.hasText(subDirectory)) { fileToAdd = enhanceNameWithSubDirectory(file, subDirectory); } - if (this.filterAfterEnhancement) { - if (!this.filter.accept(fileToAdd)) { - return; - } + if (this.filterAfterEnhancement && !this.filter.accept(fileToAdd)) { + return; } String fileName = getFilename(fileToAdd); final boolean isDirectory = isDirectory(file); @@ -1020,8 +1018,8 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply } if (recursion && isDirectory && !isDots) { - lsFiles.addAll(listFilesInRemoteDir(session, directory, fileName + - this.remoteFileTemplate.getRemoteFileSeparator())); + lsFiles.addAll(listFilesInRemoteDir(session, directory, + fileName + this.remoteFileTemplate.getRemoteFileSeparator())); } }