You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stormcrawler.apache.org by jn...@apache.org on 2026/04/02 17:46:53 UTC
(stormcrawler) branch fix-opensearch-connection-resource-leak created (now 1661088b)
This is an automated email from the ASF dual-hosted git repository.
jnioche pushed a change to branch fix-opensearch-connection-resource-leak
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
at 1661088b Fix client leak on BulkProcessor or Sniffer construction failure
This branch includes the following new commits:
new 1661088b Fix client leak on BulkProcessor or Sniffer construction failure
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(stormcrawler) 01/01: Fix client leak on BulkProcessor or Sniffer construction failure
Posted by jn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jnioche pushed a commit to branch fix-opensearch-connection-resource-leak
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit 1661088b48a2bc4421b8f543371bc251d2c70510
Author: Julien Nioche <ju...@digitalpebble.com>
AuthorDate: Thu Apr 2 18:46:42 2026 +0100
Fix client leak on BulkProcessor or Sniffer construction failure
In getConnection(), if BulkProcessor.builder().build() or
Sniffer.builder().build() throws, the already-created RestHighLevelClient
(and its connection pool/threads) is leaked. Similarly, if Sniffer
construction fails, the BulkProcessor is also leaked.
Wrap construction in try-catch to close already-created resources on failure.
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
---
.../opensearch/OpenSearchConnection.java | 56 ++++++++++++++--------
1 file changed, 37 insertions(+), 19 deletions(-)
diff --git a/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java b/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java
index c3662a09..58d38df2 100644
--- a/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java
+++ b/external/opensearch/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java
@@ -286,27 +286,45 @@ public final class OpenSearchConnection {
new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(
bufferSize * 1024 * 1024));
- final BulkProcessor bulkProcessor =
- BulkProcessor.builder(
- (request, bulkListener) ->
- client.bulkAsync(
- request,
- requestOptionsBuilder.build(),
- bulkListener),
- listener)
- .setFlushInterval(flushInterval)
- .setBulkActions(bulkActions)
- .setConcurrentRequests(concurrentRequests)
- .build();
-
- boolean sniff =
- ConfUtils.getBoolean(stormConf, Constants.PARAMPREFIX, dottedType, "sniff", true);
+ BulkProcessor bulkProcessor = null;
Sniffer sniffer = null;
- if (sniff) {
- sniffer = Sniffer.builder(client.getLowLevelClient()).build();
- }
+ try {
+ bulkProcessor =
+ BulkProcessor.builder(
+ (request, bulkListener) ->
+ client.bulkAsync(
+ request,
+ requestOptionsBuilder.build(),
+ bulkListener),
+ listener)
+ .setFlushInterval(flushInterval)
+ .setBulkActions(bulkActions)
+ .setConcurrentRequests(concurrentRequests)
+ .build();
+
+ boolean sniff =
+ ConfUtils.getBoolean(
+ stormConf, Constants.PARAMPREFIX, dottedType, "sniff", true);
+ if (sniff) {
+ sniffer = Sniffer.builder(client.getLowLevelClient()).build();
+ }
- return new OpenSearchConnection(client, bulkProcessor, sniffer);
+ return new OpenSearchConnection(client, bulkProcessor, sniffer);
+ } catch (Exception e) {
+ if (bulkProcessor != null) {
+ try {
+ bulkProcessor.close();
+ } catch (Exception suppressed) {
+ e.addSuppressed(suppressed);
+ }
+ }
+ try {
+ client.close();
+ } catch (IOException suppressed) {
+ e.addSuppressed(suppressed);
+ }
+ throw e;
+ }
}
private boolean isClosed = false;