HSEARCH-2761 Elasticsearch purges may fail when purging multiple classes of the same inheritance tree in the same transaction

elasticsearch/src/main/java/org/hibernate/search/elasticsearch/impl/ElasticsearchIndexWorkVisitor.java

@@ -7,7 +7,6 @@
 package org.hibernate.search.elasticsearch.impl;
 
 import java.util.List;
-import java.util.Set;
 
 import org.apache.lucene.document.Document;
 import org.apache.lucene.facet.FacetsConfig;
@@ -113,10 +112,11 @@ public ElasticsearchWork<?> visitPurgeAllWork(PurgeAllLuceneWork work, IndexingM
 				.luceneWork( work )
 				.markIndexDirty( refreshAfterWrite );
 
-		Set<Class<?>> typesToDelete = searchIntegrator.getIndexedTypesPolymorphic( new Class<?>[] { work.getEntityClass() } );
-		for ( Class<?> typeToDelete : typesToDelete ) {
-			builder.type( URLEncodedString.fromString( typeToDelete.getName() ) );
-		}
+		/*
+		 * Deleting only the given type.
+		 * Inheritance trees are handled at a higher level by creating multiple purge works.
+		 */
+		builder.type( URLEncodedString.fromString( work.getEntityClass().getName() ) );
 
 		return builder.build();
 	}

elasticsearch/src/main/java/org/hibernate/search/elasticsearch/processor/impl/ElasticsearchWorkProcessor.java

@@ -102,7 +102,10 @@ public <T> T executeSyncUnsafe(ElasticsearchWork<T> work) {
 	 * @param works The works to be executed.
 	 */
 	public void executeSyncSafe(Iterable<ElasticsearchWork<?>> works) {
-		executeSafe( works, true );
+		SequentialWorkExecutionContext context = new SequentialWorkExecutionContext(
+				client, gsonProvider, workFactory, this, errorHandler );
+		executeSafe( context, works, true );
+		context.flush();
 	}
 
 	/**
@@ -149,9 +152,8 @@ public void awaitAsyncProcessingCompletion() {
 	 * @param nonBulkedWorks The works to be bulked (as much as possible) and executed
 	 * @param refreshInBulkAPICall The parameter to pass to {@link #createRequestGroups(Iterable, boolean)}.
 	 */
-	private void executeSafe(Iterable<ElasticsearchWork<?>> nonBulkedWorks, boolean refreshInBulkAPICall) {
-		SequentialWorkExecutionContext context = new SequentialWorkExecutionContext(
-				client, gsonProvider, workFactory, this, errorHandler );
+	private void executeSafe(SequentialWorkExecutionContext context, Iterable<ElasticsearchWork<?>> nonBulkedWorks,
+			boolean refreshInBulkAPICall) {
 		ErrorContextBuilder errorContextBuilder = new ErrorContextBuilder();
 
 		for ( ElasticsearchWork<?> work : createRequestGroups( nonBulkedWorks, refreshInBulkAPICall ) ) {
@@ -183,8 +185,6 @@ private void executeSafe(Iterable<ElasticsearchWork<?>> nonBulkedWorks, boolean
 				break;
 			}
 		}
-
-		context.flush();
 	}
 
 	private void executeUnsafe(ElasticsearchWork<?> work, ElasticsearchWorkExecutionContext context) {
@@ -344,7 +344,7 @@ private void processAsyncWork() {
 						return;
 					}
 					Iterable<ElasticsearchWork<?>> flattenedWorks = CollectionHelper.flatten( works );
-					executeSafe( flattenedWorks, false );
+					executeSafe( context, flattenedWorks, false );
 				}
 			}
 		}

elasticsearch/src/main/java/org/hibernate/search/elasticsearch/work/impl/ES2DeleteByQueryWork.java

@@ -17,6 +17,8 @@
 import org.hibernate.search.elasticsearch.logging.impl.Log;
 import org.hibernate.search.elasticsearch.util.impl.ElasticsearchClientUtils;
 import org.hibernate.search.elasticsearch.work.impl.builder.DeleteByQueryWorkBuilder;
+import org.hibernate.search.elasticsearch.work.impl.builder.RefreshWorkBuilder;
+import org.hibernate.search.elasticsearch.work.impl.factory.ElasticsearchWorkFactory;
 import org.hibernate.search.exception.AssertionFailure;
 import org.hibernate.search.exception.SearchException;
 import org.hibernate.search.util.logging.impl.LoggerFactory;
