diff --git a/external/opensearch-java/README.md b/external/opensearch-java/README.md index 5e0c33b80..4b63e0f0c 100644 --- a/external/opensearch-java/README.md +++ b/external/opensearch-java/README.md @@ -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..bulkActions`, when the buffered operations reach 5 MiB of + JSON (counted before compression), or when `opensearch..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..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 a6fd19cb2..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; @@ -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. * *

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. + * + *

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); @@ -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; @@ -70,27 +87,30 @@ public interface Listener { private final ReentrantLock lock = new ReentrantLock(); private List 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, @@ -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 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 { @@ -142,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(); @@ -164,36 +217,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 +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 @@ -240,6 +338,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 +393,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 +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); } 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/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/AsyncBulkProcessorHttpTest.java b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorHttpTest.java new file mode 100644 index 000000000..a9c8cc554 --- /dev/null +++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorHttpTest.java @@ -0,0 +1,233 @@ +/* + * 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.assertTrue; +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.time.Duration; +import java.util.Arrays; +import java.util.LinkedHashMap; +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.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; +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()); + } + + @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 new file mode 100644 index 000000000..f5ff10be5 --- /dev/null +++ b/external/opensearch-java/src/test/java/org/apache/stormcrawler/opensearch/AsyncBulkProcessorRetryTest.java @@ -0,0 +1,436 @@ +/* + * 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.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; +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(); + + /** + * 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) { + p.awaitClose(100, TimeUnit.MILLISECONDS); + } + } + + 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 -> { + 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))); + } + + private StatusUpdaterBolt bolt(int bulkActions) throws IOException { + return prepared(new StatusUpdaterBolt(), bulkActions); + } + + /** + * 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 B prepared(B bolt, int bulkActions) throws IOException { + AsyncBulkProcessor p = processor(bulkActions, 60_000, true); + OpenSearchConnection connection = mock(OpenSearchConnection.class); + doAnswer( + inv -> { + if (addFailure != null) { + RuntimeException e = addFailure; + addFailure = null; + throw e; + } + 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"); + 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()); + } + + @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()); + } + + @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. + */ + @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()); + } +} 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 =