[
https://issues.apache.org/jira/browse/FLINK-3857?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16056969#comment-16056969
]
ASF GitHub Bot commented on FLINK-3857:
---------------------------------------
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/1962#discussion_r123149965
--- Diff:
flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java
---
@@ -234,4 +255,36 @@ private void checkErrorAndRethrow() {
throw new RuntimeException("An error occured in
ElasticsearchSink.", cause);
}
}
+
+ private void retry(T value) throws Exception {
+ int retryCounter = 1;
+
+ while (retryCounter <= connectionRetries) {
+ if (bulkProcessor != null) {
+ bulkProcessor.close();
+ bulkProcessor = null;
+ }
+
+ if (client != null) {
+ client.close();
+ }
+
+ try {
+ open(null);
+ elasticsearchSinkFunction.process(value,
getRuntimeContext(), requestIndexer);
+ } catch (Exception ex) {
+ if (client instanceof TransportClient &&
!callBridge.isConnected(((TransportClient) client))) {
+ TimeUnit.SECONDS.sleep(3);
--- End diff --
Should this be configurable?
Also, could you explain a bit on why you've chosen 3 seconds?
> Add reconnect attempt to Elasticsearch host
> -------------------------------------------
>
> Key: FLINK-3857
> URL: https://issues.apache.org/jira/browse/FLINK-3857
> Project: Flink
> Issue Type: Improvement
> Components: Streaming Connectors
> Affects Versions: 1.1.0, 1.0.2
> Reporter: Fabian Hueske
> Assignee: Subhobrata Dey
>
> Currently, the connection to the Elasticsearch host is opened in
> {{ElasticsearchSink.open()}}. In case the connection is lost (maybe due to a
> changed DNS entry), the sink fails.
> I propose to catch the Exception for lost connections in the {{invoke()}}
> method and try to re-open the connection for a configurable number of times
> with a certain delay.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
