Do not attempt to delete phantom key if ShadowCopy is OFF

This commit makes sure to consider the ShadowCopy flag for expiring key events and prevents delete attempts if shadow copy is off.

Closes: #2954
Original Pull Request: #2955

Author: ltorje

src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java

@@ -99,6 +99,7 @@
  * @author Mark Paluch
  * @author Andrey Muchnik
  * @author John Blum
+ * @author Lucian Torje
  * @since 1.7
  */
 public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
@@ -735,9 +736,8 @@ private void initMessageListenerContainer() {
 	private void initKeyExpirationListener() {
 
 		if (this.expirationListener.get() == null) {
-
 			MappingExpirationListener listener = new MappingExpirationListener(this.messageListenerContainer, this.redisOps,
-					this.converter);
+					this.converter, this.shadowCopy);
 
 			listener.setKeyspaceNotificationsConfigParameter(keyspaceNotificationsConfigParameter);
 
@@ -763,16 +763,17 @@ static class MappingExpirationListener extends KeyExpirationEventMessageListener
 
 		private final RedisOperations<?, ?> ops;
 		private final RedisConverter converter;
-
+		private final ShadowCopy shadowCopy;
 		/**
 		 * Creates new {@link MappingExpirationListener}.
 		 */
 		MappingExpirationListener(RedisMessageListenerContainer listenerContainer, RedisOperations<?, ?> ops,
-				RedisConverter converter) {
+				RedisConverter converter, ShadowCopy shadowCopy) {
 
 			super(listenerContainer);
 			this.ops = ops;
 			this.converter = converter;
+			this.shadowCopy = shadowCopy;
 		}
 
 		@Override
@@ -783,22 +784,25 @@ public void onMessage(Message message, @Nullable byte[] pattern) {
 			}
 
 			byte[] key = message.getBody();
+			Object value = null;
 
-			byte[] phantomKey = ByteUtils.concat(key,
-					converter.getConversionService().convert(KeyspaceIdentifier.PHANTOM_SUFFIX, byte[].class));
+			if (shadowCopy != ShadowCopy.OFF) {
+				byte[] phantomKey = ByteUtils.concat(key,
+						converter.getConversionService().convert(KeyspaceIdentifier.PHANTOM_SUFFIX, byte[].class));
 
-			Map<byte[], byte[]> hash = ops.execute((RedisCallback<Map<byte[], byte[]>>) connection -> {
+				Map<byte[], byte[]> hash = ops.execute((RedisCallback<Map<byte[], byte[]>>) connection -> {
 
-				Map<byte[], byte[]> phantomValue = connection.hGetAll(phantomKey);
+					Map<byte[], byte[]> phantomValue = connection.hGetAll(phantomKey);
 
-				if (!CollectionUtils.isEmpty(phantomValue)) {
-					connection.del(phantomKey);
-				}
+					if (!CollectionUtils.isEmpty(phantomValue)) {
+						connection.del(phantomKey);
+					}
 
-				return phantomValue;
-			});
+					return phantomValue;
+				});
 
-			Object value = CollectionUtils.isEmpty(hash) ? null : converter.read(Object.class, new RedisData(hash));
+				value = CollectionUtils.isEmpty(hash) ? null : converter.read(Object.class, new RedisData(hash));
+			}
 
 			byte[] channelAsBytes = message.getChannel();
 

src/test/java/org/springframework/data/redis/core/MappingExpirationListenerTest.java

@@ -0,0 +1,74 @@
+package org.springframework.data.redis.core;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.core.convert.RedisConverter;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+/**
+ * @author Lucian Torje
+ */
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class MappingExpirationListenerTest {
+
+    @Mock
+    private RedisOperations<?, ?> redisOperations;
+    @Mock
+    private RedisConverter redisConverter;
+    @Mock
+    private RedisMessageListenerContainer listenerContainer;
+    @Mock
+    private Message message;
+    @Mock
+    private RedisKeyExpiredEvent<?> event;
+    @Mock
+    private ConversionService conversionService;
+
+    private RedisKeyValueAdapter.MappingExpirationListener listener;
+
+    @Test
+    void testOnNonKeyExpiration() {
+        byte[] key = "testKey".getBytes();
+        when(message.getBody()).thenReturn(key);
+        listener = new RedisKeyValueAdapter.MappingExpirationListener(listenerContainer, redisOperations, redisConverter, RedisKeyValueAdapter.ShadowCopy.ON);
+
+        listener.onMessage(message, null);
+
+        verify(redisOperations, times(0)).execute(any(RedisCallback.class));
+    }
+
+    @Test
+    void testOnValidKeyExpiration() {
+        List<Object> eventList = new ArrayList<>();
+
+        byte[] key = "abc:testKey".getBytes();
+        when(message.getBody()).thenReturn(key);
+
+        listener = new RedisKeyValueAdapter.MappingExpirationListener(listenerContainer, redisOperations, redisConverter, RedisKeyValueAdapter.ShadowCopy.OFF);
+        listener.setApplicationEventPublisher(eventList::add);
+        listener.onMessage(message, null);
+
+        verify(redisOperations, times(1)).execute(any(RedisCallback.class));
+        assertThat(eventList).hasSize(1);
+        assertThat(eventList.get(0)).isInstanceOf(RedisKeyExpiredEvent.class);
+        assertThat(((RedisKeyExpiredEvent) (eventList.get(0))).getKeyspace()).isEqualTo("abc");
+        assertThat(((RedisKeyExpiredEvent) (eventList.get(0))).getId()).isEqualTo("testKey".getBytes());
+    }
+}