nadav fux created FLINK-40381:
---------------------------------
Summary: Fix thread leak in Elasticsearch8AsyncWriter by closing
HTTP transport on sink close
Key: FLINK-40381
URL: https://issues.apache.org/jira/browse/FLINK-40381
Project: Flink
Issue Type: Bug
Components: Connectors / ElasticSearch
Affects Versions: elasticsearch-4.0.0, elasticsearch-3.0.1,
elasticsearch-3.1.0, elasticsearch-3.0.0
Reporter: nadav fux
In `flink-connector-elasticsearch8`, running jobs on a persistent Flink Session
Cluster causes a thread leak of `I/O dispatcher` threads per job run. When jobs
are submitted and completed repeatedly, zombie reactor worker threads
accumulate in the TaskManager JVM until the TaskManager process crashes or
exhausts system thread limits.
*Root Cause*
In `Elasticsearch8AsyncWriter.java`, the `close()` method contains:
public void close() {
if (!this.close) {
this.close = true;
this.esClient.shutdown();
}
}
In the official Elastic Java API Client
({{{}co.elastic.clients.elasticsearch.ElasticsearchAsyncClient{}}}),
{{esClient.shutdown()}} returns an {{ElasticsearchShutdownAsyncClient}} builder
for making requests to Elasticsearch Cluster Shutdown REST API endpoints. It
does *NOT* close the underlying {{RestClient}} or HTTP Async Client IO reactor
thread pool.
h3. Solution
Update {{Elasticsearch8AsyncWriter.close()}} to invoke
{{{}esClient._transport().close(){}}}, which closes the underlying
{{RestClient}} and terminates the {{I/O dispatcher}} worker thread pool upon
sink teardown.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
