Clear indexed data if property value is set to null.

Closes #2882
Original pull request: #2895

Author: christophstrobl

Diff for: src/main/java/org/springframework/data/redis/core/convert/PathIndexResolver.java

@@ -122,6 +122,7 @@ public void doWithPersistentProperty(RedisPersistentProperty persistentProperty)
 				Object propertyValue = accessor.getProperty(persistentProperty);
 
 				if (propertyValue == null) {
+					indexes.addAll(resolveIndex(keyspace, currentPath, persistentProperty, null));
 					return;
 				}
 
@@ -212,21 +213,32 @@ protected Set<IndexedData> resolveIndex(String keyspace, String propertyPath,
 
 				IndexedData indexedData = null;
 				if (transformedValue == null) {
-					indexedData = new RemoveIndexedData(indexedData);
+
+					indexedData = new RemoveIndexedData(new IndexedData() {
+						@Override
+						public String getIndexName() {
+							return indexDefinition.getIndexName();
+						}
+
+						@Override
+						public String getKeyspace() {
+							return indexDefinition.getKeyspace();
+						}
+					});
 				} else {
 					indexedData = indexedDataFactoryProvider.getIndexedDataFactory(indexDefinition).createIndexedDataFor(value);
 				}
 				data.add(indexedData);
 			}
 		}
 
-		else if (property != null && property.isAnnotationPresent(Indexed.class)) {
+		else if (property != null && value != null && property.isAnnotationPresent(Indexed.class)) {
 
 			SimpleIndexDefinition indexDefinition = new SimpleIndexDefinition(keyspace, path);
 			indexConfiguration.addIndexDefinition(indexDefinition);
 
 			data.add(indexedDataFactoryProvider.getIndexedDataFactory(indexDefinition).createIndexedDataFor(value));
-		} else if (property != null && property.isAnnotationPresent(GeoIndexed.class)) {
+		} else if (property != null &&  value != null && property.isAnnotationPresent(GeoIndexed.class)) {
 
 			GeoIndexDefinition indexDefinition = new GeoIndexDefinition(keyspace, path);
 			indexConfiguration.addIndexDefinition(indexDefinition);

Diff for: src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java

@@ -147,6 +147,25 @@ void putWritesSimpleIndexDataCorrectly() {
 		assertThat(template.opsForSet().members("persons:firstname:rand")).contains("1");
 	}
 
+	@Test // GH-2882
+	void indexDataShouldBeCleardIfPropertyValueIsSetToNull() {
+
+		Person rand = new Person();
+		rand.firstname = "rand";
+
+		adapter.put("1", rand, "persons");
+
+		assertThat(template.keys("persons*")).contains("persons:firstname:rand");
+		assertThat(template.opsForSet().members("persons:firstname:rand")).contains("1");
+
+		rand.id = "1";
+		rand.firstname = null;
+		adapter.put("1", rand, "persons");
+
+		assertThat(template.keys("persons*")).doesNotContain("persons:firstname:rand");
+		assertThat(template.opsForSet().members("persons:firstname:rand")).doesNotContain("1");
+	}
+
 	@Test // DATAREDIS-744
 	void putWritesSimpleIndexDataWithColonCorrectly() {