DATAES-751 - Introduce ClientCallback for the rest client.

Original PR: #401

src/main/java/org/springframework/data/elasticsearch/ElasticsearchException.java

@@ -25,7 +25,9 @@ import org.springframework.lang.Nullable;
  * @author Rizwan Idrees
  * @author Mohsin Husen
  * @author Peter-Josef Meisch
+ * @deprecated since 4.0, use {@link org.springframework.dao.UncategorizedDataAccessException}
  */
+@Deprecated
 public class ElasticsearchException extends RuntimeException {
 
 	@Nullable private Map<String, String> failedDocuments;

src/main/java/org/springframework/data/elasticsearch/UncategorizedElasticsearchException.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch;
+
+import org.springframework.dao.UncategorizedDataAccessException;
+
+/**
+ * @author Peter-Josef Meisch
+ * @since 4.0
+ */
+public class UncategorizedElasticsearchException extends UncategorizedDataAccessException {
+	public UncategorizedElasticsearchException(String msg, Throwable cause) {
+		super(msg, cause);
+	}
+}

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

@@ -33,14 +33,12 @@ import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.client.indices.CreateIndexRequest;
 import org.elasticsearch.client.indices.GetIndexRequest;
 import org.elasticsearch.client.indices.PutMappingRequest;
 import org.elasticsearch.cluster.metadata.AliasMetaData;
-import org.springframework.data.elasticsearch.ElasticsearchException;
+import org.springframework.data.elasticsearch.UncategorizedElasticsearchException;
 import org.springframework.data.elasticsearch.core.client.support.AliasData;
-import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.document.Document;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.query.AliasQuery;
@@ -60,29 +58,22 @@ import com.fasterxml.jackson.databind.ObjectMapper;
  */
 class DefaultIndexOperations extends AbstractDefaultIndexOperations implements IndexOperations {
 
-	private RestHighLevelClient client;
+	private ElasticsearchRestTemplate restTemplate;
 
-	public DefaultIndexOperations(RestHighLevelClient client, ElasticsearchConverter elasticsearchConverter,
+	public DefaultIndexOperations(ElasticsearchRestTemplate restTemplate, Class<?> boundClass) {
-			Class<?> boundClass) {
+		super(restTemplate.getElasticsearchConverter(), boundClass);
-		super(elasticsearchConverter, boundClass);
+		this.restTemplate = restTemplate;
-		this.client = client;
 	}
 
-	public DefaultIndexOperations(RestHighLevelClient client, ElasticsearchConverter elasticsearchConverter,
+	public DefaultIndexOperations(ElasticsearchRestTemplate restTemplate, IndexCoordinates boundIndex) {
-			IndexCoordinates boundIndex) {
+		super(restTemplate.getElasticsearchConverter(), boundIndex);
-		super(elasticsearchConverter, boundIndex);
+		this.restTemplate = restTemplate;
-		this.client = client;
 	}
 
 	@Override
 	protected boolean doCreate(String indexName, @Nullable Document settings) {
 		CreateIndexRequest request = requestFactory.createIndexRequest(indexName, settings);
-		try {
+		return restTemplate.execute(client -> client.indices().create(request, RequestOptions.DEFAULT).isAcknowledged());
-			return client.indices().create(request, RequestOptions.DEFAULT).isAcknowledged();
-		} catch (IOException e) {
-			throw new ElasticsearchException(
-					"Error for creating index: " + indexName + ", client: " + client.getLowLevelClient().getNodes(), e);
-		}
 	}
 
 	@Override
@@ -92,11 +83,7 @@ class DefaultIndexOperations extends AbstractDefaultIndexOperations implements I
 
 		if (doExists(indexName)) {
 			DeleteIndexRequest request = new DeleteIndexRequest(indexName);
-			try {
+			return restTemplate.execute(client -> client.indices().delete(request, RequestOptions.DEFAULT).isAcknowledged());
-				return client.indices().delete(request, RequestOptions.DEFAULT).isAcknowledged();
-			} catch (IOException e) {
-				throw new ElasticsearchException("Error while deleting index request: " + request.toString(), e);
-			}
 		}
 		return false;
 	}
@@ -104,11 +91,7 @@ class DefaultIndexOperations extends AbstractDefaultIndexOperations implements I
 	@Override
 	protected boolean doExists(String indexName) {
 		GetIndexRequest request = new GetIndexRequest(indexName);
-		try {
+		return restTemplate.execute(client -> client.indices().exists(request, RequestOptions.DEFAULT));
-			return client.indices().exists(request, RequestOptions.DEFAULT);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error while for indexExists request: " + request.toString(), e);
-		}
 	}
 
 	@Override
@@ -117,11 +100,8 @@ class DefaultIndexOperations extends AbstractDefaultIndexOperations implements I
 		Assert.notNull(index, "No index defined for putMapping()");
 
 		PutMappingRequest request = requestFactory.putMappingRequest(index, mapping);
-		try {
+		return restTemplate
-			return client.indices().putMapping(request, RequestOptions.DEFAULT).isAcknowledged();
+				.execute(client -> client.indices().putMapping(request, RequestOptions.DEFAULT).isAcknowledged());
-		} catch (IOException e) {
-			throw new ElasticsearchException("Failed to put mapping for " + index.getIndexName(), e);
-		}
 	}
 
 	@Override
@@ -129,24 +109,19 @@ class DefaultIndexOperations extends AbstractDefaultIndexOperations implements I
 
 		Assert.notNull(index, "No index defined for getMapping()");
 
-		RestClient restClient = client.getLowLevelClient();
+		return restTemplate.execute(client -> {
-		try {
+			RestClient restClient = client.getLowLevelClient();
 			Request request = new Request("GET", '/' + index.getIndexName() + "/_mapping");
 			Response response = restClient.performRequest(request);
 			return convertMappingResponse(EntityUtils.toString(response.getEntity()));
-		} catch (Exception e) {
+		});
-			throw new ElasticsearchException("Error while getting mapping for indexName : " + index.getIndexName(), e);
-		}
 	}
 
 	@Override
 	protected boolean doAddAlias(AliasQuery query, IndexCoordinates index) {
 		IndicesAliasesRequest request = requestFactory.indicesAddAliasesRequest(query, index);
-		try {
+		return restTemplate
-			return client.indices().updateAliases(request, RequestOptions.DEFAULT).isAcknowledged();
+				.execute(client -> client.indices().updateAliases(request, RequestOptions.DEFAULT).isAcknowledged());
-		} catch (IOException e) {
-			throw new ElasticsearchException("failed to update aliases with request: " + request, e);
-		}
 	}
 
 	@Override
@@ -156,29 +131,23 @@ class DefaultIndexOperations extends AbstractDefaultIndexOperations implements I
 		Assert.notNull(query.getAliasName(), "No alias defined");
 
 		IndicesAliasesRequest indicesAliasesRequest = requestFactory.indicesRemoveAliasesRequest(query, index);
-		try {
+		return restTemplate.execute(
-			return client.indices().updateAliases(indicesAliasesRequest, RequestOptions.DEFAULT).isAcknowledged();
+				client -> client.indices().updateAliases(indicesAliasesRequest, RequestOptions.DEFAULT).isAcknowledged());
-		} catch (IOException e) {
-			throw new ElasticsearchException(
-					"failed to update aliases with indicesRemoveAliasesRequest: " + indicesAliasesRequest, e);
-		}
 	}
 
 	@Override
 	protected List<AliasMetaData> doQueryForAlias(String indexName) {
 		List<AliasMetaData> aliases = null;
-		RestClient restClient = client.getLowLevelClient();
+		return restTemplate.execute(client -> {
-		Response response;
+			RestClient restClient = client.getLowLevelClient();
-		String aliasResponse;
+			Response response;
+			String aliasResponse;
 
-		try {
 			response = restClient.performRequest(new Request("GET", '/' + indexName + "/_alias/*"));
 			aliasResponse = EntityUtils.toString(response.getEntity());
-		} catch (Exception e) {
-			throw new ElasticsearchException("Error while getting mapping for indexName : " + indexName, e);
-		}
 
-		return convertAliasResponse(aliasResponse);
+			return convertAliasResponse(aliasResponse);
+		});
 	}
 
 	@Override
