Implementation:Risingwavelabs Risingwave ElasticRestHighLevelClientAdapter

Property	Value
File	java/connector-node/risingwave-sink-es-7/src/main/java/com/risingwave/connector/ElasticRestHighLevelClientAdapter.java
Language	Java
Lines	89
Category	Adapter
Package	com.risingwave.connector

Overview

ElasticRestHighLevelClientAdapter is an adapter wrapping the Elasticsearch RestHighLevelClient with authentication configuration support. It manages the Elasticsearch 7 client lifecycle, constructing the client with optional basic credentials (username/password) and API compatibility mode enabled. The adapter exposes methods for pinging the cluster, performing asynchronous bulk operations, and executing search requests.

Code Reference

Source Location

java/connector-node/risingwave-sink-es-7/src/main/java/com/risingwave/connector/ElasticRestHighLevelClientAdapter.java

Signature

public class ElasticRestHighLevelClientAdapter implements AutoCloseable {
    public ElasticRestHighLevelClientAdapter(HttpHost host, EsSinkConfig config);
    public void close() throws IOException;
    public boolean ping(RequestOptions options) throws IOException;
    public Cancellable bulkAsync(BulkRequest bulkRequest, RequestOptions options, ActionListener<BulkResponse> listener);
    public SearchResponse search(SearchRequest searchRequest, RequestOptions options) throws IOException;
}

Imports

import java.io.IOException;
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.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Cancellable;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.RestHighLevelClientBuilder;

I/O Contract

Constructor

• Input: An HttpHost specifying the Elasticsearch cluster endpoint and an EsSinkConfig with optional credentials.
• Behavior: Builds a RestHighLevelClient with API compatibility mode enabled. If both username and password are present in the config, configures HTTP basic authentication via BasicCredentialsProvider.

Methods

Method	Input	Output	Description
ping	RequestOptions	boolean	Tests connectivity to the Elasticsearch cluster
bulkAsync	BulkRequest, RequestOptions, ActionListener	Cancellable	Submits an asynchronous bulk request; used by ElasticBulkProcessorAdapter
search	SearchRequest, RequestOptions	SearchResponse	Executes a synchronous search query
close	(none)	(void)	Closes the underlying REST client and releases resources

Usage Examples

// Create the adapter with authentication
HttpHost host = HttpHost.create("http://localhost:9200");
EsSinkConfig config = new EsSinkConfig("http://localhost:9200")
    .withUsername("elastic")
    .withPassword("changeme");
ElasticRestHighLevelClientAdapter client = new ElasticRestHighLevelClientAdapter(host, config);

// Verify connectivity
boolean connected = client.ping(RequestOptions.DEFAULT);

// Close when done
client.close();

Related Pages

• ElasticBulkProcessorAdapter - Uses this client adapter for async bulk operations
• EsSinkFactory - Creates this adapter during validation
• EsSinkConfig - Configuration providing credentials and connection details
• OpensearchRestHighLevelClientAdapter - OpenSearch counterpart of this adapter