DATAES-799 - Support optimistic locking for full update scenario using seq_no + primary_term

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

@@ -53,6 +53,7 @@
 import org.springframework.data.elasticsearch.core.query.MoreLikeThisQuery;
 import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
 import org.springframework.data.elasticsearch.core.query.Query;
+import org.springframework.data.elasticsearch.core.query.SeqNoPrimaryTerm;
 import org.springframework.data.elasticsearch.support.VersionInfo;
 import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.util.CloseableIterator;
@@ -438,18 +439,40 @@ private Long getEntityVersion(Object entity) {
 		return null;
 	}
 
+	@Nullable
+	private SeqNoPrimaryTerm getEntitySeqNoPrimaryTerm(Object entity) {
+		ElasticsearchPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
+		ElasticsearchPersistentProperty property = persistentEntity.getSeqNoPrimaryTermProperty();
+
+		if (property != null) {
+			Object seqNoPrimaryTerm = persistentEntity.getPropertyAccessor(entity).getProperty(property);
+
+			if (seqNoPrimaryTerm != null && SeqNoPrimaryTerm.class.isAssignableFrom(seqNoPrimaryTerm.getClass())) {
+				return (SeqNoPrimaryTerm) seqNoPrimaryTerm;
+			}
+		}
+
+		return null;
+	}
+
 	private <T> IndexQuery getIndexQuery(T entity) {
 		String id = getEntityId(entity);
 
 		if (id != null) {
 			id = elasticsearchConverter.convertId(id);
 		}
 
-		return new IndexQueryBuilder() //
+		IndexQueryBuilder builder = new IndexQueryBuilder() //
 				.withId(id) //
-				.withVersion(getEntityVersion(entity)) //
-				.withObject(entity) //
-				.build();
+				.withObject(entity);
+		SeqNoPrimaryTerm seqNoPrimaryTerm = getEntitySeqNoPrimaryTerm(entity);
+		if (seqNoPrimaryTerm != null) {
+			builder.withSeqNoPrimaryTerm(seqNoPrimaryTerm);
+		} else {
+			// version cannot be used together with seq_no and primary_term
+			builder.withVersion(getEntityVersion(entity));
+		}
+		return builder.build();
 	}
 
 	/**

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

@@ -250,7 +250,7 @@ default void bulkUpdate(List<UpdateQuery> queries, IndexCoordinates index) {
 	 * @param clazz the type of the object to be returned
 	 * @param index the index from which the object is read.
 	 * @return the found object
-	 * @deprecated since 4.0, use {@link #getById(String, Class, IndexCoordinates)}
+	 * @deprecated since 4.0, use {@link #get(String, Class, IndexCoordinates)}
 	 */
 	@Deprecated
 	@Nullable

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

@@ -22,9 +22,12 @@
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.common.ValidationException;
+import org.elasticsearch.index.engine.VersionConflictEngineException;
+import org.elasticsearch.rest.RestStatus;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.DataAccessResourceFailureException;
 import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.elasticsearch.NoSuchIndexException;
 import org.springframework.data.elasticsearch.UncategorizedElasticsearchException;
@@ -35,6 +38,7 @@
 /**
  * @author Christoph Strobl
  * @author Peter-Josef Meisch
+ * @author Roman Puchkovskiy
  * @since 3.2
  */
 public class ElasticsearchExceptionTranslator implements PersistenceExceptionTranslator {
@@ -50,6 +54,12 @@ public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
 				return new NoSuchIndexException(ObjectUtils.nullSafeToString(elasticsearchException.getMetadata("es.index")),
 						ex);
 			}
+
+			if (isSeqNoConflict(elasticsearchException)) {
+				return new OptimisticLockingFailureException("Cannot index a document due to seq_no+primary_term conflict",
+						elasticsearchException);
+			}
+
 			return new UncategorizedElasticsearchException(ex.getMessage(), ex);
 		}
 
@@ -65,6 +75,25 @@ public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
 		return null;
 	}
 
