You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "mtfelisb (via GitHub)" <gi...@apache.org> on 2023/03/25 21:06:20 UTC
[GitHub] [flink-connector-elasticsearch] mtfelisb opened a new pull request, #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
mtfelisb opened a new pull request, #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53
# 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](https://issues.apache.org/jira/browse/FLINK-26088)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1827828796
@mtfelisb Are you still active on this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reta commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1211712860
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:
##########
@@ -0,0 +1,70 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperationVariant;
+
+import java.io.*;
+import java.util.Objects;
+
+public class Operation implements Serializable {
+ private final BulkOperationVariant bulkOperationVariant;
+
+ private int retries;
+
+ public Operation(BulkOperationVariant bulkOperation, int retries) {
+ this.bulkOperationVariant = bulkOperation;
+ this.retries = retries;
+ }
+
+ public void retry() {
+ this.retries = this.retries - 1;
Review Comment:
This will be called from different threads - needs atomic counter
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] boring-cyborg[bot] commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1483922123
Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reta commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1211722856
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8Writer.java:
##########
@@ -0,0 +1,134 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
+import co.elastic.clients.elasticsearch._helpers.bulk.BulkIngester;
+import co.elastic.clients.elasticsearch._helpers.bulk.BulkListener;
+import co.elastic.clients.elasticsearch.core.BulkRequest;
+import co.elastic.clients.elasticsearch.core.BulkResponse;
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
+import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.connector.base.sink.writer.AsyncSinkWriter;
+
+import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+import org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration;
+
+import org.apache.http.HttpHost;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.function.Consumer;
+
+public class Elasticsearch8Writer<InputT> extends AsyncSinkWriter<InputT, Operation> {
+ private static final Logger LOG = LoggerFactory.getLogger(Elasticsearch8Writer.class);
+
+ private final ElasticsearchAsyncClient esClient;
+
+ public Elasticsearch8Writer(
+ ElementConverter<InputT, Operation> elementConverter,
+ Sink.InitContext context,
+ int maxBatchSize,
+ int maxInFlightRequests,
+ int maxBufferedRequests,
+ long maxBatchSizeInBytes,
+ long maxTimeInBufferMS,
+ long maxRecordSizeInBytes,
+ String username,
+ String password,
+ HttpHost[] httpHosts,
+ Collection<BufferedRequestState<Operation>> state
+ ) {
+ super(
+ elementConverter,
+ context,
+ AsyncSinkWriterConfiguration.builder()
+ .setMaxBatchSize(maxBatchSize)
+ .setMaxBatchSizeInBytes(maxBatchSizeInBytes)
+ .setMaxInFlightRequests(maxInFlightRequests)
+ .setMaxBufferedRequests(maxBufferedRequests)
+ .setMaxTimeInBufferMS(maxTimeInBufferMS)
+ .setMaxRecordSizeInBytes(maxRecordSizeInBytes)
+ .build(),
+ state
+ );
+
+ this.esClient = new NetworkConfigFactory(httpHosts, username, password).create();
+ }
+
+ @Override
+ protected void submitRequestEntries(List<Operation> requestEntries, Consumer<List<Operation>> requestResult) {
+ LOG.debug("submitRequestEntries with {} items", requestEntries.size());
+
+ BulkListener<Operation> listener = new BulkListener<Operation>() {
+ @Override
+ public void beforeBulk(long executionId, BulkRequest request, List<Operation> contexts) {}
+
+ @Override
+ public void afterBulk(long executionId, BulkRequest request, List<Operation> contexts, BulkResponse response) {
+ LOG.debug("Bulk request " + executionId + " completed");
+ ArrayList<Operation> retriableEntries = new ArrayList<>();
+
+ for (int i = 0; i < contexts.size(); i++) {
+ BulkResponseItem item = response.items().get(i);
+ Operation operation = contexts.get(i);
+
+ if (item.error() != null) {
+ LOG.error("Failed operation " + operation + " - " + item.error().reason());
+ }
+
+ if (item.error() != null && operation.isRetriable()) {
+ retriableEntries.add(operation);
+ operation.retry();
+ }
+ }
+
+ requestResult.accept(retriableEntries);
+ }
+
+ @Override
+ public void afterBulk(long executionId, BulkRequest request, List<Operation> contexts, Throwable failure) {
+ LOG.debug("Bulk request " + executionId + " failed", failure);
+ requestResult.accept(requestEntries);
+ }
+ };
+
+ try (BulkIngester<Operation> ingester = BulkIngester.of(b -> b
Review Comment:
I think this breaks the semantics of the `submitRequestEntries` operation, from the javadocs (there is also an example there):
> NOTE: The client must support asynchronous requests and the method called to persist the records must asynchronously execute and return a future with the results of that request.
The `BulkIngester::close` will block and wait for all requests to be completed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reswqa commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1691672096
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 :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa merged PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-2009477539
> Thanks @mtfelisb, looks good to me. Could you please rebase and squash all commits?
Thank you for the review, @reswqa! Just did it :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "SpaceMiao (via GitHub)" <gi...@apache.org>.
SpaceMiao commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1963974534
@mtfelisb Hi, I want read data from Kafka as byte[], but when i write these byte[] to elasticsearch with your "Elasticsearch8AsyncSinkBuilder". Here are some core codes of mine:
```
KafkaSource<byte[]> bytesKafkasSource = KafkaSource.<byte[]>builder().setValueOnlyDeserializer(new ByteDeserializationSchema()).......build();
StreamExecutionEnvironment env = StreamExecutionEnvironment .createLocalEnvironmentWithWebUI();
env.formSource(bytesKafkasSource ).sinkTo(Elasticsearch8AsyncSinkBuilder.<byte[]>builder().setElementConverter((element, ctx) -> new IndexOperation.Builder<>().index().document(element).build()).build());
```
I got unexpected result, it just write small parts of the full data from the begining to the ES.
Did you test this case before? I not sure if it is because of the deserializer "Kryo" can't handle with byte[] data?
Or can you give me some advice about it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1925293405
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "YegorMedvedev (via GitHub)" <gi...@apache.org>.
YegorMedvedev commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-2621108461
Amazing work! Thank you!
However, I can't find the package in Maven, can someone share how shall I add the dependency to my app?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reta commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1211757737
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializer.java:
##########
@@ -0,0 +1,60 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.objenesis.strategy.StdInstantiatorStrategy;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+
+public class OperationSerializer {
+ private final Kryo kryo = new Kryo();
+
+ public OperationSerializer() {
+ kryo.setRegistrationRequired(false);
+ kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+ }
+
+ public void serialize(Operation request, DataOutputStream out) {
+ Output output = new Output(out);
+ kryo.writeObject(output, request);
+ output.flush();
+ }
+
+ public Operation deserialize(long requestSize, DataInputStream in) {
+ Input input = new Input(in, (int) requestSize);
Review Comment:
```suggestion
try (Input input = new Input(in, (int) requestSize)) {
return kryo.readObject(input, Operation.class);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] MartijnVisser commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1557720158
@mtfelisb What's the status of this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] mtfelisb commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1316465669
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:
##########
@@ -0,0 +1,70 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperationVariant;
+
+import java.io.*;
+import java.util.Objects;
+
+public class Operation implements Serializable {
+ private final BulkOperationVariant bulkOperationVariant;
+
+ private int retries;
+
+ public Operation(BulkOperationVariant bulkOperation, int retries) {
+ this.bulkOperationVariant = bulkOperation;
+ this.retries = retries;
+ }
+
+ public void retry() {
+ this.retries = this.retries - 1;
Review Comment:
Indeed!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] StefanXiepj commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "StefanXiepj (via GitHub)" <gi...@apache.org>.
StefanXiepj commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1570393677
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'。
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-2011167239
Awesome work, congrats on your first merged pull request!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1479350601
##########
flink-connector-elasticsearch8/pom.xml:
##########
@@ -0,0 +1,139 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<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/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-elasticsearch-parent</artifactId>
+ <version>4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>flink-connector-elasticsearch8</artifactId>
+ <name>Flink : Connectors : Elasticsearch 8</name>
+
+ <packaging>jar</packaging>
+
+ <!-- Allow users to pass custom connector versions -->
+ <properties>
+ <elasticsearch.version>8.7.0</elasticsearch.version>
+ </properties>
+
+ <dependencies>
+
+ <!-- Core -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Table ecosystem -->
+
+ <!-- Projects depending on this project won't depend on flink-table-*. -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java-bridge</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.9.0</version>
Review Comment:
why are we sticked to so old jackson which even has several CVEs?
I guess we could use 2.15.x or 2.14.x
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] MartijnVisser commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1569617360
> 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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reta commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1211695458
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:
##########
@@ -0,0 +1,70 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperationVariant;
+
+import java.io.*;
+import java.util.Objects;
+
+public class Operation implements Serializable {
+ private final BulkOperationVariant bulkOperationVariant;
Review Comment:
This is not going to work, the `BulkOperationVariant` is not serializable, I believe you need to have custom serialization here. Just to give you an idea from OpenSearch [1] implementation (not the same but very close): https://github.com/apache/flink-connector-opensearch/pull/5/files#diff-1c079dfa499f278474b93e29b5dd776722feaae97461612fec46a8d9d24b5585R43
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reta commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1211756809
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializer.java:
##########
@@ -0,0 +1,60 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.objenesis.strategy.StdInstantiatorStrategy;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+
+public class OperationSerializer {
+ private final Kryo kryo = new Kryo();
+
+ public OperationSerializer() {
+ kryo.setRegistrationRequired(false);
+ kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+ }
+
+ public void serialize(Operation request, DataOutputStream out) {
+ Output output = new Output(out);
+ kryo.writeObject(output, request);
+ output.flush();
+ }
+
+ public Operation deserialize(long requestSize, DataInputStream in) {
+ Input input = new Input(in, (int) requestSize);
+ return kryo.readObject(input, Operation.class);
+ }
+
+ public int size(Operation operation) {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ Output output = new Output(byteArrayOutputStream);
Review Comment:
```suggestion
try (Output output = new Output(byteArrayOutputStream)) {
kryo.writeObject(output, operation);
output.flush();
return (int) output.total();
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1925311387
> 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1924270597
@MartijnVisser @mtfelisb it looks great, I have no more comments, it would be great to have e2e tests added but we could also do that in separate pull request.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "drorventura (via GitHub)" <gi...@apache.org>.
drorventura commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1530068146
##########
flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBaseITCase.java:
##########
@@ -0,0 +1,107 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import org.apache.http.HttpHost;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.IOException;
+import java.time.Duration;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
Review Comment:
cc @mtfelisb
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1530090871
##########
flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBaseITCase.java:
##########
@@ -0,0 +1,107 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import org.apache.http.HttpHost;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.IOException;
+import java.time.Duration;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.junit.Assert.assertEquals;
Review Comment:
Done! Thank you, @reswqa ☺️
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "rinkako (via GitHub)" <gi...@apache.org>.
rinkako commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1751562763
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? 😄
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "drorventura (via GitHub)" <gi...@apache.org>.
drorventura commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1530067738
##########
flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBaseITCase.java:
##########
@@ -0,0 +1,107 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import org.apache.http.HttpHost;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.IOException;
+import java.time.Duration;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.junit.Assert.assertEquals;
Review Comment:
cc @mtfelisb
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1923933841
> @reta @snuyanzin Do you think either of you could help with the review of this? It seems that #74 is stale, and it would be great if we could move forward with adding ES 8
@MartijnVisser sure, looking into it now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1933568415
> 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 https://github.com/apache/flink-connector-kafka/pull/84
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reta commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1211676310
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8Sink.java:
##########
@@ -0,0 +1,119 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.flink.connector.elasticsearch.sink;
+
+import org.apache.flink.connector.base.sink.AsyncSinkBase;
+import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.http.HttpHost;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.Collections;
+
+public class Elasticsearch8Sink<InputT> extends AsyncSinkBase<InputT, Operation> {
+ private static final Logger LOG = LoggerFactory.getLogger(Elasticsearch8Sink.class);
+
+ private final String username;
+
+ private final String password;
+
+ private final HttpHost[] httpHosts;
+
+ protected Elasticsearch8Sink(
+ ElementConverter<InputT, Operation> converter,
+ int maxBatchSize,
+ int maxInFlightRequests,
+ int maxBufferedRequests,
+ long maxBatchSizeInBytes,
+ long maxTimeInBufferMS,
+ long maxRecordSizeInByte,
+ String username,
+ String password,
+ HttpHost[] httpHosts
+ ) {
+ super(
+ converter,
+ maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBatchSizeInBytes,
+ maxTimeInBufferMS,
+ maxRecordSizeInByte
+ );
+
+ this.username = username;
+ this.password = password;
+ this.httpHosts = httpHosts;
+ }
+
+ @Override
+ public StatefulSinkWriter<InputT, BufferedRequestState<Operation>> createWriter(
+ InitContext context
+ ) {
+ LOG.debug("creating writer");
Review Comment:
Personally, I don't think you need this debug logging here and in `restoreWriter`, it does not bring any useful information (just my opinion)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reta commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1211729913
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/INetworkConfigFactory.java:
##########
@@ -0,0 +1,29 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.flink.connector.elasticsearch.sink;
+
+import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
+
+import java.io.Serializable;
+
+public interface INetworkConfigFactory extends Serializable {
Review Comment:
Why do you need `INetworkConfigFactory`? It is not used anywhere (as far as I can tell), the concrete class `NetworkConfigFactory` is used instead
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] mtfelisb commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1316470031
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:
##########
@@ -0,0 +1,70 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperationVariant;
+
+import java.io.*;
+import java.util.Objects;
+
+public class Operation implements Serializable {
+ private final BulkOperationVariant bulkOperationVariant;
+
+ private int retries;
+
+ public Operation(BulkOperationVariant bulkOperation, int retries) {
+ this.bulkOperationVariant = bulkOperation;
+ this.retries = retries;
+ }
+
+ public void retry() {
+ this.retries = this.retries - 1;
+ }
+
+ public boolean isRetriable() {
Review Comment:
Definitely! I was trying an approach for retry unsuccessful operations first, before going deep into what is an retryable operation per se. I totally agree that ErrorCause should be taking into account. 😀
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:
##########
@@ -0,0 +1,70 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperationVariant;
+
+import java.io.*;
+import java.util.Objects;
+
+public class Operation implements Serializable {
+ private final BulkOperationVariant bulkOperationVariant;
+
+ private int retries;
+
+ public Operation(BulkOperationVariant bulkOperation, int retries) {
+ this.bulkOperationVariant = bulkOperation;
+ this.retries = retries;
+ }
+
+ public void retry() {
+ this.retries = this.retries - 1;
+ }
+
+ public boolean isRetriable() {
Review Comment:
Definitely! I was trying an approach for retry unsuccessful operations first, before going deep into what is an retryable operation per se. I totally agree that ErrorCause should be taking into account. 😀
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-2621941619
> However, I can't find the package in Maven, can someone share how shall I add the dependency to my app?
It seems like 4.0 hasn't been released yet, please check https://cwiki.apache.org/confluence/display/FLINK/Creating+a+flink-connector+release for the way to ask for a release, thank you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] mtfelisb commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1225894224
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8Writer.java:
##########
@@ -0,0 +1,134 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
+import co.elastic.clients.elasticsearch._helpers.bulk.BulkIngester;
+import co.elastic.clients.elasticsearch._helpers.bulk.BulkListener;
+import co.elastic.clients.elasticsearch.core.BulkRequest;
+import co.elastic.clients.elasticsearch.core.BulkResponse;
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
+import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.connector.base.sink.writer.AsyncSinkWriter;
+
+import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+import org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration;
+
+import org.apache.http.HttpHost;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.function.Consumer;
+
+public class Elasticsearch8Writer<InputT> extends AsyncSinkWriter<InputT, Operation> {
+ private static final Logger LOG = LoggerFactory.getLogger(Elasticsearch8Writer.class);
+
+ private final ElasticsearchAsyncClient esClient;
+
+ public Elasticsearch8Writer(
+ ElementConverter<InputT, Operation> elementConverter,
+ Sink.InitContext context,
+ int maxBatchSize,
+ int maxInFlightRequests,
+ int maxBufferedRequests,
+ long maxBatchSizeInBytes,
+ long maxTimeInBufferMS,
+ long maxRecordSizeInBytes,
+ String username,
+ String password,
+ HttpHost[] httpHosts,
+ Collection<BufferedRequestState<Operation>> state
+ ) {
+ super(
+ elementConverter,
+ context,
+ AsyncSinkWriterConfiguration.builder()
+ .setMaxBatchSize(maxBatchSize)
+ .setMaxBatchSizeInBytes(maxBatchSizeInBytes)
+ .setMaxInFlightRequests(maxInFlightRequests)
+ .setMaxBufferedRequests(maxBufferedRequests)
+ .setMaxTimeInBufferMS(maxTimeInBufferMS)
+ .setMaxRecordSizeInBytes(maxRecordSizeInBytes)
+ .build(),
+ state
+ );
+
+ this.esClient = new NetworkConfigFactory(httpHosts, username, password).create();
+ }
+
+ @Override
+ protected void submitRequestEntries(List<Operation> requestEntries, Consumer<List<Operation>> requestResult) {
+ LOG.debug("submitRequestEntries with {} items", requestEntries.size());
+
+ BulkListener<Operation> listener = new BulkListener<Operation>() {
+ @Override
+ public void beforeBulk(long executionId, BulkRequest request, List<Operation> contexts) {}
+
+ @Override
+ public void afterBulk(long executionId, BulkRequest request, List<Operation> contexts, BulkResponse response) {
+ LOG.debug("Bulk request " + executionId + " completed");
+ ArrayList<Operation> retriableEntries = new ArrayList<>();
+
+ for (int i = 0; i < contexts.size(); i++) {
+ BulkResponseItem item = response.items().get(i);
+ Operation operation = contexts.get(i);
+
+ if (item.error() != null) {
+ LOG.error("Failed operation " + operation + " - " + item.error().reason());
+ }
+
+ if (item.error() != null && operation.isRetriable()) {
+ retriableEntries.add(operation);
+ operation.retry();
+ }
+ }
+
+ requestResult.accept(retriableEntries);
+ }
+
+ @Override
+ public void afterBulk(long executionId, BulkRequest request, List<Operation> contexts, Throwable failure) {
+ LOG.debug("Bulk request " + executionId + " failed", failure);
+ requestResult.accept(requestEntries);
+ }
+ };
+
+ try (BulkIngester<Operation> ingester = BulkIngester.of(b -> b
Review Comment:
I believe you're right. Thank you for pointing it out! 🤝
I've made a couple of tests, and indeed the `BulkIngester::close` was blocking. I've refactored to use `BulkRequest`, which returns a `CompletableFuture<BulkResponse>` as the `submitRequestEntries` specifies and it worked like a charm.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reta commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1211711218
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:
##########
@@ -0,0 +1,70 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperationVariant;
+
+import java.io.*;
+import java.util.Objects;
+
+public class Operation implements Serializable {
+ private final BulkOperationVariant bulkOperationVariant;
+
+ private int retries;
+
+ public Operation(BulkOperationVariant bulkOperation, int retries) {
+ this.bulkOperationVariant = bulkOperation;
+ this.retries = retries;
+ }
+
+ public void retry() {
+ this.retries = this.retries - 1;
+ }
+
+ public boolean isRetriable() {
Review Comment:
It is likely that decision to retry or not may need to have the failure cause (`ErrorCause`) into account
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] StefanXiepj commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "StefanXiepj (via GitHub)" <gi...@apache.org>.
StefanXiepj commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1712529841
> 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] mtfelisb commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1690822793
> 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1482924435
##########
flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriterITCase.java:
##########
@@ -0,0 +1,276 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import org.apache.flink.connector.base.sink.writer.TestSinkInitContext;
+import org.apache.flink.connector.elasticsearch.sink.Elasticsearch8AsyncSinkITCase.DummyData;
+import org.apache.flink.metrics.Gauge;
+
+import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
+import co.elastic.clients.elasticsearch.core.bulk.UpdateOperation;
+import org.apache.http.HttpHost;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Integration tests for {@link Elasticsearch8AsyncWriter}. */
+@Testcontainers
+public class Elasticsearch8AsyncWriterITCase extends ElasticsearchSinkBaseITCase {
+ private TestSinkInitContext context;
+
+ private final Lock lock = new ReentrantLock();
+
+ private final Condition completed = lock.newCondition();
+
+ @BeforeEach
+ void setUp() {
+ this.context = new TestSinkInitContext();
+ this.client = getRestClient();
+ }
+
+ @AfterEach
+ void shutdown() throws IOException {
+ if (client != null) {
+ client.close();
+ }
+ }
+
+ @Test
+ @Timeout(5)
+ public void testBulkOnFlush() throws IOException, InterruptedException {
+ String index = "test-bulk-on-flush";
+ int maxBatchSize = 2;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+
+ writer.flush(false);
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2"});
+
+ writer.write(new DummyData("3", "test-3"), null);
+
+ writer.flush(true);
+ assertIdsAreWritten(index, new String[] {"test-3"});
+ }
+ }
+
+ @Test
+ @Timeout(5)
+ public void testBulkOnBufferTimeFlush() throws Exception {
+ String index = "test-bulk-on-time-in-buffer";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.flush(true);
+
+ assertIdsAreWritten(index, new String[] {"test-1"});
+
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ assertIdsAreNotWritten(index, new String[] {"test-2", "test-3"});
+ context.getTestProcessingTimeService().advance(6000L);
+
+ await();
+ }
+
+ assertIdsAreWritten(index, new String[] {"test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testBytesSentMetric() throws Exception {
+ String index = "test-bytes-sent-metrics";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ assertThat(context.getNumBytesOutCounter().getCount()).isEqualTo(0);
+
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ await();
+ }
+
+ assertThat(context.getNumBytesOutCounter().getCount()).isGreaterThan(0);
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testRecordsSentMetric() throws Exception {
+ String index = "test-records-sent-metric";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ assertThat(context.getNumRecordsOutCounter().getCount()).isEqualTo(0);
+
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ await();
+ }
+
+ assertThat(context.getNumRecordsOutCounter().getCount()).isEqualTo(3);
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testSendTimeMetric() throws Exception {
+ String index = "test-send-time-metric";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ final Optional<Gauge<Long>> currentSendTime = context.getCurrentSendTimeGauge();
+
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ await();
+
+ assertThat(currentSendTime).isPresent();
+ assertThat(currentSendTime.get().getValue()).isGreaterThan(0L);
+ }
+
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testHandlePartiallyFailedBulk() throws Exception {
+ String index = "test-partially-failed-bulk";
+ int maxBatchSize = 2;
+
+ Elasticsearch8AsyncSinkBuilder.OperationConverter<DummyData> elementConverter =
+ new Elasticsearch8AsyncSinkBuilder.OperationConverter<>(
+ (element, ctx) ->
+ new UpdateOperation.Builder<>()
+ .id(element.getId())
+ .index(index)
+ .action(
+ ac ->
+ ac.doc(element)
+ .docAsUpsert(
+ element.getId()
+ .equals("test-2")))
+ .build());
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize, elementConverter)) {
+ writer.write(new DummyData("test-1", "test-1-updated"), null);
+ writer.write(new DummyData("test-2", "test-2-updated"), null);
+ }
+
+ await();
+
+ assertThat(context.metricGroup().getNumRecordsOutErrorsCounter().getCount()).isEqualTo(1);
+ assertIdsAreWritten(index, new String[] {"test-2"});
+ assertIdsAreNotWritten(index, new String[] {"test-1"});
+ }
+
+ private Elasticsearch8AsyncSinkBuilder.OperationConverter<DummyData>
+ getDefaultTestElementConverter(String index) {
+ return new Elasticsearch8AsyncSinkBuilder.OperationConverter<>(
+ (element, ctx) ->
+ new IndexOperation.Builder<DummyData>()
+ .id(element.getId())
+ .document(element)
+ .index(index)
+ .build());
+ }
+
+ private Elasticsearch8AsyncWriter<DummyData> createWriter(String index, int maxBatchSize) {
+ return createWriter(index, maxBatchSize, getDefaultTestElementConverter(index));
+ }
+
+ private Elasticsearch8AsyncWriter<DummyData> createWriter(
+ String index,
+ int maxBatchSize,
+ Elasticsearch8AsyncSinkBuilder.OperationConverter<DummyData> elementConverter) {
+ return new Elasticsearch8AsyncWriter<DummyData>(
Review Comment:
`Elasticsearch8AsyncSink` has two methods.
```java
createWriter(WriterInitContext);
createWriter(InitContext);
```
Whether `TestinitContext` inherits from `WriterInitContext` or `InitContext`, we can use the overloading method of `Elasticsearch8AsyncSink.createWriter` to maintain compatibility.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1484848274
##########
flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriterITCase.java:
##########
@@ -0,0 +1,276 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import org.apache.flink.connector.base.sink.writer.TestSinkInitContext;
+import org.apache.flink.connector.elasticsearch.sink.Elasticsearch8AsyncSinkITCase.DummyData;
+import org.apache.flink.metrics.Gauge;
+
+import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
+import co.elastic.clients.elasticsearch.core.bulk.UpdateOperation;
+import org.apache.http.HttpHost;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Integration tests for {@link Elasticsearch8AsyncWriter}. */
+@Testcontainers
+public class Elasticsearch8AsyncWriterITCase extends ElasticsearchSinkBaseITCase {
+ private TestSinkInitContext context;
+
+ private final Lock lock = new ReentrantLock();
+
+ private final Condition completed = lock.newCondition();
+
+ @BeforeEach
+ void setUp() {
+ this.context = new TestSinkInitContext();
+ this.client = getRestClient();
+ }
+
+ @AfterEach
+ void shutdown() throws IOException {
+ if (client != null) {
+ client.close();
+ }
+ }
+
+ @Test
+ @Timeout(5)
+ public void testBulkOnFlush() throws IOException, InterruptedException {
+ String index = "test-bulk-on-flush";
+ int maxBatchSize = 2;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+
+ writer.flush(false);
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2"});
+
+ writer.write(new DummyData("3", "test-3"), null);
+
+ writer.flush(true);
+ assertIdsAreWritten(index, new String[] {"test-3"});
+ }
+ }
+
+ @Test
+ @Timeout(5)
+ public void testBulkOnBufferTimeFlush() throws Exception {
+ String index = "test-bulk-on-time-in-buffer";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.flush(true);
+
+ assertIdsAreWritten(index, new String[] {"test-1"});
+
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ assertIdsAreNotWritten(index, new String[] {"test-2", "test-3"});
+ context.getTestProcessingTimeService().advance(6000L);
+
+ await();
+ }
+
+ assertIdsAreWritten(index, new String[] {"test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testBytesSentMetric() throws Exception {
+ String index = "test-bytes-sent-metrics";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ assertThat(context.getNumBytesOutCounter().getCount()).isEqualTo(0);
+
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ await();
+ }
+
+ assertThat(context.getNumBytesOutCounter().getCount()).isGreaterThan(0);
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testRecordsSentMetric() throws Exception {
+ String index = "test-records-sent-metric";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ assertThat(context.getNumRecordsOutCounter().getCount()).isEqualTo(0);
+
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ await();
+ }
+
+ assertThat(context.getNumRecordsOutCounter().getCount()).isEqualTo(3);
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testSendTimeMetric() throws Exception {
+ String index = "test-send-time-metric";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ final Optional<Gauge<Long>> currentSendTime = context.getCurrentSendTimeGauge();
+
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ await();
+
+ assertThat(currentSendTime).isPresent();
+ assertThat(currentSendTime.get().getValue()).isGreaterThan(0L);
+ }
+
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testHandlePartiallyFailedBulk() throws Exception {
+ String index = "test-partially-failed-bulk";
+ int maxBatchSize = 2;
+
+ Elasticsearch8AsyncSinkBuilder.OperationConverter<DummyData> elementConverter =
+ new Elasticsearch8AsyncSinkBuilder.OperationConverter<>(
+ (element, ctx) ->
+ new UpdateOperation.Builder<>()
+ .id(element.getId())
+ .index(index)
+ .action(
+ ac ->
+ ac.doc(element)
+ .docAsUpsert(
+ element.getId()
+ .equals("test-2")))
+ .build());
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize, elementConverter)) {
+ writer.write(new DummyData("test-1", "test-1-updated"), null);
+ writer.write(new DummyData("test-2", "test-2-updated"), null);
+ }
+
+ await();
+
+ assertThat(context.metricGroup().getNumRecordsOutErrorsCounter().getCount()).isEqualTo(1);
+ assertIdsAreWritten(index, new String[] {"test-2"});
+ assertIdsAreNotWritten(index, new String[] {"test-1"});
+ }
+
+ private Elasticsearch8AsyncSinkBuilder.OperationConverter<DummyData>
+ getDefaultTestElementConverter(String index) {
+ return new Elasticsearch8AsyncSinkBuilder.OperationConverter<>(
+ (element, ctx) ->
+ new IndexOperation.Builder<DummyData>()
+ .id(element.getId())
+ .document(element)
+ .index(index)
+ .build());
+ }
+
+ private Elasticsearch8AsyncWriter<DummyData> createWriter(String index, int maxBatchSize) {
+ return createWriter(index, maxBatchSize, getDefaultTestElementConverter(index));
+ }
+
+ private Elasticsearch8AsyncWriter<DummyData> createWriter(
+ String index,
+ int maxBatchSize,
+ Elasticsearch8AsyncSinkBuilder.OperationConverter<DummyData> elementConverter) {
+ return new Elasticsearch8AsyncWriter<DummyData>(
Review Comment:
Hey, @Jiabao-Sun I've requested your review. I had to rebase the main here to get [these changes](https://github.com/apache/flink-connector-elasticsearch/commit/52f2afdda90d4d8881d64e6d13c8c8f8b94eb922). Essentially, I'm creating the writer via sink as you pointed out https://github.com/apache/flink-connector-elasticsearch/pull/53/commits/2869a87a4df3823010b963571a0aa1ea63fe0a0e. Once again, thank you! 👊
@MartijnVisser its indeed working for Flink 1.17/1.18 and the upcoming 1.19 (in a single branch approach) 🙌
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1487031922
##########
flink-connector-elasticsearch8/pom.xml:
##########
@@ -0,0 +1,141 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<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/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-elasticsearch-parent</artifactId>
+ <version>4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>flink-connector-elasticsearch8</artifactId>
+ <name>Flink : Connectors : Elasticsearch 8</name>
+
+ <packaging>jar</packaging>
+
+ <!-- Allow users to pass custom connector versions -->
+ <properties>
+ <elasticsearch.version>8.7.0</elasticsearch.version>
Review Comment:
Probably we could use latest one (as of today):
```suggestion
<elasticsearch.version>8.12.1</elasticsearch.version>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1923472262
@reta @snuyanzin Do you think either of you could help with the review of this? It seems that https://github.com/apache/flink-connector-elasticsearch/pull/74 is stale, and it would be great if we could move forward with adding ES 8
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1928954595
good to know, thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "rinkako (via GitHub)" <gi...@apache.org>.
rinkako commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1862265394
> > @mtfelisb Are you still active on this PR?
>
> Hi, @MartijnVisser. Yes, I'm waiting for review. Also, I'll not implement the `DynamicTableSink` because @rinkako started [here](https://github.com/rinkako/flink-connector-elasticsearch_es8/tree/FLINK-26088/flink-connector-elasticsearch8/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table) based on my pull request. Thanks by the way @rinkako! ☺️
thanks @mtfelisb , I'm willing to PR my code. 😊
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1479350601
##########
flink-connector-elasticsearch8/pom.xml:
##########
@@ -0,0 +1,139 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<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/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-elasticsearch-parent</artifactId>
+ <version>4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>flink-connector-elasticsearch8</artifactId>
+ <name>Flink : Connectors : Elasticsearch 8</name>
+
+ <packaging>jar</packaging>
+
+ <!-- Allow users to pass custom connector versions -->
+ <properties>
+ <elasticsearch.version>8.7.0</elasticsearch.version>
+ </properties>
+
+ <dependencies>
+
+ <!-- Core -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Table ecosystem -->
+
+ <!-- Projects depending on this project won't depend on flink-table-*. -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java-bridge</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.9.0</version>
Review Comment:
why are we sticked to so old jackson which even has several CVE?
I guess we could use 2.15.x or 2.14.x
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1483760795
##########
flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriterITCase.java:
##########
@@ -0,0 +1,276 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import org.apache.flink.connector.base.sink.writer.TestSinkInitContext;
+import org.apache.flink.connector.elasticsearch.sink.Elasticsearch8AsyncSinkITCase.DummyData;
+import org.apache.flink.metrics.Gauge;
+
+import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
+import co.elastic.clients.elasticsearch.core.bulk.UpdateOperation;
+import org.apache.http.HttpHost;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Integration tests for {@link Elasticsearch8AsyncWriter}. */
+@Testcontainers
+public class Elasticsearch8AsyncWriterITCase extends ElasticsearchSinkBaseITCase {
+ private TestSinkInitContext context;
+
+ private final Lock lock = new ReentrantLock();
+
+ private final Condition completed = lock.newCondition();
+
+ @BeforeEach
+ void setUp() {
+ this.context = new TestSinkInitContext();
+ this.client = getRestClient();
+ }
+
+ @AfterEach
+ void shutdown() throws IOException {
+ if (client != null) {
+ client.close();
+ }
+ }
+
+ @Test
+ @Timeout(5)
+ public void testBulkOnFlush() throws IOException, InterruptedException {
+ String index = "test-bulk-on-flush";
+ int maxBatchSize = 2;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+
+ writer.flush(false);
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2"});
+
+ writer.write(new DummyData("3", "test-3"), null);
+
+ writer.flush(true);
+ assertIdsAreWritten(index, new String[] {"test-3"});
+ }
+ }
+
+ @Test
+ @Timeout(5)
+ public void testBulkOnBufferTimeFlush() throws Exception {
+ String index = "test-bulk-on-time-in-buffer";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.flush(true);
+
+ assertIdsAreWritten(index, new String[] {"test-1"});
+
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ assertIdsAreNotWritten(index, new String[] {"test-2", "test-3"});
+ context.getTestProcessingTimeService().advance(6000L);
+
+ await();
+ }
+
+ assertIdsAreWritten(index, new String[] {"test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testBytesSentMetric() throws Exception {
+ String index = "test-bytes-sent-metrics";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ assertThat(context.getNumBytesOutCounter().getCount()).isEqualTo(0);
+
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ await();
+ }
+
+ assertThat(context.getNumBytesOutCounter().getCount()).isGreaterThan(0);
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testRecordsSentMetric() throws Exception {
+ String index = "test-records-sent-metric";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ assertThat(context.getNumRecordsOutCounter().getCount()).isEqualTo(0);
+
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ await();
+ }
+
+ assertThat(context.getNumRecordsOutCounter().getCount()).isEqualTo(3);
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testSendTimeMetric() throws Exception {
+ String index = "test-send-time-metric";
+ int maxBatchSize = 3;
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize)) {
+ final Optional<Gauge<Long>> currentSendTime = context.getCurrentSendTimeGauge();
+
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "test-3"), null);
+
+ await();
+
+ assertThat(currentSendTime).isPresent();
+ assertThat(currentSendTime.get().getValue()).isGreaterThan(0L);
+ }
+
+ assertIdsAreWritten(index, new String[] {"test-1", "test-2", "test-3"});
+ }
+
+ @Test
+ @Timeout(5)
+ public void testHandlePartiallyFailedBulk() throws Exception {
+ String index = "test-partially-failed-bulk";
+ int maxBatchSize = 2;
+
+ Elasticsearch8AsyncSinkBuilder.OperationConverter<DummyData> elementConverter =
+ new Elasticsearch8AsyncSinkBuilder.OperationConverter<>(
+ (element, ctx) ->
+ new UpdateOperation.Builder<>()
+ .id(element.getId())
+ .index(index)
+ .action(
+ ac ->
+ ac.doc(element)
+ .docAsUpsert(
+ element.getId()
+ .equals("test-2")))
+ .build());
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(index, maxBatchSize, elementConverter)) {
+ writer.write(new DummyData("test-1", "test-1-updated"), null);
+ writer.write(new DummyData("test-2", "test-2-updated"), null);
+ }
+
+ await();
+
+ assertThat(context.metricGroup().getNumRecordsOutErrorsCounter().getCount()).isEqualTo(1);
+ assertIdsAreWritten(index, new String[] {"test-2"});
+ assertIdsAreNotWritten(index, new String[] {"test-1"});
+ }
+
+ private Elasticsearch8AsyncSinkBuilder.OperationConverter<DummyData>
+ getDefaultTestElementConverter(String index) {
+ return new Elasticsearch8AsyncSinkBuilder.OperationConverter<>(
+ (element, ctx) ->
+ new IndexOperation.Builder<DummyData>()
+ .id(element.getId())
+ .document(element)
+ .index(index)
+ .build());
+ }
+
+ private Elasticsearch8AsyncWriter<DummyData> createWriter(String index, int maxBatchSize) {
+ return createWriter(index, maxBatchSize, getDefaultTestElementConverter(index));
+ }
+
+ private Elasticsearch8AsyncWriter<DummyData> createWriter(
+ String index,
+ int maxBatchSize,
+ Elasticsearch8AsyncSinkBuilder.OperationConverter<DummyData> elementConverter) {
+ return new Elasticsearch8AsyncWriter<DummyData>(
Review Comment:
Hey, @Jiabao-Sun thank you for sharing. You are right, I must use the `Elasticsearch8AsyncSink` to work. I have tested it locally and it's working smoothly. I look forward to your review after I commit the changes. 🙂
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reta commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1211695458
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Operation.java:
##########
@@ -0,0 +1,70 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperationVariant;
+
+import java.io.*;
+import java.util.Objects;
+
+public class Operation implements Serializable {
+ private final BulkOperationVariant bulkOperationVariant;
Review Comment:
~This is not going to work, the `BulkOperationVariant` is not serializable, I believe you need to have custom serialization here. Just to give you an idea from OpenSearch [1] implementation (not the same but very close): https://github.com/apache/flink-connector-opensearch/pull/5/files#diff-1c079dfa499f278474b93e29b5dd776722feaae97461612fec46a8d9d24b5585R43~
UPD: correcting my self, since you are using Kryo, it has no requirements on `Serializable`, should be fine
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] mtfelisb commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1579604638
> 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] mtfelisb commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1579602598
> > 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] mtfelisb commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1322154473
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/INetworkConfigFactory.java:
##########
@@ -0,0 +1,29 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.flink.connector.elasticsearch.sink;
+
+import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
+
+import java.io.Serializable;
+
+public interface INetworkConfigFactory extends Serializable {
Review Comment:
You're right. Just removed it! Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] mtfelisb commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1722311553
Guys, I wrote a bunch of tests heavily inspired by https://github.com/apache/flink-connector-opensearch/pull/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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1933166244
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](https://issues.apache.org/jira/browse/FLINK-34113?page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel&focusedCommentId=17811151#comment-17811151), 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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1934031525
> 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](https://issues.apache.org/jira/browse/FLINK-34113?page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel&focusedCommentId=17811151#comment-17811151), 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "drorventura (via GitHub)" <gi...@apache.org>.
drorventura commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-2006807958
Thank you @mtfelisb
@reswqa could you please approve? 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-2011649582
> 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "ebdxflr (via GitHub)" <gi...@apache.org>.
ebdxflr commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-2017413587
> > 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1774999640
There's now this PR by @mtfelisb and PR https://github.com/apache/flink-connector-elasticsearch/pull/74 by @StefanXiepj - How can we consolidate into one PR? @reta could you help with reviews if necessary?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1837517174
> @mtfelisb Are you still active on this PR?
Hi, @MartijnVisser. Yes, I'm waiting for review.
Also, I'll not implement the `DynamicTableSink` because @rinkako started [here](https://github.com/rinkako/flink-connector-elasticsearch_es8/tree/FLINK-26088/flink-connector-elasticsearch8/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table) based on my pull request. Thanks by the way @rinkako! ☺️
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reta commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1570176136
The top level `pom.xml` misses the new module:
```
<module>flink-connector-elasticsearch8</module>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] mtfelisb commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1569396046
@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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1935170882
> > 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](https://github.com/apache/flink-connector-kafka/pull/84)
I understand now, it makes total sense. I appreciate the detailed information you provided, @MartijnVisser 😀
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1479594864
##########
flink-connector-elasticsearch8/pom.xml:
##########
@@ -0,0 +1,139 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<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/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-elasticsearch-parent</artifactId>
+ <version>4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>flink-connector-elasticsearch8</artifactId>
+ <name>Flink : Connectors : Elasticsearch 8</name>
+
+ <packaging>jar</packaging>
+
+ <!-- Allow users to pass custom connector versions -->
+ <properties>
+ <elasticsearch.version>8.7.0</elasticsearch.version>
+ </properties>
+
+ <dependencies>
+
+ <!-- Core -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Table ecosystem -->
+
+ <!-- Projects depending on this project won't depend on flink-table-*. -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java-bridge</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.9.0</version>
Review Comment:
Good call, @snuyanzin thanks. I'm going to update it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "drorventura (via GitHub)" <gi...@apache.org>.
drorventura commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-2006100743
Hi
Is this going to be merged soon?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1529874870
##########
flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBaseITCase.java:
##########
@@ -0,0 +1,107 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import org.apache.http.HttpHost;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.IOException;
+import java.time.Duration;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.junit.Assert.assertEquals;
Review Comment:
We should avoid mixing the two assertions(junit & assertJ). AssertJ should be the only appropriate assertion.
##########
flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/ElasticsearchSinkBaseITCase.java:
##########
@@ -0,0 +1,107 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import org.apache.http.HttpHost;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.IOException;
+import java.time.Duration;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
Review Comment:
`org.assertj.core.api.Assertions.assertThat`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "drorventura (via GitHub)" <gi...@apache.org>.
drorventura commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-2011373385
when is the next release planned?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reswqa commented on pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1687418291
What's the progress of this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] reta commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1211758453
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/OperationSerializer.java:
##########
@@ -0,0 +1,60 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.flink.connector.elasticsearch.sink;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.objenesis.strategy.StdInstantiatorStrategy;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+
+public class OperationSerializer {
+ private final Kryo kryo = new Kryo();
+
+ public OperationSerializer() {
+ kryo.setRegistrationRequired(false);
+ kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+ }
+
+ public void serialize(Operation request, DataOutputStream out) {
+ Output output = new Output(out);
Review Comment:
```suggestion
try (Output output = new Output(out)) {
kryo.writeObject(output, request);
output.flush();
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1753824138
> 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 😃
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support [flink-connector-elasticsearch]
Posted by "reta (via GitHub)" <gi...@apache.org>.
reta commented on PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#issuecomment-1775197561
> @reta could you help with reviews if necessary?
@MartijnVisser absolutely, how would you like to proceed? I believe @mtfelisb did a great job addressing all comments and concerns, so did @StefanXiepj, which one we should consolidate into?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-elasticsearch] mtfelisb commented on a diff in pull request #53: [FLINK-26088][Connectors/ElasticSearch] Add Elasticsearch 8.0 support
Posted by "mtfelisb (via GitHub)" <gi...@apache.org>.
mtfelisb commented on code in PR #53:
URL: https://github.com/apache/flink-connector-elasticsearch/pull/53#discussion_r1298739730
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8Sink.java:
##########
@@ -0,0 +1,119 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.flink.connector.elasticsearch.sink;
+
+import org.apache.flink.connector.base.sink.AsyncSinkBase;
+import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.http.HttpHost;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.Collections;
+
+public class Elasticsearch8Sink<InputT> extends AsyncSinkBase<InputT, Operation> {
+ private static final Logger LOG = LoggerFactory.getLogger(Elasticsearch8Sink.class);
+
+ private final String username;
+
+ private final String password;
+
+ private final HttpHost[] httpHosts;
+
+ protected Elasticsearch8Sink(
+ ElementConverter<InputT, Operation> converter,
+ int maxBatchSize,
+ int maxInFlightRequests,
+ int maxBufferedRequests,
+ long maxBatchSizeInBytes,
+ long maxTimeInBufferMS,
+ long maxRecordSizeInByte,
+ String username,
+ String password,
+ HttpHost[] httpHosts
+ ) {
+ super(
+ converter,
+ maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBatchSizeInBytes,
+ maxTimeInBufferMS,
+ maxRecordSizeInByte
+ );
+
+ this.username = username;
+ this.password = password;
+ this.httpHosts = httpHosts;
+ }
+
+ @Override
+ public StatefulSinkWriter<InputT, BufferedRequestState<Operation>> createWriter(
+ InitContext context
+ ) {
+ LOG.debug("creating writer");
Review Comment:
I agree. Just removed it :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org