[FLINK-3857][Streaming Connectors]Add reconnect attempt to Elasticsearch host #1962

sbcd90 · 2016-05-04T02:27:29Z

General
- The pull request references the related JIRA issue ("[FLINK-3857] Add reconnect attempt to Elasticsearch host")
Documentation
- Documentation added based on the changes made.
Tests & Build
- Functionality added by the pull request is covered by tests
- mvn clean verify has been executed successfully locally or a Travis build has passed

sbcd90 · 2016-05-04T14:03:16Z

Hello @fhueske , The tests do not fail because of the changes made in the PR. I tested the Junits for elasticsearch connector & all of them runs fine.
Can you kindly have a look?

rmetzger · 2016-05-05T10:50:38Z

...h2/src/main/java/org/apache/flink/streaming/connectors/elasticsearch2/ElasticsearchSink.java

@@ -86,6 +88,7 @@
 	public static final String CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS = "bulk.flush.max.actions";
 	public static final String CONFIG_KEY_BULK_FLUSH_MAX_SIZE_MB = "bulk.flush.max.size.mb";
 	public static final String CONFIG_KEY_BULK_FLUSH_INTERVAL_MS = "bulk.flush.interval.ms";
+	public static final String CONFIG_NO_OF_CONN_RETRIES = "conn.retries";


We usually don't abbreviate configuration keys. Can you rename it to "CONFIG_KEY_CONNECTION_RETRIES" ?

rmetzger · 2016-05-05T11:05:03Z

Thank you for opening the pull request.

I made some inline comments.
I don't think the proposed changes fix the issue described in the JIRA.
I would check on each invoke() if hasFailure is set. If that's the case, you can reconnect to EL.

rmetzger · 2016-05-05T11:06:28Z

The change is missing documentation updates & test cases.

sbcd90 · 2016-05-05T22:28:01Z

Hello @rmetzger ,

Thanks a lot for reviewing the PR.
I have made all the changes mentioned by you as inline comments as well as added some documentation.
Kindly have a look now.

sbcd90 · 2016-05-06T03:59:12Z

Hello @rmetzger ,

Looking at the test case ElasticsearchSinkItCase.testTransportClient, I think to test the re-connect scenario the hasFailure may need to be made public so that the test-method can set it.
Can you kindly provide some suggestions?

rmetzger · 2016-05-06T07:53:01Z

How did you test/try out the code you've implemented in this pull request?

sbcd90 · 2016-05-06T19:43:28Z

Hello @rmetzger ,

I added a testcase now to the ElasticsearchSinkITCase.java list of tests. Can you kindly have a look once?

StephanEwen · 2016-06-07T17:49:09Z

I took a quick look at this. I am wondering if this actually needs an extra timer service for retries.

Can this be solved without a timer? The failures could be detected in the invoke(...) method, and the retry done directly there (with some minimal backoff or so).

Triggering asynchronous timers is very complex and easily creates leaks, races, or leftover work / tasks at shutdown.

sbcd90 · 2016-06-09T04:14:51Z

Hello @StephanEwen ,

I have removed the asynchronous timer & doing the retry logic directly now. The backoff is 3s. Please have a look.

tzulitai · 2016-06-09T05:45:18Z

...h2/src/main/java/org/apache/flink/streaming/connectors/elasticsearch2/ElasticsearchSink.java

+
+		if (params.has(CONFIG_KEY_CONNECTION_RETRIES)) {
+			this.connectionRetries = params.getInt(CONFIG_KEY_CONNECTION_RETRIES);
+		}


Need to have a default value set if not specified by user? Otherwise null exception in invoke().

tzulitai · 2016-06-09T06:33:21Z

Hi @sbcd90,

Gave the changes a quick review and commented. please let me know your opinion on them. Hope they'll be helpful to get you going.

sbcd90 · 2016-06-10T02:22:29Z

Hello @tzulitai ,

I think default value for int in Java is 0.
The check if connection is lost or not & then retry for connection is a good suggestion. Made the change.
separated the methods for connection creation & connection status check.

tzulitai · 2016-06-14T04:45:39Z

Hi @sbcd90,
Sorry for the late reply, as I'm currently busy some other things. I'll be happy to help review again within the next 2~3 days.

tzulitai · 2016-06-20T04:53:05Z

...h2/src/main/java/org/apache/flink/streaming/connectors/elasticsearch2/ElasticsearchSink.java

-			LOG.info("Created Elasticsearch TransportClient {}", client);
+			if (LOG.isInfoEnabled()) {
+				LOG.info("Created Elasticsearch TransportClient {}", client);
+			}


