From 6f91929da3a23529985f13ac640b6a3bdc4cb0ab Mon Sep 17 00:00:00 2001
From: Unseok Kim <unseok.kim@linecorp.com>
Date: Wed, 9 Feb 2022 00:08:23 +0900
Subject: [PATCH] GH-3716 Fix wait for init redisMsgListenContainer

Fixes https://github.com/spring-projects/spring-integration/issues/3716

If the `redisMessageListenerContainer` is starting, waiting for it to complete without doing `subscribeUnlock()`

* Introduce `isRunningRedisMessageListenerContainer` state since the `running` in the `RedisMessageListenerContainer` is set in the beginning of the `start()` misleading on the concurrent calls to the `RedisLockRegistry`

**Cherry-pick to `5.5.x`**
---
 .../redis/util/RedisLockRegistry.java           | 17 ++++++++++++++---
 1 file changed, 14 insertions(+), 3 deletions(-)

diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
index ae90805d0f..b5836c33b2 100644
--- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
@@ -164,6 +164,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 	private boolean executorExplicitlySet;
 
 	private volatile boolean unlinkAvailable = true;
+	private volatile boolean isRunningRedisMessageListenerContainer = false;
 
 	/**
 	 * Constructs a lock registry with the default (60 second) lock expiration.
@@ -364,9 +365,9 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 				return true;
 			}
 
-			if (!RedisLockRegistry.this.redisMessageListenerContainer.isRunning()) {
-				RedisLockRegistry.this.redisMessageListenerContainer.afterPropertiesSet();
-				RedisLockRegistry.this.redisMessageListenerContainer.start();
+			if (!(RedisLockRegistry.this.isRunningRedisMessageListenerContainer
+					&& RedisLockRegistry.this.redisMessageListenerContainer.isRunning())) {
+				runRedisMessageListenerContainer();
 			}
 			while (time == -1 || expiredTime >= System.currentTimeMillis()) {
 				try {
@@ -523,6 +524,16 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 			return RedisLockRegistry.this;
 		}
 
+		private void runRedisMessageListenerContainer() {
+			synchronized (RedisLockRegistry.this.redisMessageListenerContainer) {
+				if (!(RedisLockRegistry.this.isRunningRedisMessageListenerContainer
+						&& RedisLockRegistry.this.redisMessageListenerContainer.isRunning())) {
+					RedisLockRegistry.this.redisMessageListenerContainer.afterPropertiesSet();
+					RedisLockRegistry.this.redisMessageListenerContainer.start();
+					RedisLockRegistry.this.isRunningRedisMessageListenerContainer = true;
+				}
+			}
+		}
 	}
 
 	private static final class RedisUnLockNotifyMessageListener implements MessageListener {