Use routing info on delete operations.

Original Pull Request #2755
Closes #2754

(cherry picked from commit 9abcacb)

Author: sothawo

src/main/java/org/springframework/data/elasticsearch/core/AbstractReactiveElasticsearchTemplate.java

@@ -748,6 +748,12 @@ public ElasticsearchPersistentEntity<?> getPersistentEntityFor(@Nullable Class<?
 
 	public abstract Mono<String> getClusterVersion();
 
+	@Nullable
+	public String getEntityRouting(Object entity) {
+		return entityOperations.forEntity(entity, converter.getConversionService(), routingResolver)
+				.getRouting();
+	}
+
 	/**
 	 * Value class to capture client independent information from a response to an index request.
 	 */

src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchOperations.java

@@ -117,6 +117,17 @@ public interface ReactiveElasticsearchOperations
 	ReactiveClusterOperations cluster();
 
 	// region routing
+	/**
+	 * gets the routing for an entity.
+	 *
+	 * @param entity the entity
+	 * @return the routing, may be null if not set.
+	 * @since 5.2
+	 */
+	@Nullable
+	String getEntityRouting(Object entity);
+
+	// region customizations
 	/**
 	 * Returns a copy of this instance with the same configuration, but that uses a different {@link RoutingResolver} to
 	 * obtain routing information.

src/main/java/org/springframework/data/elasticsearch/repository/support/SimpleElasticsearchRepository.java

@@ -39,6 +39,7 @@
 import org.springframework.data.elasticsearch.core.query.BaseQuery;
 import org.springframework.data.elasticsearch.core.query.MoreLikeThisQuery;
 import org.springframework.data.elasticsearch.core.query.Query;
+import org.springframework.data.elasticsearch.core.routing.RoutingResolver;
 import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
 import org.springframework.data.util.StreamUtils;
 import org.springframework.data.util.Streamable;
@@ -223,15 +224,15 @@ public void deleteById(ID id) {
 
 		Assert.notNull(id, "Cannot delete entity with id 'null'.");
 
-		doDelete(id, getIndexCoordinates());
+		doDelete(id, null, getIndexCoordinates());
 	}
 
 	@Override
 	public void delete(T entity) {
 
 		Assert.notNull(entity, "Cannot delete 'null' entity.");
 
-		doDelete(extractIdFromBean(entity), getIndexCoordinates());
+		doDelete(extractIdFromBean(entity), operations.getEntityRouting(entity), getIndexCoordinates());
 	}
 
 	@Override
@@ -271,10 +272,14 @@ public void deleteAll(Iterable<? extends T> entities) {
 		deleteAllById(ids);
 	}
 
-	private void doDelete(@Nullable ID id, IndexCoordinates indexCoordinates) {
+	private void doDelete(@Nullable ID id, @Nullable String routing, IndexCoordinates indexCoordinates) {
 
 		if (id != null) {
-			executeAndRefresh(operations -> operations.delete(stringIdRepresentation(id), indexCoordinates));
+			executeAndRefresh(operations -> {
+				var ops = routing != null ? operations.withRouting(RoutingResolver.just(routing)) : operations;
+				// noinspection DataFlowIssue
+				return ops.delete(stringIdRepresentation(id), indexCoordinates);
+			});
 		}
 	}
 

src/main/java/org/springframework/data/elasticsearch/repository/support/SimpleReactiveElasticsearchRepository.java

@@ -31,6 +31,7 @@
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.query.BaseQuery;
 import org.springframework.data.elasticsearch.core.query.Query;
+import org.springframework.data.elasticsearch.core.routing.RoutingResolver;
 import org.springframework.data.elasticsearch.repository.ReactiveElasticsearchRepository;
 import org.springframework.util.Assert;
 
@@ -196,7 +197,10 @@ public Mono<Void> deleteById(Publisher<ID> id) {
 	public Mono<Void> delete(T entity) {
 
 		Assert.notNull(entity, "Entity must not be null!");
-		return operations.delete(entity, entityInformation.getIndexCoordinates()) //
+
+		var routing = operations.getEntityRouting(entity);
+		var ops = routing != null ? operations.withRouting(RoutingResolver.just(routing)) : operations;
+		return ops.delete(entity, entityInformation.getIndexCoordinates()) //
 				.then(doRefresh());
 	}