dpol1 opened a new pull request, #2241: URL: https://github.com/apache/stormcrawler/pull/2241
@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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
