Implementation:Risingwavelabs Risingwave EsSinkConfig

Property	Value
File	java/connector-node/risingwave-sink-es-7/src/main/java/com/risingwave/connector/EsSinkConfig.java
Language	Java
Lines	157
Category	Configuration
Package	com.risingwave.connector

Overview

EsSinkConfig is a configuration class for the Elasticsearch/OpenSearch sink connector in RisingWave. It extends CommonSinkConfig and holds connection and behavior parameters including URL, index name, delimiter, authentication credentials, batch settings, retry policies, and routing configuration. The class uses Jackson @JsonProperty annotations for deserialization and provides builder-style setter methods (returning this) for fluent configuration.

Code Reference

Source Location

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

Signature

public class EsSinkConfig extends CommonSinkConfig {
    @JsonCreator
    public EsSinkConfig(@JsonProperty(value = "url") String url);
}

Imports

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.risingwave.connector.api.sink.CommonSinkConfig;

I/O Contract

Configuration Properties

Property Key	Field	Required	Default	Description
url	url	Yes	N/A	Elasticsearch/OpenSearch cluster URL
index	index	No	null	Target index name (mutually exclusive with index_column)
delimiter	delimiter	No	null	Delimiter for generating document IDs
username	username	No	null	Authentication username
password	password	No	null	Authentication password
index_column	indexColumn	No	null	Column name to derive index from (mutually exclusive with index)
retry_on_conflict	retryOnConflict	No	3	Number of retries on version conflict
batch_num_messages	batchNumMessages	No	1000	Number of messages per bulk batch
batch_size_kb	batchSizeKb	No	5120 (5 MB)	Bulk batch size limit in KB
concurrent_requests	concurrentRequests	No	1	Number of concurrent bulk requests
routing_column	routingColumn	No	null	Column name for custom document routing

Default Value Handling

Getter methods provide default values when fields are null:

• getBatchNumMessages() returns 1000 if not set
• getBatchSizeKb() returns 5120 (5 * 1024 KB) if not set
• getRetryOnConflict() returns 3 if not set
• getConcurrentRequests() returns 1 if not set

Usage Examples

// Deserialization from a properties map via Jackson
ObjectMapper mapper = new ObjectMapper();
EsSinkConfig config = mapper.convertValue(tableProperties, EsSinkConfig.class);

// Fluent builder-style configuration
EsSinkConfig config = new EsSinkConfig("http://localhost:9200")
    .withIndex("my_index")
    .withUsername("elastic")
    .withPassword("changeme")
    .withBatchNumMessages(500)
    .withRetryOnConflict(5)
    .withConcurrentRequests(2)
    .withRoutingColumn("region");

Related Pages

• EsSinkFactory - Factory that uses this configuration to create and validate ES sinks
• ElasticBulkProcessorAdapter - Uses batch and retry settings from this config
• OpensearchBulkProcessorAdapter - Uses batch and retry settings from this config
• ElasticRestHighLevelClientAdapter - Uses credentials from this config
• OpensearchRestHighLevelClientAdapter - Uses credentials from this config