Improve thread safety of RedisMessageListenerContainer

RedisMessageListenerContainer relies on 2 threads for subscription when patterns and channels topics are present. With Jedis, since the subscription thread blocks while listening for messages, an additional thread is used to subscribe to patterns while the subscription threads subscribe to channels and block.

There were some race conditions between those two threads that could corrupt the Jedis stream since operations are not synchronized in JedisSubscription. A lock on the JedisSubscription instance has been added  to enforce that operations on the Jedis stream cannot be affected by a concurrent thread.

Additionaly, there were no error handling and retry mechanism on the pattern subscription thread. Multiple conditions could trigger an unexpected behavior here, exceptions were not handled and logged to stderr with no notice. Also, if the connection was not subscribed after 3 tries, the thread would exit silently with no log. Defensive measure have been added to retry redis connection failures and the subscription will now retry indefinitely, unless canceled on shutdown and on the main subscription thread errors.

Fixes spring-projects#964 for versions before spring-projects#2256 was introduced.

Author: jebeaudet

src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java

@@ -42,11 +42,13 @@ class JedisSubscription extends AbstractSubscription {
 	 */
 	@Override
 	protected void doClose() {
-		if (!getChannels().isEmpty()) {
-			jedisPubSub.unsubscribe();
-		}
-		if (!getPatterns().isEmpty()) {
-			jedisPubSub.punsubscribe();
+		synchronized (this) {
+			if (!getChannels().isEmpty()) {
+				jedisPubSub.unsubscribe();
+			}
+			if (!getPatterns().isEmpty()) {
+				jedisPubSub.punsubscribe();
+			}
 		}
 	}
 
@@ -56,7 +58,9 @@ protected void doClose() {
 	 */
 	@Override
 	protected void doPsubscribe(byte[]... patterns) {
-		jedisPubSub.psubscribe(patterns);
+		synchronized (this) {
+			jedisPubSub.psubscribe(patterns);
+		}
 	}
 
 	/*
@@ -65,10 +69,12 @@ protected void doPsubscribe(byte[]... patterns) {
 	 */
 	@Override
 	protected void doPUnsubscribe(boolean all, byte[]... patterns) {
-		if (all) {
-			jedisPubSub.punsubscribe();
-		} else {
-			jedisPubSub.punsubscribe(patterns);
+		synchronized (this) {
+			if (all) {
+				jedisPubSub.punsubscribe();
+			} else {
+				jedisPubSub.punsubscribe(patterns);
+			}
 		}
 	}
 
@@ -78,7 +84,9 @@ protected void doPUnsubscribe(boolean all, byte[]... patterns) {
 	 */
 	@Override
 	protected void doSubscribe(byte[]... channels) {
-		jedisPubSub.subscribe(channels);
+		synchronized (this) {
+			jedisPubSub.subscribe(channels);
+		}
 	}
 
 	/*
@@ -87,10 +95,12 @@ protected void doSubscribe(byte[]... channels) {
 	 */
 	@Override
 	protected void doUnsubscribe(boolean all, byte[]... channels) {
-		if (all) {
-			jedisPubSub.unsubscribe();
-		} else {
-			jedisPubSub.unsubscribe(channels);
+		synchronized (this) {
+			if (all) {
+				jedisPubSub.unsubscribe();
+			} else {
+				jedisPubSub.unsubscribe(channels);
+			}
 		}
 	}
 }

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

@@ -25,6 +25,7 @@
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -678,7 +679,6 @@ protected void sleepBeforeRecoveryAttempt() {
 	 * @author Costin Leau
 	 */
 	private class SubscriptionTask implements SchedulingAwareRunnable {
-
 		/**
 		 * Runnable used, on a parallel thread, to do the initial pSubscribe. This is required since, during initialization,
 		 * both subscribe and pSubscribe might be needed but since the first call is blocking, the second call needs to
@@ -689,36 +689,55 @@ private class SubscriptionTask implements SchedulingAwareRunnable {
 		private class PatternSubscriptionTask implements SchedulingAwareRunnable {
 
 			private long WAIT = 500;
-			private long ROUNDS = 3;
+			private AtomicBoolean isThreadRunning = new AtomicBoolean(true);
+
 
 			public boolean isLongLived() {
 				return false;
 			}
 
+			void cancel() {
+				isThreadRunning.set(false);
+			}
+
 			public void run() {
 				// wait for subscription to be initialized
 				boolean done = false;
 				// wait 3 rounds for subscription to be initialized
-				for (int i = 0; i < ROUNDS && !done; i++) {
+				while(!done && isThreadRunning.get() && !Thread.currentThread().isInterrupted()) {
 					if (connection != null) {
-						synchronized (localMonitor) {
-							if (connection.isSubscribed()) {
-								done = true;
-								connection.getSubscription().pSubscribe(unwrap(patternMapping.keySet()));
-							} else {
-								try {
-									Thread.sleep(WAIT);
-								} catch (InterruptedException ex) {
-									Thread.currentThread().interrupt();
+							try {
+								if (connection.isSubscribed()) {
+									synchronized (localMonitor) {
+										connection.getSubscription().pSubscribe(unwrap(patternMapping.keySet()));
+										done = true;
+									}
+								} else {
+									try {
+										Thread.sleep(WAIT);
+									} catch (InterruptedException ex) {
+										logger.info("PatternSubscriptionTask was interrupted, exiting.");
+										Thread.currentThread().interrupt();
+										return;
+									}
+								}
+							} catch(Throwable e) {
+								if (e instanceof RedisConnectionFailureException) {
+									if (isRunning() && isThreadRunning.get()) {
+										logger.error("Connection failure occurred on pattern subscription task. Restarting subscription task after " + recoveryInterval + " ms");
+										sleepBeforeRecoveryAttempt();
+									}
+								} else {
+									logger.error("PatternSubscriptionTask aborted with exception:", e);
 									return;
 								}
 							}
-						}
 					}
 				}
 			}
 		}
 
+		private volatile @Nullable PatternSubscriptionTask patternSubscriptionTask;
 		private volatile @Nullable RedisConnection connection;
 		private boolean subscriptionTaskRunning = false;
 		private final Object localMonitor = new Object();
@@ -759,6 +778,7 @@ public void run() {
 					}
 				}
 			} catch (Throwable t) {
+				cancelPatternSubscriptionTask();
 				handleSubscriptionException(t);
 			} finally {
 				// this block is executed once the subscription thread has ended, this may or may not mean
@@ -789,7 +809,8 @@ private SubscriptionPresentCondition eventuallyPerformSubscription() {
 					condition = new SubscriptionPresentCondition();
 				} else {
 					// schedule the rest of the subscription
-					subscriptionExecutor.execute(new PatternSubscriptionTask());
+					patternSubscriptionTask = new PatternSubscriptionTask();
+					subscriptionExecutor.execute(patternSubscriptionTask);
 					condition = new PatternSubscriptionPresentCondition();
 				}
 
@@ -815,7 +836,7 @@ public boolean passes() {
 		 * Checks whether the current connection has an associated pattern subscription.
 		 *
 		 * @author Thomas Darimont
-		 * @see org.springframework.data.redis.listener.RedisMessageListenerContainer.SubscriptionTask.SubscriptionPresentTestCondition
+		 * @see org.springframework.data.redis.listener.RedisMessageListenerContainer.SubscriptionTask.SubscriptionPresentCondition
 		 */
 		private class PatternSubscriptionPresentCondition extends SubscriptionPresentCondition {
 
@@ -840,7 +861,17 @@ private byte[][] unwrap(Collection<ByteArrayWrapper> holders) {
 			return unwrapped;
 		}
 
+		private void cancelPatternSubscriptionTask() {
+			if(patternSubscriptionTask != null) {
+				synchronized (localMonitor) {
+					patternSubscriptionTask.cancel();
+					patternSubscriptionTask = null;
+				}
+			}
+		}
+
 		void cancel() {
+			cancelPatternSubscriptionTask();
 
 			if (!listening || connection == null) {
 				return;

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

@@ -35,6 +35,7 @@
 import org.springframework.core.task.SyncTaskExecutor;
 import org.springframework.data.redis.ObjectFactory;
 import org.springframework.data.redis.connection.ConnectionUtils;
+import org.springframework.data.redis.connection.MessageListener;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
@@ -55,12 +56,14 @@
 public class PubSubTests<T> {
 
 	private static final String CHANNEL = "pubsub::test";
+	private static final String PATTERN = "pattern::pubsub::test";
 
 	protected RedisMessageListenerContainer container;
 	protected ObjectFactory<T> factory;
 	@SuppressWarnings("rawtypes") protected RedisTemplate template;
 
 	private final BlockingDeque<Object> bag = new LinkedBlockingDeque<>(99);
+	private final BlockingDeque<Object> patternBag = new LinkedBlockingDeque<>(99);
 
 	private final Object handler = new Object() {
 		@SuppressWarnings("unused")
@@ -69,7 +72,15 @@ public void handleMessage(Object message) {
 		}
 	};
 
+	private final Object patternHandler = new Object() {
+		@SuppressWarnings("unused")
+		public void handleMessage(Object message) {
+			patternBag.add(message);
+		}
+	};
+
 	private final MessageListenerAdapter adapter = new MessageListenerAdapter(handler);
+	private final MessageListenerAdapter patternAdapter = new MessageListenerAdapter(patternHandler);
 
 	@SuppressWarnings("rawtypes")
 	public PubSubTests(ObjectFactory<T> factory, RedisTemplate template) {
@@ -84,10 +95,14 @@ public static Collection<Object[]> testParams() {
 	@BeforeEach
 	void setUp() throws Exception {
 		bag.clear();
+		patternBag.clear();
 
 		adapter.setSerializer(template.getValueSerializer());
 		adapter.afterPropertiesSet();
 
+		patternAdapter.setSerializer(template.getValueSerializer());
+		patternAdapter.afterPropertiesSet();
+
 		Phaser phaser = new Phaser(1);
 
 		container = new RedisMessageListenerContainer();
@@ -198,6 +213,34 @@ void testStartListenersToNoSpecificChannelTest() throws InterruptedException {
 		assertThat(set).contains(payload);
 	}
 
+	@SuppressWarnings("unchecked")
+	@ParameterizedRedisTest
+	void testStartListenersToBothChannelsAndPatternTopics() throws InterruptedException {
+
+		assumeThat(isClusterAware(template.getConnectionFactory())).isFalse();
+		assumeThat(ConnectionUtils.isJedis(template.getConnectionFactory())).isTrue();
+
+		PubSubAwaitUtil.runAndAwaitPatternSubscription(template.getRequiredConnectionFactory(), () -> {
+
+			container.addMessageListener(patternAdapter, Collections.singletonList(new PatternTopic(PATTERN + "*")));
+			container.start();
+		});
+
+		T payload = getT();
+
+		template.convertAndSend(PATTERN, payload);
+		template.convertAndSend(CHANNEL, payload);
+
+		Set<T> patternSet = new LinkedHashSet<>();
+		patternSet.add((T) patternBag.poll(3, TimeUnit.SECONDS));
+
+		Set<T> channelSet = new LinkedHashSet<>();
+		channelSet.add((T) bag.poll(3, TimeUnit.SECONDS));
+
+		assertThat(channelSet).contains(payload);
+		assertThat(patternSet).contains(payload);
+	}
+
 	private static boolean isClusterAware(RedisConnectionFactory connectionFactory) {
 
 		if (connectionFactory instanceof LettuceConnectionFactory) {