[jira] [Created] (FLINK-40381) Fix thread leak in Elasticsearch8AsyncWriter by closing HTTP transport on sink close

["nadav fux (Jira)" <ji...@apache.org>]

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)