[dpol1]

@rzo1 asked for 429 retries and flushing by size before we move opensearch-java into external/opensearch (#2212). This PR adds both to AsyncBulkProcessor. The move will follow in a second PR on top of this one.

Today an item rejected with 429 goes straight to the listener, so StatusUpdaterBolt, IndexerBolt and DeletionBolt fail its tuple, and bulks are flushed by count or time only. With this change:

• Items rejected with 429 are resent with the delays of the legacy BulkProcessor default backoff, 8 retries from 50 ms up to 2.7 s. Each retry carries only those items.
• The bulk keeps its permit during the backoff, and afterBulk is called once, with the items of all attempts in request order.
• A bulk is sent once its operations reach 5 MiB, the legacy default. The size is the NDJSON the transport writes, before compression.

There is one difference with the legacy module. It retries only when every failed item of the response is a 429, so a single 409, e.g. a create for a DISCOVERED URL that already exists, stops the retry of the whole bulk. Here the 429 items are resent whatever happened to the others.

I didn't use BulkIngester. It releases its in-flight slot before the listener has run and holds none while items wait for a retry, the opposite of what #2225 did for AsyncBulkProcessor. Its listener is also called once per attempt. In a local test with two writes for the same URL in one bulk (#832), a 200 then a 429, WaitAckCache acked the tuples before the second write was resent.

A few things worth a look:

• Retries run on their own thread, because the flush thread can be blocked waiting for the permit of a retrying bulk.
• Counting the size serializes each operation a second time, on the bolt's thread. A serialization error is now thrown by add(), where on main it failed the whole bulk.
• The final flush in awaitClose() takes a permit with no timeout, as in the legacy module, so while bulks are retrying it can wait about 5 s of delays plus the round trips. Retries still pending when awaitClose() times out are dropped, also as in the legacy module. The bolts that track acks have already failed those tuples in cleanup().

The new tests don't need Docker and take about 5 s, and the module suite passes locally with Docker. I haven't yet run a crawl against an OpenSearch loaded enough to return 429s, nor StatusUpdaterBenchmark against main.

[rzo1]

Thanks, looks good to me.

A few small things:

• In StatusUpdaterBolt, waitAck.addTuple() runs before connection.addToProcessor(op). Now that add() can throw on a serialization error, the tuple gets failed but its entry stays in the cache. A later bulk for the same URL would then ack a tuple that's already been failed, or the cache expiry would fail it a second time. IndexerBolt already calls waitAck.invalidate(docID) in its catch block. Could you do the same here?
• The README says items rejected with 429 are resent. If OpenSearch rejects the whole request with a 429, it isn't retried, same as in the legacy module. Maybe say "items rejected with 429 in the bulk response", so nobody expects more than that.
• If a retry hits a connection error, items that already got a 200 are failed as well. The legacy module does the same, and your test covers it, but one line in the README would help.
• In duplicateIdIsAckedWhenOneWriteSucceeds, both tuples get acked even though the second write never went through. That's how it already worked since #832, so it's fine, but a short comment in the test saying so would save the next reader some head scratching.

[dpol1]

@rzo1 I was going to ping you after the weekend tests, but you beat me to it :) pushed the fixes, thanks! StatusUpdaterBolt now drops the cache entry when add() throws, like IndexerBolt does, plus a test for it. README now also covers whole-request rejections and retries that fail as a whole, and the duplicate test has its comment.

[jnioche]

thanks @dpol1, 2 observations:

Every operation is now serialized twice. The second time happens on the bolt’s thread, for size counting. That’s cheap for status and delete operations, but IndexerBolt sends full page text, so it may cost noticeable CPU. The legacy module estimated the size instead of serializing.

DeletionBolt doesn’t handle the new exception from add(). At DeletionBolt.java:112-114 it calls waitAck.addTuple and then addToProcessor with no try/catch. A serialization error now escapes execute(), which in Storm kills the worker, and leaves a stale entry in WaitAckCache. Serializing a delete operation is unlikely to fail, but for consistency it should do what IndexerBolt and now StatusUpdaterBolt do: invalidate the entry, then fail the tuple.

[dpol1]

@jnioche thanks, DeletionBolt now invalidates the entry and fails the tuple, with a test.

On the double serialization: in a microbenchmark on my laptop the size count costs about 1.5 µs per KB of JSON, so 16 µs for a page with 10 KB of text and 160 µs for 100 KB. That's roughly 70% of what the transport spends serializing the same operation. I'd keep the exact count so the 5 MiB limit matches the bulk body. If you prefer an estimate, as the legacy module did, I can do it in a follow-up.