GH-3434: Optimize unlinkAvailable in Redis

Fixes #3434

Each `RedisLock` has their own `unlinkAvailable` property
therefore we try to `unlink` on every unlock call and also log a WARN
about unavailability of this command in Redis, plus exception

* Move `unlinkAvailable` property to the `RedisLockRegistry` level
to have a check only once on a first `unlock()` call
* Move the whole exception logging onto the DEBUG level, leaving the WARN
only with the `ex.getMessage()`
* Optimize logging the same way in the `RedisMessageStore`

Cherry-pick to `5.3.x & 5.2.x`

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2007-2019 the original author or authors.
+ * Copyright 2007-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -139,9 +139,7 @@ protected Object doRemove(Object id) {
 					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;
+					unlinkUnavailable(ex);
 					this.redisTemplate.delete(id);
 				}
 			}
@@ -152,16 +150,26 @@ protected Object doRemove(Object id) {
 		return removedObject;
 	}
 
+	private void unlinkUnavailable(Exception ex) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("The UNLINK command has failed (not supported on the Redis server?); " +
+					"falling back to the regular DELETE command", ex);
+		}
+		else {
+			logger.warn("The UNLINK command has failed (not supported on the Redis server?); " +
+					"falling back to the regular DELETE command: " + ex.getMessage());
+		}
+		this.unlinkAvailable = false;
+	}
+
 	@Override
 	protected void doRemoveAll(Collection<Object> ids) {
 		if (this.unlinkAvailable) {
 			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;
+				unlinkUnavailable(ex);
 				this.redisTemplate.delete(ids);
 			}
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 the original author or authors.
+ * Copyright 2014-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,7 +19,6 @@
 import java.text.SimpleDateFormat;
 import java.util.Collections;
 import java.util.Date;
-import java.util.Iterator;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
@@ -116,6 +115,8 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 	 */
 	private boolean executorExplicitlySet;
 
+	private volatile boolean unlinkAvailable = true;
+
 	/**
 	 * Constructs a lock registry with the default (60 second) lock expiration.
 	 * @param connectionFactory The connection factory.
@@ -160,15 +161,12 @@ public Lock obtain(Object lockKey) {
 
 	@Override
 	public void expireUnusedOlderThan(long age) {
-		Iterator<Map.Entry<String, RedisLock>> iterator = this.locks.entrySet().iterator();
 		long now = System.currentTimeMillis();
-		while (iterator.hasNext()) {
-			Map.Entry<String, RedisLock> entry = iterator.next();
-			RedisLock lock = entry.getValue();
-			if (now - lock.getLockedAt() > age && !lock.isAcquiredInThisProcess()) {
-				iterator.remove();
-			}
-		}
+		this.locks.entrySet()
+				.removeIf((entry) -> {
+					RedisLock lock = entry.getValue();
+					return now - lock.getLockedAt() > age && !lock.isAcquiredInThisProcess();
+				});
 	}
 
 	@Override
@@ -184,16 +182,14 @@ private final class RedisLock implements Lock {
 
 		private final ReentrantLock localLock = new ReentrantLock();
 
-		private volatile boolean unlinkAvailable = true;
-
 		private volatile long lockedAt;
 
 		private RedisLock(String path) {
 			this.lockKey = constructLockKey(path);
 		}
 
 		private String constructLockKey(String path) {
-			return RedisLockRegistry.this.registryKey + ":" + path;
+			return RedisLockRegistry.this.registryKey + ':' + path;
 		}
 
 		public long getLockedAt() {
@@ -331,14 +327,20 @@ public void unlock() {
 		}
 
 		private void removeLockKey() {
-			if (this.unlinkAvailable) {
+			if (RedisLockRegistry.this.unlinkAvailable) {
 				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.unlinkAvailable = false;
+					if (LOGGER.isDebugEnabled()) {
+						LOGGER.debug("The UNLINK command has failed (not supported on the Redis server?); " +
+								"falling back to the regular DELETE command", ex);
+					}
+					else {
+						LOGGER.warn("The UNLINK command has failed (not supported on the Redis server?); " +
+								"falling back to the regular DELETE command: " + ex.getMessage());
+					}
 					RedisLockRegistry.this.redisTemplate.delete(this.lockKey);
 				}
 			}
@@ -359,7 +361,7 @@ public boolean isAcquiredInThisProcess() {
 
 		@Override
 		public String toString() {
-			SimpleDateFormat dateFormat = new SimpleDateFormat("YYYY-MM-dd@HH:mm:ss.SSS");
+			SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd@HH:mm:ss.SSS");
 			return "RedisLock [lockKey=" + this.lockKey
 					+ ",lockedAt=" + dateFormat.format(new Date(this.lockedAt))
 					+ ", clientId=" + RedisLockRegistry.this.clientId