Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix race condition for FluxMessageChannelTests

Browse Source 
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
This commit is contained in:
Artem Bilan
2021-02-08 10:33:01 -05:00
parent 338a26d34d
commit d55d8402d9
4 changed files with 49 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java
View File

@@ -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<S extends CorrelationHandlerSpec<S,
* @return the endpoint spec.
* @since 5.4
* @see AbstractCorrelatingMessageHandler#setExpireTimeout(long)
* @deprecated since 5.5 in favor of {@link #expireTimeout(long)}
*/
@Deprecated
public S setExpireTimeout(long expireTimeout) {
return expireTimeout(expireTimeout);
}
/**
* Configure a timeout for old groups in the store to purge.
* @param expireTimeout the timeout in milliseconds to use.
* @return the endpoint spec.
* @since 5.5
* @see AbstractCorrelatingMessageHandler#setExpireTimeout(long)
*/
public S expireTimeout(long expireTimeout) {
this.handler.setExpireTimeout(expireTimeout);
return _this();
}
@@ -348,10 +361,37 @@ public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S,
* @return the endpoint spec.
* @since 5.4
* @see AbstractCorrelatingMessageHandler#setExpireDuration(Duration)
* @deprecated since 5.5 in favor of {@link #expireDuration(Duration)}
*/
@Deprecated
public S setExpireDuration(Duration expireDuration) {
return expireDuration(expireDuration);
}
/**
* Configure a {@link Duration} how often to run a scheduled purge task.
* @param expireDuration the duration for scheduled purge task.
* @return the endpoint spec.
* @since 5.5
* @see AbstractCorrelatingMessageHandler#setExpireDuration(Duration)
*/
public S expireDuration(Duration expireDuration) {
this.handler.setExpireDuration(expireDuration);
return _this();
}
/**
* Set to true to release the message group lock before sending any output. See
* "Avoiding Deadlocks" in the Aggregator section of the reference manual for more
* information as to why this might be needed.
* @param releaseLockBeforeSend true to release the lock.
* @return the endpoint spec.
* @since 5.5
* @see AbstractCorrelatingMessageHandler#setReleaseLockBeforeSend(boolean)
*/
public S releaseLockBeforeSend(boolean releaseLockBeforeSend) {
this.handler.setReleaseLockBeforeSend(releaseLockBeforeSend);
return _this();
}
}

4
spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java
View File

@@ -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.
@@ -127,7 +127,7 @@ public class FluxMessageChannelTests {
IntegrationFlow testFlow = f -> f
.<String>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 =