[FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support #53

mtfelisb · 2023-03-25T21:06:18Z

Description

The eighth version was released in February last year and has a lot of internal changes, such as releasing a whole new Java API Client. Therefore, we can create a full new integration based on the latest Java API, and rely on the compatibility mode to previous versions. That is the goal of this pull request.

Jira

FLINK-26088

boring-cyborg · 2023-03-25T21:06:21Z

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

MartijnVisser · 2023-05-22T18:43:39Z

@mtfelisb What's the status of this PR?

mtfelisb · 2023-05-31T02:12:10Z

@MartijnVisser I'm wrapping up, and just need to write a few more tests. Even though I'm concerned about the implementation using Kryo, I'll mark it as ready to review and get your feedback.

MartijnVisser · 2023-05-31T07:08:21Z

Even though I'm concerned about the implementation using Kryo, I'll mark it as ready to review and get your feedback.

@reta Could you help with a review?

reta · 2023-05-31T12:51:23Z

The top level pom.xml misses the new module:

		<module>flink-connector-elasticsearch8</module>

...csearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8Sink.java

...or-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java

...earch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8Writer.java

...arch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/INetworkConfigFactory.java

...search8/src/main/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializer.java

StefanXiepj · 2023-05-31T14:54:28Z

I try to compile your code, but throwing follow errors:

[INFO] There are 69 errors reported by Checkstyle 8.14 with /tools/maven/checkstyle.xml ruleset.
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializer.java:[33] (javadoc) JavadocType: 缺少 Javadoc 。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:[26] (imports) AvoidStarImport: 不应使用 '.*' 形式的导入 - java.io.* 。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:[29] (javadoc) JavadocType: 缺少 Javadoc 。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:[53] (blocks) NeedBraces: 'if' 结构必须使用大括号 '{}'。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:[54] (blocks) NeedBraces: 'if' 结构必须使用大括号 '{}'。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/INetworkConfigFactory.java:[21] (whitespace) EmptyLineSeparator: 'package 前应有空行。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/INetworkConfigFactory.java:[27] (javadoc) JavadocType: 缺少 Javadoc 。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8Writer.java:[32] (imports) ImportOrder: Import org.apache.flink.api.connector.sink2.Sink appears after other imports that it should precede
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8Writer.java:[35] (imports) ImportOrder: 导入组之前的额外空行 'org.apache.flink.connector.base.sink.writer.BufferedRequestState'。

mtfelisb · 2023-06-06T23:40:36Z

Even though I'm concerned about the implementation using Kryo, I'll mark it as ready to review and get your feedback.

@reta Could you help with a review?

@reta thank you for the review. I appreciate that. I'm going to work on what you pointed out.

mtfelisb · 2023-06-06T23:43:58Z

I try to compile your code, but throwing follow errors:

[INFO] There are 69 errors reported by Checkstyle 8.14 with /tools/maven/checkstyle.xml ruleset.
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializer.java:[33] (javadoc) JavadocType: 缺少 Javadoc 。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:[26] (imports) AvoidStarImport: 不应使用 '.*' 形式的导入 - java.io.* 。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:[29] (javadoc) JavadocType: 缺少 Javadoc 。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:[53] (blocks) NeedBraces: 'if' 结构必须使用大括号 '{}'。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:[54] (blocks) NeedBraces: 'if' 结构必须使用大括号 '{}'。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/INetworkConfigFactory.java:[21] (whitespace) EmptyLineSeparator: 'package 前应有空行。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/INetworkConfigFactory.java:[27] (javadoc) JavadocType: 缺少 Javadoc 。
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8Writer.java:[32] (imports) ImportOrder: Import org.apache.flink.api.connector.sink2.Sink appears after other imports that it should precede
[ERROR] src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8Writer.java:[35] (imports) ImportOrder: 导入组之前的额外空行 'org.apache.flink.connector.base.sink.writer.BufferedRequestState'。

@StefanXiepj I'll fix these Checkstyle errors, and once I do, I'll let you know.

reswqa · 2023-08-22T04:54:06Z

What's the progress of this?

mtfelisb · 2023-08-24T00:45:13Z

What's the progress of this?

Hi @reswqa, thanks for asking.
I'm writing the retry mechanism inspired by the PR that @reta shared here. After that, I'll write tests and mark the PR as ready for review.

reswqa · 2023-08-24T13:25:00Z

Thanks @mtfelisb! I have seen many people's urgent need for this, so feel free to ping me for reviewing to accelerate the progress. And I can provide any possible help :)

StefanXiepj · 2023-09-09T14:51:45Z

Thanks @mtfelisb! I have seen many people's urgent need for this, so feel free to ping me for reviewing to accelerate the progress. And I can provide any possible help :)

@reswqa can you help me to review my another pr for this issue ? thanks

mtfelisb · 2023-09-16T20:39:25Z

Guys, I wrote a bunch of tests heavily inspired by apache/flink-connector-opensearch#5 since they have very similar implementations. Thanks @reta for sharing, it helped a lot!
I need to create the documentation and would appreciate some guidance on how to generate that.

I'm looking forward to your feedback :)
cc @MartijnVisser @reswqa

rinkako · 2023-10-07T02:12:51Z

hi @mtfelisb , I read the code and found that it has no implementation for DynamicTableSink, hence this es8 sink cannot be used in flink-sql. Would you have any plan to work on this point? 😄

mtfelisb · 2023-10-09T20:49:56Z

hi @mtfelisb , I read the code and found that it has no implementation for DynamicTableSink, hence this es8 sink cannot be used in flink-sql. Would you have any plan to work on this point? 😄

Hi, @rinkako! I wasn't planning to do it. To be honest with you I'm not quite familiar with it. However, it could be a significant increment, indeed. I'm open to collaborations 😃

snuyanzin · 2024-02-03T11:37:28Z

I have a question about license
it looks like we don't have any info about Elastic 8.x in NOTICE, should we add it or what is the best approach here?
Based on the info [1] it is no longer produced under Apache 2.0 license after 7.10

[1] https://www.elastic.co/pricing/faq/licensing

reta · 2024-02-03T12:43:22Z

Based on the info [1] it is no longer produced under Apache 2.0 license after 7.10

The ES Java Client license is still Apache [1] (it is an exemption)

[1] https://github.com/elastic/elasticsearch-java

snuyanzin · 2024-02-06T07:51:40Z

good to know, thanks

flink-connector-elasticsearch8/pom.xml

mtfelisb · 2024-02-08T00:28:06Z

Hi, @MartijnVisser. I have a question regarding compatibility. This pull request is working on every Flink version but 1.19-SNAPSHOT. I've checked this issue, however, I'm already using TestSinkInitContext here. It's unclear if I should consider 1.19-SNAPSHOT and upward only. Am I missing something here?

MartijnVisser · 2024-02-08T08:19:42Z

This pull request is working on every Flink version but 1.19-SNAPSHOT

I'm suspecting that's because of https://issues.apache.org/jira/browse/FLINK-25857 and https://issues.apache.org/jira/browse/FLINK-33972 which have introduced new changes to the Sink API.

It's unclear if I should consider 1.19-SNAPSHOT and upward only.

There's only thing that the community agreed on and that is that the last two minor version of Flink need to be supported. At this moment of writing, that means that Flink 1.17 and Flink 1.18 need to be supported. At the moment Flink 1.19 gets officially released, we need to have a connector that supports both Flink 1.18 and Flink 1.19.

However, that doesn't mean that connector support needs to come from one connector version. It's perfectly fine to say "There's the Flink Elastichsearch connector v3.0 that supports Flink 1.17 and 1.18, and we'll have a new Flink Elasticsearch connector v3.1 / v4.0 that supports only Flink 1.19 and upwards". It just means that there's more work involved in backporting any bugfixes (and potentially new features) to two branches. It's explained in more depth at https://cwiki.apache.org/confluence/display/FLINK/Externalized+Connector+development#ExternalizedConnectordevelopment-Support

In this case, I do think that it would be possible to add support for Flink 1.17/1.18 and the upcoming Flink 1.19 in one branch. At least, that's also what's being tried in the Flink Kafka connector, see apache/flink-connector-kafka#84

Jiabao-Sun · 2024-02-08T12:38:44Z

Hi, @MartijnVisser. I have a question regarding compatibility. This pull request is working on every Flink version but 1.19-SNAPSHOT. I've checked this issue, however, I'm already using TestSinkInitContext here. It's unclear if I should consider 1.19-SNAPSHOT and upward only. Am I missing something here?

TestSinkInitContext is ok if we use Sink.createWriter method to create a writer.
https://github.com/apache/flink/pull/24180/files#r1467200868

...test/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriterITCase.java

mtfelisb · 2024-02-09T01:02:52Z

This pull request is working on every Flink version but 1.19-SNAPSHOT

I'm suspecting that's because of https://issues.apache.org/jira/browse/FLINK-25857 and https://issues.apache.org/jira/browse/FLINK-33972 which have introduced new changes to the Sink API.

It's unclear if I should consider 1.19-SNAPSHOT and upward only.

There's only thing that the community agreed on and that is that the last two minor version of Flink need to be supported. At this moment of writing, that means that Flink 1.17 and Flink 1.18 need to be supported. At the moment Flink 1.19 gets officially released, we need to have a connector that supports both Flink 1.18 and Flink 1.19.

However, that doesn't mean that connector support needs to come from one connector version. It's perfectly fine to say "There's the Flink Elastichsearch connector v3.0 that supports Flink 1.17 and 1.18, and we'll have a new Flink Elasticsearch connector v3.1 / v4.0 that supports only Flink 1.19 and upwards". It just means that there's more work involved in backporting any bugfixes (and potentially new features) to two branches. It's explained in more depth at https://cwiki.apache.org/confluence/display/FLINK/Externalized+Connector+development#ExternalizedConnectordevelopment-Support

In this case, I do think that it would be possible to add support for Flink 1.17/1.18 and the upcoming Flink 1.19 in one branch. At least, that's also what's being tried in the Flink Kafka connector, see apache/flink-connector-kafka#84

I understand now, it makes total sense. I appreciate the detailed information you provided, @MartijnVisser 😀

flink-connector-elasticsearch8/pom.xml

...src/test/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBaseITCase.java

drorventura · 2024-03-19T07:31:58Z

Hi
@snuyanzin , @Jiabao-Sun
Is this planned to be merged soon?

...src/test/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBaseITCase.java

drorventura · 2024-03-19T10:49:27Z

Thank you @mtfelisb
@reswqa could you please approve? 🙏

reswqa

Thanks @mtfelisb, looks good to me. Could you please rebase and squash all commits?

mtfelisb · 2024-03-20T12:41:50Z

Thanks @mtfelisb, looks good to me. Could you please rebase and squash all commits?

Thank you for the review, @reswqa! Just did it :)

reswqa

Thanks @mtfelisb, merging...

boring-cyborg · 2024-03-21T03:46:46Z

Awesome work, congrats on your first merged pull request!

drorventura · 2024-03-21T07:27:47Z

when is the next release planned?

reswqa · 2024-03-21T08:43:04Z

when is the next release planned?

TBH, I'm not really sure. But I think we will probably release a series of connectors that supporting flink-1.19 in the near future. You can make a release request on the flink mailing list.

ebdxflr · 2024-03-25T07:56:37Z

when is the next release planned?

TBH, I'm not really sure. But I think we will probably release a series of connectors that supporting flink-1.19 in the near future. You can make a release request on the flink mailing list.

the flink mailing list is what was missing, since requests were made all over other mediums, from git to jira... can we speed things up @Flink

boring-cyborg bot added the component=Connectors/Elasticsearch label Mar 25, 2023

mtfelisb force-pushed the FLINK-26088 branch from 6c27742 to fd565a9 Compare March 25, 2023 21:09