aljoscha commented on a change in pull request #10936: [FLINK-13689]
[Connectors/ElasticSearch] Fix thread leak in Elasticsearch connector when
cluster is down
URL: https://github.com/apache/flink/pull/10936#discussion_r370527156
##########
File path:
flink-connectors/flink-connector-elasticsearch5/src/main/java/org/apache/flink/streaming/connectors/elasticsearch5/Elasticsearch5ApiCallBridge.java
##########
@@ -132,4 +119,19 @@ public void configureBulkProcessorBackoff(
builder.setBackoffPolicy(backoffPolicy);
}
+
+ @Override
+ public void verifyClientConnection(TransportClient client) {
+ // verify that we actually are connected to a cluster
+ if (client.connectedNodes().isEmpty()) {
+ // close the transportClient here
+ IOUtils.closeQuietly(client);
+
+ throw new RuntimeException("Elasticsearch client is not
connected to any Elasticsearch nodes!");
+ }
+
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Created Elasticsearch TransportClient with
connected nodes {}", client.connectedNodes());
Review comment:
```suggestion
LOG.info("Elasticsearch TransportClient is connected to
nodes {}", client.connectedNodes());
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
