Skip to content

#2212 retry 429 items and flush bulk requests by size #2241

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jnioche merged 5 commits into from
Oct 21, 2026
Merged

#2212 retry 429 items and flush bulk requests by size #2241

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3a04d64
retry the bulk items rejected with status 429
dpol1 Oct 16, 2026
bf7b5a2
flush the bulk requests by size
dpol1 Oct 16, 2026
2bccebb
drop a status tuple from WaitAckCache when adding it to the bulk fails
dpol1 Oct 19, 2026
410382c
document which bulk failures are not retried
dpol1 Oct 19, 2026
026ae44
fail a deletion tuple when adding it to the bulk fails
dpol1 Oct 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 20 additions & 10 deletions external/opensearch-java/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,16 +10,26 @@ Apache HttpClient 5:
* [MetricsConsumer](https://github.com/apache/stormcrawler/blob/master/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/metrics/MetricsConsumer.java)
* [StatusMetricsBolt](https://github.com/apache/stormcrawler/blob/master/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/metrics/StatusMetricsBolt.java) for sending the breakdown of URLs per status as metrics and display its evolution over time.

This module is functionally equivalent to the legacy `external/opensearch` module
(which is based on the deprecated `RestHighLevelClient` and HttpClient 4), but
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). 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.
The bolts write to OpenSearch through `AsyncBulkProcessor`, which groups their
operations into bulk requests and:

* sends a bulk request when the number of buffered operations reaches
`opensearch.<bolt>.bulkActions`, when the buffered operations reach 5 MiB of
JSON (counted before compression), or when `opensearch.<bolt>.flushInterval`
elapses.
* resends the items rejected with status 429 (too many requests) in the bulk
response up to 8 times, after 50, 60, 80, 150, 280, 580, 1250 and 2740 ms.
A request that OpenSearch rejects as a whole, with status 429 or any other
error, is not resent. `IndexerBolt`, `DeletionBolt` and `StatusUpdaterBolt`
fail the tuple of an item still rejected after the last retry.
* fails the whole bulk when a retry fails as a whole, for example on a
connection error: `IndexerBolt`, `DeletionBolt` and `StatusUpdaterBolt` fail
the tuples of all its items, including those stored by an earlier attempt.
* sends the requests without blocking the bolt. The bolt waits only when
`opensearch.<bolt>.concurrentRequests` bulks are in flight, including bulks
waiting for a retry.
* reports the outcome of a bulk to the bolt once, after its last attempt, with
the items in request order.

Getting started
---------------------
Expand Down
223 changes: 188 additions & 35 deletions .../opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,11 @@

package org.apache.stormcrawler.opensearch;

import com.google.common.io.CountingOutputStream;
import jakarta.json.stream.JsonGenerator;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
Expand All @@ -29,29 +33,41 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.IntStream;
import org.opensearch.client.json.JsonpMapper;
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;
import org.opensearch.client.opensearch.core.bulk.BulkResponseItem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* Replacement for the legacy {@code org.opensearch.action.bulk.BulkProcessor} that works with the
* new opensearch-java client. Accumulates {@link BulkOperation} instances and flushes them to
* OpenSearch either when the configured number of actions is reached or when a periodic timer
* fires.
* OpenSearch when the configured number of actions is reached, when the buffered operations reach 5
* MiB, or when a periodic timer fires.
*
* <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.
*
* <p>Items rejected with status 429 are sent again up to 8 times, while the bulk keeps its permit.
* The listener's {@code afterBulk} is called once per bulk, with the items of all attempts in
* request order.
*/
public final class AsyncBulkProcessor implements AutoCloseable {

private static final Logger LOG = LoggerFactory.getLogger(AsyncBulkProcessor.class);

private static final long[] RETRY_DELAYS_MILLIS = {50, 60, 80, 150, 280, 580, 1250, 2740};

/** Flush threshold, in bytes of the serialized operations before compression. */
static final long BULK_SIZE_BYTES = 5L * 1024 * 1024;

/** Listener interface equivalent to the legacy {@code BulkProcessor.Listener}. */
public interface Listener {
void beforeBulk(long executionId, BulkRequest request);
Expand All @@ -62,6 +78,7 @@ public interface Listener {
}

private final OpenSearchAsyncClient client;
private final JsonpMapper mapper;
private final Listener listener;
private final int bulkActions;
private final int concurrentRequests;
Expand All @@ -70,27 +87,30 @@ public interface Listener {

private final ReentrantLock lock = new ReentrantLock();
private List<BulkOperation> buffer;
private long bufferedBytes;

private final ScheduledExecutorService scheduler;
private final ScheduledFuture<?> flushTask;

// separate from the flush thread, which can block on a permit that a retrying bulk holds
private final ScheduledExecutorService retryScheduler;

private final AtomicBoolean closed = new AtomicBoolean(false);

private AsyncBulkProcessor(Builder builder) {
this.client = builder.client;
this.mapper = client._transport().jsonpMapper();
this.listener = builder.listener;
this.bulkActions = builder.bulkActions;
this.concurrentRequests = builder.concurrentRequests;
this.concurrencyPermits = new Semaphore(this.concurrentRequests);
this.buffer = new ArrayList<>(bulkActions);

this.scheduler =
Executors.newSingleThreadScheduledExecutor(
r -> {
Thread t = new Thread(r, "AsyncBulkProcessor-flush");
t.setDaemon(true);
return t;
});
this.scheduler = daemonScheduler("AsyncBulkProcessor-flush");
this.retryScheduler =
builder.retryScheduler != null
? builder.retryScheduler
: daemonScheduler("AsyncBulkProcessor-retry");
this.flushTask =
scheduler.scheduleWithFixedDelay(
this::flushIfNeeded,
Expand All @@ -99,16 +119,31 @@ private AsyncBulkProcessor(Builder builder) {
TimeUnit.MILLISECONDS);
}

/** Adds a single bulk operation. Triggers a flush when {@code bulkActions} is reached. */
private static ScheduledExecutorService daemonScheduler(String name) {
return Executors.newSingleThreadScheduledExecutor(
r -> {
Thread t = new Thread(r, name);
t.setDaemon(true);
return t;
});
}

/**
* Adds a single bulk operation. Triggers a flush when {@code bulkActions} or 5 MiB of buffered
* operations is reached. The operation is serialized on the calling thread to count its size,
* so a serialization error is thrown from this method.
*/
public void add(BulkOperation operation) {
if (closed.get()) {
throw new IllegalStateException("BulkProcessor is closed");
}
final long size = serializedSize(mapper, operation);
List<BulkOperation> toFlush = null;
lock.lock();
try {
buffer.add(operation);
if (buffer.size() >= bulkActions) {
bufferedBytes += size;
if (buffer.size() >= bulkActions || bufferedBytes >= BULK_SIZE_BYTES) {
toFlush = swapBuffer();
}
} finally {
Expand Down Expand Up @@ -142,9 +177,27 @@ private void flushIfNeeded() {
private List<BulkOperation> swapBuffer() {
List<BulkOperation> old = buffer;
buffer = new ArrayList<>(bulkActions);
bufferedBytes = 0;
return old;
}

/**
* Bytes the transport writes for the operation: each NDJSON line (action, then document or
* update body) serialized with the transport's mapper, plus a newline.
*/
static long serializedSize(JsonpMapper mapper, BulkOperation operation) {
long size = 0;
for (Iterator<?> it = operation._serializables(); it.hasNext(); ) {
final CountingOutputStream line =
new CountingOutputStream(OutputStream.nullOutputStream());
try (JsonGenerator generator = mapper.jsonProvider().createGenerator(line)) {
mapper.serialize(it.next(), generator);
}
size += line.getCount() + 1;
}
return size;
}

/** Builds the request, acquires a concurrency permit, and executes asynchronously. */
private void executeBulk(List<BulkOperation> operations) {
final long executionId = executionIdGen.incrementAndGet();
Expand All @@ -164,36 +217,77 @@ private void executeBulk(List<BulkOperation> operations) {
LOG.warn("beforeBulk callback threw exception", e);
}

send(
new Bulk(executionId, request),
request,
IntStream.range(0, operations.size()).toArray());
}

/**
* Sends one attempt of the bulk. {@code positions[i]} is the position in the original request
* of the i-th operation of {@code attempt}.
*/
private void send(Bulk bulk, BulkRequest attempt, int[] positions) {
CompletableFuture<BulkResponse> future;
try {
future = client.bulk(request);
future = client.bulk(attempt);
} 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);
Throwable failure =
throwable instanceof CompletionException && throwable.getCause() != null
? throwable.getCause()
: throwable;
BulkResponse result = null;
if (failure == null) {
try {
final int[] rejected = bulk.merge(response, positions);
if (rejected.length > 0 && bulk.retries < RETRY_DELAYS_MILLIS.length) {
retry(bulk, rejected);
return;
}
result = bulk.result(response);
} catch (RuntimeException e) {
failure = e;
}
} catch (Exception e) {
LOG.warn("afterBulk callback threw exception", e);
} finally {
concurrencyPermits.release();
}
finish(bulk, result, failure);
});
}

private void retry(Bulk bulk, int[] positions) {
final long delay = RETRY_DELAYS_MILLIS[bulk.retries++];
LOG.debug(
"Bulk [{}]: retrying {} items rejected with status 429 in {} ms",
bulk.executionId,
positions.length,
delay);
final List<BulkOperation> operations = new ArrayList<>(positions.length);
for (int position : positions) {
operations.add(bulk.request.operations().get(position));
}
final BulkRequest attempt = new BulkRequest.Builder().operations(operations).build();
retryScheduler.schedule(() -> send(bulk, attempt, positions), delay, TimeUnit.MILLISECONDS);
}

// 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
private void finish(Bulk bulk, BulkResponse response, Throwable failure) {
try {
if (failure != null) {
listener.afterBulk(bulk.executionId, bulk.request, failure);
} else {
listener.afterBulk(bulk.executionId, bulk.request, response);
}
} catch (Exception e) {
LOG.warn("afterBulk callback threw exception", e);
} finally {
concurrencyPermits.release();
}
}

/**
* Drains pending operations and waits for all in-flight bulk requests to complete, up to the
* given timeout. Equivalent to the legacy {@code BulkProcessor.awaitClose()}.
Expand Down Expand Up @@ -222,13 +316,17 @@ public boolean awaitClose(long timeout, TimeUnit unit) throws InterruptedExcepti
executeBulk(remaining);
}

// Wait for all in-flight requests to finish by acquiring all permits
boolean acquired = concurrencyPermits.tryAcquire(concurrentRequests, timeout, unit);
if (acquired) {
concurrencyPermits.release(concurrentRequests);
// Wait for all in-flight requests to finish, retries included, by acquiring all permits
try {
boolean acquired = concurrencyPermits.tryAcquire(concurrentRequests, timeout, unit);
if (acquired) {
concurrencyPermits.release(concurrentRequests);
}
return acquired;
} finally {
// retries still waiting after the timeout are dropped
retryScheduler.shutdownNow();
}

return acquired;
}

@Override
Expand All @@ -240,13 +338,62 @@ public void close() {
}
}

/** One bulk and the merged items of its attempts. */
private static final class Bulk {
final long executionId;
final BulkRequest request;

// items is set once a response has items to retry. The attempts of a bulk run one after
// the other, so these fields need no lock.
List<BulkResponseItem> items;
long took;
int retries;

Bulk(long executionId, BulkRequest request) {
this.executionId = executionId;
this.request = request;
}

/** Merges the items of an attempt and returns the positions still rejected with 429. */
int[] merge(BulkResponse response, int[] positions) {
final List<BulkResponseItem> received = response.items();
if (items == null && received.stream().noneMatch(i -> i.status() == 429)) {
return new int[0];
}
if (received.size() != positions.length) {
throw new IllegalStateException("Bulk response items do not match the request");
}
if (items == null) {
items = new ArrayList<>(received);
} else {
for (int i = 0; i < positions.length; i++) {
items.set(positions[i], received.get(i));
}
}
took += response.took();
return IntStream.range(0, items.size())
.filter(i -> items.get(i).status() == 429)
.toArray();
}

/** The response for the listener: the last one, or the merged items after retries. */
BulkResponse result(BulkResponse last) {
if (items == null) {
return last;
}
final boolean errors = items.stream().anyMatch(i -> i.error() != null);
return BulkResponse.of(b -> b.took(took).errors(errors).items(items));
}
}

/** Builder for {@link AsyncBulkProcessor}. */
public static final class Builder {
private final OpenSearchAsyncClient client;
private final Listener listener;
private int bulkActions = 50;
private long flushIntervalMillis = 5000;
private int concurrentRequests = 1;
private ScheduledExecutorService retryScheduler;

public Builder(OpenSearchAsyncClient client, Listener listener) {
this.client = client;
Expand All @@ -268,6 +415,12 @@ public Builder setConcurrentRequests(int concurrentRequests) {
return this;
}

/** Test hook: schedules the retries on the given executor. */
Builder retryScheduler(ScheduledExecutorService retryScheduler) {
this.retryScheduler = retryScheduler;
return this;
}

public AsyncBulkProcessor build() {
return new AsyncBulkProcessor(this);
}
Expand Down
Loading
Loading