(flink-connector-elasticsearch) branch main updated: [FLINK-40006] ElasticsearchRowDataLookupFunction leaks REST client / HTTP connections on task close

[guoweijie]

This is an automated email from the ASF dual-hosted git repository.

reswqa pushed a commit to branch main
in repository 
https://gitbox.apache.org/repos/asf/flink-connector-elasticsearch.git


The following commit(s) were added to refs/heads/main by this push:
     new 1339e8b  [FLINK-40006] ElasticsearchRowDataLookupFunction leaks REST 
client / HTTP connections on task close
1339e8b is described below

commit 1339e8bdf29999905b273bd8aa4b67912d2b502c
Author: reswqa <res...@163.com>
AuthorDate: Mon Jun 29 15:20:23 2026 +0800

    [FLINK-40006] ElasticsearchRowDataLookupFunction leaks REST client / HTTP 
connections on task close
---
 .../lookup/ElasticsearchRowDataLookupFunction.java      | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git 
a/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/lookup/ElasticsearchRowDataLookupFunction.java
 
b/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/lookup/ElasticsearchRowDataLookupFunction.java
index afec68a..0186e75 100644
--- 
a/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/lookup/ElasticsearchRowDataLookupFunction.java
+++ 
b/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/lookup/ElasticsearchRowDataLookupFunction.java
@@ -191,4 +191,21 @@ public class ElasticsearchRowDataLookupFunction<C extends 
AutoCloseable> extends
 
         return row;
     }
+
+    @Override
+    public void close() throws Exception {
+        try {
+            if (client != null) {
+                try {
+                    callBridge.close(client);
+                } catch (Exception e) {
+                    LOG.warn("Failed to close Elasticsearch client in lookup 
function.", e);
+                } finally {
+                    client = null;
+                }
+            }
+        } finally {
+            super.close();
+        }
+    }
 }