From d34c45bfe44d2beab62fb1987f235fa7b882d3b2 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Thu, 17 Feb 2022 16:24:39 +0100 Subject: [PATCH] Polishing. Allow Jedis to exist pub/sub loop gracefully. Await unsubscribe. Reduce retry backoff to allow from broken connection recovery. Do not use pooling to avoid using broken connections. See #2256 --- .../RedisMessageListenerContainer.java | 97 ++++++++++++++----- .../JedisConnectionFactoryExtension.java | 2 +- ...sageListenerContainerIntegrationTests.java | 1 + ...edisMessageListenerContainerUnitTests.java | 4 +- 4 files changed, 76 insertions(+), 28 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java index e579cf712..b4b13849c 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java @@ -265,7 +265,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } try { - futureToAwait.get(getMaxSubscriptionRegistrationWaitingTime(), TimeUnit.SECONDS); + futureToAwait.get(getMaxSubscriptionRegistrationWaitingTime(), TimeUnit.MILLISECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } catch (ExecutionException e) { @@ -391,15 +391,13 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab return true; } - try { - listenFuture.join(); - } catch (Exception e) { - // ignore, just await completion here. - } + awaitRegistrationTime(listenFuture); if (this.state.compareAndSet(state, State.prepareUnsubscribe())) { - getRequiredSubscriber().cancel(); + getRequiredSubscriber().unsubscribeAll(); + + awaitRegistrationTime(this.unsubscribeFuture); this.state.set(State.notListening()); @@ -416,6 +414,16 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } } + private void awaitRegistrationTime(CompletableFuture future) { + try { + future.get(getMaxSubscriptionRegistrationWaitingTime(), TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (ExecutionException | TimeoutException e) { + // ignore + } + } + @Override public boolean isRunning() { return this.started.get(); @@ -876,7 +884,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab long recoveryInterval = backOffExecution.nextBackOff(); if (recoveryInterval != BackOffExecution.STOP) { - logger.error(String.format("Connection failure occurred: %s. Restarting subscription task after %s ms.", ex, recoveryInterval)); + logger.error(String.format("Connection failure occurred: %s. Restarting subscription task after %s ms.", ex, + recoveryInterval), ex); } return recoveryInterval; @@ -897,7 +906,9 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab return; } - logger.error("SubscriptionTask aborted with exception:", ex); + if (isRunning()) { // log only if the container is still running to prevent close errors from logging + logger.error("SubscriptionTask aborted with exception:", ex); + } future.completeExceptionally(ex); } @@ -1216,6 +1227,26 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab this.synchronizingMessageListener.addSynchronization(synchronizer); } + public void unsubscribeAll() { + + synchronized (localMonitor) { + + RedisConnection connection = this.connection; + if (connection == null) { + return; + } + + doUnsubscribe(connection); + } + } + + void doUnsubscribe(RedisConnection connection) { + closeSubscription(connection); + closeConnection(); + + unsubscribeFuture.complete(null); + } + /** * Cancel all subscriptions and close the connection. */ @@ -1228,26 +1259,34 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab return; } + doCancel(connection); + } + } + + void doCancel(RedisConnection connection) { + closeSubscription(connection); + closeConnection(); + } + + void closeSubscription(RedisConnection connection) { + + if (logger.isTraceEnabled()) { + logger.trace("Cancelling Redis subscription..."); + } + + Subscription sub = connection.getSubscription(); + + if (sub != null) { + if (logger.isTraceEnabled()) { - logger.trace("Cancelling Redis subscription..."); + logger.trace("Unsubscribing from all channels"); } - Subscription sub = connection.getSubscription(); - - if (sub != null) { - - if (logger.isTraceEnabled()) { - logger.trace("Unsubscribing from all channels"); - } - - try { - sub.close(); - } catch (Exception e) { - logger.warn("Unable to unsubscribe from subscriptions", e); - } + try { + sub.close(); + } catch (Exception e) { + logger.warn("Unable to unsubscribe from subscriptions", e); } - - closeConnection(); } } @@ -1324,6 +1363,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } } } + } /** @@ -1341,6 +1381,11 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab this.executor = executor; } + @Override + void doUnsubscribe(RedisConnection connection) { + closeSubscription(connection); // connection will be closed after exiting the doSubscribe method + } + @Override protected void eventuallyPerformSubscription(RedisConnection connection, BackOffExecution backOffExecution, CompletableFuture subscriptionDone, Collection patterns, Collection channels) { @@ -1369,6 +1414,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab try { doSubscribe(connection, patterns, initiallySubscribeToChannels); + closeConnection(); + unsubscribeFuture.complete(null); } catch (Throwable t) { handleSubscriptionException(subscriptionDone, backOffExecution, t); } diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/extension/JedisConnectionFactoryExtension.java b/src/test/java/org/springframework/data/redis/connection/jedis/extension/JedisConnectionFactoryExtension.java index d6269f7a9..e59479ceb 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/extension/JedisConnectionFactoryExtension.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/extension/JedisConnectionFactoryExtension.java @@ -56,7 +56,7 @@ public class JedisConnectionFactoryExtension implements ParameterResolver { .create(JedisConnectionFactoryExtension.class); private static final JedisClientConfiguration CLIENT_CONFIGURATION = JedisClientConfiguration.builder() - .clientName("jedis-client").usePooling().build(); + .clientName("jedis-client").build(); private static final NewableLazy STANDALONE = NewableLazy.of(() -> { diff --git a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerIntegrationTests.java index fff82ee9c..9685ea7b0 100644 --- a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerIntegrationTests.java @@ -63,6 +63,7 @@ class RedisMessageListenerContainerIntegrationTests { void setUp() { container = new RedisMessageListenerContainer(); + container.setRecoveryInterval(100); container.setConnectionFactory(connectionFactory); container.setBeanName("container"); container.afterPropertiesSet(); diff --git a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java index b7ec21aee..09031c276 100644 --- a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java +++ b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java @@ -81,7 +81,7 @@ class RedisMessageListenerContainerUnitTests { doAnswer(it -> { Runnable r = it.getArgument(0); - new Thread(r).start(); + r.run(); return null; }).when(executorMock).execute(any()); @@ -103,7 +103,7 @@ class RedisMessageListenerContainerUnitTests { container.stop(); assertThat(container.isRunning()).isFalse(); - verify(subscriptionMock).close(); + verify(connectionMock).close(); } @Test // GH-964