@@ -190,14 +159,11 @@ class DefaultIndexOperations extends AbstractDefaultIndexOperations implements I
 				.indices(indexName) //
 				.includeDefaults(includeDefaults);
 
-		try {
+		//
-			GetSettingsResponse response = client.indices() //
+		GetSettingsResponse response = restTemplate.execute(client -> client.indices() //
-					.getSettings(request, RequestOptions.DEFAULT);
+				.getSettings(request, RequestOptions.DEFAULT));
 
-			return convertSettingsResponseToMap(response, indexName);
+		return convertSettingsResponseToMap(response, indexName);
-		} catch (IOException e) {
-			throw new ElasticsearchException("failed to get settings for index: " + indexName, e);
-		}
 	}
 
 	@Override
@@ -205,11 +171,8 @@ class DefaultIndexOperations extends AbstractDefaultIndexOperations implements I
 
 		Assert.notNull(index, "No index defined for refresh()");
 
-		try {
+		restTemplate
-			client.indices().refresh(refreshRequest(index.getIndexNames()), RequestOptions.DEFAULT);
+				.execute(client -> client.indices().refresh(refreshRequest(index.getIndexNames()), RequestOptions.DEFAULT));
-		} catch (IOException e) {
-			throw new ElasticsearchException("failed to refresh index: " + index, e);
-		}
 	}
 
 	// region Helper methods
