This is an automated email from the ASF dual-hosted git repository.
dybyte pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 3d55c3d763 [Fix][Connector-V2] Fix Elasticsearch connector connection
pool leak … (#10857)
3d55c3d763 is described below
commit 3d55c3d763a0218227ce325f3703688dd0a5e872
Author: cosmosni <[email protected]>
AuthorDate: Tue May 12 17:31:13 2026 +0800
[Fix][Connector-V2] Fix Elasticsearch connector connection pool leak …
(#10857)
---
.../elasticsearch/client/EsRestClient.java | 98 ++++++-------
.../elasticsearch/client/EsRestClientTest.java | 155 +++++++++++++++++++++
2 files changed, 206 insertions(+), 47 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java
index 21b3c0eae6..6c57433ae9 100644
---
a/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java
@@ -130,8 +130,8 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.BULK_RESPONSE_ERROR,
"bulk es Response is null");
}
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
JsonNode json = OBJECT_MAPPER.readTree(entity);
int took = json.get("took").asInt();
boolean errors = json.get("errors").asBoolean();
@@ -140,8 +140,9 @@ public class EsRestClient implements Closeable {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.BULK_RESPONSE_ERROR,
String.format(
- "bulk es response status=%s,request
body(truncate)=%s",
- response,
+ "bulk es response status=%s, response body=%s,
request body(truncate)=%s",
+ response.getStatusLine().getStatusCode(),
+ entity,
requestBody.substring(0, Math.min(1000,
requestBody.length()))));
}
} catch (IOException e) {
@@ -316,16 +317,17 @@ public class EsRestClient implements Closeable {
log.warn("DELETE {} response null for scroll ID: {}",
endpoint, scrollId);
return false;
}
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
JsonNode jsonNode = JsonUtils.parseObject(entity);
boolean succeeded = jsonNode.get("succeeded").asBoolean();
return succeeded;
} else {
log.warn(
- "DELETE {} response status code={} for scroll ID: {}",
+ "DELETE {} response status code={}, body={} for scroll
ID: {}",
endpoint,
response.getStatusLine().getStatusCode(),
+ entity,
scrollId);
return false;
}
@@ -346,16 +348,19 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.SCROLL_REQUEST_ERROR,
"POST " + endpoint + " response null");
}
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
ObjectNode responseJson = JsonUtils.parseObject(entity);
return getDocsFromSqlResponse(responseJson, columnNodes);
} else {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.SCROLL_REQUEST_ERROR,
String.format(
- "POST %s response status code=%d,request
body=%s",
- endpoint,
response.getStatusLine().getStatusCode(), requestBody));
+ "POST %s response status code=%d, body=%s,
request body=%s",
+ endpoint,
+ response.getStatusLine().getStatusCode(),
+ entity,
+ requestBody));
}
} catch (IOException e) {
throw new ElasticsearchConnectorException(
@@ -375,8 +380,8 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.SCROLL_REQUEST_ERROR,
"POST " + endpoint + " response null");
}
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
ObjectNode responseJson = JsonUtils.parseObject(entity);
JsonNode shards = responseJson.get("_shards");
@@ -393,8 +398,11 @@ public class EsRestClient implements Closeable {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.SCROLL_REQUEST_ERROR,
String.format(
- "POST %s response status code=%d,request
body=%s",
- endpoint,
response.getStatusLine().getStatusCode(), requestBody));
+ "POST %s response status code=%d, body=%s,
request body=%s",
+ endpoint,
+ response.getStatusLine().getStatusCode(),
+ entity,
+ requestBody));
}
} catch (IOException e) {
throw new ElasticsearchConnectorException(
@@ -502,15 +510,15 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.GET_INDEX_DOCS_COUNT_FAILED,
"GET " + endpoint + " response null");
}
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
return JsonUtils.toList(entity, IndexDocsCount.class);
} else {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.GET_INDEX_DOCS_COUNT_FAILED,
String.format(
- "GET %s response status code=%d",
- endpoint,
response.getStatusLine().getStatusCode()));
+ "GET %s response status code=%d, body=%s",
+ endpoint,
response.getStatusLine().getStatusCode(), entity));
}
} catch (IOException ex) {
throw new ElasticsearchConnectorException(
@@ -528,8 +536,8 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.LIST_INDEX_FAILED,
"GET " + endpoint + " response null");
}
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
return JsonUtils.toList(entity, Map.class).stream()
.map(map -> map.get("index").toString())
.collect(Collectors.toList());
@@ -537,8 +545,8 @@ public class EsRestClient implements Closeable {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.LIST_INDEX_FAILED,
String.format(
- "GET %s response status code=%d",
- endpoint,
response.getStatusLine().getStatusCode()));
+ "GET %s response status code=%d, body=%s",
+ endpoint,
response.getStatusLine().getStatusCode(), entity));
}
} catch (IOException ex) {
throw new ElasticsearchConnectorException(
@@ -563,12 +571,13 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.CREATE_INDEX_FAILED,
"PUT " + endpoint + " response null");
}
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.CREATE_INDEX_FAILED,
String.format(
- "PUT %s response status code=%d",
- endpoint,
response.getStatusLine().getStatusCode()));
+ "PUT %s response status code=%d, body=%s",
+ endpoint,
response.getStatusLine().getStatusCode(), entity));
}
} catch (IOException ex) {
throw new ElasticsearchConnectorException(
@@ -586,15 +595,13 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.DROP_INDEX_FAILED,
"DELETE " + endpoint + " response null");
}
- // todo: if the index doesn't exist, the response status code is
200?
- if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- return;
- } else {
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
+ if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.DROP_INDEX_FAILED,
String.format(
- "DELETE %s response status code=%d",
- endpoint,
response.getStatusLine().getStatusCode()));
+ "DELETE %s response status code=%d, body=%s",
+ endpoint,
response.getStatusLine().getStatusCode(), entity));
}
} catch (IOException ex) {
throw new ElasticsearchConnectorException(
@@ -615,15 +622,13 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.CLEAR_INDEX_DATA_FAILED,
"POST " + endpoint + " response null");
}
- // todo: if the index doesn't exist, the response status code is
200?
- if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- return;
- } else {
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
+ if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.CLEAR_INDEX_DATA_FAILED,
String.format(
- "POST %s response status code=%d",
- endpoint,
response.getStatusLine().getStatusCode()));
+ "POST %s response status code=%d, body=%s",
+ endpoint,
response.getStatusLine().getStatusCode(), entity));
}
} catch (IOException ex) {
throw new ElasticsearchConnectorException(
@@ -650,11 +655,12 @@ public class EsRestClient implements Closeable {
"GET " + endpoint + " response null");
}
if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.GET_INDEX_DOCS_COUNT_FAILED,
String.format(
- "GET %s response status code=%d",
- endpoint,
response.getStatusLine().getStatusCode()));
+ "GET %s response status code=%d, body=%s",
+ endpoint,
response.getStatusLine().getStatusCode(), entity));
}
String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
log.info(String.format("GET %s respnse=%s", endpoint, entity));
@@ -841,15 +847,13 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.ADD_FIELD_FAILED,
"PUT " + endpoint + " response null");
}
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.ADD_FIELD_FAILED,
String.format(
- "PUT %s response status code=%d, response=%s",
- endpoint,
- response.getStatusLine().getStatusCode(),
- EntityUtils.toString(
- response.getEntity(),
StandardCharsets.UTF_8)));
+ "PUT %s response status code=%d, body=%s",
+ endpoint,
response.getStatusLine().getStatusCode(), entity));
}
} catch (IOException ex) {
throw new ElasticsearchConnectorException(
@@ -877,16 +881,16 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.CREATE_PIT_FAILED,
"POST " + endpoint + " response null");
}
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
JsonNode jsonNode = JsonUtils.parseObject(entity);
return jsonNode.get("id").asText();
} else {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.CREATE_PIT_FAILED,
String.format(
- "POST %s response status code=%d",
- endpoint,
response.getStatusLine().getStatusCode()));
+ "POST %s response status code=%d, body=%s",
+ endpoint,
response.getStatusLine().getStatusCode(), entity));
}
} catch (IOException ex) {
throw new ElasticsearchConnectorException(
@@ -913,16 +917,16 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.DELETE_PIT_FAILED,
"DELETE " + endpoint + " response null");
}
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
JsonNode jsonNode = JsonUtils.parseObject(entity);
return jsonNode.get("succeeded").asBoolean();
} else {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.DELETE_PIT_FAILED,
String.format(
- "DELETE %s response status code=%d",
- endpoint,
response.getStatusLine().getStatusCode()));
+ "DELETE %s response status code=%d, body=%s",
+ endpoint,
response.getStatusLine().getStatusCode(), entity));
}
} catch (IOException ex) {
throw new ElasticsearchConnectorException(
@@ -1010,15 +1014,15 @@ public class EsRestClient implements Closeable {
ElasticsearchConnectorErrorCode.SEARCH_WITH_PIT_FAILED,
"POST " + endpoint + " response null");
}
+ String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
- String entity = EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8);
return parsePointInTimeResponse(entity, pitId);
} else {
throw new ElasticsearchConnectorException(
ElasticsearchConnectorErrorCode.SEARCH_WITH_PIT_FAILED,
String.format(
- "POST %s response status code=%d",
- endpoint,
response.getStatusLine().getStatusCode()));
+ "POST %s response status code=%d, body=%s",
+ endpoint,
response.getStatusLine().getStatusCode(), entity));
}
} catch (IOException ex) {
throw new ElasticsearchConnectorException(
diff --git
a/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClientTest.java
b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClientTest.java
new file mode 100644
index 0000000000..03a4ddbb0f
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClientTest.java
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.elasticsearch.client;
+
+import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.BulkResponse;
+import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.ElasticsearchConnectorErrorCode;
+import
org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.ElasticsearchConnectorException;
+
+import org.apache.http.StatusLine;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.lang.reflect.Constructor;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class EsRestClientTest {
+
+ @Mock private RestClient mockRestClient;
+
+ // ---- bulk: non-200 path ----
+
+ @Test
+ void testBulkIncludesEsErrorBodyOnNon200Response() throws Exception {
+ Response mockResponse = mock(Response.class);
+ StatusLine mockStatusLine = mock(StatusLine.class);
+ String esErrorJson =
"{\"error\":{\"root_cause\":[{\"type\":\"rate_limit_exceeded\"}]}}";
+
+
when(mockRestClient.performRequest(any(Request.class))).thenReturn(mockResponse);
+ when(mockResponse.getStatusLine()).thenReturn(mockStatusLine);
+ when(mockStatusLine.getStatusCode()).thenReturn(429);
+ when(mockResponse.getEntity())
+ .thenReturn(
+ new org.apache.http.entity.StringEntity(
+ esErrorJson,
org.apache.http.entity.ContentType.APPLICATION_JSON));
+
+ EsRestClient client = createEsRestClient(mockRestClient);
+
+ ElasticsearchConnectorException exception =
+ Assertions.assertThrows(
+ ElasticsearchConnectorException.class,
+ () -> client.bulk("{\"index\":{}}\n{}\n"));
+
+ Assertions.assertEquals(
+ ElasticsearchConnectorErrorCode.BULK_RESPONSE_ERROR,
+ exception.getSeaTunnelErrorCode());
+ Assertions.assertTrue(
+ exception.getMessage().contains(esErrorJson),
+ "Error message should contain ES response body for debugging");
+ }
+
+ // ---- bulk: 200 success path ----
+
+ @Test
+ void testBulkSucceedsOn200Response() throws Exception {
+ Response mockResponse = mock(Response.class);
+ StatusLine mockStatusLine = mock(StatusLine.class);
+
+
when(mockRestClient.performRequest(any(Request.class))).thenReturn(mockResponse);
+ when(mockResponse.getStatusLine()).thenReturn(mockStatusLine);
+ when(mockStatusLine.getStatusCode()).thenReturn(200);
+ when(mockResponse.getEntity())
+ .thenReturn(
+ new org.apache.http.entity.StringEntity(
+ "{\"took\":1,\"errors\":false,\"items\":[]}",
+
org.apache.http.entity.ContentType.APPLICATION_JSON));
+
+ EsRestClient client = createEsRestClient(mockRestClient);
+
+ BulkResponse result = client.bulk("{\"index\":{}}\n{}\n");
+ Assertions.assertFalse(result.isErrors());
+ }
+
+ // ---- createIndex: 200 success path (void method, must still consume
entity) ----
+
+ @Test
+ void testCreateIndexConsumesEntityOn200Response() throws Exception {
+ Response mockResponse = mock(Response.class);
+ StatusLine mockStatusLine = mock(StatusLine.class);
+
+
when(mockRestClient.performRequest(any(Request.class))).thenReturn(mockResponse);
+ when(mockResponse.getStatusLine()).thenReturn(mockStatusLine);
+ when(mockStatusLine.getStatusCode()).thenReturn(200);
+ when(mockResponse.getEntity())
+ .thenReturn(
+ new org.apache.http.entity.StringEntity(
+ "{\"acknowledged\":true}",
+
org.apache.http.entity.ContentType.APPLICATION_JSON));
+
+ EsRestClient client = createEsRestClient(mockRestClient);
+
+ // Should not throw — entity is consumed internally
+ Assertions.assertDoesNotThrow(() -> client.createIndex("test_index"));
+ }
+
+ // ---- createIndex: non-200 path includes ES error body ----
+
+ @Test
+ void testCreateIndexIncludesEsErrorBodyOnNon200Response() throws Exception
{
+ Response mockResponse = mock(Response.class);
+ StatusLine mockStatusLine = mock(StatusLine.class);
+ String esErrorJson =
"{\"error\":{\"type\":\"resource_already_exists_exception\"}}";
+
+
when(mockRestClient.performRequest(any(Request.class))).thenReturn(mockResponse);
+ when(mockResponse.getStatusLine()).thenReturn(mockStatusLine);
+ when(mockStatusLine.getStatusCode()).thenReturn(400);
+ when(mockResponse.getEntity())
+ .thenReturn(
+ new org.apache.http.entity.StringEntity(
+ esErrorJson,
org.apache.http.entity.ContentType.APPLICATION_JSON));
+
+ EsRestClient client = createEsRestClient(mockRestClient);
+
+ ElasticsearchConnectorException exception =
+ Assertions.assertThrows(
+ ElasticsearchConnectorException.class,
+ () -> client.createIndex("test_index"));
+
+ Assertions.assertTrue(
+ exception.getMessage().contains(esErrorJson),
+ "Error message should contain ES response body for debugging");
+ }
+
+ private EsRestClient createEsRestClient(RestClient restClient) throws
Exception {
+ Constructor<EsRestClient> constructor =
+ EsRestClient.class.getDeclaredConstructor(RestClient.class);
+ constructor.setAccessible(true);
+ return constructor.newInstance(restClient);
+ }
+}
