From 3a04d6438a9d509edad42d37340236cb3ba6d72c Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Fri, 16 Oct 2026 10:20:27 +0200 Subject: [PATCH 1/5] retry the bulk items rejected with status 429 AsyncBulkProcessor passed the items that OpenSearch rejected with 429 straight to the listener, so the bolts failed their tuples at once. The legacy BulkProcessor resends them with its default exponential backoff when every failed item of the response has status 429. AsyncBulkProcessor now resends the 429 items after the same delays (50, 60, 80, 150, 280, 580, 1250 and 2740 ms), also when other items failed with another status, for example a 409 from the create of a DISCOVERED URL that is already known. Each retry carries only the items still rejected, and the bulk keeps its concurrency permit until the listener has run. afterBulk is called once per bulk, with the items of all attempts in request order. If a retry request fails as a whole, the whole bulk fails, as on the first attempt. Retries run on their own thread, because the flush thread can be blocked waiting for the permit that a retrying bulk holds. awaitClose() waits for pending retries within its timeout and then drops the ones still waiting, as the legacy BulkProcessor does. Signed-off-by: Davide Polato --- .../opensearch/AsyncBulkProcessor.java | 180 ++++++-- .../AsyncBulkProcessorHttpTest.java | 154 +++++++ .../AsyncBulkProcessorRetryTest.java | 386 ++++++++++++++++++ 3 files changed, 689 insertions(+), 31 deletions(-) create mode 100644 external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorHttpTest.java create mode 100644 external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java 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 a6fd19cb2..0f15c6a76 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 @@ -29,10 +29,12 @@ 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.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; @@ -47,11 +49,17 @@ * 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. + * + *

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}; + /** Listener interface equivalent to the legacy {@code BulkProcessor.Listener}. */ public interface Listener { void beforeBulk(long executionId, BulkRequest request); @@ -74,6 +82,9 @@ public interface Listener { 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) { @@ -84,13 +95,11 @@ private AsyncBulkProcessor(Builder builder) { 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, @@ -99,6 +108,15 @@ private AsyncBulkProcessor(Builder builder) { TimeUnit.MILLISECONDS); } + 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} is reached. */ public void add(BulkOperation operation) { if (closed.get()) { @@ -164,36 +182,77 @@ private void executeBulk(List 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 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 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()}. @@ -222,13 +281,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 @@ -240,6 +303,54 @@ 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 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 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; @@ -247,6 +358,7 @@ public static final class Builder { private int bulkActions = 50; private long flushIntervalMillis = 5000; private int concurrentRequests = 1; + private ScheduledExecutorService retryScheduler; public Builder(OpenSearchAsyncClient client, Listener listener) { this.client = client; @@ -268,6 +380,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); } diff --git a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorHttpTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorHttpTest.java new file mode 100644 index 000000000..cae75ccae --- /dev/null +++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorHttpTest.java @@ -0,0 +1,154 @@ +/* + * 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.stormcrawler.opensearch; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; + +import java.io.IOException; +import java.net.InetAddress; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import org.apache.hc.core5.http.ContentType; +import org.apache.hc.core5.http.HttpHost; +import org.apache.hc.core5.http.impl.bootstrap.HttpServer; +import org.apache.hc.core5.http.impl.bootstrap.ServerBootstrap; +import org.apache.hc.core5.http.io.entity.EntityUtils; +import org.apache.hc.core5.http.io.entity.StringEntity; +import org.apache.hc.core5.io.CloseMode; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.mockito.ArgumentCaptor; +import org.opensearch.client.json.jackson.JacksonJsonpMapper; +import org.opensearch.client.opensearch.OpenSearchAsyncClient; +import org.opensearch.client.opensearch.core.BulkResponse; +import org.opensearch.client.opensearch.core.bulk.BulkOperation; +import org.opensearch.client.transport.OpenSearchTransport; +import org.opensearch.client.transport.httpclient5.ApacheHttpClient5TransportBuilder; + +/** + * Real client and HTTP transport against a local server that records the request bodies and answers + * with the queued responses, then with an empty bulk response. + */ +class AsyncBulkProcessorHttpTest { + + private final List bodies = new CopyOnWriteArrayList<>(); + + private final Queue responses = new ConcurrentLinkedQueue<>(); + + private final AsyncBulkProcessor.Listener listener = mock(AsyncBulkProcessor.Listener.class); + + private HttpServer server; + + private OpenSearchTransport transport; + + private AsyncBulkProcessor processor; + + @BeforeEach + void start() throws IOException { + server = + ServerBootstrap.bootstrap() + .setLocalAddress(InetAddress.getByName("127.0.0.1")) + .setListenerPort(0) + .setCanonicalHostName("127.0.0.1") + .register( + "*", + (request, response, context) -> { + bodies.add(EntityUtils.toByteArray(request.getEntity())); + String next = responses.poll(); + response.setCode(200); + response.setEntity( + new StringEntity( + next != null + ? next + : "{\"took\":1,\"errors\":false,\"items\":[]}", + ContentType.APPLICATION_JSON)); + }) + .create(); + server.start(); + transport = + ApacheHttpClient5TransportBuilder.builder( + new HttpHost("http", "127.0.0.1", server.getLocalPort())) + .setMapper(new JacksonJsonpMapper()) + .build(); + } + + @AfterEach + void stop() throws Exception { + if (processor != null) { + processor.awaitClose(100, TimeUnit.MILLISECONDS); + } + transport.close(); + server.close(CloseMode.IMMEDIATE); + } + + private AsyncBulkProcessor processor(int bulkActions) { + processor = + new AsyncBulkProcessor.Builder(new OpenSearchAsyncClient(transport), listener) + .setBulkActions(bulkActions) + .setFlushIntervalMillis(60_000) + .build(); + return processor; + } + + private List lines(int request) { + return new String(bodies.get(request), StandardCharsets.UTF_8).lines().toList(); + } + + private static BulkOperation index(String id) { + return BulkOperation.of(b -> b.index(i -> i.index("idx").id(id).document(Map.of("k", id)))); + } + + /** The first response rejects the second item with 429; the retry is accepted. */ + @Test + @Timeout(30) + void resendsOnlyTheRejectedItem() { + responses.add( + "{\"took\":1,\"errors\":true,\"items\":[" + + "{\"index\":{\"_index\":\"idx\",\"_id\":\"a\",\"status\":200}}," + + "{\"index\":{\"_index\":\"idx\",\"_id\":\"b\",\"status\":429," + + "\"error\":{\"type\":\"rejected_execution_exception\"," + + "\"reason\":\"rejected\"}}}]}"); + responses.add( + "{\"took\":1,\"errors\":false,\"items\":[" + + "{\"index\":{\"_index\":\"idx\",\"_id\":\"b\",\"status\":200}}]}"); + AsyncBulkProcessor p = processor(2); + p.add(index("a")); + p.add(index("b")); + + ArgumentCaptor response = ArgumentCaptor.forClass(BulkResponse.class); + verify(listener, timeout(10_000)).afterBulk(anyLong(), any(), response.capture()); + assertEquals(2, bodies.size()); + // action and document lines of "b" + assertEquals(lines(0).subList(2, 4), lines(1)); + assertEquals( + List.of("a:200", "b:200"), + response.getValue().items().stream().map(i -> i.id() + ":" + i.status()).toList()); + } +} diff --git a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java new file mode 100644 index 000000000..b06642ef0 --- /dev/null +++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java @@ -0,0 +1,386 @@ +/* + * 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.stormcrawler.opensearch; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.net.ConnectException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.tuple.Tuple; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.TestOutputCollector; +import org.apache.stormcrawler.TestUtil; +import org.apache.stormcrawler.opensearch.persistence.StatusUpdaterBolt; +import org.apache.stormcrawler.persistence.AbstractStatusUpdaterBolt; +import org.apache.stormcrawler.persistence.DefaultScheduler; +import org.apache.stormcrawler.persistence.Scheduler; +import org.apache.stormcrawler.persistence.Status; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.mockito.MockedStatic; +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.opensearch.client.opensearch.core.bulk.OperationType; + +/** Retry of items rejected with status 429. Runs without OpenSearch. */ +class AsyncBulkProcessorRetryTest { + + /** Each request gets a future that the test completes. */ + private final List requests = new CopyOnWriteArrayList<>(); + + private final List> futures = new CopyOnWriteArrayList<>(); + + /** Retry tasks with their delays, run by the test. */ + private final List retries = new CopyOnWriteArrayList<>(); + + private final List delays = new CopyOnWriteArrayList<>(); + + private final List outcomes = new CopyOnWriteArrayList<>(); + + private AsyncBulkProcessor.Listener delegate; + + private final AsyncBulkProcessor.Listener listener = + new AsyncBulkProcessor.Listener() { + @Override + public void beforeBulk(long id, BulkRequest request) { + if (delegate != null) { + delegate.beforeBulk(id, request); + } + } + + @Override + public void afterBulk(long id, BulkRequest request, BulkResponse response) { + outcomes.add(response); + if (delegate != null) { + delegate.afterBulk(id, request, response); + } + } + + @Override + public void afterBulk(long id, BulkRequest request, Throwable failure) { + outcomes.add(failure); + if (delegate != null) { + delegate.afterBulk(id, request, failure); + } + } + }; + + private final List processors = new ArrayList<>(); + + private final TestOutputCollector output = new TestOutputCollector(); + + @AfterEach + void close() throws Exception { + for (AsyncBulkProcessor p : processors) { + p.awaitClose(100, TimeUnit.MILLISECONDS); + } + } + + private AsyncBulkProcessor processor(int bulkActions, long flushMillis, boolean manualRetries) + throws IOException { + OpenSearchAsyncClient client = mock(OpenSearchAsyncClient.class); + when(client.bulk(any(BulkRequest.class))) + .thenAnswer( + inv -> { + CompletableFuture future = new CompletableFuture<>(); + requests.add(inv.getArgument(0)); + futures.add(future); + return future; + }); + AsyncBulkProcessor.Builder builder = + new AsyncBulkProcessor.Builder(client, listener) + .setBulkActions(bulkActions) + .setFlushIntervalMillis(flushMillis); + if (manualRetries) { + ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class); + when(scheduler.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))) + .thenAnswer( + inv -> { + retries.add(inv.getArgument(0)); + delays.add(inv.getArgument(1)); + return mock(ScheduledFuture.class); + }); + builder.retryScheduler(scheduler); + } + AsyncBulkProcessor p = builder.build(); + processors.add(p); + return p; + } + + private void reply(int call, int... statuses) { + futures.get(call).complete(response(requests.get(call), statuses)); + } + + private void runRetry() { + retries.remove(0).run(); + } + + private List ids(int call) { + return requests.get(call).operations().stream() + .map(OpenSearchConnection::getBulkOperationId) + .toList(); + } + + private static List statuses(Object response) { + return ((BulkResponse) response) + .items().stream().map(i -> i.id() + ":" + i.status()).toList(); + } + + private static BulkOperation index(String id) { + return BulkOperation.of(b -> b.index(i -> i.index("idx").id(id).document(Map.of("k", id)))); + } + + /** One item per operation, in request order. */ + private static BulkResponse response(BulkRequest request, int... statuses) { + List items = new ArrayList<>(); + for (int i = 0; i < statuses.length; i++) { + BulkOperation op = request.operations().get(i); + int status = statuses[i]; + items.add( + BulkResponseItem.of( + b -> { + b.operationType(OperationType.valueOf(op._kind().name())) + .index("idx") + .id(OpenSearchConnection.getBulkOperationId(op)) + .status(status); + if (status >= 300) { + b.error(e -> e.type("rejected_execution_exception")); + } + return b; + })); + } + boolean errors = Arrays.stream(statuses).anyMatch(s -> s >= 300); + return BulkResponse.of(b -> b.took(1).errors(errors).items(items)); + } + + @Test + void onlyItemsRejectedWith429AreResentAndTheMergedResponseKeepsRequestOrder() + throws IOException { + AsyncBulkProcessor p = processor(3, 60_000, true); + p.add(index("a")); + p.add(index("b")); + p.add(index("c")); + + reply(0, 429, 409, 429); + runRetry(); + assertEquals(List.of("a", "c"), ids(1)); + reply(1, 200, 429); + runRetry(); + assertEquals(List.of("c"), ids(2)); + reply(2, 200); + + assertEquals(1, outcomes.size()); + assertEquals(List.of("a:200", "b:409", "c:200"), statuses(outcomes.get(0))); + assertTrue(((BulkResponse) outcomes.get(0)).errors()); + assertEquals(List.of(50L, 60L), delays); + } + + @Test + void transportErrorDuringARetryFailsTheWholeBulk() throws IOException { + AsyncBulkProcessor p = processor(2, 60_000, true); + p.add(index("a")); + p.add(index("b")); + + reply(0, 200, 429); + runRetry(); + futures.get(1).completeExceptionally(new ConnectException("refused")); + + assertEquals(1, outcomes.size()); + assertInstanceOf(ConnectException.class, outcomes.get(0)); + } + + @Test + @Timeout(10) + void failureWhileHandlingTheResponseFailsTheBulkAndReleasesThePermit() throws IOException { + AsyncBulkProcessor p = processor(1, 60_000, true); + p.add(index("a")); + BulkResponse broken = mock(BulkResponse.class); + IllegalStateException failure = new IllegalStateException("broken response"); + when(broken.items()).thenThrow(failure); + futures.get(0).complete(broken); + p.add(index("b")); + + assertSame(failure, outcomes.get(0)); + assertEquals(2, requests.size()); + } + + /** + * A bulk waiting for its retry keeps the only permit, so the timer flush of the next bulk + * blocks the flush thread. The second retry is scheduled while that thread is blocked and still + * runs, on the processor's own retry thread; the next bulk goes out after the listener. + */ + @Test + @Timeout(10) + void permitIsHeldDuringTheBackoffWithoutBlockingTheRetry() throws Exception { + AsyncBulkProcessor p = processor(2, 300, false); + p.add(index("a1")); + p.add(index("a2")); + reply(0, 200, 429); + await().atMost(Duration.ofSeconds(5)).until(() -> requests.size() == 2); + p.add(index("b1")); + await().atMost(Duration.ofSeconds(5)) + .until(AsyncBulkProcessorRetryTest::flushThreadWaitsForAPermit); + + reply(1, 429); + await().atMost(Duration.ofSeconds(5)).until(() -> requests.size() == 3); + assertEquals(List.of("a2"), ids(2)); + reply(2, 200); + + await().atMost(Duration.ofSeconds(5)).until(() -> requests.size() == 4); + assertEquals(List.of("b1"), ids(3)); + assertEquals(List.of("a1:200", "a2:200"), statuses(outcomes.get(0))); + } + + private static boolean flushThreadWaitsForAPermit() { + return Thread.getAllStackTraces().keySet().stream() + .anyMatch( + t -> + t.getName().equals("AsyncBulkProcessor-flush") + && t.getState() == Thread.State.WAITING); + } + + @Test + @Timeout(10) + void closeWaitsForAPendingRetry() throws Exception { + AsyncBulkProcessor p = processor(1, 60_000, false); + p.add(index("a")); + reply(0, 429); + CompletableFuture.runAsync( + () -> { + await().atMost(Duration.ofSeconds(5)).until(() -> requests.size() == 2); + reply(1, 200); + }); + + assertTrue(p.awaitClose(5, TimeUnit.SECONDS)); + assertEquals(List.of("a:200"), statuses(outcomes.get(0))); + } + + /** + * StatusUpdaterBolt with the real processor and WaitAckCache. OpenSearchConnection is mocked + * statically so that prepare() gets a connection which feeds the processor, and the listener + * the bolt passes to it is captured as {@code delegate}, so that the outcomes reach the bolt. + */ + private StatusUpdaterBolt bolt(int bulkActions) throws IOException { + AsyncBulkProcessor p = processor(bulkActions, 60_000, true); + OpenSearchConnection connection = mock(OpenSearchConnection.class); + doAnswer( + inv -> { + p.add(inv.getArgument(0)); + return null; + }) + .when(connection) + .addToProcessor(any(BulkOperation.class)); + Map conf = new HashMap<>(); + conf.put(Scheduler.schedulerClassParamName, DefaultScheduler.class.getName()); + conf.put( + AbstractStatusUpdaterBolt.cacheConfigParamName, + "maximumSize=10000,expireAfterAccess=1h"); + StatusUpdaterBolt bolt = new StatusUpdaterBolt(); + try (MockedStatic factory = mockStatic(OpenSearchConnection.class); + MockedStatic ignored = mockStatic(IndexCreation.class)) { + factory.when(() -> OpenSearchConnection.getBulkOperationId(any())).thenCallRealMethod(); + factory.when(() -> OpenSearchConnection.getConnection(any(), anyString(), any())) + .thenAnswer( + inv -> { + delegate = inv.getArgument(2); + return connection; + }); + bolt.prepare(conf, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + } + return bolt; + } + + private static Tuple tuple(String url) { + return TestUtil.getMockedTestTuple( + Map.of("url", url, "status", Status.FETCHED, "metadata", new Metadata())); + } + + @Test + void retried429IsAckedAfterSuccess() throws IOException { + StatusUpdaterBolt bolt = bolt(1); + Tuple t = tuple("https://a/"); + bolt.execute(t); + + reply(0, 429); + assertTrue(output.getAckedTuples().isEmpty() && output.getFailedTuples().isEmpty()); + runRetry(); + reply(1, 200); + + assertEquals(List.of(t), output.getAckedTuples()); + } + + @Test + void tupleFailsWhenRetriesRunOut() throws IOException { + StatusUpdaterBolt bolt = bolt(1); + Tuple t = tuple("https://a/"); + bolt.execute(t); + + reply(0, 429); + for (int call = 1; call <= 8; call++) { + runRetry(); + reply(call, 429); + } + + assertEquals(List.of(50L, 60L, 80L, 150L, 280L, 580L, 1250L, 2740L), delays); + assertEquals(List.of(t), output.getFailedTuples()); + } + + /** Two tuples for one URL in one bulk: WaitAckCache acks both when one write succeeded. */ + @Test + void duplicateIdIsAckedWhenOneWriteSucceeds() throws IOException { + StatusUpdaterBolt bolt = bolt(2); + bolt.execute(tuple("https://a/")); + bolt.execute(tuple("https://a/")); + + reply(0, 200, 429); + for (int call = 1; call <= 8; call++) { + runRetry(); + assertEquals(1, requests.get(call).operations().size()); + reply(call, 429); + } + + assertEquals(2, output.getAckedTuples().size()); + } +} From bf7b5a2ff7f60797cd4576ca4a9deb0293576dc5 Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Fri, 16 Oct 2026 10:21:44 +0200 Subject: [PATCH 2/5] flush the bulk requests by size The legacy BulkProcessor sends a bulk once its estimated size reaches 5 MiB. AsyncBulkProcessor flushed by count and time only, so large documents could produce very large requests. The README claimed size-based flushing. When an operation is added, AsyncBulkProcessor now computes the bytes that the transport writes for it: each NDJSON line serialized with the transport's mapper, before compression, plus a newline. This serializes each operation one more time, on the caller's thread. The bulk is sent once the buffered operations reach 5 MiB, including the operation that crosses the threshold. The README now lists what AsyncBulkProcessor does. Signed-off-by: Davide Polato --- external/opensearch-java/README.md | 26 +++--- .../opensearch/AsyncBulkProcessor.java | 43 +++++++++- .../AsyncBulkProcessorHttpTest.java | 79 +++++++++++++++++++ .../AsyncBulkProcessorRetryTest.java | 1 + .../opensearch/AsyncBulkProcessorTest.java | 12 +++ 5 files changed, 147 insertions(+), 14 deletions(-) diff --git a/external/opensearch-java/README.md b/external/opensearch-java/README.md index 5e0c33b80..d1cf8e8eb 100644 --- a/external/opensearch-java/README.md +++ b/external/opensearch-java/README.md @@ -10,16 +10,22 @@ 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..bulkActions`, when the buffered operations reach 5 MiB of + JSON (counted before compression), or when `opensearch..flushInterval` + elapses. +* resends the items that OpenSearch rejects with status 429 (too many requests) + up to 8 times, after 50, 60, 80, 150, 280, 580, 1250 and 2740 ms. + `IndexerBolt`, `DeletionBolt` and `StatusUpdaterBolt` fail the tuple of an + item still rejected after the last retry. +* sends the requests without blocking the bolt. The bolt waits only when + `opensearch..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 --------------------- 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 0f15c6a76..85866fa09 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 @@ -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; @@ -30,6 +34,7 @@ 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; @@ -41,8 +46,8 @@ /** * 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. * *

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 @@ -60,6 +65,9 @@ public final class AsyncBulkProcessor implements AutoCloseable { 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); @@ -70,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; @@ -78,6 +87,7 @@ public interface Listener { private final ReentrantLock lock = new ReentrantLock(); private List buffer; + private long bufferedBytes; private final ScheduledExecutorService scheduler; private final ScheduledFuture flushTask; @@ -89,6 +99,7 @@ public interface Listener { 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; @@ -117,16 +128,22 @@ private static ScheduledExecutorService daemonScheduler(String name) { }); } - /** Adds a single bulk operation. Triggers a flush when {@code bulkActions} is reached. */ + /** + * 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 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 { @@ -160,9 +177,27 @@ private void flushIfNeeded() { private List swapBuffer() { List 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 operations) { final long executionId = executionIdGen.incrementAndGet(); diff --git a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorHttpTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorHttpTest.java index cae75ccae..a9c8cc554 100644 --- a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorHttpTest.java +++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorHttpTest.java @@ -17,7 +17,9 @@ package org.apache.stormcrawler.opensearch; +import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; @@ -27,6 +29,9 @@ import java.io.IOException; import java.net.InetAddress; import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Arrays; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Queue; @@ -45,8 +50,10 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; import org.mockito.ArgumentCaptor; +import org.opensearch.client.json.JsonData; import org.opensearch.client.json.jackson.JacksonJsonpMapper; import org.opensearch.client.opensearch.OpenSearchAsyncClient; +import org.opensearch.client.opensearch._types.Script; import org.opensearch.client.opensearch.core.BulkResponse; import org.opensearch.client.opensearch.core.bulk.BulkOperation; import org.opensearch.client.transport.OpenSearchTransport; @@ -151,4 +158,76 @@ void resendsOnlyTheRejectedItem() { List.of("a:200", "b:200"), response.getValue().items().stream().map(i -> i.id() + ":" + i.status()).toList()); } + + @Test + @Timeout(30) + void countedSizeEqualsTheBytesTheTransportSends() { + AsyncBulkProcessor p = processor(1); + Map metadata = new LinkedHashMap<>(); + metadata.put("list", Arrays.asList(null, "ünïcödé 日本語 😀")); + Map doc = new LinkedHashMap<>(); + doc.put("content", "line1\nline2\t\u0001 \"q\" \\"); + doc.put("metadata", metadata); + Script script = + Script.of( + s -> + s.inline( + in -> + in.source("ctx._source.n += params.n") + .params("n", JsonData.of(5)))); + List operations = + List.of( + BulkOperation.of(b -> b.index(i -> i.index("idx").id("é").document(doc))), + BulkOperation.of(b -> b.create(c -> c.index("idx").id("c").document(doc))), + BulkOperation.of(b -> b.update(u -> u.index("idx").id("v").document(doc))), + BulkOperation.of( + b -> + b.update( + u -> + u.index("idx") + .id("u") + .document(doc) + .upsert(doc))), + BulkOperation.of( + b -> + b.update( + u -> + u.index("idx") + .id("s") + .script(script) + .upsert(doc))), + BulkOperation.of(b -> b.delete(d -> d.index("idx").id("d")))); + + for (int i = 0; i < operations.size(); i++) { + p.add(operations.get(i)); + final int sent = i + 1; + await().atMost(Duration.ofSeconds(5)).until(() -> bodies.size() == sent); + assertEquals( + bodies.get(i).length, + AsyncBulkProcessor.serializedSize(new JacksonJsonpMapper(), operations.get(i))); + } + } + + @Test + @Timeout(30) + void flushesWhenTheBufferedBytesReachTheThreshold() { + AsyncBulkProcessor p = processor(1000); + Map doc = Map.of("content", "x".repeat(1_000_000)); + for (int i = 0; i < 12; i++) { + final String id = Integer.toString(i); + p.add(BulkOperation.of(b -> b.index(x -> x.index("idx").id(id).document(doc)))); + // five documents of 1 MB stay below 5 MiB, the sixth crosses it + if (i == 4) { + assertTrue(bodies.isEmpty(), "flushed before reaching 5 MiB"); + } + if (i == 5) { + await().atMost(Duration.ofSeconds(10)).until(() -> bodies.size() == 1); + } + } + await().atMost(Duration.ofSeconds(10)).until(() -> bodies.size() == 2); + for (byte[] body : bodies) { + assertEquals(12, new String(body, StandardCharsets.UTF_8).lines().count()); + assertTrue(body.length >= AsyncBulkProcessor.BULK_SIZE_BYTES); + } + } } diff --git a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java index b06642ef0..3ec00d0aa 100644 --- a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java +++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java @@ -121,6 +121,7 @@ void close() throws Exception { private AsyncBulkProcessor processor(int bulkActions, long flushMillis, boolean manualRetries) throws IOException { OpenSearchAsyncClient client = mock(OpenSearchAsyncClient.class); + AsyncBulkProcessorTest.stubMapper(client); when(client.bulk(any(BulkRequest.class))) .thenAnswer( inv -> { 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 fefcca56c..0c91e5a88 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 @@ -31,10 +31,12 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; +import org.opensearch.client.json.jackson.JacksonJsonpMapper; 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.transport.OpenSearchTransport; class AsyncBulkProcessorTest { @@ -50,8 +52,16 @@ private static BulkResponse emptyBulkResponse() { .build(); } + /** The processor takes the JSON mapper from the client's transport. */ + static void stubMapper(OpenSearchAsyncClient client) { + OpenSearchTransport transport = mock(OpenSearchTransport.class); + when(transport.jsonpMapper()).thenReturn(new JacksonJsonpMapper()); + when(client._transport()).thenReturn(transport); + } + private static OpenSearchAsyncClient mockClient() throws IOException { OpenSearchAsyncClient client = mock(OpenSearchAsyncClient.class); + stubMapper(client); when(client.bulk(any(BulkRequest.class))) .thenAnswer(invocation -> CompletableFuture.supplyAsync(() -> emptyBulkResponse())); return client; @@ -153,6 +163,7 @@ void concurrentRequestLimiting() throws Exception { CountDownLatch allDone = new CountDownLatch(3); OpenSearchAsyncClient client = mock(OpenSearchAsyncClient.class); + stubMapper(client); when(client.bulk(any(BulkRequest.class))) .thenAnswer( invocation -> { @@ -259,6 +270,7 @@ void addDoesNotWaitForBulkResponse() throws Exception { CountDownLatch afterBulkLatch = new CountDownLatch(1); OpenSearchAsyncClient client = mock(OpenSearchAsyncClient.class); + stubMapper(client); when(client.bulk(any(BulkRequest.class))).thenReturn(pending); AsyncBulkProcessor.Listener listener = From 2bccebb03203f92fb7e404cdf98c8cb755fe5b4e Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Mon, 19 Oct 2026 10:29:53 +0200 Subject: [PATCH 3/5] drop a status tuple from WaitAckCache when adding it to the bulk fails AsyncBulkProcessor.add() serializes the operation and can throw. StatusUpdaterBolt put the tuple in WaitAckCache before calling it, so when add() threw, AbstractStatusUpdaterBolt failed the tuple while its entry stayed in the cache. A later bulk for the same document ID then acked the failed tuple, or the cache expiry failed it a second time. StatusUpdaterBolt now invalidates the entry when add() throws, as IndexerBolt already does. Signed-off-by: Davide Polato --- .../persistence/StatusUpdaterBolt.java | 8 +++++- .../AsyncBulkProcessorRetryTest.java | 26 +++++++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java index bc0c59ccf..86f373fee 100644 --- a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java +++ b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/persistence/StatusUpdaterBolt.java @@ -298,7 +298,13 @@ public void store( LOG.debug("Sending to OpenSearch buffer {} with ID {}", url, documentID); - connection.addToProcessor(op); + try { + connection.addToProcessor(op); + } catch (Exception e) { + // the caller fails the tuple, so no bulk response may ack or fail it again + waitAck.invalidate(documentID); + throw e; + } } @Override diff --git a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java index 3ec00d0aa..3dfa5718c 100644 --- a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java +++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java @@ -111,6 +111,12 @@ public void afterBulk(long id, BulkRequest request, Throwable failure) { private final TestOutputCollector output = new TestOutputCollector(); + /** + * Thrown by the next addToProcessor() of the bolt connection, the way a serialization error + * would be. + */ + private RuntimeException addFailure; + @AfterEach void close() throws Exception { for (AsyncBulkProcessor p : processors) { @@ -308,6 +314,11 @@ private StatusUpdaterBolt bolt(int bulkActions) throws IOException { OpenSearchConnection connection = mock(OpenSearchConnection.class); doAnswer( inv -> { + if (addFailure != null) { + RuntimeException e = addFailure; + addFailure = null; + throw e; + } p.add(inv.getArgument(0)); return null; }) @@ -368,6 +379,21 @@ void tupleFailsWhenRetriesRunOut() throws IOException { assertEquals(List.of(t), output.getFailedTuples()); } + @Test + void tupleFailedWhileAddingIsNotAckedByALaterWrite() throws IOException { + StatusUpdaterBolt bolt = bolt(1); + Tuple failed = tuple("https://a/"); + addFailure = new IllegalStateException("cannot serialize"); + bolt.execute(failed); + Tuple next = tuple("https://a/"); + bolt.execute(next); + + reply(0, 200); + + assertEquals(List.of(failed), output.getFailedTuples()); + assertEquals(List.of(next), output.getAckedTuples()); + } + /** Two tuples for one URL in one bulk: WaitAckCache acks both when one write succeeded. */ @Test void duplicateIdIsAckedWhenOneWriteSucceeds() throws IOException { From 410382c2b681a99767d767ae1e74b6dabaca119f Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Mon, 19 Oct 2026 10:30:11 +0200 Subject: [PATCH 4/5] document which bulk failures are not retried The README now says that only the items rejected with 429 in the bulk response are resent, and that a retry that fails as a whole, for example on a connection error, makes IndexerBolt, DeletionBolt and StatusUpdaterBolt fail the tuples of the whole bulk. The duplicate ID test says why both tuples are acked. Signed-off-by: Davide Polato --- external/opensearch-java/README.md | 12 ++++++++---- .../opensearch/AsyncBulkProcessorRetryTest.java | 5 ++++- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/external/opensearch-java/README.md b/external/opensearch-java/README.md index d1cf8e8eb..4b63e0f0c 100644 --- a/external/opensearch-java/README.md +++ b/external/opensearch-java/README.md @@ -17,10 +17,14 @@ operations into bulk requests and: `opensearch..bulkActions`, when the buffered operations reach 5 MiB of JSON (counted before compression), or when `opensearch..flushInterval` elapses. -* resends the items that OpenSearch rejects with status 429 (too many requests) - up to 8 times, after 50, 60, 80, 150, 280, 580, 1250 and 2740 ms. - `IndexerBolt`, `DeletionBolt` and `StatusUpdaterBolt` fail the tuple of an - item still rejected after the last retry. +* 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..concurrentRequests` bulks are in flight, including bulks waiting for a retry. diff --git a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java index 3dfa5718c..1a79c41a6 100644 --- a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java +++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java @@ -394,7 +394,10 @@ void tupleFailedWhileAddingIsNotAckedByALaterWrite() throws IOException { assertEquals(List.of(next), output.getAckedTuples()); } - /** Two tuples for one URL in one bulk: WaitAckCache acks both when one write succeeded. */ + /** + * Two tuples for one URL in one bulk: WaitAckCache acks both when one write succeeded, although + * the second write was never stored. + */ @Test void duplicateIdIsAckedWhenOneWriteSucceeds() throws IOException { StatusUpdaterBolt bolt = bolt(2); From 026ae4492f79a9e92feb43a7137c85f4830ca563 Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Tue, 20 Oct 2026 16:01:16 +0200 Subject: [PATCH 5/5] fail a deletion tuple when adding it to the bulk fails AsyncBulkProcessor.add() serializes the operation and can throw. DeletionBolt called it without a try/catch, so the exception escaped execute(), which kills the Storm worker, and the tuple's entry stayed in WaitAckCache. DeletionBolt now logs the error, invalidates the entry and fails the tuple, as IndexerBolt and StatusUpdaterBolt do. Signed-off-by: Davide Polato --- .../opensearch/bolt/DeletionBolt.java | 8 +++++- .../AsyncBulkProcessorRetryTest.java | 26 ++++++++++++++++--- 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/bolt/DeletionBolt.java b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/bolt/DeletionBolt.java index f77662ea6..f7f0f1730 100644 --- a/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/bolt/DeletionBolt.java +++ b/external/opensearch-java/src/main/java/org/apache/stormcrawler/opensearch/bolt/DeletionBolt.java @@ -111,7 +111,13 @@ public void execute(Tuple tuple) { waitAck.addTuple(docID, tuple); - connection.addToProcessor(op); + try { + connection.addToProcessor(op); + } catch (Exception e) { + LOG.error("Error adding the deletion of {} to the bulk", url, e); + waitAck.invalidate(docID); + _collector.fail(tuple); + } } @Override diff --git a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java index 1a79c41a6..f5ff10be5 100644 --- a/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java +++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java @@ -43,11 +43,13 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import org.apache.storm.task.IBolt; import org.apache.storm.task.OutputCollector; import org.apache.storm.tuple.Tuple; import org.apache.stormcrawler.Metadata; import org.apache.stormcrawler.TestOutputCollector; import org.apache.stormcrawler.TestUtil; +import org.apache.stormcrawler.opensearch.bolt.DeletionBolt; import org.apache.stormcrawler.opensearch.persistence.StatusUpdaterBolt; import org.apache.stormcrawler.persistence.AbstractStatusUpdaterBolt; import org.apache.stormcrawler.persistence.DefaultScheduler; @@ -304,12 +306,16 @@ void closeWaitsForAPendingRetry() throws Exception { assertEquals(List.of("a:200"), statuses(outcomes.get(0))); } + private StatusUpdaterBolt bolt(int bulkActions) throws IOException { + return prepared(new StatusUpdaterBolt(), bulkActions); + } + /** - * StatusUpdaterBolt with the real processor and WaitAckCache. OpenSearchConnection is mocked + * The bolt prepared with the real processor and WaitAckCache. OpenSearchConnection is mocked * statically so that prepare() gets a connection which feeds the processor, and the listener * the bolt passes to it is captured as {@code delegate}, so that the outcomes reach the bolt. */ - private StatusUpdaterBolt bolt(int bulkActions) throws IOException { + private B prepared(B bolt, int bulkActions) throws IOException { AsyncBulkProcessor p = processor(bulkActions, 60_000, true); OpenSearchConnection connection = mock(OpenSearchConnection.class); doAnswer( @@ -329,7 +335,6 @@ private StatusUpdaterBolt bolt(int bulkActions) throws IOException { conf.put( AbstractStatusUpdaterBolt.cacheConfigParamName, "maximumSize=10000,expireAfterAccess=1h"); - StatusUpdaterBolt bolt = new StatusUpdaterBolt(); try (MockedStatic factory = mockStatic(OpenSearchConnection.class); MockedStatic ignored = mockStatic(IndexCreation.class)) { factory.when(() -> OpenSearchConnection.getBulkOperationId(any())).thenCallRealMethod(); @@ -394,6 +399,21 @@ void tupleFailedWhileAddingIsNotAckedByALaterWrite() throws IOException { assertEquals(List.of(next), output.getAckedTuples()); } + @Test + void deletionFailedWhileAddingIsNotAckedByALaterDeletion() throws IOException { + DeletionBolt bolt = prepared(new DeletionBolt(), 1); + Tuple failed = tuple("https://a/"); + addFailure = new IllegalStateException("cannot serialize"); + bolt.execute(failed); + Tuple next = tuple("https://a/"); + bolt.execute(next); + + reply(0, 200); + + assertEquals(List.of(failed), output.getFailedTuples()); + assertEquals(List.of(next), output.getAckedTuples()); + } + /** * Two tuples for one URL in one bulk: WaitAckCache acks both when one write succeeded, although * the second write was never stored.