abhinav-phi opened a new pull request, #2117: URL: https://github.com/apache/stormcrawler/pull/2117
## Summary URLFrontier [2.6](https://github.com/crawler-commons/url-frontier/releases/tag/2.6) added a batched endpoint for ingesting discovered URLs, `PutDiscovered(stream DiscoveredBatch) returns (stream BatchAck)`. The per-message cost of the streaming `PutURLs` endpoint is what caps the ingestion rate, and discovered URLs — the outlinks of the pages being parsed — are the bulk of what a crawl writes. This PR routes them through the batched endpoint, grouping up to `urlfrontier.batch.size` URLs (default 100) into one message, while known URLs (fetched, redirections, errors...) keep using the streaming mode, as suggested in #2062. The control-flow mechanism follows the best practices of the client shipped with URLFrontier ([`PutURLs.java`](https://github.com/crawler-commons/url-frontier/blob/master/client/src/main/java/crawlercommons/urlfrontier/client/PutURLs.java)). Fixes #2062 ## What the bolt does now `StatusUpdaterBolt` keeps its overall structure: a `waitAck` cache holds the tuples per URL until the frontier acks them, and a semaphore bounds the messages in flight. What changes is the path each URL takes: **Discovered URLs → `PutDiscovered`.** A DISCOVERED URL is appended to a buffer instead of being sent on its own. The buffer leaves as one `DiscoveredBatch` message when: - it reaches `urlfrontier.batch.size` entries; - a known URL arrives — the outlinks buffered so far belong to the page whose status is now being updated, a natural end-of-page boundary for the batch (the release notes for 2.6 explicitly call the outlinks of a page a natural batch); - a partially filled batch has been open for more than a second (checked by a scheduled flusher every 100 ms), so that acks are not delayed when the crawl tails off; - the bolt is throttled because the semaphore is exhausted — pushing the batch out frees the permits sooner. Each batch carries a unique ID (`batch-N`); the frontier echoes it in the `BatchAck` with **one status per URL, in the order they were sent**, so every URL can be resolved against `waitAck` and its tuples acked (or failed on `FAIL`) exactly as the streaming path did. Permits are released in bulk per URL, unchanged. **Known URLs → `PutURLs`**, one message per URL, untouched behaviour. **Flow control without polling.** `store()` used to busy-poll `tryAcquire` with a sleep. It now waits on a monitor (`flow`) and is woken by `notifyAll()` from the places that change its conditions: permit releases (acks, evictions), transport on-ready notifications, and stream errors. The wait timeout (`urlfrontier.throttling.time.msec`, unchanged, 10 ms) is a backstop that also drives the periodic `waitAck.cleanUp()` which prevents the deadlock the old code guarded against. This mirrors the `isReady()`/on-ready pattern of the reference client instead of polling. **Fallback for older frontiers.** If the frontier does not implement `PutDiscovered` (i.e. predates 2.6), the stream fails with `UNIMPLEMENTED`. The bolt detects this, switches batching off permanently for its lifetime, and re-sends whatever was buffered or still in flight individually on the streaming endpoint — no URLs are lost, and a pre-2.6 frontier behaves exactly as before this PR. A dead batch stream (any other error) drops the pending batch IDs and opens a fresh stream on the next flush; the affected tuples are failed locally so Storm replays them. ## Configuration New key, documented in the module README: ```yaml # max number of discovered URLs per PutDiscovered message (default 100, 0 disables batching) urlfrontier.batch.size: 100 ``` The module now depends on `urlfrontier-API` **2.6** (bumped from 2.5). The schema stays wire-compatible with 2.5 apart from the semantics noted in the 2.6 release notes, and the fallback covers servers that do not implement the new RPC, so no frontier upgrade is strictly required — though the batching only pays off against a 2.6 server. ## Tests - `StatusUpdaterBoltTest.acknowledgesDiscoveredURLsSentInBatches` — 6 discovered URLs against a 2.6 frontier with batch size 2: all acked, ≥ 3 batches sent, nothing failed; - `StatusUpdaterBoltTest.sendsDiscoveredURLsIndividuallyWhenBatchingDisabled` — `urlfrontier.batch.size: 0` sends individually and never opens a batch; - `StatusUpdaterBoltTest.acksKnownURLsThroughStreamingEndpoint` — known URLs still ack through the streaming endpoint; - `StatusUpdaterBoltFallbackTest` — runs the bolt against a `crawlercommons/url-frontier:2.5` container: the `UNIMPLEMENTED` error is caught, batching is switched off, and discovered URLs are acked through the streaming fallback; - the existing `StatusUpdaterBoltTest` cases (ack with metadata, queue-stream emission, semaphore recovery after a frontier restart) pass unchanged on the new implementation. All 64 module tests pass locally against real frontier containers (testcontainers), checkstyle, forbiddenapis and the google-java-format check are clean. The `StatusUpdaterBolt` JaCoCo ratios configured for this module still pass. ## Notes for reviewers - **Dependency availability.** The url-frontier 2.6 release published its artifacts to Sonatype Central via a workflow that completed successfully on 2026-08-19, but the deployment is not yet visible on Maven Central (`urlfrontier-API` still lists 2.5 as latest; the portal deployment likely awaits the manual publish step, since `central-publishing-maven-plugin` defaults to `autoPublish=false`). This PR builds against the 2.6 API regardless — the protobuf/gRPC API is stable in the 2.6 tag — but CI will need `urlfrontier-API:2.6` to be resolvable; it may be worth pinging the crawler-commons maintainers to complete the release on the portal. If 2.6 cannot be published soon, an alternative is to keep the dependency at 2.5 and build the batch messages from the proto (not possible: `DiscoveredBatch`/`BatchAck` and the `putDiscovered` stub only exist in 2.6), so the version bump is inherent to the feature. - The `waitAck`-cache eviction semantics, the semaphore accounting, the queue-stream emission and the channel management (multi-address assignment, reconnect on `TRANSIENT_FAILURE`) are carried over unchanged from the previous implementation. - The `PutDiscovered` stream is deliberately not re-created on channel `TRANSIENT_FAILURE` (unlike the `PutURLs` stream): with `waitForReady` it survives connection blips, and abandoning it would orphan the batches already sent on it. -- 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]
