[FLINK-25756] [connectors/opensearch] Dedicated Opensearch connectors #1

reta · 2022-09-23T16:43:56Z

Signed-off-by: Andriy Redko andriy.redko@aiven.io

What is the purpose of the change

The goal of this change is to provide dedicated Opensearch connectors [1], [2], [3], [4].

Brief change log

The implementation is largely based on the existing Elasticsearch 7 connector with a few notable changes (besides the dependencies and APIs):

any mentions and uses of mapping types have been removed: it is deprecated feature, scheduled for removal (the indices with mapping types cannot be created or migrated to Opensearch 1.x and beyond)
any mentions and uses have been removed: it is deprecated feature, scheduled for removal (only HighLevelRestClient is used)
the default distributions of Opensearch come with HTTPS turned on, using self-signed certificates: to simplify the integration a new option allow-insecure has been added to suppress certificates validation for development and testing purposes
old streaming APIs are also supported to facilitate the migration of existing applications from Elasticsearch 7/6 to Opensearch (the classes will change but the familiar model will stay)

The new connector name is opensearch and it follows the existing conventions:

CREATE TABLE users ( ... ) WITH (
  'connector' = 'opensearch', 
  'hosts' = 'https://localhost:9200',
  'index' = 'users', 
  'allow-insecure' = 'true', 
  'username' = 'admin', 
  'password' = 'admin');

Verifying this change

This change added comprehensive tests and can be verified as follows (largely ported the existing unit and integration tests for Elasticsearch 7):

Added unit tests
Added integration tests for end-to-end
Added end-to-end tests
Manually verified the connector by running a node clusters

Does this pull request potentially affect one of the following parts:

Dependencies: yes (the latest Opensearch 1.2.4 APIs as of this moment)
The public API, i.e., is any changed class annotated with @Public(Evolving): yes
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? (docs - in progress, JavaDocs)

Huge thanks @snuyanzin for help.
Retargeting apache/flink#18541 to separate repository

[1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-243%3A+Dedicated+Opensearch+connectors
[2] https://www.mail-archive.com/dev@flink.apache.org/msg58911.html
[3] https://lists.apache.org/thread/jls0vqc7jb84jp14j4jok1pqfgo2cl30
[4] https://lists.apache.org/thread/4bms24983g38q956rp8qmm4bpdo4361s

reta · 2022-10-03T12:53:59Z

@MartijnVisser could you please take a look? thank you :-)

reta · 2022-11-04T14:36:35Z

@MartijnVisser doing my deligance by pinging once per month :-)

reta · 2022-11-14T19:14:54Z

@MartijnVisser the POMs and CI scripts have been updated accordingly

zentol · 2022-11-15T10:22:45Z

Can you split the commit such that 1 adds the workflows and another the rest? We can then merge the workflow commit first and actually get CI in this PR.

zentol

Made a pass over infrastructure/build/legal files.

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

flink-connector-opensearch/pom.xml

flink-sql-connector-opensearch/pom.xml

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

reta · 2022-11-15T13:47:04Z

Can you split the commit such that 1 adds the workflows and another the rest? We can then merge the workflow commit first and actually get CI in this PR.

Sure, #2, thanks @zentol

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

reta · 2022-11-28T14:23:05Z

@zentol thnaks a lot for the review, I think I addressed all your comments, please let me know if I missed some, thanks!

@reta There are some unresolved comments, like #1 (comment), #1 (comment), #1 (comment).

Thanks @zentol , addressed or/and answered

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

zentol · 2022-11-29T12:12:58Z

...earch/src/test/java/org/apache/flink/streaming/connectors/opensearch/OpensearchSinkTest.java

+        testHarness.processElement(new StreamRecord<>("msg-1"));
+
+        // Await for flush to be complete
+        awaitForFlushToFinish(1);


You can avoid this kind of busy waiting quite easily by using OneShotLatches.