@@ -225,7 +188,7 @@ class DefaultIndexOperations extends AbstractDefaultIndexOperations implements I
 
 			return result;
 		} catch (IOException e) {
-			throw new ElasticsearchException("Could not map alias response : " + mappingResponse, e);
+			throw new UncategorizedElasticsearchException("Could not map alias response : " + mappingResponse, e);
 		}
 	}
 
@@ -261,7 +224,7 @@ class DefaultIndexOperations extends AbstractDefaultIndexOperations implements I
 			}
 			return aliasMetaDataList;
 		} catch (IOException e) {
-			throw new ElasticsearchException("Could not map alias response : " + aliasResponse, e);
+			throw new UncategorizedElasticsearchException("Could not map alias response : " + aliasResponse, e);
 		}
 	}
 	// endregion

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

@@ -16,21 +16,25 @@
 
 package org.springframework.data.elasticsearch.core;
 
-import java.net.ConnectException;
+import java.io.IOException;
 import java.util.List;
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.common.ValidationException;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.elasticsearch.NoSuchIndexException;
+import org.springframework.data.elasticsearch.UncategorizedElasticsearchException;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
 /**
  * @author Christoph Strobl
+ * @author Peter-Josef Meisch
  * @since 3.2
  */
 public class ElasticsearchExceptionTranslator implements PersistenceExceptionTranslator {
@@ -46,9 +50,15 @@ public class ElasticsearchExceptionTranslator implements PersistenceExceptionTra
 				return new NoSuchIndexException(ObjectUtils.nullSafeToString(elasticsearchException.getMetadata("es.index")),
 						ex);
 			}
+			return new UncategorizedElasticsearchException(ex.getMessage(), ex);
 		}
 
-		if (ex.getCause() instanceof ConnectException) {
+		if (ex instanceof ValidationException) {
+			return new DataIntegrityViolationException(ex.getMessage(), ex);
+		}
+
+		Throwable cause = ex.getCause();
+		if (cause instanceof IOException) {
 			return new DataAccessResourceFailureException(ex.getMessage(), ex);
 		}
 
@@ -60,8 +70,9 @@ public class ElasticsearchExceptionTranslator implements PersistenceExceptionTra
 		List<String> metadata = ex.getMetadata("es.index_uuid");
 		if (metadata == null) {
 			if (ex instanceof ElasticsearchStatusException) {
-				return StringUtils.hasText(ObjectUtils.nullSafeToString(((ElasticsearchStatusException) ex).getIndex()));
+				return StringUtils.hasText(ObjectUtils.nullSafeToString(ex.getIndex()));
 			}
+			return false;
 		}
 		return !CollectionUtils.contains(metadata.iterator(), "_na_");
 	}

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

@@ -17,7 +17,6 @@ package org.springframework.data.elasticsearch.core;
 
 import java.io.IOException;
 import java.util.List;
