Bulk operations must prefer index set on individual query.

Original Pull Request #2363
Closes #2362

src/main/java/org/springframework/data/elasticsearch/client/elc/RequestConverter.java

@@ -15,8 +15,12 @@
  */
 package org.springframework.data.elasticsearch.client.elc;
 
-import static org.springframework.data.elasticsearch.client.elc.TypeUtils.*;
+import static org.springframework.data.elasticsearch.client.elc.TypeUtils.searchType;
-import static org.springframework.util.CollectionUtils.*;
+import static org.springframework.data.elasticsearch.client.elc.TypeUtils.slices;
+import static org.springframework.data.elasticsearch.client.elc.TypeUtils.time;
+import static org.springframework.data.elasticsearch.client.elc.TypeUtils.timeStringMs;
+import static org.springframework.data.elasticsearch.client.elc.TypeUtils.toFloat;
+import static org.springframework.util.CollectionUtils.isEmpty;
 
 import co.elastic.clients.elasticsearch._types.Conflicts;
 import co.elastic.clients.elasticsearch._types.FieldValue;
@@ -33,7 +37,18 @@ import co.elastic.clients.elasticsearch._types.mapping.RuntimeFieldType;
 import co.elastic.clients.elasticsearch._types.mapping.TypeMapping;
 import co.elastic.clients.elasticsearch._types.query_dsl.Like;
 import co.elastic.clients.elasticsearch.cluster.HealthRequest;
-import co.elastic.clients.elasticsearch.core.*;
+import co.elastic.clients.elasticsearch.core.BulkRequest;
+import co.elastic.clients.elasticsearch.core.ClosePointInTimeRequest;
+import co.elastic.clients.elasticsearch.core.DeleteByQueryRequest;
+import co.elastic.clients.elasticsearch.core.DeleteRequest;
+import co.elastic.clients.elasticsearch.core.GetRequest;
+import co.elastic.clients.elasticsearch.core.IndexRequest;
+import co.elastic.clients.elasticsearch.core.MgetRequest;
+import co.elastic.clients.elasticsearch.core.MsearchRequest;
+import co.elastic.clients.elasticsearch.core.OpenPointInTimeRequest;
+import co.elastic.clients.elasticsearch.core.SearchRequest;
+import co.elastic.clients.elasticsearch.core.UpdateByQueryRequest;
+import co.elastic.clients.elasticsearch.core.UpdateRequest;
 import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
 import co.elastic.clients.elasticsearch.core.bulk.CreateOperation;
 import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
@@ -43,8 +58,17 @@ import co.elastic.clients.elasticsearch.core.msearch.MultisearchBody;
 import co.elastic.clients.elasticsearch.core.search.Highlight;
 import co.elastic.clients.elasticsearch.core.search.Rescore;
 import co.elastic.clients.elasticsearch.core.search.SourceConfig;
-import co.elastic.clients.elasticsearch.indices.*;
+import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
+import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest;
 import co.elastic.clients.elasticsearch.indices.ExistsRequest;
+import co.elastic.clients.elasticsearch.indices.GetAliasRequest;
+import co.elastic.clients.elasticsearch.indices.GetIndexRequest;
+import co.elastic.clients.elasticsearch.indices.GetIndicesSettingsRequest;
+import co.elastic.clients.elasticsearch.indices.GetMappingRequest;
+import co.elastic.clients.elasticsearch.indices.IndexSettings;
+import co.elastic.clients.elasticsearch.indices.PutMappingRequest;
+import co.elastic.clients.elasticsearch.indices.RefreshRequest;
+import co.elastic.clients.elasticsearch.indices.UpdateAliasesRequest;
 import co.elastic.clients.elasticsearch.indices.update_aliases.Action;
 import co.elastic.clients.json.JsonData;
 import co.elastic.clients.json.JsonpDeserializer;
@@ -82,7 +106,19 @@ import org.springframework.data.elasticsearch.core.index.PutTemplateRequest;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentEntity;
 import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentProperty;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
