You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2022/04/25 11:52:38 UTC
[flink-connector-elasticsearch] 15/21: [FLINK-26281][connectors/elasticsearch] setting default delivery guarantee to AT_LEAST_ONCE
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-elasticsearch.git
commit 07e4d6644026e0928ade6404746fb4f5259bff28
Author: Alexander Preuß <11...@users.noreply.github.com>
AuthorDate: Thu Mar 10 09:48:58 2022 +0100
[FLINK-26281][connectors/elasticsearch] setting default delivery guarantee to AT_LEAST_ONCE
---
.../docs/connectors/datastream/elasticsearch.md | 52 ++++------------------
.../content/docs/connectors/table/elasticsearch.md | 2 +-
2 files changed, 9 insertions(+), 45 deletions(-)
diff --git a/docs/content/docs/connectors/datastream/elasticsearch.md b/docs/content/docs/connectors/datastream/elasticsearch.md
index c8342ae..eb49392 100644
--- a/docs/content/docs/connectors/datastream/elasticsearch.md
+++ b/docs/content/docs/connectors/datastream/elasticsearch.md
@@ -214,10 +214,7 @@ flushes of the buffered actions in progress.
### Elasticsearch Sinks and Fault Tolerance
-By default, the Flink Elasticsearch Sink will not provide any strong delivery guarantees.
-Users have the option to enable at-least-once semantics for the Elasticsearch sink.
-
-With Flink’s checkpointing enabled, the Flink Elasticsearch Sink can guarantee
+With Flink’s checkpointing enabled, the Flink Elasticsearch Sink guarantees
at-least-once delivery of action requests to Elasticsearch clusters. It does
so by waiting for all pending action requests in the `BulkProcessor` at the
time of checkpoints. This effectively assures that all requests before the
@@ -226,34 +223,13 @@ proceeding to process more records sent to the sink.
More details on checkpoints and fault tolerance are in the [fault tolerance docs]({{< ref "docs/learn-flink/fault_tolerance" >}}).
-To use fault tolerant Elasticsearch Sinks, at-least-once delivery has to be configured and checkpointing of the topology needs to be enabled at the execution environment:
+To use fault tolerant Elasticsearch Sinks, checkpointing of the topology needs to be enabled at the execution environment:
{{< tabs "d00d1e93-4844-40d7-b0ec-9ec37e73145e" >}}
{{< tab "Java" >}}
-Elasticsearch 6:
```java
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(5000); // checkpoint every 5000 msecs
-
-Elasticsearch6SinkBuilder sinkBuilder = new Elasticsearch6SinkBuilder<String>()
- .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
- .setHosts(new HttpHost("127.0.0.1", 9200, "http"))
- .setEmitter(
- (element, context, indexer) ->
- indexer.add(createIndexRequest(element)));
-```
-
-Elasticsearch 7:
-```java
-final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-env.enableCheckpointing(5000); // checkpoint every 5000 msecs
-
-Elasticsearch7SinkBuilder sinkBuilder = new Elasticsearch7SinkBuilder<String>()
- .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
- .setHosts(new HttpHost("127.0.0.1", 9200, "http"))
- .setEmitter(
- (element, context, indexer) ->
- indexer.add(createIndexRequest(element)));
```
{{< /tab >}}
{{< tab "Scala" >}}
@@ -261,28 +237,16 @@ Elasticsearch 6:
```scala
val env = StreamExecutionEnvironment.getExecutionEnvironment()
env.enableCheckpointing(5000) // checkpoint every 5000 msecs
-
-val sinkBuilder = new Elasticsearch6SinkBuilder[String]
- .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
- .setHosts(new HttpHost("127.0.0.1", 9200, "http"))
- .setEmitter((element: String, context: SinkWriter.Context, indexer: RequestIndexer) =>
- indexer.add(createIndexRequest(element)))
-```
-
-Elasticsearch 7:
-```scala
-val env = StreamExecutionEnvironment.getExecutionEnvironment()
-env.enableCheckpointing(5000) // checkpoint every 5000 msecs
-
-val sinkBuilder = new Elasticsearch7SinkBuilder[String]
- .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
- .setHosts(new HttpHost("127.0.0.1", 9200, "http"))
- .setEmitter((element: String, context: SinkWriter.Context, indexer: RequestIndexer) =>
- indexer.add(createIndexRequest(element)))
```
{{< /tab >}}
{{< /tabs >}}
+<p style="border-radius: 5px; padding: 5px" class="bg-info">
+<b>IMPORTANT</b>: Checkpointing is not enabled by default but the default delivery guarantee is AT_LEAST_ONCE.
+This causes the sink to buffer requests until it either finishes or the BulkProcessor flushes automatically.
+By default, the BulkProcessor will flush after 1000 added Actions. To configure the processor to flush more frequently, please refer to the <a href="#configuring-the-internal-bulk-processor">BulkProcessor configuration section</a>.
+</p>
+
### Handling Failing Elasticsearch Requests
Elasticsearch action requests may fail due to a variety of reasons, including
diff --git a/docs/content/docs/connectors/table/elasticsearch.md b/docs/content/docs/connectors/table/elasticsearch.md
index 78d0ce9..4d92903 100644
--- a/docs/content/docs/connectors/table/elasticsearch.md
+++ b/docs/content/docs/connectors/table/elasticsearch.md
@@ -143,7 +143,7 @@ Connector Options
<td><h5>sink.delivery-guarantee</h5></td>
<td>optional</td>
<td>no</td>
- <td style="word-wrap: break-word;">NONE</td>
+ <td style="word-wrap: break-word;">AT_LEAST_ONCE</td>
<td>String</td>
<td>Optional delivery guarantee when committing. Valid values are <code>NONE</code> or <code>AT_LEAST_ONCE</code>.</td>
</tr>