-import java.util.Optional;
 
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.delete.DeleteRequest;
@@ -40,7 +39,6 @@ import org.elasticsearch.index.reindex.DeleteByQueryRequest;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.springframework.data.domain.Pageable;
-import org.springframework.data.elasticsearch.ElasticsearchException;
 import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.document.DocumentAdapters;
 import org.springframework.data.elasticsearch.core.document.SearchDocumentResponse;
@@ -90,25 +88,29 @@ import org.springframework.util.Assert;
 public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 
 	private RestHighLevelClient client;
+	private ElasticsearchExceptionTranslator exceptionTranslator;
 
 	// region Initialization
 	public ElasticsearchRestTemplate(RestHighLevelClient client) {
-		this.client = client;
-		initialize(client, createElasticsearchConverter());
-	}
-
-	public ElasticsearchRestTemplate(RestHighLevelClient client, ElasticsearchConverter elasticsearchConverter) {
-		this.client = client;
-		initialize(client, elasticsearchConverter);
-	}
-
-	private void initialize(RestHighLevelClient client, ElasticsearchConverter elasticsearchConverter) {
 
 		Assert.notNull(client, "Client must not be null!");
 
 		this.client = client;
+		this.exceptionTranslator = new ElasticsearchExceptionTranslator();
+
+		initialize(createElasticsearchConverter());
+	}
+
+	public ElasticsearchRestTemplate(RestHighLevelClient client, ElasticsearchConverter elasticsearchConverter) {
+
+		Assert.notNull(client, "Client must not be null!");
+
+		this.client = client;
+		this.exceptionTranslator = new ElasticsearchExceptionTranslator();
+
 		initialize(elasticsearchConverter);
 	}
+
 	// endregion
 
 	// region IndexOperations
@@ -117,7 +119,7 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 
 		Assert.notNull(clazz, "clazz must not be null");
 
-		return new DefaultIndexOperations(client, elasticsearchConverter, clazz);
+		return new DefaultIndexOperations(this, clazz);
 	}
 
 	@Override
@@ -125,7 +127,7 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 
 		Assert.notNull(index, "index must not be null");
 
-		return new DefaultIndexOperations(client, elasticsearchConverter, index);
+		return new DefaultIndexOperations(this, index);
 	}
 	// endregion
 
@@ -133,29 +135,21 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 	@Override
 	public String index(IndexQuery query, IndexCoordinates index) {
 		IndexRequest request = requestFactory.indexRequest(query, index);
-		try {
+		String documentId = execute(client -> client.index(request, RequestOptions.DEFAULT).getId());
-			String documentId = client.index(request, RequestOptions.DEFAULT).getId();
 
-			// We should call this because we are not going through a mapper.
+		// We should call this because we are not going through a mapper.
-			if (query.getObject() != null) {
+		if (query.getObject() != null) {
-				setPersistentEntityId(query.getObject(), documentId);
+			setPersistentEntityId(query.getObject(), documentId);
-			}
-			return documentId;
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error while index for request: " + request.toString(), e);
 		}
+		return documentId;
 	}
 
 	@Override
 	@Nullable
 	public <T> T get(String id, Class<T> clazz, IndexCoordinates index) {
 		GetRequest request = requestFactory.getRequest(id, index);
-		try {
+		GetResponse response = execute(client -> client.get(request, RequestOptions.DEFAULT));
-			GetResponse response = client.get(request, RequestOptions.DEFAULT);
+		return elasticsearchConverter.mapDocument(DocumentAdapters.from(response), clazz);
-			return elasticsearchConverter.mapDocument(DocumentAdapters.from(response), clazz);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error while getting for request: " + request.toString(), e);
-		}
 	}
 
 	@Override
@@ -165,22 +159,14 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 		Assert.notEmpty(query.getIds(), "No Id define for Query");
 
 		MultiGetRequest request = requestFactory.multiGetRequest(query, index);
-		try {
+		MultiGetResponse result = execute(client -> client.mget(request, RequestOptions.DEFAULT));
-			MultiGetResponse result = client.mget(request, RequestOptions.DEFAULT);
+		return elasticsearchConverter.mapDocuments(DocumentAdapters.from(result), clazz);
-			return elasticsearchConverter.mapDocuments(DocumentAdapters.from(result), clazz);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error while multiget for request: " + request.toString(), e);
-		}
 	}
 
 	@Override
 	protected boolean doExists(String id, IndexCoordinates index) {
 		GetRequest request = requestFactory.getRequest(id, index);
-		try {
+		return execute(client -> client.get(request, RequestOptions.DEFAULT).isExists());
-			return client.get(request, RequestOptions.DEFAULT).isExists();
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error while getting for request: " + request.toString(), e);
-		}
 	}
 
 	@Override
