Implementation:Risingwavelabs Risingwave OpensearchRestHighLevelClientAdapter

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

Overview

OpensearchRestHighLevelClientAdapter is an adapter wrapping the OpenSearch RestHighLevelClient with authentication support. It manages the OpenSearch client lifecycle, constructing the client with optional basic credentials (username/password). The adapter exposes methods for pinging the cluster and performing asynchronous bulk operations. It mirrors the design of ElasticRestHighLevelClientAdapter but uses OpenSearch client libraries instead of Elasticsearch ones.

Code Reference

Source Location

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

Signature

public class OpensearchRestHighLevelClientAdapter implements AutoCloseable {
    public OpensearchRestHighLevelClientAdapter(HttpHost host, EsSinkConfig config);
    public void close() throws IOException;
    public boolean ping(org.opensearch.client.RequestOptions options) throws IOException;
    public Cancellable bulkAsync(BulkRequest bulkRequest, RequestOptions options, ActionListener<BulkResponse> listener);
}

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.opensearch.action.bulk.BulkRequest;
import org.opensearch.action.bulk.BulkResponse;
import org.opensearch.client.Cancellable;
import org.opensearch.client.RequestOptions;
import org.opensearch.client.RestClientBuilder;
import org.opensearch.client.RestHighLevelClient;
import org.opensearch.core.action.ActionListener;

I/O Contract

Constructor

• Input: An HttpHost specifying the OpenSearch cluster endpoint and an EsSinkConfig with optional credentials.
• Behavior: Builds an OpenSearch RestHighLevelClient. 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 OpenSearch cluster
bulkAsync	BulkRequest, RequestOptions, ActionListener	Cancellable	Submits an asynchronous bulk request; used by OpensearchBulkProcessorAdapter
close	(none)	(void)	Closes the underlying REST client and releases resources

Differences from ElasticRestHighLevelClientAdapter

• Does not enable API compatibility mode (not needed for OpenSearch).
• Does not expose a synchronous search() method.
• Uses org.opensearch.client package classes instead of org.elasticsearch.client.

Usage Examples

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

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

// Close when done
client.close();

Related Pages

• OpensearchBulkProcessorAdapter - Uses this client adapter for async bulk operations
• EsSinkFactory - Creates this adapter during validation
• EsSinkConfig - Configuration providing credentials and connection details
• ElasticRestHighLevelClientAdapter - Elasticsearch counterpart of this adapter