[FLINK-38962][connectors/opensearch] Add OpenSearch 3.x support with opensearch-java client by uliandim · Pull Request #56 · apache/flink-connector-opensearch

uliandim · 2026-01-22T13:56:11Z

Add flink-connector-opensearch3 module using opensearch-java 3.0.0 client
Add flink-sql-connector-opensearch3 shaded JAR module
Add flink-connector-opensearch3-e2e-tests module
Add OPENSEARCH_3 docker image version for testing
Use new BulkOperation wrapper for serializable bulk operations
Configure httpcore5/httpcore5-h2 5.3.1 for compatibility

The new connector uses the official opensearch-java client instead of the deprecated rest-high-level-client, providing better compatibility with OpenSearch 3.x and future versions.

boring-cyborg · 2026-01-22T13:56:17Z

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

uliandim · 2026-01-22T14:33:28Z

Hi, I don't have a JIRA account. Could a committer please create a JIRA ticket for this OpenSearch 3.x support feature, or let me know if there's an existing one I should reference? I'll then update the PR title accordingly.

...rch3/src/main/java/org/apache/flink/connector/opensearch/sink/Opensearch3RequestIndexer.java

flink-connector-opensearch3/pom.xml

reta

Thanks @uliandim , this is largely adapted from flink-connector-elasticsearch, right? (no issues here, just the fact it is the existing code that already works) Thank you

flink-connector-opensearch3/pom.xml

...-opensearch3/src/main/java/org/apache/flink/connector/opensearch/sink/Opensearch3Writer.java

...or-opensearch3/src/main/java/org/apache/flink/connector/opensearch/sink/Opensearch3Sink.java

…port

reta · 2026-02-15T21:10:37Z

@snuyanzin could you help us here please? the implementation is pretty much aligned with https://github.com/apache/flink-connector-elasticsearch, thank you

snuyanzin · 2026-02-15T22:06:02Z

flink-sql-connector-opensearch3/src/main/resources/META-INF/NOTICE

@@ -0,0 +1,5 @@
+flink-sql-connector-opensearch3
+Copyright 2014-2024 The Apache Software Foundation


Suggested change

Copyright 2014-2024 The Apache Software Foundation

Copyright 2014-2026 The Apache Software Foundation

?

snuyanzin · 2026-02-15T22:08:11Z

flink-connector-opensearch3/pom.xml

+<version>${httpclient5.version}</version>
+</dependency>
+
+<!-- Explicit httpcore5 to avoid version conflicts -->
+<dependency>
+<groupId>org.apache.httpcomponents.core5</groupId>
+<artifactId>httpcore5</artifactId>
+<version>5.3.1</version>
+</dependency>
+<dependency>
+<groupId>org.apache.httpcomponents.core5</groupId>
+<artifactId>httpcore5-h2</artifactId>
+<version>5.3.1</version>
+</dependency>


finally we have same version for all, should we reuse ${httpclient5.version} or have a separate property?

httpclient5 and httpcore5 do evolve separately, there is no version alignment

snuyanzin · 2026-02-15T22:11:24Z

flink-connector-opensearch3/pom.xml

+<optional>true</optional>
+</dependency>
+
+<!-- Opensearch Java Client (new official client) -->


what is new?
after half a year/1 or 2 years will it be still new?
Or should we have a bit different comment here, may be with mentioning a major version

snuyanzin · 2026-02-15T22:13:21Z

...nsearch3/src/test/java/org/apache/flink/connector/opensearch/sink/Opensearch3TestClient.java