@@ -208,53 +194,33 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 		Assert.notNull(index, "index must not be null");
 
 		DeleteRequest request = new DeleteRequest(index.getIndexName(), elasticsearchConverter.convertId(id));
-		try {
+		return execute(client -> client.delete(request, RequestOptions.DEFAULT).getId());
-			return client.delete(request, RequestOptions.DEFAULT).getId();
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error while deleting item request: " + request.toString(), e);
-		}
 	}
 
 	@Override
 	public void delete(Query query, Class<?> clazz, IndexCoordinates index) {
 		DeleteByQueryRequest deleteByQueryRequest = requestFactory.deleteByQueryRequest(query, clazz, index);
-		try {
+		execute(client -> client.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT));
-			client.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error for delete request: " + deleteByQueryRequest.toString(), e);
-		}
 	}
 
 	@Override
 	@Deprecated
 	public void delete(DeleteQuery deleteQuery, IndexCoordinates index) {
 		DeleteByQueryRequest deleteByQueryRequest = requestFactory.deleteByQueryRequest(deleteQuery, index);
-		try {
+		execute(client -> client.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT));
-			client.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error for delete request: " + deleteByQueryRequest.toString(), e);
-		}
 	}
 
 	@Override
 	public UpdateResponse update(UpdateQuery query, IndexCoordinates index) {
 		UpdateRequest request = requestFactory.updateRequest(query, index);
-		try {
+		UpdateResponse.Result result = UpdateResponse.Result
-			org.elasticsearch.action.update.UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
+				.valueOf(execute(client -> client.update(request, RequestOptions.DEFAULT)).getResult().name());
-			UpdateResponse.Result result = UpdateResponse.Result.valueOf(updateResponse.getResult().name());
+		return new UpdateResponse(result);
-			return new UpdateResponse(result);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error while update for request: " + request.toString(), e);
-		}
 	}
 
 	private List<String> doBulkOperation(List<?> queries, BulkOptions bulkOptions, IndexCoordinates index) {
 		BulkRequest bulkRequest = requestFactory.bulkRequest(queries, bulkOptions, index);
-		try {
+		return checkForBulkOperationFailure(execute(client -> client.bulk(bulkRequest, RequestOptions.DEFAULT)));
-			return checkForBulkOperationFailure(client.bulk(bulkRequest, RequestOptions.DEFAULT));
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error while bulk for request: " + bulkRequest.toString(), e);
-		}
 	}
 	// endregion
 
@@ -272,22 +238,14 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 
 		searchRequest.source().size(0);
 
-		try {
+		return SearchHitsUtil
-			return SearchHitsUtil.getTotalCount(client.search(searchRequest, RequestOptions.DEFAULT).getHits());
+				.getTotalCount(execute(client -> client.search(searchRequest, RequestOptions.DEFAULT).getHits()));
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error for search request: " + searchRequest.toString(), e);
-		}
 	}
 
 	@Override
 	public <T> SearchHits<T> search(Query query, Class<T> clazz, IndexCoordinates index) {
 		SearchRequest searchRequest = requestFactory.searchRequest(query, clazz, index);
-		SearchResponse response;
+		SearchResponse response = execute(client -> client.search(searchRequest, RequestOptions.DEFAULT));
-		try {
-			response = client.search(searchRequest, RequestOptions.DEFAULT);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error for search request: " + searchRequest.toString(), e);
-		}
 		return elasticsearchConverter.read(clazz, SearchDocumentResponse.from(response));
 	}
 
@@ -299,13 +257,8 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 
 		SearchRequest searchRequest = requestFactory.searchRequest(query, clazz, index);
 		searchRequest.scroll(TimeValue.timeValueMillis(scrollTimeInMillis));