+	private boolean isSeqNoConflict(ElasticsearchException exception) {
+
+		if (exception instanceof ElasticsearchStatusException) {
+			ElasticsearchStatusException statusException = (ElasticsearchStatusException) exception;
+			return statusException.status() == RestStatus.CONFLICT
+					&& statusException.getMessage() != null
+					&& statusException.getMessage().contains("type=version_conflict_engine_exception")
+					&& statusException.getMessage().contains("version conflict, required seqNo");
+		}
+
+		if (exception instanceof VersionConflictEngineException) {
+			VersionConflictEngineException versionConflictEngineException = (VersionConflictEngineException) exception;
+			return versionConflictEngineException.getMessage() != null
+					&& versionConflictEngineException.getMessage().contains("version conflict, required seqNo");
+		}
+
+		return false;
+	}
+
 	private boolean indexAvailable(ElasticsearchException ex) {
 
 		List<String> metadata = ex.getMetadata("es.index_uuid");

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

@@ -27,6 +27,7 @@
 import org.elasticsearch.action.get.GetResponse;
 import org.elasticsearch.action.get.MultiGetRequestBuilder;
 import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.search.MultiSearchRequest;
 import org.elasticsearch.action.search.MultiSearchResponse;
 import org.elasticsearch.action.search.SearchRequestBuilder;
@@ -89,6 +90,8 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate {
 	private Client client;
 	@Nullable private String searchTimeout;
 
+	private final ElasticsearchExceptionTranslator exceptionTranslator = new ElasticsearchExceptionTranslator();
+
 	// region Initialization
 	public ElasticsearchTemplate(Client client) {
 		this.client = client;
@@ -145,7 +148,14 @@ public String index(IndexQuery query, IndexCoordinates index) {
 		maybeCallbackBeforeConvertWithQuery(query, index);
 
 		IndexRequestBuilder indexRequestBuilder = requestFactory.indexRequestBuilder(client, query, index);
-		String documentId = indexRequestBuilder.execute().actionGet().getId();
+		ActionFuture<IndexResponse> future = indexRequestBuilder.execute();
+		IndexResponse response;
+		try {
+			response = future.actionGet();
+		} catch (RuntimeException e) {
+			throw translateException(e);
+		}
+		String documentId = response.getId();
 
 		// We should call this because we are not going through a mapper.
 		Object queryObject = query.getObject();
@@ -360,4 +370,22 @@ public Client getClient() {
 		return client;
 	}
 	// endregion
+
+	/**
+	 * translates an Exception if possible. Exceptions that are no {@link RuntimeException}s are wrapped in a
+	 * RuntimeException
+	 *
+	 * @param exception the Exception to map
+	 * @return the potentially translated RuntimeException.
+	 * @since 4.0
+	 */
+	private RuntimeException translateException(Exception exception) {
+
+		RuntimeException runtimeException = exception instanceof RuntimeException ? (RuntimeException) exception
+				: new RuntimeException(exception.getMessage(), exception);
+		RuntimeException potentiallyTranslatedException = exceptionTranslator
+				.translateExceptionIfPossible(runtimeException);
+
+		return potentiallyTranslatedException != null ? potentiallyTranslatedException : runtimeException;
+	}
 }

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

@@ -21,6 +21,7 @@
 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.SeqNoPrimaryTerm;
 import org.springframework.data.mapping.IdentifierAccessor;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.context.MappingContext;
@@ -35,6 +36,7 @@
  * @author Mark Paluch
  * @author Christoph Strobl
  * @author Peter-Josef Meisch
+ * @author Roman Puchkovskiy
  * @since 3.2
  */
 class EntityOperations {
@@ -256,6 +258,21 @@ interface AdaptibleEntity<T> extends Entity<T> {
 		@Override
 		@Nullable
 		Number getVersion();
+
+		/**
+		 * Returns whether there is a property with type SeqNoPrimaryTerm in this entity.
+		 *
+		 * @return true if there is SeqNoPrimaryTerm property
+		 * @since 4.0
+		 */
+		boolean hasSeqNoPrimaryTerm();
+
+		/**
+		 * Returns SeqNoPropertyTerm for this entity.
+		 *
+		 * @return SeqNoPrimaryTerm, may be {@literal null}
+		 */
+		@Nullable SeqNoPrimaryTerm getSeqNoPrimaryTerm();
 	}
 
 	/**
@@ -333,6 +350,16 @@ public Number getVersion() {
 			return null;
 		}
 
+		@Override
+		public boolean hasSeqNoPrimaryTerm() {
+			return false;
+		}
+
+		@Override
+		public SeqNoPrimaryTerm getSeqNoPrimaryTerm() {
+			return null;
+		}
+
 		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.elasticsearch.core.EntityOperations.AdaptibleEntity#incrementVersion()
@@ -588,6 +615,19 @@ public Number getVersion() {
 			return propertyAccessor.getProperty(versionProperty, Number.class);
 		}
 
+		@Override
+		public boolean hasSeqNoPrimaryTerm() {
+			return entity.hasSeqNoPrimaryTermProperty();
+		}
+
+		@Override
+		public SeqNoPrimaryTerm getSeqNoPrimaryTerm() {
+
+			ElasticsearchPersistentProperty seqNoPrimaryTermProperty = entity.getRequiredSeqNoPrimaryTermProperty();
+
+			return propertyAccessor.getProperty(seqNoPrimaryTermProperty, SeqNoPrimaryTerm.class);
+		}
+
 		/* 
 		 * (non-Javadoc)
 		 * @see org.springframework.data.elasticsearch.core.EntityOperations.AdaptibleEntity#initializeVersionProperty()

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

@@ -84,6 +84,7 @@
 import org.springframework.data.elasticsearch.core.query.IndexQuery;
 import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
 import org.springframework.data.elasticsearch.core.query.Query;
+import org.springframework.data.elasticsearch.core.query.SeqNoPrimaryTerm;
 import org.springframework.data.elasticsearch.core.query.StringQuery;
 import org.springframework.data.elasticsearch.core.query.UpdateQuery;
 import org.springframework.data.elasticsearch.support.VersionInfo;
@@ -347,7 +348,19 @@ private IndexRequest getIndexRequest(Object value, AdaptibleEntity<?> entity, In
 
 		request.source(converter.mapObject(value).toJson(), Requests.INDEX_CONTENT_TYPE);
 
-		if (entity.isVersionedEntity()) {
+		boolean usingSeqNo = false;
+		if (entity.hasSeqNoPrimaryTerm()) {
+			SeqNoPrimaryTerm seqNoPrimaryTerm = entity.getSeqNoPrimaryTerm();
+
+			if (seqNoPrimaryTerm != null) {
+				request.setIfSeqNo(seqNoPrimaryTerm.getSequenceNumber());
+				request.setIfPrimaryTerm(seqNoPrimaryTerm.getPrimaryTerm());
+				usingSeqNo = true;
+			}
+		}
+
+		// seq_no and version are incompatible in the same request
+		if (!usingSeqNo && entity.isVersionedEntity()) {
 
 			Number version = entity.getVersion();
 
@@ -356,6 +369,7 @@ private IndexRequest getIndexRequest(Object value, AdaptibleEntity<?> entity, In
 				request.versionType(EXTERNAL);
 			}
 		}
+
 		return request;
 	}
 

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

@@ -83,6 +83,7 @@
  *
  * @author Peter-Josef Meisch
  * @author Sascha Woo
+ * @author Roman Puchkovskiy
  * @since 4.0
  */
 class RequestFactory {
@@ -342,6 +343,13 @@ public IndexRequest indexRequest(IndexQuery query, IndexCoordinates index) {
 			indexRequest.versionType(versionType);
 		}
 
+		if (query.getSeqNo() != null) {
+			indexRequest.setIfSeqNo(query.getSeqNo());
+		}
+		if (query.getPrimaryTerm() != null) {
+			indexRequest.setIfPrimaryTerm(query.getPrimaryTerm());
+		}
+
 		return indexRequest;
 	}
 
@@ -374,6 +382,13 @@ public IndexRequestBuilder indexRequestBuilder(Client client, IndexQuery query,
 			indexRequestBuilder.setVersionType(versionType);
 		}
 
+		if (query.getSeqNo() != null) {
+			indexRequestBuilder.setIfSeqNo(query.getSeqNo());
+		}
+		if (query.getPrimaryTerm() != null) {
+			indexRequestBuilder.setIfPrimaryTerm(query.getPrimaryTerm());
+		}
+
 		return indexRequestBuilder;
 	}
 
@@ -618,6 +633,9 @@ private SearchRequest prepareSearchRequest(Query query, @Nullable Class<?> clazz
 		SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
 		sourceBuilder.version(true);
 		sourceBuilder.trackScores(query.getTrackScores());
+		if (hasSeqNoPrimaryTermProperty(clazz)) {
+			sourceBuilder.seqNoAndPrimaryTerm(true);
+		}
 
 		if (query.getSourceFilter() != null) {
 			SourceFilter sourceFilter = query.getSourceFilter();
@@ -681,7 +699,20 @@ private SearchRequest prepareSearchRequest(Query query, @Nullable Class<?> clazz
 		return request;
 	}
 
-	@SuppressWarnings("unchecked")
+	private boolean hasSeqNoPrimaryTermProperty(@Nullable Class<?> entityClass) {
+
+		if (entityClass == null) {
+			return false;
+		}
+
+		if (!elasticsearchConverter.getMappingContext().hasPersistentEntityFor(entityClass)) {
+			return false;
+		}
+
+		ElasticsearchPersistentEntity<?> entity = elasticsearchConverter.getMappingContext().getRequiredPersistentEntity(entityClass);
+		return entity.hasSeqNoPrimaryTermProperty();
+	}
+
 	public PutMappingRequest putMappingRequest(IndexCoordinates index, Document mapping) {
 		PutMappingRequest request = new PutMappingRequest(index.getIndexName());
 		request.source(mapping);
@@ -784,6 +815,9 @@ private SearchRequestBuilder prepareSearchRequestBuilder(Query query, Client cli
 				.setSearchType(query.getSearchType()) //
 				.setVersion(true) //
 				.setTrackScores(query.getTrackScores());
+		if (hasSeqNoPrimaryTermProperty(clazz)) {
+			searchRequestBuilder.seqNoAndPrimaryTerm(true);
+		}
 
 		if (query.getSourceFilter() != null) {
 			SourceFilter sourceFilter = query.getSourceFilter();