INT-4507: Use UNLINK in RedisMessageStore if any

JIRA: https://jira.spring.io/browse/INT-4507

* Add `RedisUtils` with the `isUnlinkAvailable()` to check the Redis
server version to be sure that `UNLINK` is available or not
* Use `RedisUtils.isUnlinkAvailable()` in the `RedisMessageStore` and
`RedisLockRegistry` when a removal functionality is performed
* Add `AbstractKeyValueMessageStore.doRemoveAll(Collection<Object> ids)`
for optimization
* Implement `doRemoveAll()` in the `RedisMessageStore` and
`GemfireMessageStore`

**Cherry-pick to 5.0.x**

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -20,8 +20,10 @@
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Set;
 import java.util.UUID;
+import java.util.stream.Collectors;
 
 import org.springframework.jmx.export.annotation.ManagedAttribute;
 import org.springframework.messaging.Message;
@@ -243,11 +245,21 @@ public void removeMessagesFromGroup(Object groupId, Collection<Message<?>> messa
 		if (mgm != null) {
 			Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
 			MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
-			for (Message<?> messageToRemove : messages) {
-				UUID messageId = messageToRemove.getHeaders().getId();
-				messageGroupMetadata.remove(messageId);
-				doRemove(this.messagePrefix + messageId);
-			}
+
+			List<UUID> ids =
+					messages.stream()
+							.map(messageToRemove -> messageToRemove.getHeaders().getId())
+							.collect(Collectors.toList());
+
+			messageGroupMetadata.removeAll(ids);
+
+			List<Object> messageIds =
+					ids.stream()
+							.map(id -> this.messagePrefix + id)
+							.collect(Collectors.toList());
+
+			doRemoveAll(messageIds);
+
 			messageGroupMetadata.setLastModified(System.currentTimeMillis());
 			doStore(this.groupPrefix + groupId, messageGroupMetadata);
 		}
@@ -275,10 +287,13 @@ public void removeMessageGroup(Object groupId) {
 			Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
 			MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
 
-			Iterator<UUID> messageIds = messageGroupMetadata.messageIdIterator();
-			while (messageIds.hasNext()) {
-				removeMessage(messageIds.next());
-			}
+			List<Object> messageIds =
+					messageGroupMetadata.getMessageIds()
+							.stream()
+							.map(id -> this.messagePrefix + id)
+							.collect(Collectors.toList());
+
+			doRemoveAll(messageIds);
 		}
 	}
 