-
+		SearchResponse result = execute(client -> client.search(searchRequest, RequestOptions.DEFAULT));
-		try {
+		return elasticsearchConverter.mapResults(SearchDocumentResponse.from(result), clazz, null);
-			SearchResponse result = client.search(searchRequest, RequestOptions.DEFAULT);
-			return elasticsearchConverter.mapResults(SearchDocumentResponse.from(result), clazz, null);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error for search request with scroll: " + searchRequest.toString(), e);
-		}
 	}
 
 	@Override
@@ -313,12 +266,7 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 			Class<T> clazz) {
 		SearchScrollRequest request = new SearchScrollRequest(scrollId);
 		request.scroll(TimeValue.timeValueMillis(scrollTimeInMillis));
-		SearchResponse response;
+		SearchResponse response = execute(client -> client.searchScroll(request, RequestOptions.DEFAULT));
-		try {
-			response = client.searchScroll(request, RequestOptions.DEFAULT);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error for search request with scroll: " + request.toString(), e);
-		}
 		return elasticsearchConverter.mapResults(SearchDocumentResponse.from(response), clazz, Pageable.unpaged());
 	}
 
@@ -326,11 +274,7 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 	public void searchScrollClear(String scrollId) {
 		ClearScrollRequest request = new ClearScrollRequest();
 		request.addScrollId(scrollId);
-		try {
+		execute(client -> client.clearScroll(request, RequestOptions.DEFAULT));
-			client.clearScroll(request, RequestOptions.DEFAULT);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error for search request with scroll: " + request.toString(), e);
-		}
 	}
 
 	@Override
@@ -339,26 +283,66 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 		SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
 		sourceBuilder.suggest(suggestion);
 		searchRequest.source(sourceBuilder);
-
+		return execute(client -> client.search(searchRequest, RequestOptions.DEFAULT));
-		try {
-			return client.search(searchRequest, RequestOptions.DEFAULT);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Could not execute search request : " + searchRequest.toString(), e);
-		}
-
 	}
 
 	@Override
 	protected MultiSearchResponse.Item[] getMultiSearchResult(MultiSearchRequest request) {
-		MultiSearchResponse response;
+		MultiSearchResponse response = execute(client -> client.multiSearch(request, RequestOptions.DEFAULT));
-		try {
-			response = client.multiSearch(request, RequestOptions.DEFAULT);
-		} catch (IOException e) {
-			throw new ElasticsearchException("Error for search request: " + request.toString(), e);
-		}
 		MultiSearchResponse.Item[] items = response.getResponses();
 		Assert.isTrue(items.length == request.requests().size(), "Response should has same length with queries");
 		return items;
 	}
 	// endregion
+
+	// region clientcallback
+	/**
+	 * Callback interface to be used with {@link #execute(ClientCallback)} for operating directly on
+	 * {@link RestHighLevelClient}.
+	 *
+	 * @since 4.0
+	 */
+	@FunctionalInterface
+	interface ClientCallback<T> {
+		T doWithClient(RestHighLevelClient client) throws IOException;
+	}
+
+	/**
+	 * Execute a callback with the {@link RestHighLevelClient}
+	 *
+	 * @param callback the callback to execute, must not be {@literal null}
+	 * @param <T> the type returned from the callback
+	 * @return the callback result
+	 * @since 4.0
+	 */
+	public <T> T execute(ClientCallback<T> callback) {
+
+		Assert.notNull(callback, "callback must not be null");
+
+		try {
+			return callback.doWithClient(client);
+		} catch (IOException | RuntimeException e) {
+			throw translateException(e);
+		}
+	}
+
+	/**
+	 * 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;
+	}
+
+	// endregion
 }

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

@@ -16,7 +16,6 @@
 package org.springframework.data.elasticsearch.core;
 
 import java.util.List;
-import java.util.Optional;
 
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;

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

@@ -792,15 +792,6 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 		return type != null ? mappingContext.getPersistentEntity(type) : null;
 	}
 
-	private Throwable translateException(Throwable throwable) {
-
-		RuntimeException exception = throwable instanceof RuntimeException ? (RuntimeException) throwable
-				: new RuntimeException(throwable.getMessage(), throwable);
-		RuntimeException potentiallyTranslatedException = exceptionTranslator.translateExceptionIfPossible(exception);
-
-		return potentiallyTranslatedException != null ? potentiallyTranslatedException : throwable;
-	}
-
 	/**
 	 * Obtain the {@link ReactiveElasticsearchClient} to operate upon.
 	 *
@@ -812,4 +803,22 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera
 
 	// endregion
 
+	/**
+	 * translates an Exception if possible. Exceptions that are no {@link RuntimeException}s are wrapped in a
+	 * RuntimeException
+	 *
+	 * @param throwable the Throwable to map
+	 * @return the potentially translated RuntimeException.
+	 * @since 4.0
+	 */
+	private RuntimeException translateException(Throwable throwable) {
+
+		RuntimeException runtimeException = throwable instanceof RuntimeException ? (RuntimeException) throwable
+				: new RuntimeException(throwable.getMessage(), throwable);
+		RuntimeException potentiallyTranslatedException = exceptionTranslator
+				.translateExceptionIfPossible(runtimeException);
+
+		return potentiallyTranslatedException != null ? potentiallyTranslatedException : runtimeException;
+	}
+
 }

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

