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

Author: mp911de

src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java

@@ -265,7 +265,7 @@ private void lazyListen() {
 		}
 
 		try {
-			futureToAwait.get(getMaxSubscriptionRegistrationWaitingTime(), TimeUnit.SECONDS);
+			futureToAwait.get(getMaxSubscriptionRegistrationWaitingTime(), TimeUnit.MILLISECONDS);
 		} catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
 		} catch (ExecutionException e) {
@@ -391,15 +391,13 @@ private boolean doUnsubscribe() {
 			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 @@ private boolean doUnsubscribe() {
 		}
 	}
 
+	private void awaitRegistrationTime(CompletableFuture<Void> 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 @@ protected void handleSubscriptionException(CompletableFuture<Void> future, BackO
 				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 @@ protected void handleSubscriptionException(CompletableFuture<Void> future, BackO
 			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 @@ void addSynchronization(SynchronizingMessageListener.SubscriptionSynchronizion s
 			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 void cancel() {
 					return;
 				}
 
-				if (logger.isTraceEnabled()) {
-					logger.trace("Cancelling Redis subscription...");
-				}
+				doCancel(connection);
+			}
+		}
 
-				Subscription sub = connection.getSubscription();
+		void doCancel(RedisConnection connection) {
+			closeSubscription(connection);
+			closeConnection();
+		}
 
-				if (sub != null) {
+		void closeSubscription(RedisConnection connection) {
 
-					if (logger.isTraceEnabled()) {
-						logger.trace("Unsubscribing from all channels");
-					}
+			if (logger.isTraceEnabled()) {
+				logger.trace("Cancelling Redis subscription...");
+			}
 
-					try {
-						sub.close();
-					} catch (Exception e) {
-						logger.warn("Unable to unsubscribe from subscriptions", e);
-					}
+			Subscription sub = connection.getSubscription();
+
+			if (sub != null) {
+
+				if (logger.isTraceEnabled()) {
+					logger.trace("Unsubscribing from all channels");
 				}
 
-				closeConnection();
+				try {
+					sub.close();
+				} catch (Exception e) {
+					logger.warn("Unable to unsubscribe from subscriptions", e);
+				}
 			}
 		}
 
@@ -1324,6 +1363,7 @@ private void doWithSubscription(byte[][] data, BiConsumer<Subscription, byte[][]
 				}
 			}
 		}
+
 	}
 
 	/**
@@ -1341,6 +1381,11 @@ class BlockingSubscriber extends Subscriber {
 			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<Void> subscriptionDone, Collection<byte[]> patterns, Collection<byte[]> channels) {
@@ -1369,6 +1414,8 @@ protected void eventuallyPerformSubscription(RedisConnection connection, BackOff
 
 				try {
 					doSubscribe(connection, patterns, initiallySubscribeToChannels);
+					closeConnection();
+					unsubscribeFuture.complete(null);
 				} catch (Throwable t) {
 					handleSubscriptionException(subscriptionDone, backOffExecution, t);
 				}

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<JedisConnectionFactory> STANDALONE = NewableLazy.of(() -> {
 

src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerIntegrationTests.java

@@ -63,6 +63,7 @@ public RedisMessageListenerContainerIntegrationTests(RedisConnectionFactory conn
 	void setUp() {
 
 		container = new RedisMessageListenerContainer();
+		container.setRecoveryInterval(100);
 		container.setConnectionFactory(connectionFactory);
 		container.setBeanName("container");
 		container.afterPropertiesSet();

src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java

@@ -81,7 +81,7 @@ void containerShouldStopGracefullyOnUnsubscribeErrors() {
 		doAnswer(it -> {
 
 			Runnable r = it.getArgument(0);
-			new Thread(r).start();
+			r.run();
 			return null;
 		}).when(executorMock).execute(any());
 
@@ -103,7 +103,7 @@ void containerShouldStopGracefullyOnUnsubscribeErrors() {
 		container.stop();
 
 		assertThat(container.isRunning()).isFalse();
-		verify(subscriptionMock).close();
+		verify(connectionMock).close();
 	}
 
 	@Test // GH-964