[GitHub] flink issue #1962: [FLINK-3857][Streaming Connectors]Add reconnect attempt t...
Github user sbcd90 commented on the issue: https://github.com/apache/flink/pull/1962 Hello @tzulitai , I have rebased the changes. Can you please review? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #1962: [FLINK-3857][Streaming Connectors]Add reconnect attempt t...
Github user tzulitai commented on the issue: https://github.com/apache/flink/pull/1962 Thank you for picking this up again @sbcd90. I would wait until #3112 is merged before rebasing. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #1962: [FLINK-3857][Streaming Connectors]Add reconnect attempt t...
Github user sbcd90 commented on the issue: https://github.com/apache/flink/pull/1962 Hi @tzulitai , thanks for the updates. I'll refactor the code & will rebase the PR. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #1962: [FLINK-3857][Streaming Connectors]Add reconnect attempt t...
Github user tzulitai commented on the issue: https://github.com/apache/flink/pull/1962 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 :) --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #1962: [FLINK-3857][Streaming Connectors]Add reconnect attempt t...
Github user tzulitai commented on the issue: https://github.com/apache/flink/pull/1962 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. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #1962: [FLINK-3857][Streaming Connectors]Add reconnect attempt t...
Github user HungUnicorn commented on the issue:
https://github.com/apache/flink/pull/1962
Using the sniffing feature of transport client can achieve this.
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
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #1962: [FLINK-3857][Streaming Connectors]Add reconnect attempt t...
Github user tzulitai commented on the issue: https://github.com/apache/flink/pull/1962 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? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #1962: [FLINK-3857][Streaming Connectors]Add reconnect attempt t...
Github user tzulitai commented on the issue: https://github.com/apache/flink/pull/1962 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. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #1962: [FLINK-3857][Streaming Connectors]Add reconnect attempt t...
Github user sbcd90 commented on the issue: https://github.com/apache/flink/pull/1962 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. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #1962: [FLINK-3857][Streaming Connectors]Add reconnect attempt t...
Github user tzulitai commented on the issue: https://github.com/apache/flink/pull/1962 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. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #1962: [FLINK-3857][Streaming Connectors]Add reconnect attempt t...
Github user StephanEwen commented on the issue: https://github.com/apache/flink/pull/1962 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. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