@@ -25,7 +25,6 @@ import java.util.List;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
-import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.index.query.IdsQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.slf4j.Logger;
@@ -93,8 +92,8 @@ public abstract class AbstractElasticsearchRepository<T, ID> implements Elastics
 				createIndex();
 				putMapping();
 			}
-		} catch (ElasticsearchException exception) {
+		} catch (Exception exception) {
-			LOGGER.warn("Cannot create index: {}", exception.getDetailedMessage());
+			LOGGER.warn("Cannot create index: {}", exception.getMessage());
 		}
 	}
 

src/test/java/org/springframework/data/elasticsearch/config/nested/EnableNestedElasticsearchRepositoriesTests.java

@@ -24,6 +24,8 @@ import lombok.Data;
 import java.lang.Double;
 import java.lang.Long;
 
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
@@ -34,6 +36,7 @@ import org.springframework.data.elasticsearch.annotations.Document;
 import org.springframework.data.elasticsearch.annotations.Field;
 import org.springframework.data.elasticsearch.annotations.Score;
 import org.springframework.data.elasticsearch.annotations.ScriptedField;
+import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
 import org.springframework.data.elasticsearch.core.geo.GeoPoint;
 import org.springframework.data.elasticsearch.junit.jupiter.ElasticsearchRestTemplateConfiguration;
 import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