Log message here doesn't match the purpose of this method

tzulitai · 2016-06-20T05:24:11Z

Hi @sbcd90,
I think to address "add reconnect attempt" alone, checking whether or not the transport client is connected to nodes and retry connect if lost connection in invoke() before processing the element should be fine.

On the other hand, another problem that raises if we are to add reconnect attempt for the ES sink is that failing records due to connection errors also need to be caught in the BulkProcessor afterBulk() callback and re-processed. I wonder if we should be solving this together to resolve this issue. @rmetzger, what's your opinion?

HungUnicorn · 2016-08-25T14:13:18Z

Using the sniffing feature of transport client can achieve this with the current ES2 connector in master branch.
The client will connect to all existing nodes and the connected list is updated every 5 seconds. It can fit our case because we will only have to specify one ip, and we will obtain a list of ips which updated periodically. It's done by
Settings settings = Settings.settingsBuilder().put(userConfig).put("client.transport.sniff", true).build();

Explanation:

The Transport client comes with a cluster sniffing feature which allows it to dynamically add new hosts and remove old ones. When sniffing is enabled the the transport client will connect to the nodes in its internal node list, which is built via calls to addTransportAddress. After this, the client will call the internal cluster state API on those nodes to discover available data nodes. The internal node list of the client will be replaced with those data nodes only. This list is refreshed every five seconds by default.

Source:
https://www.elastic.co/guide/en/elasticsearch/client/java-api/2.3//transport-client.html

tzulitai · 2016-09-15T06:46:36Z

Thanks @HungUnicorn, thats useful info. I wonder though if this config should be set by the user, instead of letting the connector internally set this.

tzulitai · 2017-01-23T08:17:36Z

Hi @sbcd90, will you like to continue working on this PR?

There's going to be a restructuring of the ES connectors (#3112) perhaps soon after the 1.2 release, and this PR will very likely need a rebase. I'd like to include this fix after the restructuring, so please let me know on how you'd like to proceed with this contribution :-) Thank you!

sbcd90 · 2017-01-26T03:23:15Z

Hi @tzulitai , thanks for the updates. I'll refactor the code & will rebase the PR.

tzulitai · 2017-01-26T06:23:23Z

Thank you for picking this up again @sbcd90. I would wait until #3112 is merged before rebasing.

…rch host

sbcd90 · 2017-02-17T08:37:50Z

Hello @tzulitai ,

I have rebased the changes. Can you please review?

tzulitai

Thanks a lot for the rebase, and really sorry for the long delay on the review here @sbcd90.
I finally have some time to revisit the Elasticsearch connectors :) I have some comments, could you please let me know what you think?

tzulitai · 2017-06-21T04:06:20Z

...src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java

