Implementation:Risingwavelabs Risingwave EsSinkFactory

Property	Value
File	java/connector-node/risingwave-sink-es-7/src/main/java/com/risingwave/connector/EsSinkFactory.java
Language	Java
Lines	123
Category	Factory
Package	com.risingwave.connector

Overview

EsSinkFactory is the factory class that creates and validates EsSink instances for both Elasticsearch and OpenSearch backends. It implements the SinkFactory interface and serves as the entry point for the RisingWave connector framework to instantiate Elasticsearch/OpenSearch sink writers. The factory performs validation of the URL, column configurations, index/index_column mutual exclusivity, and cluster connectivity by pinging the target cluster.

Code Reference

Source Location

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

Signature

public class EsSinkFactory implements SinkFactory {
    public SinkWriter createWriter(TableSchema tableSchema, Map<String, String> tableProperties);
    public void validate(TableSchema tableSchema, Map<String, String> tableProperties, Catalog.SinkType sinkType);
    private void checkColumn(String column, TableSchema tableSchema, Data.DataType.TypeName typeName);
}

Imports

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.risingwave.connector.api.TableSchema;
import com.risingwave.connector.api.sink.SinkFactory;
import com.risingwave.connector.api.sink.SinkWriter;
import com.risingwave.connector.api.sink.SinkWriterV1;
import com.risingwave.proto.Catalog;
import com.risingwave.proto.Data;
import io.grpc.Status;
import java.util.Map;
import org.apache.http.HttpHost;

I/O Contract

createWriter

• Input: A TableSchema and a Map<String, String> of table properties.
• Output: Returns a SinkWriter wrapping a new EsSink instance via SinkWriterV1.Adapter.

validate

Performs the following validation steps:

1. URL validation: Parses the URL via HttpHost.create(); throws INVALID_ARGUMENT on failure.
2. Routing column check: If routing_column is set, verifies it exists in the schema and is of type VARCHAR.
3. Index column check: If index_column is set, verifies it exists in the schema, is of type VARCHAR, and that index is not simultaneously set (mutual exclusivity).
4. Index requirement: If neither index_column nor index is set, throws INVALID_ARGUMENT.
5. Connection check: Based on the connector type:

• • "elasticsearch_v1": Creates an ElasticRestHighLevelClientAdapter and pings the cluster.
  • "opensearch_v1": Creates an OpensearchRestHighLevelClientAdapter and pings the cluster.
  • Otherwise: Throws a RuntimeException.

checkColumn (private)

• Input: A column name, the TableSchema, and the expected TypeName.
• Output: Void on success; throws INVALID_ARGUMENT if the column is not found or has the wrong type.

Usage Examples

// Creating an ES sink writer through the factory
EsSinkFactory factory = new EsSinkFactory();
SinkWriter writer = factory.createWriter(tableSchema, tableProperties);

// Validating configuration before use
factory.validate(tableSchema, tableProperties, Catalog.SinkType.SINK_TYPE_UPSERT);

Related Pages

• EsSinkConfig - Configuration class deserialized by this factory
• ElasticRestHighLevelClientAdapter - Elasticsearch client created during validation
• OpensearchRestHighLevelClientAdapter - OpenSearch client created during validation
• SinkFactory CreateWriter - General sink factory pattern