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

GH-2907: Use CF.closeTimeout for confirms wait

Browse Source 
Fixes: #2907

Issue link: https://github.com/spring-projects/spring-amqp/issues/2907

The current hard-coded `5 seconds` is not enough in real applications under heavy load

* Fix `CachingConnectionFactory` to use `getCloseTimeout()` for `publisherCallbackChannel.waitForConfirms()`
which is `30 seconds` by default, but can be modified via `CachingConnectionFactory.setCloseTimeout()`

(cherry picked from commit 562bc772c4)
This commit is contained in:
Artem Bilan
2024-11-18 14:49:30 -05:00
committed by Spring Builds
parent f366cc3073
commit d2e5a0dfdc
2 changed files with 6 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java
View File

@@ -478,8 +478,9 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
}
/**
* How long to wait (milliseconds) for a response to a connection close operation from the broker; default 30000 (30
* seconds).
* How long to wait (milliseconds) for a response to a connection close operation from the broker;
* default 30000 (30 seconds).
* Also used for {@link com.rabbitmq.client.Channel#waitForConfirms()}.
* @param closeTimeout the closeTimeout to set.
*/
public void setCloseTimeout(int closeTimeout) {

8
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
View File

@@ -1087,8 +1087,6 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
private final class CachedChannelInvocationHandler implements InvocationHandler {
private static final int ASYNC_CLOSE_TIMEOUT = 5_000;
private final ChannelCachingConnectionProxy theConnection;
private final Deque<ChannelProxy> channelList;
@@ -1302,7 +1300,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
getChannelsExecutor()
.execute(() -> {
try {
publisherCallbackChannel.waitForConfirms(ASYNC_CLOSE_TIMEOUT);
publisherCallbackChannel.waitForConfirms(getCloseTimeout());
}
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
@@ -1426,10 +1424,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
executorService.execute(() -> {
try {
if (ConfirmType.CORRELATED.equals(CachingConnectionFactory.this.confirmType)) {
channel.waitForConfirmsOrDie(ASYNC_CLOSE_TIMEOUT);
channel.waitForConfirmsOrDie(getCloseTimeout());
}
else {
Thread.sleep(ASYNC_CLOSE_TIMEOUT);
Thread.sleep(5_000); // NOSONAR - some time to give the channel a chance to ack
}
}
catch (@SuppressWarnings(UNUSED) InterruptedException e1) {