@@ -30,8 +32,19 @@
  */
 public class ES2DeleteByQueryWork extends SimpleElasticsearchWork<Void> {
 
+	private final ElasticsearchWork<?> refreshWork;
+
 	protected ES2DeleteByQueryWork(Builder builder) {
 		super( builder );
+		this.refreshWork = builder.buildRefreshWork();
+	}
+
+	@Override
+	protected void beforeExecute(ElasticsearchWorkExecutionContext executionContext, ElasticsearchRequest request) {
+		/*
+		 * Refresh the index so as to minimize the risk of version conflict
+		 */
+		refreshWork.execute( executionContext );
 	}
 
 	@Override
@@ -46,10 +59,13 @@ public static class Builder
 		private final JsonObject payload;
 		private final Set<URLEncodedString> typeNames = new HashSet<>();
 
-		public Builder(URLEncodedString indexName, JsonObject payload) {
+		private final RefreshWorkBuilder refreshWorkBuilder;
+
+		public Builder(URLEncodedString indexName, JsonObject payload, ElasticsearchWorkFactory workFactory) {
 			super( indexName, SuccessAssessor.INSTANCE );
 			this.indexName = indexName;
 			this.payload = payload;
+			this.refreshWorkBuilder = workFactory.refresh().index( indexName );
 		}
 
 		@Override
@@ -74,6 +90,10 @@ protected ElasticsearchRequest buildRequest() {
 			return builder.build();
 		}
 
+		protected ElasticsearchWork<?> buildRefreshWork() {
+			return refreshWorkBuilder.build();
+		}
+
 		@Override
 		public ES2DeleteByQueryWork build() {
 			return new ES2DeleteByQueryWork( this );

elasticsearch/src/main/java/org/hibernate/search/elasticsearch/work/impl/ES5DeleteByQueryWork.java

@@ -14,6 +14,8 @@
 import org.hibernate.search.elasticsearch.client.impl.Paths;
 import org.hibernate.search.elasticsearch.client.impl.URLEncodedString;
 import org.hibernate.search.elasticsearch.work.impl.builder.DeleteByQueryWorkBuilder;
+import org.hibernate.search.elasticsearch.work.impl.builder.RefreshWorkBuilder;
+import org.hibernate.search.elasticsearch.work.impl.factory.ElasticsearchWorkFactory;
 
 import com.google.gson.JsonObject;
 
@@ -24,8 +26,19 @@
  */
 public class ES5DeleteByQueryWork extends SimpleElasticsearchWork<Void> {
 
+	private final ElasticsearchWork<?> refreshWork;
+
 	protected ES5DeleteByQueryWork(Builder builder) {
 		super( builder );
+		this.refreshWork = builder.buildRefreshWork();
+	}
+
+	@Override
+	protected void beforeExecute(ElasticsearchWorkExecutionContext executionContext, ElasticsearchRequest request) {
+		/*
+		 * Refresh the index so as to minimize the risk of version conflict
+		 */
+		refreshWork.execute( executionContext );
 	}
 
 	@Override
@@ -40,10 +53,13 @@ public static class Builder
 		private final JsonObject payload;
 		private final Set<URLEncodedString> typeNames = new HashSet<>();
 
-		public Builder(URLEncodedString indexName, JsonObject payload) {
+		private final RefreshWorkBuilder refreshWorkBuilder;
+
+		public Builder(URLEncodedString indexName, JsonObject payload, ElasticsearchWorkFactory workFactory) {
 			super( indexName, DefaultElasticsearchRequestSuccessAssessor.INSTANCE );
 			this.indexName = indexName;
 			this.payload = payload;
+			this.refreshWorkBuilder = workFactory.refresh().index( indexName );
 		}
 
 		@Override
@@ -56,7 +72,12 @@ public Builder type(URLEncodedString typeName) {
 		protected ElasticsearchRequest buildRequest() {
 			ElasticsearchRequest.Builder builder =
 					ElasticsearchRequest.post()
-					.pathComponent( indexName );
+					.pathComponent( indexName )
+					/*
+					 * Ignore conflicts: if we wrote to a document concurrently,
+					 * we just want to keep it as is.
+					 */
+					.param( "conflicts", "proceed" );
 
 			if ( !typeNames.isEmpty() ) {
 				builder.multiValuedPathComponent( typeNames );
@@ -68,6 +89,10 @@ protected ElasticsearchRequest buildRequest() {
 			return builder.build();
 		}
 
+		protected ElasticsearchWork<?> buildRefreshWork() {
+			return refreshWorkBuilder.build();
+		}
+
 		@Override
 		public ES5DeleteByQueryWork build() {
 			return new ES5DeleteByQueryWork( this );

elasticsearch/src/main/java/org/hibernate/search/elasticsearch/work/impl/factory/Elasticsearch2WorkFactory.java

@@ -83,7 +83,7 @@ public DeleteWorkBuilder delete(URLEncodedString indexName, URLEncodedString typ
 
 	@Override
 	public DeleteByQueryWorkBuilder deleteByQuery(URLEncodedString indexName, JsonObject payload) {
-		return new ES2DeleteByQueryWork.Builder( indexName, payload );
+		return new ES2DeleteByQueryWork.Builder( indexName, payload, this );
 	}
 
 	@Override

elasticsearch/src/main/java/org/hibernate/search/elasticsearch/work/impl/factory/Elasticsearch5WorkFactory.java

@@ -38,7 +38,7 @@ public OptimizeWorkBuilder optimize() {
 
 	@Override
 	public DeleteByQueryWorkBuilder deleteByQuery(URLEncodedString indexName, JsonObject payload) {
-		return new ES5DeleteByQueryWork.Builder( indexName, payload );
+		return new ES5DeleteByQueryWork.Builder( indexName, payload, this );
 	}
 
 }