The Elasticsearch REST Client provides a powerful way to interact with Elasticsearch from Java applications. It offers high-level abstraction while maintaining flexibility and performance.
Setup and Dependencies
Maven Dependencies
<properties>
<elasticsearch.version>8.11.0</elasticsearch.version>
</properties>
<dependencies>
<!-- Elasticsearch Java Client -->
<dependency>
<groupId>co.elastic.clients</groupId>
<artifactId>elasticsearch-java</artifactId>
<version>${elasticsearch.version}</version>
</dependency>
<!-- Jackson for JSON processing -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.15.2</version>
</dependency>
<!-- Alternatively, use the legacy client -->
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-client</artifactId>
<version>7.17.0</version>
</dependency>
<!-- Spring Boot Starter Data Elasticsearch -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>
</dependencies>
Configuration and Setup
Basic REST Client Configuration
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.client.RestClient;
@Configuration
public class ElasticsearchConfig {
@Value("${elasticsearch.host:localhost}")
private String host;
@Value("${elasticsearch.port:9200}")
private int port;
@Value("${elasticsearch.username:}")
private String username;
@Value("${elasticsearch.password:}")
private String password;
@Bean
public RestClient restClient() {
HttpHost httpHost = new HttpHost(host, port);
RestClient.Builder builder = RestClient.builder(httpHost);
// Configure authentication if provided
if (!username.isEmpty() && !password.isEmpty()) {
CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(
AuthScope.ANY,
new UsernamePasswordCredentials(username, password)
);
builder.setHttpClientConfigCallback(httpClientBuilder ->
httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
);
}
// Configure default headers
builder.setDefaultHeaders(new Header[]{
new BasicHeader("Content-Type", "application/json"),
new BasicHeader("User-Agent", "Java-ES-Client/1.0")
});
// Configure connection timeout and retries
builder.setRequestConfigCallback(requestConfigBuilder ->
requestConfigBuilder
.setConnectTimeout(5000)
.setSocketTimeout(60000)
);
return builder.build();
}
@Bean
public ElasticsearchTransport elasticsearchTransport() {
return new RestClientTransport(restClient(), new JacksonJsonpMapper());
}
@Bean
public ElasticsearchClient elasticsearchClient() {
return new ElasticsearchClient(elasticsearchTransport());
}
}
Advanced Configuration with Multiple Nodes
@Configuration
public class AdvancedElasticsearchConfig {
@Bean
public RestClient elasticsearchRestClient() {
return RestClient.builder(
new HttpHost("es-node1", 9200, "http"),
new HttpHost("es-node2", 9200, "http"),
new HttpHost("es-node3", 9200, "http")
)
.setHttpClientConfigCallback(httpClientBuilder -> {
// Configure connection pooling
httpClientBuilder.setMaxConnTotal(100);
httpClientBuilder.setMaxConnPerRoute(50);
// Configure keep-alive strategy
httpClientBuilder.setKeepAliveStrategy((response, context) -> 60000);
return httpClientBuilder;
})
.setRequestConfigCallback(requestConfigBuilder ->
requestConfigBuilder
.setConnectTimeout(5000)
.setSocketTimeout(30000)
.setConnectionRequestTimeout(1000)
)
.setFailureListener(new RestClient.FailureListener() {
@Override
public void onFailure(HttpHost host) {
System.err.println("Connection failed to host: " + host);
}
})
.build();
}
@Bean
public ElasticsearchClient elasticsearchClient() {
ElasticsearchTransport transport = new RestClientTransport(
elasticsearchRestClient(),
new JacksonJsonpMapper()
);
return new ElasticsearchClient(transport);
}
}
Document Model Classes
// Product document
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class Product {
private String id;
private String name;
private String description;
private Double price;
private String category;
private List<String> tags;
private Integer stock;
private Boolean active;
private LocalDateTime createdAt;
private Map<String, Object> attributes;
}
// User document
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class User {
private String id;
private String username;
private String email;
private String firstName;
private String lastName;
private Integer age;
private Address address;
private List<String> roles;
private LocalDateTime registeredAt;
private Boolean enabled;
}
// Address embedded object
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class Address {
private String street;
private String city;
private String state;
private String zipCode;
private String country;
}
// Search response wrapper
@Data
@AllArgsConstructor
@NoArgsConstructor
public class SearchResponse<T> {
private List<T> hits;
private long totalHits;
private long took;
private boolean timedOut;
}
CRUD Operations
Basic CRUD Service
@Service
@Slf4j
public class ElasticsearchService {
private final ElasticsearchClient client;
private final ObjectMapper objectMapper;
public ElasticsearchService(ElasticsearchClient client, ObjectMapper objectMapper) {
this.client = client;
this.objectMapper = objectMapper;
}
// Index (Create/Update) a document
public <T> String index(String indexName, T document) throws IOException {
IndexResponse response = client.index(i -> i
.index(indexName)
.document(document)
);
log.info("Document indexed: {}", response.id());
return response.id();
}
// Index with custom ID
public <T> String indexWithId(String indexName, String id, T document) throws IOException {
IndexResponse response = client.index(i -> i
.index(indexName)
.id(id)
.document(document)
);
log.info("Document indexed with ID: {}", response.id());
return response.id();
}
// Get document by ID
public <T> Optional<T> getById(String indexName, String id, Class<T> clazz) throws IOException {
GetResponse<T> response = client.get(g -> g
.index(indexName)
.id(id),
clazz
);
if (response.found()) {
return Optional.of(response.source());
}
return Optional.empty();
}
// Check if document exists
public boolean exists(String indexName, String id) throws IOException {
return client.exists(e -> e
.index(indexName)
.id(id)
).value();
}
// Update document
public <T> String update(String indexName, String id, T document) throws IOException {
UpdateResponse<T> response = client.update(u -> u
.index(indexName)
.id(id)
.doc(document),
clazz
);
log.info("Document updated: {}", response.id());
return response.id();
}
// Partial update using script
public String partialUpdate(String indexName, String id, Map<String, Object> fields) throws IOException {
UpdateResponse<Object> response = client.update(u -> u
.index(indexName)
.id(id)
.doc(fields),
Object.class
);
return response.id();
}
// Delete document
public boolean delete(String indexName, String id) throws IOException {
DeleteResponse response = client.delete(d -> d
.index(indexName)
.id(id)
);
log.info("Document deleted: {}", response.id());
return response.result().jsonValue().equals("deleted");
}
// Bulk operations
public BulkResponse bulkIndex(String indexName, List<Product> products) throws IOException {
BulkRequest.Builder br = new BulkRequest.Builder();
for (Product product : products) {
br.operations(op -> op
.index(idx -> idx
.index(indexName)
.id(product.getId())
.document(product)
)
);
}
return client.bulk(br.build());
}
}
Search Operations
Advanced Search Service
@Service
@Slf4j
public class ProductSearchService {
private final ElasticsearchClient client;
private static final String PRODUCT_INDEX = "products";
public ProductSearchService(ElasticsearchClient client) {
this.client = client;
}
// Simple search
public SearchResponse<Product> searchProducts(String query) throws IOException {
SearchResponse<Product> response = client.search(s -> s
.index(PRODUCT_INDEX)
.query(q -> q
.multiMatch(t -> t
.fields("name", "description", "category")
.query(query)
)
),
Product.class
);
List<Product> products = response.hits().hits().stream()
.map(hit -> hit.source())
.collect(Collectors.toList());
return new SearchResponse<>(
products,
response.hits().total().value(),
response.took(),
response.timedOut()
);
}
// Filtered search with range and term filters
public SearchResponse<Product> searchWithFilters(ProductSearchRequest searchRequest) throws IOException {
List<Query> mustQueries = new ArrayList<>();
List<Query> shouldQueries = new ArrayList<>();
List<Query> filterQueries = new ArrayList<>();
// Text search
if (searchRequest.getQuery() != null && !searchRequest.getQuery().isEmpty()) {
mustQueries.add(Query.of(q -> q
.multiMatch(m -> m
.fields("name^2", "description", "category")
.query(searchRequest.getQuery())
.fuzziness("AUTO")
)
));
}
// Category filter
if (searchRequest.getCategory() != null) {
filterQueries.add(Query.of(q -> q
.term(t -> t
.field("category")
.value(searchRequest.getCategory())
)
));
}
// Price range filter
if (searchRequest.getMinPrice() != null || searchRequest.getMaxPrice() != null) {
RangeQuery.Builder rangeQuery = new RangeQuery.Builder();
rangeQuery.field("price");
if (searchRequest.getMinPrice() != null) {
rangeQuery.gte(JsonData.of(searchRequest.getMinPrice()));
}
if (searchRequest.getMaxPrice() != null) {
rangeQuery.lte(JsonData.of(searchRequest.getMaxPrice()));
}
filterQueries.add(Query.of(q -> q.range(rangeQuery.build())));
}
// Active products only
filterQueries.add(Query.of(q -> q
.term(t -> t
.field("active")
.value(true)
)
));
// Tags search (should query for better scoring)
if (searchRequest.getTags() != null && !searchRequest.getTags().isEmpty()) {
for (String tag : searchRequest.getTags()) {
shouldQueries.add(Query.of(q -> q
.term(t -> t
.field("tags")
.value(tag)
)
));
}
}
// Build the final query
Query finalQuery = Query.of(q -> q
.bool(b -> b
.must(mustQueries)
.should(shouldQueries)
.filter(filterQueries)
.minimumShouldMatch(searchRequest.getTags() != null ? "1" : "0")
)
);
SearchResponse<Product> response = client.search(s -> s
.index(PRODUCT_INDEX)
.query(finalQuery)
.from(searchRequest.getFrom())
.size(searchRequest.getSize())
.sort(so -> so
.field(f -> f
.field("_score")
.order(SortOrder.Desc)
)
)
.highlight(h -> h
.fields("name", f -> f
.type(HighlighterType.Unified)
)
.fields("description", f -> f
.type(HighlighterType.Unified)
)
),
Product.class
);
List<Product> products = response.hits().hits().stream()
.map(hit -> {
Product product = hit.source();
// You can process highlights here
return product;
})
.collect(Collectors.toList());
return new SearchResponse<>(
products,
response.hits().total().value(),
response.took(),
response.timedOut()
);
}
// Aggregation search
public Map<String, Object> getProductAggregations() throws IOException {
SearchResponse<Product> response = client.search(s -> s
.index(PRODUCT_INDEX)
.size(0)
.aggregations("categories", a -> a
.terms(t -> t.field("category"))
)
.aggregations("price_stats", a -> a
.stats(st -> st.field("price"))
)
.aggregations("price_histogram", a -> a
.histogram(h -> h
.field("price")
.interval(50.0)
)
),
Product.class
);
Map<String, Aggregate> aggs = response.aggregations();
Map<String, Object> result = new HashMap<>();
// Process category aggregation
if (aggs.containsKey("categories")) {
List<? extends StringTermsBucket> buckets = aggs.get("categories").sterms().buckets().array();
Map<String, Long> categories = buckets.stream()
.collect(Collectors.toMap(
StringTermsBucket::key,
StringTermsBucket::docCount
));
result.put("categories", categories);
}
// Process price stats
if (aggs.containsKey("price_stats")) {
StatsAggregate stats = aggs.get("price_stats").stats();
result.put("price_stats", Map.of(
"count", stats.count(),
"min", stats.min(),
"max", stats.max(),
"avg", stats.avg(),
"sum", stats.sum()
));
}
return result;
}
// Auto-complete/suggest search
public List<String> suggestProductNames(String prefix) throws IOException {
SearchResponse<Product> response = client.search(s -> s
.index(PRODUCT_INDEX)
.suggest(su -> su
.suggesters("product_suggest", sug -> sug
.prefix(prefix)
.completion(c -> c
.field("name.suggest")
.skipDuplicates(true)
.size(5)
)
)
),
Product.class
);
return response.suggest().get("product_suggest").stream()
.flatMap(suggestion -> suggestion.completion().options().stream())
.map(option -> option.text())
.collect(Collectors.toList());
}
// More Like This search
public SearchResponse<Product> findSimilarProducts(String productId) throws IOException {
SearchResponse<Product> response = client.search(s -> s
.index(PRODUCT_INDEX)
.query(q -> q
.moreLikeThis(m -> m
.like(l -> l
.document(d -> d
.index(PRODUCT_INDEX)
.id(productId)
)
)
.fields("name", "description", "category", "tags")
.minTermFreq(1)
.maxQueryTerms(12)
)
)
.size(5),
Product.class
);
List<Product> products = response.hits().hits().stream()
.map(hit -> hit.source())
.collect(Collectors.toList());
return new SearchResponse<>(
products,
response.hits().total().value(),
response.took(),
response.timedOut()
);
}
}
// Search request DTO
@Data
class ProductSearchRequest {
private String query;
private String category;
private Double minPrice;
private Double maxPrice;
private List<String> tags;
private Integer from = 0;
private Integer size = 10;
}
Index Management
Index Operations Service
@Service
@Slf4j
public class IndexManagementService {
private final ElasticsearchClient client;
public IndexManagementService(ElasticsearchClient client) {
this.client = client;
}
// Create index with mapping
public boolean createProductIndex() throws IOException {
CreateIndexResponse response = client.indices().create(c -> c
.index("products")
.mappings(m -> m
.properties("id", p -> p.keyword(k -> k))
.properties("name", p -> p
.text(t -> t
.analyzer("standard")
.fields("suggest", f -> f.completion(cp -> cp))
)
)
.properties("description", p -> p.text(t -> t.analyzer("standard")))
.properties("category", p -> p.keyword(k -> k))
.properties("price", p -> p.double_(d -> d))
.properties("tags", p -> p.keyword(k -> k))
.properties("stock", p -> p.integer(i -> i))
.properties("active", p -> p.boolean_(b -> b))
.properties("createdAt", p -> p.date(d -> d.format("strict_date_optional_time||epoch_millis")))
)
.settings(s -> s
.numberOfShards("2")
.numberOfReplicas("1")
)
);
log.info("Index created: {}", response.index());
return response.acknowledged();
}
// Check if index exists
public boolean indexExists(String indexName) throws IOException {
return client.indices().exists(e -> e.index(indexName)).value();
}
// Delete index
public boolean deleteIndex(String indexName) throws IOException {
DeleteIndexResponse response = client.indices().delete(d -> d.index(indexName));
return response.acknowledged();
}
// Get index mapping
public Map<String, Property> getIndexMapping(String indexName) throws IOException {
GetMappingResponse response = client.indices().getMapping(m -> m.index(indexName));
return response.get("products").mappings().properties();
}
// Update index settings
public boolean updateIndexSettings(String indexName) throws IOException {
UpdateIndicesSettingsResponse response = client.indices().putSettings(s -> s
.index(indexName)
.settings(settings -> settings
.numberOfReplicas("2")
.refreshInterval(i -> i.time("30s"))
)
);
return response.acknowledged();
}
// Create alias
public boolean createAlias(String indexName, String aliasName) throws IOException {
UpdateAliasesResponse response = client.indices().updateAliases(u -> u
.actions(a -> a
.add(add -> add
.index(indexName)
.alias(aliasName)
)
)
);
return response.acknowledged();
}
// Refresh index
public boolean refreshIndex(String indexName) throws IOException {
RefreshResponse response = client.indices().refresh(r -> r.index(indexName));
return response.shards().total().intValue() > 0;
}
}
Advanced Features
Bulk Operations Service
@Service
@Slf4j
public class BulkOperationsService {
private final ElasticsearchClient client;
public BulkOperationsService(ElasticsearchClient client) {
this.client = client;
}
// Bulk index with error handling
public BulkResponse bulkIndexProducts(List<Product> products) throws IOException {
BulkRequest.Builder br = new BulkRequest.Builder();
for (Product product : products) {
br.operations(op -> op
.index(idx -> idx
.index("products")
.id(product.getId())
.document(product)
)
);
}
BulkResponse response = client.bulk(br.build());
// Check for errors
if (response.errors()) {
log.error("Bulk operation had errors:");
for (BulkResponseItem item : response.items()) {
if (item.error() != null) {
log.error("Error for item {}: {}", item.id(), item.error().reason());
}
}
}
return response;
}
// Bulk update
public BulkResponse bulkUpdateProducts(Map<String, Map<String, Object>> updates) throws IOException {
BulkRequest.Builder br = new BulkRequest.Builder();
for (Map.Entry<String, Map<String, Object>> entry : updates.entrySet()) {
br.operations(op -> op
.update(u -> u
.index("products")
.id(entry.getKey())
.action(a -> a
.doc(entry.getValue())
)
)
);
}
return client.bulk(br.build());
}
// Bulk delete
public BulkResponse bulkDeleteProducts(List<String> productIds) throws IOException {
BulkRequest.Builder br = new BulkRequest.Builder();
for (String productId : productIds) {
br.operations(op -> op
.delete(d -> d
.index("products")
.id(productId)
)
);
}
return client.bulk(br.build());
}
}
Async Operations Service
@Service
@Slf4j
public class AsyncElasticsearchService {
private final ElasticsearchClient client;
private final ExecutorService executorService;
public AsyncElasticsearchService(ElasticsearchClient client) {
this.client = client;
this.executorService = Executors.newFixedThreadPool(10);
}
// Async search
public CompletableFuture<SearchResponse<Product>> searchAsync(String query) {
return CompletableFuture.supplyAsync(() -> {
try {
return client.search(s -> s
.index("products")
.query(q -> q
.match(m -> m
.field("name")
.query(query)
)
),
Product.class
);
} catch (IOException e) {
throw new CompletionException(e);
}
}, executorService);
}
// Async bulk operations
public CompletableFuture<BulkResponse> bulkIndexAsync(List<Product> products) {
return CompletableFuture.supplyAsync(() -> {
try {
BulkRequest.Builder br = new BulkRequest.Builder();
for (Product product : products) {
br.operations(op -> op
.index(idx -> idx
.index("products")
.id(product.getId())
.document(product)
)
);
}
return client.bulk(br.build());
} catch (IOException e) {
throw new CompletionException(e);
}
}, executorService);
}
// Multiple async searches
public CompletableFuture<List<SearchResponse<Product>>> multiSearchAsync(List<String> queries) {
List<CompletableFuture<SearchResponse<Product>>> futures = queries.stream()
.map(this::searchAsync)
.collect(Collectors.toList());
return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
.thenApply(v -> futures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList())
);
}
}
Error Handling and Retry Mechanism
@Service
@Slf4j
public class ResilientElasticsearchService {
private final ElasticsearchClient client;
private final RetryTemplate retryTemplate;
public ResilientElasticsearchService(ElasticsearchClient client) {
this.client = client;
this.retryTemplate = createRetryTemplate();
}
private RetryTemplate createRetryTemplate() {
RetryTemplate template = new RetryTemplate();
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
retryPolicy.setMaxAttempts(3);
FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
backOffPolicy.setBackOffPeriod(1000L); // 1 second
template.setRetryPolicy(retryPolicy);
template.setBackOffPolicy(backOffPolicy);
return template;
}
public <T> T executeWithRetry(ElasticsearchOperation<T> operation) {
return retryTemplate.execute(context -> {
try {
return operation.execute();
} catch (ElasticsearchException e) {
log.warn("Elasticsearch operation failed, retry attempt: {}", context.getRetryCount());
throw e;
} catch (IOException e) {
log.error("IO error during Elasticsearch operation", e);
throw new RuntimeException(e);
}
});
}
// Search with retry
public SearchResponse<Product> searchWithRetry(String query) {
return executeWithRetry(() ->
client.search(s -> s
.index("products")
.query(q -> q.match(m -> m.field("name").query(query))),
Product.class
)
);
}
// Index with retry
public String indexWithRetry(Product product) {
return executeWithRetry(() ->
client.index(i -> i
.index("products")
.id(product.getId())
.document(product)
).id()
);
}
@FunctionalInterface
public interface ElasticsearchOperation<T> {
T execute() throws IOException;
}
}
Spring Data Elasticsearch Integration
// Spring Data Repository
@Repository
public interface ProductRepository extends ElasticsearchRepository<Product, String> {
List<Product> findByName(String name);
List<Product> findByCategoryAndPriceBetween(String category, Double minPrice, Double maxPrice);
List<Product> findByActiveTrue();
@Query("{\"match\": {\"name\": \"?0\"}}")
List<Product> findByNameCustom(String name);
Page<Product> findByCategory(String category, Pageable pageable);
List<Product> findByTagsIn(List<String> tags);
Long countByCategory(String category);
}
// Custom repository implementation
public interface CustomProductRepository {
SearchResponse<Product> searchComplex(ProductSearchRequest request);
}
@Repository
@RequiredArgsConstructor
class CustomProductRepositoryImpl implements CustomProductRepository {
private final ElasticsearchClient client;
@Override
public SearchResponse<Product> searchComplex(ProductSearchRequest request) {
// Implement complex search logic using the native client
// This allows you to use features not available in Spring Data
return null; // Implementation similar to previous examples
}
}
Monitoring and Metrics
@Component
@Slf4j
public class ElasticsearchMetrics {
private final ElasticsearchClient client;
private final MeterRegistry meterRegistry;
private final Counter requestCounter;
private final Timer requestTimer;
public ElasticsearchMetrics(ElasticsearchClient client, MeterRegistry meterRegistry) {
this.client = client;
this.meterRegistry = meterRegistry;
this.requestCounter = Counter.builder("elasticsearch.requests")
.description("Number of Elasticsearch requests")
.register(meterRegistry);
this.requestTimer = Timer.builder("elasticsearch.request.duration")
.description("Duration of Elasticsearch requests")
.register(meterRegistry);
}
public <T> T executeWithMetrics(String operation, Supplier<T> operationSupplier) {
requestCounter.increment();
return requestTimer.record(() -> {
try {
return operationSupplier.get();
} catch (Exception e) {
meterRegistry.counter("elasticsearch.errors", "operation", operation).increment();
throw e;
}
});
}
// Get cluster health
public ClusterHealthResponse getClusterHealth() throws IOException {
return executeWithMetrics("cluster_health",
() -> client.cluster().health());
}
// Get node stats
public NodesStatsResponse getNodeStats() throws IOException {
return executeWithMetrics("node_stats",
() -> client.nodes().stats(n -> n));
}
// Get index stats
public IndicesStatsResponse getIndexStats(String indexName) throws IOException {
return executeWithMetrics("index_stats",
() -> client.indices().stats(s -> s.index(indexName)));
}
}
Best Practices
1. Connection Management
- Use connection pooling
- Set appropriate timeouts
- Monitor connection health
2. Error Handling
- Implement retry mechanisms
- Handle specific Elasticsearch exceptions
- Log errors appropriately
3. Performance
- Use bulk operations for multiple documents
- Implement async operations for non-blocking calls
- Monitor query performance
4. Security
- Use HTTPS for connections
- Implement authentication
- Secure sensitive data
This comprehensive Elasticsearch REST Client implementation provides a solid foundation for building robust search functionality in Java applications, with features covering CRUD operations, advanced search, bulk processing, error handling, and monitoring.