private OneShotLatch addResponse(Consumer<HttpResponse> consumer) { OneShotLatch oneShotLatch = new OneShotLatch(); responses.add( response -> { consumer.accept(response); oneShotLatch.trigger(); }); return oneShotLatch; } .... OneShotLatch firstResponse = responses.add( createResponse( new BulkItemResponse( 1, OpType.INDEX, new IndexResponse( new ShardId("test", "-", 0), "_doc", "1", 0, 0, 1, true)))); ... firstResponse.await();

That being said, this whole "wait for server response to be consumed" approach is a bit sketchy?
We're only waiting for the server to send the response, not for the connector to process it.

I would argue there is definitely no need for additional latch: the queue is the "latch" by itself - we know when it is drained. We could also use https://github.com/awaitility/awaitility to make the waiting logic clean.

That being said, this whole "wait for server response to be consumed" approach is a bit sketchy?

We are dealing with async system so I think this is kind of expectable: we are waiting for some events to happen at some point (be it latch, queue or other means). The sink does not provide any means to hook inside and, for example, we could observe the "process it" part using getNumPendingRequests only.

The point was to not wait by looping over a condition to save CI resources.
This is especially a problem when the condition is never fulfilled and we just churn through CPU cycles for no good reason.

The point was to not wait by looping over a condition to save CI resources.

Sure, replaced with Condition, should be simpler than per-response monitors.

This is especially a problem when the condition is never fulfilled and we just churn through CPU cycles for no good reason.

All tests are timeboxed to 5 seconds

All tests are timeboxed to 5 seconds

a) In our experience such small timeouts actually cause tests to be unstable.
b) code has a tendency to travel. Arguing that "this code is fine because of that code over there" basically means "this code is unsafe to copy", which isn't a good state to be in.

The tests seem to be very stable, I was running them continuously for 2h and have not seen any flakyness.

It's more of a CI problem. Time jumping ahead and things like that. 😩

I think we could always update the test cases, if the instability comes in, the 5 seconds buffer is 3x buffer for what test case should take.

zentol · 2022-11-29T12:14:52Z

...earch/src/test/java/org/apache/flink/streaming/connectors/opensearch/OpensearchSinkTest.java

+        testHarness.processElement(new StreamRecord<>("msg"));
+
+        // current number of pending request should be 1 due to the re-add
+        assertThat(sink.getNumPendingRequests()).isEqualTo(1);


This assertion sometimes fails locally when the test is run in a loop.

expected: 1L but was: 0L

Aha, I think I know why, will fix that

...earch/src/test/java/org/apache/flink/streaming/connectors/opensearch/OpensearchSinkTest.java

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

reta · 2022-11-29T15:46:45Z

...earch/src/test/java/org/apache/flink/streaming/connectors/opensearch/OpensearchSinkTest.java