+
+ public void assertThatIdsAreWritten(String index, Integer... ids) throws Exception {
+ // Wait for documents to be indexed
+ Thread.sleep(2000);


are we sure it is enough, ideally we should avoid such timeouts...

snuyanzin · 2026-02-15T22:16:21Z

...nsearch3/src/test/java/org/apache/flink/connector/opensearch/sink/Opensearch3SinkITCase.java

+
+ @Override
+ public Long map(Long value) throws Exception {
+ Thread.sleep(50);


why do we need this?

snuyanzin · 2026-02-15T22:19:41Z

...search3/src/main/java/org/apache/flink/connector/opensearch/table/RowOpensearch3Emitter.java

+ throw new FlinkRuntimeException("Failed to initialize serialization schema.", e);
+ }
+ indexGenerator.open();
+ objectMapper = new ObjectMapper();


is there a reason we need to create it each time?
Can we instead having just one for all (static final) ?

Jackson javadoc says it should be ok if configuration is same https://github.com/FasterXML/jackson-databind/blob/16187a7ecf69fda4eacfab742ab74b07bd3d96d2/src/main/java/com/fasterxml/jackson/databind/ObjectMapper.java#L83-L88

snuyanzin · 2026-02-15T22:28:14Z

...search3/src/main/java/org/apache/flink/connector/opensearch/sink/Opensearch3SinkBuilder.java

+ public Opensearch3SinkBuilder<IN> setHosts(HttpHost... hosts) {
+ checkNotNull(hosts);
+ checkState(hosts.length > 0, "Hosts cannot be empty.");
+ this.hosts = Arrays.asList(hosts);
+ return self();
+ }


Do I understand it correctly: right now it validates that

hosts is not null

hosts is not empty

how about case when hosts is not null and have 2 elements one of which is null and second is not null?

snuyanzin · 2026-02-15T22:29:16Z

...search3/src/main/java/org/apache/flink/connector/opensearch/sink/Opensearch3SinkBuilder.java

+ */
+ public Opensearch3SinkBuilder<IN> setConnectionUsername(String username) {
+ checkNotNull(username);
+ this.username = username;


can user name be empty or blank?

snuyanzin · 2026-02-15T22:35:05Z

...search3/src/main/java/org/apache/flink/connector/opensearch/sink/Opensearch3SinkBuilder.java

+ * @param maxSizeMb the maximum size of buffered actions, in mb.
+ * @return this builder
+ */
+ public Opensearch3SinkBuilder<IN> setBulkFlushMaxSizeMb(int maxSizeMb) {


should we use something like memoryUnit here https://github.com/apache/flink/blob/4d4c30933543fb9796a941c356cb01ee84903da9/flink-core-api/src/main/java/org/apache/flink/configuration/MemorySize.java#L358-L363
?

snuyanzin · 2026-02-15T22:39:35Z

...search3/src/main/java/org/apache/flink/connector/opensearch/sink/Opensearch3SinkBuilder.java

+ private int bulkFlushMaxActions = 1000;
+ private int bulkFlushMaxMb = -1;
+ private long bulkFlushInterval = -1;
+ private FlushBackoffType bulkFlushBackoffType = FlushBackoffType.NONE;
+ private int bulkFlushBackoffRetries = -1;
+ private long bulkFlushBackOffDelay = -1;
+ private DeliveryGuarantee deliveryGuarantee = DeliveryGuarantee.AT_LEAST_ONCE;
+ private List<HttpHost> hosts;
+ protected Opensearch3Emitter<? super IN> emitter;
+ private String username;
+ private String password;
+ private String connectionPathPrefix;
+ private Integer connectionTimeout;
+ private Integer connectionRequestTimeout;
+ private Integer socketTimeout;
+ private Boolean allowInsecure;
+ private Opensearch3FailureHandler failureHandler = new DefaultFailureHandler();


idea for follow up PR: I wonder if there should be a possibility to set these params via config options

snuyanzin · 2026-02-15T22:46:25Z

flink-connector-opensearch3-e2e-tests/dependency-reduced-pom.xml

@@ -0,0 +1,382 @@
+<?xml version="1.0" encoding="UTF-8"?>


why do we have dependency-reduced-pom.xml here?
it should be in .gitignore

snuyanzin · 2026-02-15T22:46:55Z

flink-connector-opensearch2-e2e-tests/dependency-reduced-pom.xml

@@ -0,0 +1,382 @@
+<?xml version="1.0" encoding="UTF-8"?>


drop this file, it should be in .gitignore

snuyanzin · 2026-02-15T22:47:18Z

flink-connector-opensearch-e2e-tests/dependency-reduced-pom.xml

@@ -0,0 +1,382 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">


drop this file, it should be in .gitignore

snuyanzin · 2026-02-15T22:49:14Z

I put some minor comments
also it seems CI is not happy yet

SubNader · 2026-03-10T09:24:08Z

Hello everyone, we are very interested in this. Is there any blockers holding the release back?
Let me know if I can support.

boring-cyborg bot added the component=Connectors/Opensearch label Jan 22, 2026

uliandim changed the title ~~Add OpenSearch 3.x support with opensearch-java client~~ [hotfix][connectors/opensearch] Add OpenSearch 3.x support with opensearch-java client Jan 22, 2026

uliandim force-pushed the main branch from c652af5 to 8aa9015 Compare January 22, 2026 14:35

uliandim changed the title ~~[hotfix][connectors/opensearch] Add OpenSearch 3.x support with opensearch-java client~~ [FLINK-38962] Add OpenSearch 3.x support with opensearch-java client Jan 22, 2026

uliandim changed the title ~~[FLINK-38962] Add OpenSearch 3.x support with opensearch-java client~~ [FLINK-38962][connectors/opensearch] Add OpenSearch 3.x support with opensearch-java client Jan 22, 2026

uliandim force-pushed the main branch from 8aa9015 to 55d95b9 Compare January 22, 2026 16:34

mishail reviewed Jan 23, 2026

View reviewed changes

...rch3/src/main/java/org/apache/flink/connector/opensearch/sink/Opensearch3RequestIndexer.java Show resolved Hide resolved

reta reviewed Feb 3, 2026

View reviewed changes

flink-connector-opensearch3/pom.xml Outdated Show resolved Hide resolved

reta approved these changes Feb 3, 2026

View reviewed changes

mishail reviewed Feb 3, 2026

View reviewed changes

flink-connector-opensearch3/pom.xml Outdated Show resolved Hide resolved

uliandim force-pushed the main branch 2 times, most recently from 488195d to 2deaf99 Compare February 4, 2026 15:26

mishail reviewed Feb 4, 2026

View reviewed changes

...-opensearch3/src/main/java/org/apache/flink/connector/opensearch/sink/Opensearch3Writer.java Outdated Show resolved Hide resolved

...or-opensearch3/src/main/java/org/apache/flink/connector/opensearch/sink/Opensearch3Sink.java Outdated Show resolved Hide resolved

[FLINK-38962][connectors/opensearch] Add OpenSearch 3.x connector sup…

86379b0

…port

uliandim force-pushed the main branch from 2deaf99 to 86379b0 Compare February 5, 2026 08:30

snuyanzin reviewed Feb 15, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-38962][connectors/opensearch] Add OpenSearch 3.x support with opensearch-java client#56

[FLINK-38962][connectors/opensearch] Add OpenSearch 3.x support with opensearch-java client#56
uliandim wants to merge 1 commit intoapache:mainfrom
uliandim:main

uliandim commented Jan 22, 2026

boring-cyborg bot commented Jan 22, 2026

uliandim commented Jan 22, 2026

Uh oh!

Uh oh!

reta left a comment

Uh oh!

Uh oh!

Uh oh!

reta commented Feb 15, 2026

snuyanzin Feb 15, 2026

snuyanzin Feb 15, 2026

reta Feb 15, 2026

snuyanzin Feb 15, 2026 •

edited

Loading

snuyanzin Feb 15, 2026

snuyanzin Feb 15, 2026

snuyanzin Feb 15, 2026 •

edited

Loading

snuyanzin Feb 15, 2026

snuyanzin Feb 15, 2026

snuyanzin Feb 15, 2026

snuyanzin Feb 15, 2026

snuyanzin Feb 15, 2026

snuyanzin Feb 15, 2026

snuyanzin Feb 15, 2026

snuyanzin commented Feb 15, 2026

SubNader commented Mar 10, 2026

Labels

6 participants

		@@ -0,0 +1,5 @@
		flink-sql-connector-opensearch3
		Copyright 2014-2024 The Apache Software Foundation

	Copyright 2014-2024 The Apache Software Foundation
	Copyright 2014-2026 The Apache Software Foundation

		@@ -0,0 +1,382 @@
		<?xml version="1.0" encoding="UTF-8"?>
		<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

Conversation

uliandim commented Jan 22, 2026

boring-cyborg bot commented Jan 22, 2026

uliandim commented Jan 22, 2026

Uh oh!

Uh oh!

reta left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

reta commented Feb 15, 2026

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

snuyanzin Feb 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

snuyanzin Feb 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

snuyanzin commented Feb 15, 2026

SubNader commented Mar 10, 2026

Labels

6 participants

snuyanzin Feb 15, 2026 •

edited

Loading

snuyanzin Feb 15, 2026 •

edited

Loading