+				elasticsearchSinkFunction.process(value, getRuntimeContext(), requestIndexer);
+			} catch (Exception ex) {
+				if (client instanceof TransportClient && !callBridge.isConnected(((TransportClient) client))) {
+					TimeUnit.SECONDS.sleep(3);


Should this be configurable?
Also, could you explain a bit on why you've chosen 3 seconds?

tzulitai · 2017-06-21T04:06:52Z

...src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java

+
+		// if there is a connectivity failure, then retry
+		if (failureThrowable.get() != null &&
+			client instanceof TransportClient &&


Why exactly does the client need to be a TransportClient?

tzulitai · 2017-06-21T04:07:24Z

...src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java

@@ -208,6 +222,13 @@ public void invoke(T value) throws Exception {
 		checkErrorAndRethrow();

 		elasticsearchSinkFunction.process(value, getRuntimeContext(), requestIndexer);
+
+		// if there is a connectivity failure, then retry
+		if (failureThrowable.get() != null &&


I'm wondering whether or not checking the exception type would be enough for verifying the connectivity failure.

tzulitai · 2017-06-21T04:08:21Z

...src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java

+			}
+
+			try {
+				open(null);


This open() call seems a bit odd to me. I don't think its a good practice to call that here, since essentially its a life cycle method used by the system.

aljoscha · 2019-10-14T11:24:11Z

I'm closing this as "Abandoned", since there is no more activity and the code base has moved on quite a bit. Please re-open this if you feel otherwise and work should continue.

rmetzger reviewed May 5, 2016
View reviewed changes

sbcd90 force-pushed the elasticSearchRetryIssue branch 2 times, most recently from d24aa6e to 9f5df39 Compare May 5, 2016 22:24

sbcd90 force-pushed the elasticSearchRetryIssue branch from 9f5df39 to 93aba33 Compare May 6, 2016 19:41

sbcd90 force-pushed the elasticSearchRetryIssue branch from 93aba33 to ee23516 Compare June 9, 2016 04:13

tzulitai reviewed Jun 9, 2016
View reviewed changes

sbcd90 force-pushed the elasticSearchRetryIssue branch from ee23516 to b835402 Compare June 10, 2016 02:18

tzulitai reviewed Jun 20, 2016
View reviewed changes

sbcd90 force-pushed the elasticSearchRetryIssue branch from b835402 to 85c45ce Compare February 17, 2017 08:31

[FLINK-3857][Streaming Connectors]Add reconnect attempt to Elasticsea…

0c8bbc7

…rch host

sbcd90 force-pushed the elasticSearchRetryIssue branch from 85c45ce to 0c8bbc7 Compare February 17, 2017 08:36

tzulitai reviewed Jun 21, 2017

View reviewed changes

rmetzger added the component=Connectors/Common label Mar 14, 2019

aljoscha closed this Oct 14, 2019

aljoscha self-assigned this Oct 20, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-3857][Streaming Connectors]Add reconnect attempt to Elasticsearch host #1962

[FLINK-3857][Streaming Connectors]Add reconnect attempt to Elasticsearch host #1962

sbcd90 commented May 4, 2016 •

edited

sbcd90 commented May 4, 2016

rmetzger May 5, 2016

rmetzger commented May 5, 2016

rmetzger commented May 5, 2016

sbcd90 commented May 5, 2016

sbcd90 commented May 6, 2016

rmetzger commented May 6, 2016 •

edited

sbcd90 commented May 6, 2016

StephanEwen commented Jun 7, 2016

sbcd90 commented Jun 9, 2016 •

edited

tzulitai Jun 9, 2016

tzulitai commented Jun 9, 2016

sbcd90 commented Jun 10, 2016

tzulitai commented Jun 14, 2016

tzulitai Jun 20, 2016

tzulitai commented Jun 20, 2016

HungUnicorn commented Aug 25, 2016 •

edited

tzulitai commented Sep 15, 2016

tzulitai commented Jan 23, 2017 •

edited

sbcd90 commented Jan 26, 2017

tzulitai commented Jan 26, 2017

sbcd90 commented Feb 17, 2017

tzulitai left a comment

tzulitai Jun 21, 2017

tzulitai Jun 21, 2017

tzulitai Jun 21, 2017

tzulitai Jun 21, 2017

aljoscha commented Oct 14, 2019

[FLINK-3857][Streaming Connectors]Add reconnect attempt to Elasticsearch host #1962

[FLINK-3857][Streaming Connectors]Add reconnect attempt to Elasticsearch host #1962

Conversation

sbcd90 commented May 4, 2016 • edited

sbcd90 commented May 4, 2016

rmetzger May 5, 2016

Choose a reason for hiding this comment

rmetzger commented May 5, 2016

rmetzger commented May 5, 2016

sbcd90 commented May 5, 2016

sbcd90 commented May 6, 2016

rmetzger commented May 6, 2016 • edited

sbcd90 commented May 6, 2016

StephanEwen commented Jun 7, 2016

sbcd90 commented Jun 9, 2016 • edited

tzulitai Jun 9, 2016

Choose a reason for hiding this comment

tzulitai commented Jun 9, 2016

sbcd90 commented Jun 10, 2016

tzulitai commented Jun 14, 2016

tzulitai Jun 20, 2016

Choose a reason for hiding this comment

tzulitai commented Jun 20, 2016

HungUnicorn commented Aug 25, 2016 • edited

tzulitai commented Sep 15, 2016

tzulitai commented Jan 23, 2017 • edited

sbcd90 commented Jan 26, 2017

tzulitai commented Jan 26, 2017

sbcd90 commented Feb 17, 2017

tzulitai left a comment

Choose a reason for hiding this comment

tzulitai Jun 21, 2017

Choose a reason for hiding this comment

tzulitai Jun 21, 2017

Choose a reason for hiding this comment

tzulitai Jun 21, 2017

Choose a reason for hiding this comment

tzulitai Jun 21, 2017

Choose a reason for hiding this comment

aljoscha commented Oct 14, 2019

sbcd90 commented May 4, 2016 •

edited

rmetzger commented May 6, 2016 •

edited

sbcd90 commented Jun 9, 2016 •

edited

HungUnicorn commented Aug 25, 2016 •

edited

tzulitai commented Jan 23, 2017 •

edited