+        };
+    }
+
+    private static void awaitForCondition(Supplier<Boolean> condition) throws InterruptedException {


There is only 1 place when we need that, due to the fact that failureRequestIndexer is not immediately re-adding the requests but does it on the next batch interval only

zentol · 2022-11-30T09:28:55Z

...earch/src/test/java/org/apache/flink/streaming/connectors/opensearch/OpensearchSinkTest.java

+                            lock.lock();
+                            try {
+                                responses.poll().accept(resp);
+                                flushed.signalAll();


Or you could just use a OneShotLatch which makes all of this logic here a one-liner and gives you more control to wait for a specific message to be consumed 🤷
(and you wouldn't need awaitForCondition)

awaitForCondition is not checking the flushing part, it is checking the numPendingRequest, which will be eventually updated, for the test it is sufficient to just know that flush has happened (at least, I don't see the reasons to complicate this part with per-response latches).

reta · 2022-12-05T15:45:56Z

@zentol thanks one more time for the review, I think none of your comments left unattended, thanks

wbeckler · 2022-12-07T18:41:47Z

It would be amazing if this could be based on the opensearch-java client rather than RHLC, since that is where all of the improvements are going. Is that how this is set up?

reta · 2022-12-07T19:13:21Z

It would be amazing if this could be based on the opensearch-java client rather than RHLC, since that is where all of the improvements are going. Is that how this is set up?

Not for now, see please opensearch-project/opensearch-java#181 (this is what the connector uses).

pom.xml

zentol · 2022-12-08T11:35:50Z

pom.xml

+				</plugin>
+			</plugins>


Suggested change

</plugin>

</plugins>

</plugin>

</plugins>

zentol

Just noticed that we're lacking documentation; like https://github.com/apache/flink-connector-elasticsearch/blob/main/docs/content/docs/connectors/datastream/elasticsearch.md.

reta · 2022-12-08T13:09:06Z

Just noticed that we're lacking documentation; like https://github.com/apache/flink-connector-elasticsearch/blob/main/docs/content/docs/connectors/datastream/elasticsearch.md.

Thanks @zentol, I have it ready and thought to open separate pull request (to split MDs from code), makes sense to you? (#3)

zentol · 2022-12-09T11:18:12Z

ok

ypark2103 · 2022-12-21T17:33:49Z

hi @zentol @reta
I see that flink-opensearch-connector was merged to master recently but I don't see it was registered in maven repository.
When will this be added to maven repository so I can use this from pom.xml? I also don't see opensearch connector in official flink documentation. Can this library be used in Java 8 also?

reta · 2022-12-21T17:43:47Z

hi @zentol @reta I see that flink-opensearch-connector was merged to master recently but I don't see it was registered in maven repository. When will this be added to maven repository so I can use this from pom.xml?

Hey @ypark2103 , the vote was ongoing and just approved [1], expect artifacts to be available in the coming days.

[1] https://lists.apache.org/list?dev@flink.apache.org:lte=1M:Opensearch

I also don't see opensearch connector in official flink documentation.

The Opensearch documentations available here [2] but I am not sure how it is going to be integrated into Flink documentation, @zentol may be you could spot some light?

[2] https://github.com/apache/flink-connector-opensearch/tree/main/docs

Can this library be used in Java 8 also?

Yes

ypark2103 · 2022-12-21T17:56:38Z

@reta Is there a way we can install this library with pom.xml?
I don't see opensearch in maven depository yet like elasticsearch connector.
https://mvnrepository.com/artifact/org.apache.flink/flink-connector-elasticsearch7
The artifacts you mentioned that will be available in the coming days is referring to registering this library to maven repo?

reta · 2022-12-21T18:06:18Z

The artifacts you mentioned that will be available in the coming days is referring to registering this library to maven repo?

@ypark2103 Correct

MartijnVisser · 2022-12-21T18:54:35Z

The Opensearch documentations available here [2] but I am not sure how it is going to be integrated into Flink documentation, @zentol may be you could spot some light?

They will be integrated via apache/flink#21518 when the final steps for the release process have been completed :)

ypark2103 · 2022-12-27T21:47:02Z

@reta @MartijnVisser Is this Flink-Connector-Opensearch only available for flink 1.16 version? What about flink 1.11 or 1.12?

reta · 2022-12-27T22:14:18Z

@ypark2103 that is correct, the Flink Opensearch Connector was following the Flink's externalization model for connectors, the necessary scaffolding is only available in Flink 1.16 and above.

ypark2103 · 2022-12-28T18:52:46Z

@reta I heard java 8 was deprecated starting Flink 1.15. Does it mean I need to upgrade to java 11 to use this connector? Or can I still use java 8 for Flink 1.16?

reta · 2022-12-28T19:24:24Z

@reta I heard java 8 was deprecated starting Flink 1.15. Does it mean I need to upgrade to java 11 to use this connector? Or can I still use java 8 for Flink 1.16?

@ypark2103 the connector itself has baseline of JDK-8 so you don't need JDK-11 to use it as-is (using 1.3.x OpenSearch client), but if you plan to upgrade to JDK-11, it is even better.

reta force-pushed the FLINK-25756 branch 3 times, most recently from 35abeac to 71d9810 Compare November 3, 2022 12:42

reta force-pushed the FLINK-25756 branch from 71d9810 to 4762bf1 Compare November 14, 2022 19:11

reta mentioned this pull request Nov 14, 2022

[FLINK-25756] [connectors/opensearch] Dedicated Opensearch connectors apache/flink#18541

Closed

zentol requested changes Nov 15, 2022

View reviewed changes

zentol self-assigned this Nov 15, 2022

reta force-pushed the FLINK-25756 branch 11 times, most recently from 2ffb58c to 8500c3e Compare November 16, 2022 13:47

reta added 3 commits November 16, 2022 11:59

[FLINK-25756] [connectors/opensearch] Dedicated Opensearch connectors

6e38dd6

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

Addressing code review comments

e3c5e27

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

Fix build and dependency convergence issues

1d9eb6c

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

reta force-pushed the FLINK-25756 branch 2 times, most recently from c8936e2 to 49d2539 Compare November 16, 2022 17:16

Added Openseach Connector e2e tests

cc2148f

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

reta force-pushed the FLINK-25756 branch from 49d2539 to cc2148f Compare November 16, 2022 18:20

boring-cyborg bot added the component=Connectors/Opensearch label Nov 28, 2022

Address code review comments. Added OpensearchSinkTest test case

45098e3

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

reta force-pushed the FLINK-25756 branch from c37dc06 to 45098e3 Compare November 28, 2022 22:30

zentol requested changes Nov 29, 2022

View reviewed changes

Address code review comments

9dc4d72

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

reta force-pushed the FLINK-25756 branch from 9c76139 to 9dc4d72 Compare November 29, 2022 14:30

Address code review comments

d8445bb

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

reta commented Nov 29, 2022

View reviewed changes

zentol reviewed Nov 30, 2022

View reviewed changes

zentol approved these changes Dec 8, 2022

View reviewed changes

zentol added 2 commits December 8, 2022 12:54

Update pom.xml

81d49d6

pom cleanup

0b8d0d9

zentol requested changes Dec 8, 2022

View reviewed changes

reta mentioned this pull request Dec 8, 2022

[FLINK-25756] [connectors/opensearch] Dedicated Opensearch connectors (documentation) #3

Merged

zentol merged commit 48b0c0f into apache:main Dec 9, 2022

[FLINK-25756] [connectors/opensearch] Dedicated Opensearch connectors #1

[FLINK-25756] [connectors/opensearch] Dedicated Opensearch connectors #1

Conversation

reta commented Sep 23, 2022 • edited Loading

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

reta commented Oct 3, 2022

reta commented Nov 4, 2022

reta commented Nov 14, 2022

zentol commented Nov 15, 2022

zentol left a comment

Choose a reason for hiding this comment

reta commented Nov 15, 2022

reta commented Nov 28, 2022

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

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

reta Nov 30, 2022 • edited Loading

Choose a reason for hiding this comment

reta commented Dec 5, 2022 • edited Loading

wbeckler commented Dec 7, 2022

reta commented Dec 7, 2022

Choose a reason for hiding this comment

zentol left a comment • edited Loading

Choose a reason for hiding this comment

reta commented Dec 8, 2022 • edited Loading

zentol commented Dec 9, 2022

ypark2103 commented Dec 21, 2022

reta commented Dec 21, 2022

ypark2103 commented Dec 21, 2022 • edited Loading

reta commented Dec 21, 2022

MartijnVisser commented Dec 21, 2022

ypark2103 commented Dec 27, 2022

reta commented Dec 27, 2022

ypark2103 commented Dec 28, 2022

reta commented Dec 28, 2022

reta commented Sep 23, 2022 •

edited

Loading

reta Nov 30, 2022 •

edited

Loading

reta commented Dec 5, 2022 •

edited

Loading

zentol left a comment •

edited

Loading

reta commented Dec 8, 2022 •

edited

Loading

ypark2103 commented Dec 21, 2022 •

edited

Loading