[FLINK-11235][es] Close transportclient if no connection could be est…

…ablished
bbejeck · Jan 2, 2019 · 4febcdc · 4febcdc
1 parent 7bb63f6
commit 4febcdc
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 0 deletions.
diff --git a/...ava/org/apache/flink/streaming/connectors/elasticsearch2/Elasticsearch2ApiCallBridge.java b/...ava/org/apache/flink/streaming/connectors/elasticsearch2/Elasticsearch2ApiCallBridge.java
@@ -21,6 +21,7 @@
 import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchApiCallBridge;
 import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
 import org.apache.flink.streaming.connectors.elasticsearch.util.ElasticsearchUtils;
+import org.apache.flink.util.IOUtils;
 import org.apache.flink.util.Preconditions;
 
 import org.elasticsearch.action.bulk.BackoffPolicy;
@@ -72,6 +73,10 @@ public TransportClient createClient(Map<String, String> clientConfig) {
 
  // verify that we actually are connected to a cluster
  if (transportClient.connectedNodes().isEmpty()) {
+
+ // close the transportClient here
+ IOUtils.closeQuietly(transportClient);
+
  throw new RuntimeException("Elasticsearch client is not connected to any Elasticsearch nodes!");
  }
 

diff --git a/...ava/org/apache/flink/streaming/connectors/elasticsearch5/Elasticsearch5ApiCallBridge.java b/...ava/org/apache/flink/streaming/connectors/elasticsearch5/Elasticsearch5ApiCallBridge.java
@@ -21,6 +21,7 @@
 import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchApiCallBridge;
 import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
 import org.apache.flink.streaming.connectors.elasticsearch.util.ElasticsearchUtils;
+import org.apache.flink.util.IOUtils;
 import org.apache.flink.util.Preconditions;
 
 import org.elasticsearch.action.bulk.BackoffPolicy;
@@ -78,6 +79,10 @@ public TransportClient createClient(Map<String, String> clientConfig) {
 
  // verify that we actually are connected to a cluster
  if (transportClient.connectedNodes().isEmpty()) {
+
+ // close the transportClient here
+ IOUtils.closeQuietly(transportClient);
+
  throw new RuntimeException("Elasticsearch client is not connected to any Elasticsearch nodes!");
  }