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/10/06 13:50:10 UTC
(stormcrawler) branch opensearch-java-async-bulk created (now d86535f7)
This is an automated email from the ASF dual-hosted git repository.
jnioche pushed a change to branch opensearch-java-async-bulk
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
at d86535f7 opensearch-java: send bulk requests with the OpenSearchAsyncClient
This branch includes the following new commits:
new d86535f7 opensearch-java: send bulk requests with the OpenSearchAsyncClient
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: opensearch-java: send bulk requests with the OpenSearchAsyncClient
Posted by jn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jnioche pushed a commit to branch opensearch-java-async-bulk
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit d86535f73caa56bd2e2c6b4255694e7d8d628a45
Author: Julien Nioche <ju...@digitalpebble.com>
AuthorDate: Tue Oct 6 14:47:57 2026 +0100
opensearch-java: send bulk requests with the OpenSearchAsyncClient
AsyncBulkProcessor ran the blocking OpenSearchClient.bulk() call on a
ThreadPoolExecutor bounded to concurrentRequests threads, with a
SynchronousQueue and CallerRunsPolicy. Since the semaphore permit was
released before the listener processed the response, the bolt could
submit the next request while the only worker was still in afterBulk;
the task was then rejected and the bulk call executed on the bolt's
own thread, blocking it for the whole HTTP round trip.
Use OpenSearchAsyncClient instead, as the legacy BulkProcessor did with
bulkAsync, and release the permit only once the listener has run.
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
external/opensearch-java/README.md | 5 +-
.../opensearch/AsyncBulkProcessor.java | 99 ++++++++--------------
.../opensearch/OpenSearchConnection.java | 4 +-
.../opensearch/AsyncBulkProcessorTest.java | 81 +++++++++++++++---
4 files changed, 112 insertions(+), 77 deletions(-)
diff --git a/external/opensearch-java/README.md b/external/opensearch-java/README.md
index 8795d2d9..5e0c33b8 100644
--- a/external/opensearch-java/README.md
+++ b/external/opensearch-java/README.md
@@ -16,7 +16,10 @@ uses the typed `OpenSearchClient` and the `ApacheHttpClient5TransportBuilder`
transport. Unlike the legacy client, the Java Client 3.x no longer ships a
sniffer nor a built-in `BulkProcessor`; this module provides an internal
`AsyncBulkProcessor` that preserves the same semantics (size/count/time based
-flushing, back-pressure, listener callbacks).
+flushing, back-pressure, listener callbacks). Like the legacy `BulkProcessor`,
+it sends the bulk requests asynchronously (via the `OpenSearchAsyncClient`), so
+the bolts never wait for the HTTP round trip unless the maximum number of
+concurrent requests is reached.
Getting started
---------------------
diff --git a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessor.java b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessor.java
index 3f8f22d9..a6fd19cb 100644
--- a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessor.java
+++ b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessor.java
@@ -20,18 +20,16 @@ package org.apache.stormcrawler.opensearch;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
-import org.opensearch.client.opensearch.OpenSearchClient;
+import org.opensearch.client.opensearch.OpenSearchAsyncClient;
import org.opensearch.client.opensearch.core.BulkRequest;
import org.opensearch.client.opensearch.core.BulkResponse;
import org.opensearch.client.opensearch.core.bulk.BulkOperation;
@@ -44,8 +42,10 @@ import org.slf4j.LoggerFactory;
* OpenSearch either when the configured number of actions is reached or when a periodic timer
* fires.
*
- * <p>Concurrency is controlled via a {@link Semaphore}: each in-flight bulk request acquires a
- * permit, which provides natural back-pressure towards the Storm topology when the cluster slows
+ * <p>Bulk requests are sent with the {@link OpenSearchAsyncClient}, so the calling thread never
+ * waits for the HTTP round trip. Concurrency is controlled via a {@link Semaphore}: each in-flight
+ * bulk request acquires a permit, which is only released once the listener has processed the
+ * outcome. This provides natural back-pressure towards the Storm topology when the cluster slows
* down.
*/
public final class AsyncBulkProcessor implements AutoCloseable {
@@ -61,7 +61,7 @@ public final class AsyncBulkProcessor implements AutoCloseable {
void afterBulk(long executionId, BulkRequest request, Throwable failure);
}
- private final OpenSearchClient client;
+ private final OpenSearchAsyncClient client;
private final Listener listener;
private final int bulkActions;
private final int concurrentRequests;
@@ -74,9 +74,6 @@ public final class AsyncBulkProcessor implements AutoCloseable {
private final ScheduledExecutorService scheduler;
private final ScheduledFuture<?> flushTask;
- /** Dedicated executor for bulk HTTP calls -- avoids starvation of ForkJoinPool.commonPool(). */
- private final ExecutorService bulkExecutor;
-
private final AtomicBoolean closed = new AtomicBoolean(false);
private AsyncBulkProcessor(Builder builder) {
@@ -87,20 +84,6 @@ public final class AsyncBulkProcessor implements AutoCloseable {
this.concurrencyPermits = new Semaphore(this.concurrentRequests);
this.buffer = new ArrayList<>(bulkActions);
- this.bulkExecutor =
- new ThreadPoolExecutor(
- 1,
- this.concurrentRequests,
- 60L,
- TimeUnit.SECONDS,
- new SynchronousQueue<>(),
- r -> {
- Thread t = new Thread(r, "AsyncBulkProcessor-bulk");
- t.setDaemon(true);
- return t;
- },
- new ThreadPoolExecutor.CallerRunsPolicy());
-
this.scheduler =
Executors.newSingleThreadScheduledExecutor(
r -> {
@@ -181,32 +164,34 @@ public final class AsyncBulkProcessor implements AutoCloseable {
LOG.warn("beforeBulk callback threw exception", e);
}
- CompletableFuture.supplyAsync(
- () -> {
- try {
- return client.bulk(request);
- } catch (Exception e) {
- throw new BulkExecutionException(e);
- }
- },
- bulkExecutor)
- .whenComplete(
- (response, throwable) -> {
- concurrencyPermits.release();
- try {
- if (throwable != null) {
- Throwable cause =
- throwable instanceof BulkExecutionException
- ? throwable.getCause()
- : throwable;
- listener.afterBulk(executionId, request, cause);
- } else {
- listener.afterBulk(executionId, request, response);
- }
- } catch (Exception e) {
- LOG.warn("afterBulk callback threw exception", e);
- }
- });
+ CompletableFuture<BulkResponse> future;
+ try {
+ future = client.bulk(request);
+ } catch (Exception e) {
+ future = CompletableFuture.failedFuture(e);
+ }
+
+ // the permit is released only after the listener has run, so that the caller can't
+ // start a new bulk request while the outcome of the previous one is still being processed
+ future.whenComplete(
+ (response, throwable) -> {
+ try {
+ if (throwable != null) {
+ Throwable cause =
+ throwable instanceof CompletionException
+ && throwable.getCause() != null
+ ? throwable.getCause()
+ : throwable;
+ listener.afterBulk(executionId, request, cause);
+ } else {
+ listener.afterBulk(executionId, request, response);
+ }
+ } catch (Exception e) {
+ LOG.warn("afterBulk callback threw exception", e);
+ } finally {
+ concurrencyPermits.release();
+ }
+ });
}
/**
@@ -243,9 +228,6 @@ public final class AsyncBulkProcessor implements AutoCloseable {
concurrencyPermits.release(concurrentRequests);
}
- bulkExecutor.shutdown();
- bulkExecutor.awaitTermination(timeout, unit);
-
return acquired;
}
@@ -260,13 +242,13 @@ public final class AsyncBulkProcessor implements AutoCloseable {
/** Builder for {@link AsyncBulkProcessor}. */
public static final class Builder {
- private final OpenSearchClient client;
+ private final OpenSearchAsyncClient client;
private final Listener listener;
private int bulkActions = 50;
private long flushIntervalMillis = 5000;
private int concurrentRequests = 1;
- public Builder(OpenSearchClient client, Listener listener) {
+ public Builder(OpenSearchAsyncClient client, Listener listener) {
this.client = client;
this.listener = listener;
}
@@ -290,11 +272,4 @@ public final class AsyncBulkProcessor implements AutoCloseable {
return new AsyncBulkProcessor(this);
}
}
-
- /** Unchecked wrapper for checked exceptions thrown during bulk execution. */
- private static final class BulkExecutionException extends RuntimeException {
- BulkExecutionException(Throwable cause) {
- super(cause);
- }
- }
}
diff --git a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java
index 5a1ef1fa..eeb47362 100644
--- a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java
+++ b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/OpenSearchConnection.java
@@ -48,6 +48,7 @@ import org.apache.hc.core5.util.Timeout;
import org.apache.stormcrawler.util.ConfUtils;
import org.jetbrains.annotations.NotNull;
import org.opensearch.client.json.jackson.JacksonJsonpMapper;
+import org.opensearch.client.opensearch.OpenSearchAsyncClient;
import org.opensearch.client.opensearch.OpenSearchClient;
import org.opensearch.client.opensearch.core.bulk.BulkOperation;
import org.opensearch.client.transport.OpenSearchTransport;
@@ -147,7 +148,8 @@ public final class OpenSearchConnection {
AsyncBulkProcessor bulkProcessor = null;
try {
bulkProcessor =
- new AsyncBulkProcessor.Builder(cr.client(), listener)
+ new AsyncBulkProcessor.Builder(
+ new OpenSearchAsyncClient(cr.transport()), listener)
.setBulkActions(bulkActions)
.setFlushIntervalMillis(flushIntervalMillis)
.setConcurrentRequests(concurrentRequests)
diff --git a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorTest.java
index 2fff8e15..fefcca56 100644
--- a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorTest.java
+++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorTest.java
@@ -25,12 +25,13 @@ import static org.mockito.Mockito.when;
import java.io.IOException;
import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
-import org.opensearch.client.opensearch.OpenSearchClient;
+import org.opensearch.client.opensearch.OpenSearchAsyncClient;
import org.opensearch.client.opensearch.core.BulkRequest;
import org.opensearch.client.opensearch.core.BulkResponse;
import org.opensearch.client.opensearch.core.bulk.BulkOperation;
@@ -49,9 +50,10 @@ class AsyncBulkProcessorTest {
.build();
}
- private static OpenSearchClient mockClient() throws IOException {
- OpenSearchClient client = mock(OpenSearchClient.class);
- when(client.bulk(any(BulkRequest.class))).thenReturn(emptyBulkResponse());
+ private static OpenSearchAsyncClient mockClient() throws IOException {
+ OpenSearchAsyncClient client = mock(OpenSearchAsyncClient.class);
+ when(client.bulk(any(BulkRequest.class)))
+ .thenAnswer(invocation -> CompletableFuture.supplyAsync(() -> emptyBulkResponse()));
return client;
}
@@ -79,7 +81,7 @@ class AsyncBulkProcessorTest {
long executionId, BulkRequest request, Throwable failure) {}
};
- OpenSearchClient client = mockClient();
+ OpenSearchAsyncClient client = mockClient();
// bulkActions = 3, long flush interval so only threshold triggers
AsyncBulkProcessor processor =
@@ -122,7 +124,7 @@ class AsyncBulkProcessorTest {
long executionId, BulkRequest request, Throwable failure) {}
};
- OpenSearchClient client = mockClient();
+ OpenSearchAsyncClient client = mockClient();
// bulkActions very high, short flush interval
AsyncBulkProcessor processor =
@@ -150,17 +152,24 @@ class AsyncBulkProcessorTest {
AtomicInteger maxConcurrent = new AtomicInteger(0);
CountDownLatch allDone = new CountDownLatch(3);
- OpenSearchClient client = mock(OpenSearchClient.class);
+ OpenSearchAsyncClient client = mock(OpenSearchAsyncClient.class);
when(client.bulk(any(BulkRequest.class)))
.thenAnswer(
invocation -> {
int current = concurrentCalls.incrementAndGet();
maxConcurrent.updateAndGet(prev -> Math.max(prev, current));
- // simulate some work
- Thread.sleep(200);
- concurrentCalls.decrementAndGet();
- allDone.countDown();
- return emptyBulkResponse();
+ return CompletableFuture.supplyAsync(
+ () -> {
+ // simulate some work
+ try {
+ Thread.sleep(200);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ concurrentCalls.decrementAndGet();
+ allDone.countDown();
+ return emptyBulkResponse();
+ });
});
AsyncBulkProcessor.Listener listener =
@@ -219,7 +228,7 @@ class AsyncBulkProcessorTest {
long executionId, BulkRequest request, Throwable failure) {}
};
- OpenSearchClient client = mockClient();
+ OpenSearchAsyncClient client = mockClient();
// bulkActions very high so nothing auto-flushes, long interval
AsyncBulkProcessor processor =
@@ -238,4 +247,50 @@ class AsyncBulkProcessorTest {
assertTrue(closed, "awaitClose should return true");
assertEquals(1, totalBulkCalls.get(), "buffered operations should have been flushed");
}
+
+ /**
+ * Verify that the calling thread does not wait for the HTTP round trip: adding operations while
+ * a bulk request is in flight only blocks once all the permits are taken.
+ */
+ @Test
+ @Timeout(10)
+ void addDoesNotWaitForBulkResponse() throws Exception {
+ CompletableFuture<BulkResponse> pending = new CompletableFuture<>();
+ CountDownLatch afterBulkLatch = new CountDownLatch(1);
+
+ OpenSearchAsyncClient client = mock(OpenSearchAsyncClient.class);
+ when(client.bulk(any(BulkRequest.class))).thenReturn(pending);
+
+ AsyncBulkProcessor.Listener listener =
+ new AsyncBulkProcessor.Listener() {
+ @Override
+ public void beforeBulk(long executionId, BulkRequest request) {}
+
+ @Override
+ public void afterBulk(
+ long executionId, BulkRequest request, BulkResponse response) {
+ afterBulkLatch.countDown();
+ }
+
+ @Override
+ public void afterBulk(
+ long executionId, BulkRequest request, Throwable failure) {}
+ };
+
+ AsyncBulkProcessor processor =
+ new AsyncBulkProcessor.Builder(client, listener)
+ .setBulkActions(1)
+ .setFlushIntervalMillis(60_000)
+ .setConcurrentRequests(1)
+ .build();
+
+ // triggers a bulk request which never completes until we say so
+ processor.add(dummyOp());
+ assertEquals(1, afterBulkLatch.getCount(), "bulk response should still be pending");
+
+ pending.complete(emptyBulkResponse());
+ assertTrue(afterBulkLatch.await(5, TimeUnit.SECONDS));
+
+ processor.awaitClose(5, TimeUnit.SECONDS);
+ }
}