Remove phantom copy of expiring data when source gets persisted.

This commit makes sure to clean up resources when a previously expiring entity is persisted by setting the time to live to zero or negative.
In case a phantom copy exists for the changed entity, it is removed to free space on the server and prevent expiration events from being sent.

See #1955
Original Pull Request: #1961

# Conflicts:
#	src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java

Author: muchnik

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

@@ -99,6 +99,7 @@
  *
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Andrey Muchnik
  * @since 1.7
  */
 public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
@@ -245,6 +246,11 @@ public Object put(Object id, Object item, String keyspace) {
 				connection.expire(phantomKey, rdo.getTimeToLive() + PHANTOM_KEY_TTL);
 			}
 
+			boolean isNoExpire = rdo.getTimeToLive() == null || rdo.getTimeToLive() != null && rdo.getTimeToLive() < 0;
+			if (isNoExpire && !isNew && keepShadowCopy()) {
+				connection.del(ByteUtils.concat(objectKey, BinaryKeyspaceIdentifier.PHANTOM_SUFFIX));
+			}
+
 			connection.sAdd(toBytes(rdo.getKeyspace()), key);
 
 			IndexWriter indexWriter = new IndexWriter(connection, converter);
@@ -487,7 +493,7 @@ public void update(PartialUpdate<?> update) {
 				} else {
 
 					connection.persist(redisKey);
-					connection.persist(ByteUtils.concat(redisKey, BinaryKeyspaceIdentifier.PHANTOM_SUFFIX));
+					connection.del(ByteUtils.concat(redisKey, BinaryKeyspaceIdentifier.PHANTOM_SUFFIX));
 				}
 			}
 
@@ -736,6 +742,10 @@ private void initKeyExpirationListener() {
 		}
 	}
 
+	private boolean keepShadowCopy() {
+		return this.expirationListener.get() != null;
+	}
+
 	/**
 	 * {@link MessageListener} implementation used to capture Redis keyspace notifications. Tries to read a previously
 	 * created phantom key {@code keyspace:id:phantom} to provide the expired object as part of the published
@@ -777,7 +787,8 @@ public void onMessage(Message message, @Nullable byte[] pattern) {
 
 			byte[] key = message.getBody();
 
-			byte[] phantomKey = ByteUtils.concat(key, converter.getConversionService().convert(KeyspaceIdentifier.PHANTOM_SUFFIX, byte[].class));
+			byte[] phantomKey = ByteUtils.concat(key,
+					converter.getConversionService().convert(KeyspaceIdentifier.PHANTOM_SUFFIX, byte[].class));
 
 			Map<byte[], byte[]> hash = ops.execute((RedisCallback<Map<byte[], byte[]>>) connection -> {
 
@@ -794,7 +805,8 @@ public void onMessage(Message message, @Nullable byte[] pattern) {
 
 			byte[] channelAsBytes = message.getChannel();
 			String channel = !ObjectUtils.isEmpty(channelAsBytes)
-					? converter.getConversionService().convert(channelAsBytes, String.class) : null;
+					? converter.getConversionService().convert(channelAsBytes, String.class)
+					: null;
 
 			RedisKeyExpiredEvent event = new RedisKeyExpiredEvent(channel, key, value);
 

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

@@ -60,6 +60,7 @@
 /**
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Andrey Muchnik
  */
 @RunWith(Parameterized.class)
 public class RedisKeyValueAdapterTests {
@@ -687,6 +688,41 @@ public void updateShouldAlterGeoIndexCorrectlyOnUpdate() {
 		assertThat(updatedLocation.getY()).isCloseTo(18D, offset(0.005));
 	}
 
+	@Test // DATAREDIS-1955
+	public void phantomKeyIsDeletedWhenPutWithNegativeTimeToLiveAndOldEntryTimeToLiveWasPositiveAndWhenShadowCopyIsTurnedOn() {
+		ExpiringPerson rand = new ExpiringPerson();
+		rand.id = "1";
+		rand.ttl = 3000L;
+
+		adapter.put("1", rand, "persons");
+
+		assertThat(template.getExpire("persons:1:phantom")).isPositive();
+
+		rand.ttl = -1L;
+
+		adapter.put("1", rand, "persons");
+
+		assertThat(template.hasKey("persons:1:phantom")).isFalse();
+	}
+
+	@Test // DATAREDIS-1955
+	public void updateWithRefreshTtlAndWithoutPositiveTtlShouldDeletePhantomKey() {
+		ExpiringPerson person = new ExpiringPerson();
+		person.id = "1";
+		person.ttl = 100L;
+
+		adapter.put("1", person, "persons");
+
+		assertThat(template.getExpire("persons:1:phantom")).isPositive();
+
+		PartialUpdate<ExpiringPerson> update = new PartialUpdate<>("1", ExpiringPerson.class) //
+				.refreshTtl(true);
+
+		adapter.update(update);
+
+		assertThat(template.hasKey("persons:1:phantom")).isFalse();
+	}
+
 	/**
 	 * Wait up to 5 seconds until {@code key} is no longer available in Redis.
 	 *
@@ -776,6 +812,11 @@ static class TaVeren extends Person {
 
 	}
 
+	static class ExpiringPerson extends Person {
+
+		@TimeToLive Long ttl;
+	}
+
 	@KeySpace("locations")
 	static class Location {