@@ -56,6 +59,17 @@ public class EnableNestedElasticsearchRepositoriesTests {
 	static class Config {}
 
 	@Autowired(required = false) private SampleRepository nestedRepository;
+	@Autowired ElasticsearchOperations operations;
+
+	@BeforeEach
+	void setUp() {
+		operations.indexOps(SampleEntity.class).delete();
+	}
+
+	@AfterEach
+	void tearDown() {
+		operations.indexOps(SampleEntity.class).delete();
+	}
 
 	@Test
 	public void hasNestedRepository() {

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

@@ -22,9 +22,9 @@ import static org.springframework.data.elasticsearch.annotations.FieldType.*;
 import lombok.Builder;
 import lombok.Data;
 
-import org.elasticsearch.ElasticsearchStatusException;
 import org.junit.jupiter.api.Test;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.elasticsearch.UncategorizedElasticsearchException;
 import org.springframework.data.elasticsearch.annotations.Document;
 import org.springframework.data.elasticsearch.annotations.Field;
 import org.springframework.data.elasticsearch.core.query.UpdateQuery;
@@ -57,7 +57,8 @@ public class ElasticsearchRestTemplateTests extends ElasticsearchTemplateTests {
 		org.springframework.data.elasticsearch.core.document.Document document = org.springframework.data.elasticsearch.core.document.Document
 				.create();
 		UpdateQuery updateQuery = UpdateQuery.builder(randomNumeric(5)).withDocument(document).build();
-		assertThatThrownBy(() -> operations.update(updateQuery, index)).isInstanceOf(ElasticsearchStatusException.class);
+		assertThatThrownBy(() -> operations.update(updateQuery, index))
+				.isInstanceOf(UncategorizedElasticsearchException.class);
 	}
 
 	@Data

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

@@ -26,13 +26,13 @@ import java.text.SimpleDateFormat;
 import java.util.Arrays;
 import java.util.Date;
 
-import org.elasticsearch.ElasticsearchException;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.dao.DataAccessException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.elasticsearch.annotations.Document;
 import org.springframework.data.elasticsearch.annotations.Field;
@@ -102,6 +102,10 @@ public class LogEntityTests {
 		assertThat(entities).isNotNull().hasSize(1);
 	}
 
+	protected Class<? extends Exception> invalidIpExceptionClass() {
+		return DataAccessException.class;
+	}
+
 	@Test // DATAES-66
 	public void shouldThrowExceptionWhenInvalidIPGivenForSearchQuery() {
 
@@ -110,7 +114,7 @@ public class LogEntityTests {
 
 		assertThatThrownBy(() -> {
 			SearchHits<LogEntity> entities = operations.search(searchQuery, LogEntity.class, index);
-		}).isInstanceOf(ElasticsearchException.class);
+		}).isInstanceOf(invalidIpExceptionClass());
 	}
 
 	@Test // DATAES-66

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

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.elasticsearch.core;
 
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ElasticsearchSecurityException;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.data.elasticsearch.junit.jupiter.ElasticsearchTemplateConfiguration;
@@ -28,4 +30,9 @@ public class LogEntityTransportTests extends LogEntityTests {
 	@Configuration
 	@Import({ ElasticsearchTemplateConfiguration.class })
 	static class Config {}
+
+	@Override
+	protected Class<? extends Exception> invalidIpExceptionClass() {
+		return ElasticsearchException.class;
+	}
 }

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

@@ -39,7 +39,6 @@ import java.util.UUID;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
-import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.search.sort.FieldSortBuilder;
 import org.elasticsearch.search.sort.SortOrder;
 import org.junit.jupiter.api.AfterEach;
@@ -52,6 +51,7 @@ import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.elasticsearch.TestUtils;
+import org.springframework.data.elasticsearch.UncategorizedElasticsearchException;
 import org.springframework.data.elasticsearch.annotations.Document;
 import org.springframework.data.elasticsearch.annotations.Field;
 import org.springframework.data.elasticsearch.annotations.Score;
@@ -436,7 +436,7 @@ public class ReactiveElasticsearchTemplateTests {
 
 		template.search(queryWithInvalidPreference, SampleEntity.class) //
 				.as(StepVerifier::create) //
-				.expectError(ElasticsearchStatusException.class).verify();
+				.expectError(UncategorizedElasticsearchException.class).verify();
 	}
 
 	@Test // DATAES-504

src/test/java/org/springframework/data/elasticsearch/immutable/ImmutableElasticsearchRepositoryTests.java

@@ -22,6 +22,7 @@ import lombok.NoArgsConstructor;
 
 import java.util.Optional;
 
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -64,6 +65,12 @@ public class ImmutableElasticsearchRepositoryTests {
 		indexOperations.refresh();
 	}
 
+	@AfterEach
+	void tearDown() {
+		IndexOperations indexOperations = operations.indexOps(ImmutableEntity.class);
+		indexOperations.delete();
+	}
+
 	@Test // DATAES-281
 	public void shouldSaveAndFindImmutableDocument() {
 

src/test/java/org/springframework/data/elasticsearch/repository/support/simple/SimpleElasticsearchRepositoryTests.java

@@ -32,13 +32,13 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
-import org.elasticsearch.action.ActionRequestValidationException;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.annotation.Version;
 import org.springframework.data.domain.Page;
@@ -148,7 +148,7 @@ public class SimpleElasticsearchRepositoryTests {
 		sampleEntity.setVersion(System.currentTimeMillis());
 
 		// when
-		assertThatThrownBy(() -> repository.save(sampleEntity)).isInstanceOf(ActionRequestValidationException.class);
+		assertThatThrownBy(() -> repository.save(sampleEntity)).isInstanceOf(DataIntegrityViolationException.class);
 	}
 
 	@Test