GH-3041: Deprecate RedisUtils.isUnlinkAvailable()

Fixes #3041

Some Redis clients/servers don't allow to perform an `INFO` command,
therefore we are not able to determine if we can perform `UNLINK` or
not.

* Deprecate `RedisUtils.isUnlinkAvailable()` as not reliable source of
through; use trial with fallback algorithm in the target logic around
`UNLINK` command calls.

**Cherry-pick to 5.1.x**

spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java

@@ -26,7 +26,6 @@
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.serializer.SerializationException;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
-import org.springframework.integration.redis.util.RedisUtils;
 import org.springframework.integration.store.AbstractKeyValueMessageStore;
 import org.springframework.util.Assert;
 
@@ -45,10 +44,10 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore implements B
 
 	private final RedisTemplate<Object, Object> redisTemplate;
 
-	private final boolean unlinkAvailable;
-
 	private boolean valueSerializerSet;
 
+	private volatile boolean unlinkAvailable = true;
+
 	/**
 	 * Construct {@link RedisMessageStore} based on the provided
 	 * {@link RedisConnectionFactory} and default empty prefix.
@@ -74,7 +73,6 @@ public RedisMessageStore(RedisConnectionFactory connectionFactory, String prefix
 		this.redisTemplate.setKeySerializer(new StringRedisSerializer());
 		this.redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
 		this.redisTemplate.afterPropertiesSet();
-		this.unlinkAvailable = RedisUtils.isUnlinkAvailable(this.redisTemplate);
 	}
 
 	@Override
@@ -135,7 +133,15 @@ protected Object doRemove(Object id) {
 		Object removedObject = this.doRetrieve(id);
 		if (removedObject != null) {
 			if (this.unlinkAvailable) {
-				this.redisTemplate.unlink(id);
+				try {
+					this.redisTemplate.unlink(id);
+				}
+				catch (Exception ex) {
+					logger.warn("The UNLINK command has failed (not supported on the Redis server?); " +
+							"falling back to the regular DELETE command", ex);
+					this.unlinkAvailable = false;
+					this.redisTemplate.delete(id);
+				}
 			}
 			else {
 				this.redisTemplate.delete(id);
@@ -147,7 +153,15 @@ protected Object doRemove(Object id) {
 	@Override
 	protected void doRemoveAll(Collection<Object> ids) {
 		if (this.unlinkAvailable) {
-			this.redisTemplate.unlink(ids);
+			try {
+				this.redisTemplate.unlink(ids);
+			}
+			catch (Exception ex) {
+				logger.warn("The UNLINK command has failed (not supported on the Redis server?); " +
+						"falling back to the regular DELETE command", ex);
+				this.unlinkAvailable = false;
+				this.redisTemplate.delete(ids);
+			}
 		}
 		else {
 			this.redisTemplate.delete(ids);

spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java

@@ -75,7 +75,7 @@
  */
 public final class RedisLockRegistry implements ExpirableLockRegistry, DisposableBean {
 
-	private static final Log logger = LogFactory.getLog(RedisLockRegistry.class);
+	private static final Log LOGGER = LogFactory.getLog(RedisLockRegistry.class);
 
 	private static final long DEFAULT_EXPIRE_AFTER = 60000L;
 
@@ -97,16 +97,14 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 
 	private final String registryKey;
 
-	private final boolean unlinkAvailable;
-
 	private final StringRedisTemplate redisTemplate;
 
 	private final RedisScript<Boolean> obtainLockScript;
 
 	private final long expireAfter;
 
 	/**
-	 * An {@link ExecutorService} to call {@link StringRedisTemplate#delete(Object)} in
+	 * An {@link ExecutorService} to call {@link StringRedisTemplate#delete} in
 	 * the separate thread when the current one is interrupted.
 	 */
 	private Executor executor =
@@ -140,7 +138,6 @@ public RedisLockRegistry(RedisConnectionFactory connectionFactory, String regist
 		this.obtainLockScript = new DefaultRedisScript<>(OBTAIN_LOCK_SCRIPT, Boolean.class);
 		this.registryKey = registryKey;
 		this.expireAfter = expireAfter;
-		this.unlinkAvailable = RedisUtils.isUnlinkAvailable(this.redisTemplate);
 	}
 
 	/**
@@ -187,7 +184,7 @@ private final class RedisLock implements Lock {
 
 		private final ReentrantLock localLock = new ReentrantLock();
 
-		private final boolean unlinkAvailable = RedisLockRegistry.this.unlinkAvailable;
+		private volatile boolean unlinkAvailable = true;
 
 		private volatile long lockedAt;
 
@@ -321,8 +318,8 @@ public void unlock() {
 					removeLockKey();
 				}
 
-				if (logger.isDebugEnabled()) {
-					logger.debug("Released lock; " + this);
+				if (LOGGER.isDebugEnabled()) {
+					LOGGER.debug("Released lock; " + this);
 				}
 			}
 			catch (Exception e) {
@@ -335,7 +332,15 @@ public void unlock() {
 
 		private void removeLockKey() {
 			if (this.unlinkAvailable) {
-				RedisLockRegistry.this.redisTemplate.unlink(this.lockKey);
+				try {
+					RedisLockRegistry.this.redisTemplate.unlink(this.lockKey);
+				}
+				catch (Exception ex) {
+					LOGGER.warn("The UNLINK command has failed (not supported on the Redis server?); " +
+							"falling back to the regular DELETE command", ex);
+					this.unlinkAvailable = false;
+					RedisLockRegistry.this.redisTemplate.delete(this.lockKey);
+				}
 			}
 			else {
 				RedisLockRegistry.this.redisTemplate.delete(this.lockKey);

spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisUtils.java

@@ -57,7 +57,11 @@ protected boolean removeEldestEntry(Entry<RedisOperations<?, ?>, Boolean> eldest
 	 * @param redisOperations the {@link RedisOperations} to perform {@code INFO} command.
 	 * @return true or false if {@code UNLINK} Redis command is available or not.
 	 * @throws IllegalStateException when {@code INFO} returns null from the Redis.
+	 * @deprecated since 5.1.8 in favor of explicit trials in the target code.
+	 * The INFO command might not be available on the server, but UNLINK might.
+	 * Will be removed in version 5.3.
 	 */
+	@Deprecated
 	public static boolean isUnlinkAvailable(RedisOperations<?, ?> redisOperations) {
 		return unlinkAvailable.computeIfAbsent(redisOperations, key -> {
 			Properties info = redisOperations.execute(