-import org.springframework.data.elasticsearch.core.query.*;
+import org.springframework.data.elasticsearch.core.query.BaseQuery;
+import org.springframework.data.elasticsearch.core.query.BulkOptions;
+import org.springframework.data.elasticsearch.core.query.CriteriaQuery;
+import org.springframework.data.elasticsearch.core.query.GeoDistanceOrder;
+import org.springframework.data.elasticsearch.core.query.IndexQuery;
+import org.springframework.data.elasticsearch.core.query.MoreLikeThisQuery;
+import org.springframework.data.elasticsearch.core.query.Order;
+import org.springframework.data.elasticsearch.core.query.Query;
+import org.springframework.data.elasticsearch.core.query.RescorerQuery;
+import org.springframework.data.elasticsearch.core.query.ScriptData;
+import org.springframework.data.elasticsearch.core.query.SourceFilter;
+import org.springframework.data.elasticsearch.core.query.StringQuery;
+import org.springframework.data.elasticsearch.core.query.UpdateQuery;
 import org.springframework.data.elasticsearch.core.reindex.ReindexRequest;
 import org.springframework.data.elasticsearch.core.reindex.Remote;
 import org.springframework.data.elasticsearch.support.DefaultStringObjectMap;
@@ -435,7 +471,7 @@ class RequestConverter {
 
 		IndexRequest.Builder<Object> builder = new IndexRequest.Builder<>();
 
-		builder.index(indexCoordinates.getIndexName());
+		builder.index(query.getIndexName() != null ? query.getIndexName() : indexCoordinates.getIndexName());
 
 		Object queryObject = query.getObject();
 
@@ -487,7 +523,7 @@ class RequestConverter {
 
 		IndexOperation.Builder<Object> builder = new IndexOperation.Builder<>();
 
-		builder.index(indexCoordinates.getIndexName());
+		builder.index(query.getIndexName() != null ? query.getIndexName() : indexCoordinates.getIndexName());
 
 		Object queryObject = query.getObject();
 
@@ -528,7 +564,7 @@ class RequestConverter {
 
 		CreateOperation.Builder<Object> builder = new CreateOperation.Builder<>();
 
-		builder.index(indexCoordinates.getIndexName());
+		builder.index(query.getIndexName() != null ? query.getIndexName() : indexCoordinates.getIndexName());
 
 		Object queryObject = query.getObject();
 

src/test/java/org/springframework/data/elasticsearch/core/ElasticsearchIntegrationTests.java

@@ -15,20 +15,21 @@
  */
 package org.springframework.data.elasticsearch.core;
 
-import static java.util.Collections.*;
+import static java.util.Collections.singletonList;
-import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.Assertions.assertThat;
-import static org.elasticsearch.index.query.QueryBuilders.*;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.springframework.data.elasticsearch.annotations.Document.VersionType.*;
+import static org.assertj.core.api.Assertions.fail;
-import static org.springframework.data.elasticsearch.annotations.FieldType.*;
+import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
+import static org.elasticsearch.index.query.QueryBuilders.termQuery;
+import static org.springframework.data.elasticsearch.annotations.Document.VersionType.EXTERNAL_GTE;
 import static org.springframework.data.elasticsearch.annotations.FieldType.Integer;
-import static org.springframework.data.elasticsearch.core.document.Document.*;
+import static org.springframework.data.elasticsearch.annotations.FieldType.Keyword;
-import static org.springframework.data.elasticsearch.utils.IdGenerator.*;
+import static org.springframework.data.elasticsearch.annotations.FieldType.Text;
-import static org.springframework.data.elasticsearch.utils.IndexBuilder.*;
+import static org.springframework.data.elasticsearch.core.document.Document.create;
+import static org.springframework.data.elasticsearch.core.document.Document.parse;
+import static org.springframework.data.elasticsearch.utils.IdGenerator.nextIdAsString;
+import static org.springframework.data.elasticsearch.utils.IndexBuilder.buildIndex;
 
-import java.lang.Double;
-import java.lang.Integer;
-import java.lang.Long;
-import java.lang.Object;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -78,7 +79,20 @@ import org.springframework.data.elasticsearch.core.index.AliasActions;
 import org.springframework.data.elasticsearch.core.index.Settings;
 import org.springframework.data.elasticsearch.core.join.JoinField;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
-import org.springframework.data.elasticsearch.core.query.*;
+import org.springframework.data.elasticsearch.core.query.BaseQueryBuilder;
+import org.springframework.data.elasticsearch.core.query.Criteria;
+import org.springframework.data.elasticsearch.core.query.CriteriaQuery;
+import org.springframework.data.elasticsearch.core.query.FetchSourceFilterBuilder;
+import org.springframework.data.elasticsearch.core.query.HighlightQuery;
+import org.springframework.data.elasticsearch.core.query.IndexQuery;
+import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder;
+import org.springframework.data.elasticsearch.core.query.IndicesOptions;
+import org.springframework.data.elasticsearch.core.query.MoreLikeThisQuery;
+import org.springframework.data.elasticsearch.core.query.Query;
+import org.springframework.data.elasticsearch.core.query.SeqNoPrimaryTerm;
+import org.springframework.data.elasticsearch.core.query.SourceFilter;
+import org.springframework.data.elasticsearch.core.query.StringQuery;
+import org.springframework.data.elasticsearch.core.query.UpdateQuery;
 import org.springframework.data.elasticsearch.core.query.highlight.Highlight;
 import org.springframework.data.elasticsearch.core.query.highlight.HighlightField;
 import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
@@ -480,6 +494,39 @@ public abstract class ElasticsearchIntegrationTests {
 		assertThat(searchHits.getTotalHits()).isEqualTo(2);
 	}
 
+	@Test // #2362
+	@DisplayName("should do bulk index into different indices")
+	void shouldDoBulkIndexIntoDifferentIndices() {
+
+		var indexName = indexNameProvider.indexName();
+		var documentId1 = "1";
+		var sampleEntity1 = SampleEntity.builder().id(documentId1).message("some message").build();
+		var indexQuery1 = new IndexQueryBuilder() //
+				.withId(documentId1) //
+				.withObject(sampleEntity1) //
+				.withIndex(indexName + "-" + documentId1) //
+				.build();
+		var documentId2 = "2";
+		var sampleEntity2 = SampleEntity.builder().id(documentId2).message("some message").build();
+		var indexQuery2 = new IndexQueryBuilder() //
+				.withId(documentId2) //
+				.withObject(sampleEntity2) //
+				.withIndex(indexName + "-" + documentId2) //
+				.build();
+
+		var indexQueries = Arrays.asList(indexQuery1, indexQuery2);
+
+		operations.bulkIndex(indexQueries, IndexCoordinates.of(indexName));
+
+		var searchHits = operations.search(operations.matchAllQuery(), SampleEntity.class,
+				IndexCoordinates.of(indexName + "*"));
+
+		assertThat(searchHits.getTotalHits()).isEqualTo(2);
+		searchHits.forEach(searchHit -> {
+			assertThat(searchHit.getIndex()).isEqualTo(indexName + "-" + searchHit.getId());
+		});
+	}
+
 	@Test
 	public void shouldDoBulkUpdate() {
 
@@ -4548,12 +4595,12 @@ public abstract class ElasticsearchIntegrationTests {
 		@Nullable
 		@Field(type = Text) private String lastName;
 
-@Field(type = Keyword)
+		@Field(type = Keyword)
-@WriteOnlyProperty
+		@WriteOnlyProperty
-@AccessType(AccessType.Type.PROPERTY)
+		@AccessType(AccessType.Type.PROPERTY)
-public String getFullName() {
+		public String getFullName() {
 			return sanitize(firstName) + sanitize(lastName);
-}
+		}
 
 		@Nullable
 		public String getId() {