@@ -325,7 +340,7 @@ public Message<?> getOneMessageFromGroup(Object groupId) {
 	@Override
 	public Collection<Message<?>> getMessagesForGroup(Object groupId) {
 		MessageGroupMetadata groupMetadata = getGroupMetadata(groupId);
-		ArrayList<Message<?>> messages = new ArrayList<Message<?>>();
+		ArrayList<Message<?>> messages = new ArrayList<>();
 		if (groupMetadata != null) {
 			Iterator<UUID> messageIds = groupMetadata.messageIdIterator();
 			while (messageIds.hasNext()) {
@@ -345,7 +360,7 @@ public Iterator<MessageGroup> iterator() {
 	}
 
 	private Collection<String> normalizeKeys(Collection<String> keys) {
-		Set<String> normalizedKeys = new HashSet<String>();
+		Set<String> normalizedKeys = new HashSet<>();
 		for (Object key : keys) {
 			String strKey = (String) key;
 			if (strKey.startsWith(this.groupPrefix)) {
@@ -378,6 +393,8 @@ public int messageGroupSize(Object groupId) {
 
 	protected abstract Object doRemove(Object id);
 
+	protected abstract void doRemoveAll(Collection<Object> ids);
+
 	protected abstract Collection<?> doListKeys(String keyPattern);
 
 	private final class MessageGroupIterator implements Iterator<MessageGroup> {

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.store;
 
 import java.io.Serializable;
+import java.util.Collection;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
@@ -39,7 +40,7 @@ public class MessageGroupMetadata implements Serializable {
 
 	private static final long serialVersionUID = 1L;
 
-	private List<UUID> messageIds = new LinkedList<UUID>();
+	private List<UUID> messageIds = new LinkedList<>();
 
 	private long timestamp;
 
@@ -68,6 +69,10 @@ public void remove(UUID messageId) {
 		this.messageIds.remove(messageId);
 	}
 
+	public void removeAll(Collection<UUID> messageIds) {
+		this.messageIds.removeAll(messageIds);
+	}
+
 	boolean add(UUID messageId) {
 		return !this.messageIds.contains(messageId) && this.messageIds.add(messageId);
 	}

spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -96,6 +96,11 @@ protected Object doRemove(Object id) {
 		return this.messageStoreRegion.remove(id);
 	}
 
+	@Override
+	protected void doRemoveAll(Collection<Object> ids) {
+		this.messageStoreRegion.removeAll(ids);
+	}
+
 	@Override
 	protected Collection<?> doListKeys(String keyPattern) {
 		Assert.hasText(keyPattern, "'keyPattern' must not be empty");

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2007-2017 the original author or authors.
+ * Copyright 2007-2018 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.
@@ -26,6 +26,7 @@
 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.integration.store.MessageGroupStore;
 import org.springframework.integration.store.MessageStore;
@@ -66,7 +67,7 @@ public RedisMessageStore(RedisConnectionFactory connectionFactory) {
 	 */
 	public RedisMessageStore(RedisConnectionFactory connectionFactory, String prefix) {
 		super(prefix);
-		this.redisTemplate = new RedisTemplate<Object, Object>();
+		this.redisTemplate = new RedisTemplate<>();
 		this.redisTemplate.setConnectionFactory(connectionFactory);
 		this.redisTemplate.setKeySerializer(new StringRedisSerializer());
 		this.redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
@@ -130,11 +131,26 @@ protected Object doRemove(Object id) {
 		Assert.notNull(id, "'id' must not be null");
 		Object removedObject = this.doRetrieve(id);
 		if (removedObject != null) {
-			this.redisTemplate.delete(id);
+			if (RedisUtils.isUnlinkAvailable(this.redisTemplate)) {
+				this.redisTemplate.unlink(id);
+			}
+			else {
+				this.redisTemplate.delete(id);
+			}
 		}
 		return removedObject;
 	}
 
+	@Override
+	protected void doRemoveAll(Collection<Object> ids) {
+		if (RedisUtils.isUnlinkAvailable(this.redisTemplate)) {
+			this.redisTemplate.unlink(ids);
+		}
+		else {
+			this.redisTemplate.delete(ids);
+		}
+	}
+
 	@Override
 	protected Collection<?> doListKeys(String keyPattern) {
 		Assert.hasText(keyPattern, "'keyPattern' must not be empty");

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

@@ -302,11 +302,10 @@ public void unlock() {
 			}
 			try {
 				if (Thread.currentThread().isInterrupted()) {
-					RedisLockRegistry.this.executor.execute(() ->
-							RedisLockRegistry.this.redisTemplate.delete(this.lockKey));
+					RedisLockRegistry.this.executor.execute(this::removeLockKey);
 				}
 				else {
-					RedisLockRegistry.this.redisTemplate.delete(this.lockKey);
+					removeLockKey();
 				}
 
 				if (logger.isDebugEnabled()) {
@@ -321,6 +320,15 @@ public void unlock() {
 			}
 		}
 
+		private void removeLockKey() {
+			if (RedisUtils.isUnlinkAvailable(RedisLockRegistry.this.redisTemplate)) {
+				RedisLockRegistry.this.redisTemplate.unlink(this.lockKey);
+			}
+			else {
+				RedisLockRegistry.this.redisTemplate.delete(this.lockKey);
+			}
+		}
+
 		@Override
 		public Condition newCondition() {
 			throw new UnsupportedOperationException("Conditions are not supported");

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

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.redis.util;
+
+import java.util.Properties;
+
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.RedisOperations;
+
+/**
+ * A set of utility methods for common Redis functions.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.1
+ */
+public final class RedisUtils {
+
+	private static final String SECTION = "server";
+
+	private static final String VERSION_PROPERTY = "redis_version";
+
+	private static final int MAJOR_VERSION_TO_COMPARE = 4;
+
+	private static Boolean unlinkAvailable;
+
+	/**
+	 * Perform an {@code INFO} command on the provided {@link RedisOperations} to check
+	 * the Redis server version to be sure that {@code UNLINK} is available or not.
+	 * @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.
+	 */
+	public static boolean isUnlinkAvailable(RedisOperations<?, ?> redisOperations) {
+		if (unlinkAvailable == null) {
+			Properties info = redisOperations.execute(
+					(RedisCallback<Properties>) connection -> connection.serverCommands().info(SECTION));
+			if (info != null) {
+				int majorVersion = Integer.parseInt(info.getProperty(VERSION_PROPERTY).split("\\.")[0]);
+				unlinkAvailable = majorVersion >= MAJOR_VERSION_TO_COMPARE;
+			}
+			else {
+				throw new IllegalStateException("The INFO command cannot be used in pipeline/transaction.");
+			}
+		}
+		return unlinkAvailable;
+	}
+
+	private RedisUtils() {
+	}
+
+}