[jira] [Commented] (FLINK-3857) Add reconnect attempt to Elasticsearch host

["ASF GitHub Bot (JIRA)" <ji...@apache.org>]

    [ 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)