You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2022/04/20 15:13:38 UTC
[james-project] 01/06: JAMES-3747 Reactive implementation for RabbitMQ channelPool
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 8564a2ecf72e0a7ef404fe491799fb8140dea1f6
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu Apr 14 11:29:33 2022 +0700
JAMES-3747 Reactive implementation for RabbitMQ channelPool
---
backends-common/rabbitmq/pom.xml | 8 +-
.../rabbitmq/ReactorRabbitMQChannelPool.java | 182 +++++++--------------
.../backends/rabbitmq/ChannelPoolContract.java | 10 +-
.../rabbitmq/ReactorRabbitMQChannelPoolTest.java | 2 +-
event-bus/distributed/pom.xml | 4 -
.../org/apache/james/events/NetworkErrorTest.java | 6 +-
pom.xml | 5 -
.../modules/ROOT/pages/configure/rabbitmq.adoc | 6 +-
.../sample-configuration/rabbitmq.properties | 6 +-
.../sample-configuration/rabbitmq.properties | 6 +-
src/site/xdoc/server/config-rabbitmq.xml | 6 +-
11 files changed, 88 insertions(+), 153 deletions(-)
diff --git a/backends-common/rabbitmq/pom.xml b/backends-common/rabbitmq/pom.xml
index 8325eb8dee..d65cb6f8a1 100644
--- a/backends-common/rabbitmq/pom.xml
+++ b/backends-common/rabbitmq/pom.xml
@@ -73,6 +73,10 @@
<groupId>io.github.openfeign</groupId>
<artifactId>feign-slf4j</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.projectreactor.addons</groupId>
+ <artifactId>reactor-pool</artifactId>
+ </dependency>
<dependency>
<groupId>io.projectreactor.rabbitmq</groupId>
<artifactId>reactor-rabbitmq</artifactId>
@@ -89,10 +93,6 @@
<groupId>org.apache.commons</groupId>
<artifactId>commons-configuration2</artifactId>
</dependency>
- <dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-pool2</artifactId>
- </dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPool.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPool.java
index a98d11fa91..6c33dc77f1 100644
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPool.java
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPool.java
@@ -21,28 +21,24 @@ package org.apache.james.backends.rabbitmq;
import java.io.IOException;
import java.time.Duration;
-import java.util.Comparator;
import java.util.Map;
+import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import javax.annotation.PreDestroy;
-import org.apache.commons.pool2.BasePooledObjectFactory;
-import org.apache.commons.pool2.PooledObject;
-import org.apache.commons.pool2.impl.DefaultPooledObject;
-import org.apache.commons.pool2.impl.GenericObjectPool;
-import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.james.lifecycle.api.Startable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.github.fge.lambdas.Throwing;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
@@ -65,6 +61,9 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.core.scheduler.Schedulers;
+import reactor.pool.InstrumentedPool;
+import reactor.pool.PoolBuilder;
+import reactor.pool.PooledRef;
import reactor.rabbitmq.BindingSpecification;
import reactor.rabbitmq.ChannelPool;
import reactor.rabbitmq.QueueSpecification;
@@ -596,54 +595,6 @@ public class ReactorRabbitMQChannelPool implements ChannelPool, Startable {
}
}
- private static class ChannelClosedException extends IOException {
- ChannelClosedException(String message) {
- super(message);
- }
- }
-
- static class ChannelFactory extends BasePooledObjectFactory<Channel> {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(ChannelFactory.class);
-
- private final Mono<Connection> connectionMono;
- private final Configuration configuration;
-
- ChannelFactory(Mono<Connection> connectionMono, Configuration configuration) {
- this.connectionMono = connectionMono;
- this.configuration = configuration;
- }
-
- @Override
- public Channel create() {
- return connectionMono
- .flatMap(this::openChannel)
- .block();
- }
-
- private Mono<? extends Channel> openChannel(Connection connection) {
- return Mono.fromCallable(connection::openChannel)
- .map(maybeChannel ->
- maybeChannel.orElseThrow(() -> new RuntimeException("RabbitMQ reached to maximum opened channels, cannot get more channels")))
- .map(SelectOnceChannel::new)
- .retryWhen(configuration.backoffSpec().scheduler(Schedulers.elastic()))
- .doOnError(throwable -> LOGGER.error("error when creating new channel", throwable));
- }
-
- @Override
- public PooledObject<Channel> wrap(Channel obj) {
- return new DefaultPooledObject<>(obj);
- }
-
- @Override
- public void destroyObject(PooledObject<Channel> pooledObject) throws Exception {
- Channel channel = pooledObject.getObject();
- if (channel.isOpen()) {
- channel.close();
- }
- }
- }
-
public static class Configuration {
@FunctionalInterface
public interface RequiresRetries {
@@ -662,7 +613,7 @@ public class ReactorRabbitMQChannelPool implements ChannelPool, Startable {
public static final Configuration DEFAULT = builder()
.retries(MAX_BORROW_RETRIES)
- .minBorrowDelay(MIN_BORROW_DELAY)
+ .minBorrowDelay(MAX_BORROW_DELAY)
.maxChannel(MAX_CHANNELS_NUMBER);
public static RequiresRetries builder() {
@@ -670,28 +621,28 @@ public class ReactorRabbitMQChannelPool implements ChannelPool, Startable {
}
public static Configuration from(org.apache.commons.configuration2.Configuration configuration) {
- Duration minBorrowDelay = Optional.ofNullable(configuration.getLong("channel.pool.min.delay.ms", null))
+ Duration maxBorrowDelay = Optional.ofNullable(configuration.getLong("channel.pool.max.delay.ms", null))
.map(Duration::ofMillis)
- .orElse(MIN_BORROW_DELAY);
+ .orElse(MAX_BORROW_DELAY);
return builder()
.retries(configuration.getInt("channel.pool.retries", MAX_BORROW_RETRIES))
- .minBorrowDelay(minBorrowDelay)
+ .minBorrowDelay(maxBorrowDelay)
.maxChannel(configuration.getInt("channel.pool.size", MAX_CHANNELS_NUMBER));
}
- private final Duration minBorrowDelay;
+ private final Duration maxBorrowDelay;
private final int retries;
private final int maxChannel;
- public Configuration(Duration minBorrowDelay, int retries, int maxChannel) {
- this.minBorrowDelay = minBorrowDelay;
+ public Configuration(Duration maxBorrowDelay, int retries, int maxChannel) {
+ this.maxBorrowDelay = maxBorrowDelay;
this.retries = retries;
this.maxChannel = maxChannel;
}
private RetryBackoffSpec backoffSpec() {
- return Retry.backoff(retries, minBorrowDelay);
+ return Retry.backoff(retries, maxBorrowDelay);
}
public int getMaxChannel() {
@@ -700,26 +651,45 @@ public class ReactorRabbitMQChannelPool implements ChannelPool, Startable {
}
private static final Logger LOGGER = LoggerFactory.getLogger(ReactorRabbitMQChannelPool.class);
- private static final long MAXIMUM_BORROW_TIMEOUT_IN_MS = Duration.ofSeconds(5).toMillis();
private static final int MAX_CHANNELS_NUMBER = 3;
private static final int MAX_BORROW_RETRIES = 3;
- private static final Duration MIN_BORROW_DELAY = Duration.ofMillis(50);
+ private static final Duration MAX_BORROW_DELAY = Duration.ofSeconds(30);
+ private final ConcurrentHashMap<Integer, PooledRef<? extends Channel>> refs = new ConcurrentHashMap<>();
private final Mono<Connection> connectionMono;
- private final GenericObjectPool<Channel> pool;
- private final ConcurrentSkipListSet<Channel> borrowedChannels;
private final Configuration configuration;
+
+ private final InstrumentedPool<? extends Channel> newPool;
private Sender sender;
public ReactorRabbitMQChannelPool(Mono<Connection> connectionMono, Configuration configuration) {
this.connectionMono = connectionMono;
this.configuration = configuration;
- ChannelFactory channelFactory = new ChannelFactory(connectionMono, configuration);
- GenericObjectPoolConfig<Channel> config = new GenericObjectPoolConfig<>();
- config.setMaxTotal(configuration.getMaxChannel());
- this.pool = new GenericObjectPool<>(channelFactory, config);
- this.borrowedChannels = new ConcurrentSkipListSet<>(Comparator.comparingInt(System::identityHashCode));
+ newPool = PoolBuilder.from(connectionMono
+ .flatMap(this::openChannel))
+ .sizeBetween(1, configuration.maxChannel)
+ .acquisitionScheduler(Schedulers.elastic())
+ .evictionPredicate((channel, metadata) -> {
+ if (!channel.isOpen()) {
+ return true;
+ }
+ if (metadata.idleTime() > Duration.ofSeconds(30).toMillis()) {
+ return true;
+ }
+ return false;
+ })
+ .destroyHandler(channel -> Mono.fromRunnable(Throwing.runnable(channel::close)))
+ .buildPool();
+ }
+
+ private Mono<? extends Channel> openChannel(Connection connection) {
+ return Mono.fromCallable(connection::openChannel)
+ .map(maybeChannel ->
+ maybeChannel.orElseThrow(() -> new RuntimeException("RabbitMQ reached to maximum opened channels, cannot get more channels")))
+ .map(SelectOnceChannel::new)
+ .retryWhen(configuration.backoffSpec().scheduler(Schedulers.elastic()))
+ .doOnError(throwable -> LOGGER.error("error when creating new channel", throwable));
}
public void start() {
@@ -739,46 +709,26 @@ public class ReactorRabbitMQChannelPool implements ChannelPool, Startable {
return borrow();
}
- private Mono<Channel> borrow() {
- return tryBorrowFromPool()
- .doOnError(throwable -> LOGGER.warn("Cannot borrow channel", throwable))
- .retryWhen(configuration.backoffSpec().scheduler(Schedulers.elastic()))
- .onErrorMap(this::propagateException)
- .subscribeOn(Schedulers.elastic())
- .doOnNext(borrowedChannels::add);
- }
-
- private Mono<Channel> tryBorrowFromPool() {
- return Mono.fromCallable(this::borrowFromPool);
- }
-
- private Throwable propagateException(Throwable throwable) {
- if (throwable instanceof IllegalStateException
- && throwable.getMessage().contains("Retries exhausted")) {
- return throwable.getCause();
- }
-
- return throwable;
- }
-
- private Channel borrowFromPool() throws Exception {
- Channel channel = pool.borrowObject(MAXIMUM_BORROW_TIMEOUT_IN_MS);
- if (!channel.isOpen()) {
- invalidateObject(channel);
- throw new ChannelClosedException("borrowed channel is already closed");
- }
- return channel;
+ private Mono<? extends Channel> borrow() {
+ return newPool.acquire()
+ .timeout(configuration.maxBorrowDelay)
+ .doOnNext(ref -> refs.put(ref.poolable().getChannelNumber(), ref))
+ .map(PooledRef::poolable)
+ .onErrorMap(TimeoutException.class, e -> new NoSuchElementException("Timeout waiting for idle object"));
}
@Override
public BiConsumer<SignalType, Channel> getChannelCloseHandler() {
return (signalType, channel) -> {
- borrowedChannels.remove(channel);
+ PooledRef<? extends Channel> pooledRef = refs.remove(channel.getChannelNumber());
+
if (!channel.isOpen() || !executeWithoutError(signalType)) {
- invalidateObject(channel);
+ pooledRef.invalidate()
+ .block();
return;
}
- pool.returnObject(channel);
+ pooledRef.release()
+ .block();
};
}
@@ -836,24 +786,16 @@ public class ReactorRabbitMQChannelPool implements ChannelPool, Startable {
.then();
}
- private void invalidateObject(Channel channel) {
- try {
- pool.invalidateObject(channel);
- if (channel.isOpen()) {
- channel.close();
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
@PreDestroy
@Override
public void close() {
sender.close();
- borrowedChannels.forEach(channel -> getChannelCloseHandler().accept(SignalType.ON_NEXT, channel));
- borrowedChannels.clear();
- pool.close();
+ ImmutableList.copyOf(refs.values())
+ .stream()
+ .map(PooledRef::poolable)
+ .forEach(channel -> getChannelCloseHandler().accept(SignalType.ON_NEXT, channel));
+ refs.clear();
+ newPool.dispose();
}
public Mono<Boolean> tryChannel() {
@@ -861,8 +803,8 @@ public class ReactorRabbitMQChannelPool implements ChannelPool, Startable {
channel -> Mono.just(channel.isOpen()),
channel -> {
if (channel != null) {
- borrowedChannels.remove(channel);
- pool.returnObject(channel);
+ PooledRef<? extends Channel> pooledRef = refs.remove(channel.getChannelNumber());
+ return pooledRef.release();
}
return Mono.empty();
})
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/ChannelPoolContract.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/ChannelPoolContract.java
index 4fde3af687..195edb9892 100644
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/ChannelPoolContract.java
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/ChannelPoolContract.java
@@ -31,6 +31,7 @@ import com.rabbitmq.client.Channel;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
+import reactor.core.scheduler.Schedulers;
import reactor.rabbitmq.ChannelPool;
interface ChannelPoolContract {
@@ -64,17 +65,16 @@ interface ChannelPoolContract {
@Test
default void channelPoolShouldWaitTillTheNextReleaseWhenAllChannelsAreTaken() {
ChannelPool channelPool = getChannelPool(2);
- channelPool.getChannelMono();
- borrowChannel(channelPool);
+ Channel channel1 = borrowChannel(channelPool);
Channel channel2 = borrowChannel(channelPool);
- Mono.delay(Duration.ofSeconds(2))
- .doOnSuccess(any -> returnToThePool(channelPool, channel2))
+ Mono.delay(Duration.ofSeconds(1))
+ .then(Mono.fromRunnable(() -> returnToThePool(channelPool, channel1)).subscribeOn(Schedulers.elastic()))
.subscribe();
Channel channelAfterReturned = borrowChannel(channelPool);
assertThat(channelAfterReturned.getChannelNumber())
- .isEqualTo(channel2.getChannelNumber());
+ .isEqualTo(channel1.getChannelNumber());
}
@Test
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPoolTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPoolTest.java
index 0f259c340a..4744115517 100644
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPoolTest.java
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPoolTest.java
@@ -71,7 +71,7 @@ class ReactorRabbitMQChannelPoolTest implements ChannelPoolContract {
rabbitMQExtension.getConnectionPool().getResilientConnection(),
ReactorRabbitMQChannelPool.Configuration.builder()
.retries(2)
- .minBorrowDelay(Duration.ofMillis(5))
+ .minBorrowDelay(Duration.ofSeconds(2))
.maxChannel(poolSize));
reactorRabbitMQChannelPool.start();
return reactorRabbitMQChannelPool;
diff --git a/event-bus/distributed/pom.xml b/event-bus/distributed/pom.xml
index ab2af3aa3e..74fde0bdef 100644
--- a/event-bus/distributed/pom.xml
+++ b/event-bus/distributed/pom.xml
@@ -86,10 +86,6 @@
<groupId>io.projectreactor.rabbitmq</groupId>
<artifactId>reactor-rabbitmq</artifactId>
</dependency>
- <dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-pool2</artifactId>
- </dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
diff --git a/event-bus/distributed/src/test/java/org/apache/james/events/NetworkErrorTest.java b/event-bus/distributed/src/test/java/org/apache/james/events/NetworkErrorTest.java
index 795fc12e60..58d4ddaf04 100644
--- a/event-bus/distributed/src/test/java/org/apache/james/events/NetworkErrorTest.java
+++ b/event-bus/distributed/src/test/java/org/apache/james/events/NetworkErrorTest.java
@@ -27,6 +27,8 @@ import static org.apache.james.events.EventBusTestFixture.newListener;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.verify;
+import java.util.NoSuchElementException;
+
import org.apache.james.backends.rabbitmq.RabbitMQExtension;
import org.apache.james.backends.rabbitmq.RabbitMQFixture;
import org.apache.james.events.EventBusTestFixture.TestEventSerializer;
@@ -72,8 +74,8 @@ class NetworkErrorTest {
rabbitMQExtension.getRabbitMQ().pause();
assertThatThrownBy(() -> eventBus.dispatch(EVENT, NO_KEYS).block())
- .isInstanceOf(IllegalStateException.class)
- .hasMessageContaining("Retries exhausted");
+ .isInstanceOf(NoSuchElementException.class)
+ .hasMessageContaining("Timeout waiting for idle object");
rabbitMQExtension.getRabbitMQ().unpause();
diff --git a/pom.xml b/pom.xml
index f4e9085fde..30e177d2ef 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2500,11 +2500,6 @@
<artifactId>commons-lang3</artifactId>
<version>3.12.0</version>
</dependency>
- <dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-pool2</artifactId>
- <version>2.11.1</version>
- </dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-text</artifactId>
diff --git a/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc b/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc
index 5b91a0d935..540d4b2173 100644
--- a/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc
+++ b/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc
@@ -38,9 +38,9 @@ Optional integer, defaults to 100
| Configure retries count to retrieve a channel. Exponential backoff is performed between each retries.
Optional integer, defaults to 3
-| channel.pool.min.delay.ms
-| Configure initial duration (in ms) between two channel retries. Exponential backoff is performed between each retries.
-Optional integer, defaults to 50
+| channel.pool.max.delay.ms
+| Configure timeout duration (in ms) to obtain a rabbitmq channel. Defaults to 30 seconds.
+Optional integer, defaults to 30 seconds.
| channel.pool.size
| Configure the size of the channel pool.
diff --git a/server/apps/distributed-app/sample-configuration/rabbitmq.properties b/server/apps/distributed-app/sample-configuration/rabbitmq.properties
index b412d1d354..b63acd9d85 100644
--- a/server/apps/distributed-app/sample-configuration/rabbitmq.properties
+++ b/server/apps/distributed-app/sample-configuration/rabbitmq.properties
@@ -26,9 +26,9 @@ management.password=guest
# Configure retries count to retrieve a channel. Exponential backoff is performed between each retries.
# Optional integer, defaults to 3
#channel.pool.retries=3
-# Configure initial duration (in ms) between two channel retries. Exponential backoff is performed between each retries.
-# Optional integer, defaults to 50
-#channel.pool.min.delay.ms=50
+# Configure timeout duration (in ms) to obtain a rabbitmq channel. Defaults to 30 seconds.
+# Optional integer, defaults to 30 seconds.
+#channel.pool.max.delay.ms=30000
# Configure the size of the channel pool.
# Optional integer, defaults to 3
#channel.pool.size=3
diff --git a/server/apps/distributed-pop3-app/sample-configuration/rabbitmq.properties b/server/apps/distributed-pop3-app/sample-configuration/rabbitmq.properties
index 4b7a2caab8..d587cb647e 100644
--- a/server/apps/distributed-pop3-app/sample-configuration/rabbitmq.properties
+++ b/server/apps/distributed-pop3-app/sample-configuration/rabbitmq.properties
@@ -26,9 +26,9 @@ management.password=guest
# Configure retries count to retrieve a channel. Exponential backoff is performed between each retries.
# Optional integer, defaults to 3
#channel.pool.retries=3
-# Configure initial duration (in ms) between two channel retries. Exponential backoff is performed between each retries.
-# Optional integer, defaults to 50
-#channel.pool.min.delay.ms=50
+# Configure timeout duration (in ms) to obtain a rabbitmq channel. Defaults to 30 seconds.
+# Optional integer, defaults to 30 seconds.
+#channel.pool.max.delay.ms=30000
# Configure the size of the channel pool.
# Optional integer, defaults to 3
#channel.pool.size=3
diff --git a/src/site/xdoc/server/config-rabbitmq.xml b/src/site/xdoc/server/config-rabbitmq.xml
index 4aa279ae86..17d5781a88 100644
--- a/src/site/xdoc/server/config-rabbitmq.xml
+++ b/src/site/xdoc/server/config-rabbitmq.xml
@@ -70,9 +70,9 @@
<dd>Configure retries count to retrieve a channel. Exponential backoff is performed between each retries.
Optional integer, defaults to 3</dd>
- <dt><strong>channel.pool.min.delay.ms</strong></dt>
- <dd>Configure initial duration (in ms) between two channel retries. Exponential backoff is performed between each retries.
- Optional integer, defaults to 50</dd>
+ <dt><strong>channel.pool.max.delay.ms</strong></dt>
+ <dd>Configure timeout duration (in ms) to obtain a rabbitmq channel. Defaults to 30 seconds.
+ Optional integer, defaults to 30 seconds.</dd>
<dt><strong>channel.pool.size</strong></dt>
<dd>Configure the size of the channel pool.
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org