Do not delete phantom if exists and ShadowCopy is OFF

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());
+    }
+}