S3C-11127: add the notification delivery worker and its producer pool - #2800
Draft
anurag4DSB wants to merge 6 commits into
Draft
S3C-11127: add the notification delivery worker and its producer pool#2800anurag4DSB wants to merge 6 commits into
anurag4DSB wants to merge 6 commits into
Conversation
The delivery worker consumes a topic shared by every destination, so it needs a producer per destination created on demand rather than the single producer of the per-destination queue processor. DeliveryKafkaProducer bounds how long librdkafka retries a message, since the worker holds the consumer offset until the delivery report arrives and an unbounded retry would block that offset forever. DeliveryProducerPool creates producers on demand, serves concurrent requests made during connect from a single producer, and closes producers once idle. A producer with deliveries in flight is never closed: its delivery reports still release consumer offsets.
One worker serves every destination: the destination id and the notification configuration id ride in each record, so no bucket notification configuration lookup and no mongo or zookeeper connection is needed here. Two details are load bearing. The consumer callback is held until the delivery report arrives, so an offset is only committed once the notification has left the process, and a delivery failure is counted and dropped rather than passed back as an error, which the consumer would raise as a consumer level error. Ordering is by destination and object instead of the default kafka key, which would serialize a whole destination behind one lane. The consumer reads from the earliest offset: librdkafka defaults to latest, and a worker joining with a fresh group would skip everything already sitting in the topic.
Covers the commit timing (the callback waits for the delivery report), the drop reasons and their counters, the ordering key and the parse-once stash, and for the pool the single producer per destination during connect, idle reaping that spares busy producers, capacity eviction, and the message timeout landing in the producer topic config.
Several workers can run from a single rendered config file, on one host or inside one container, and only one process can bind a given port, so the DELIVERY_POOL_PROBE_PORT environment variable now wins over the configured port and the deployment can hand each worker process its own. A probe server that fails to bind no longer takes the worker down either. A worker that cannot serve its probe routes still delivers notifications, and dying instead would turn a port clash into a crash loop.
Aligns the environment variable with the name the federation supervisord template exports per worker program. Behaviour is unchanged: a valid integer replaces the configured probe server port and the rest of the probe server config is kept, anything else falls back to the configured port with a warning.
Restores the environment variable name the federation supervisord template exports and the contract pins. The resolution itself is unchanged: a valid integer replaces the configured probe server port and the rest of the probe server config is kept, anything else falls back to the configured port, with a warning when the value was non-empty.
Codecov Report❌ Patch coverage is Additional details and impacted files
... and 3 files with indirect coverage changes
@@ Coverage Diff @@
## development/9.3 #2800 +/- ##
===================================================
- Coverage 74.70% 74.44% -0.27%
===================================================
Files 199 204 +5
Lines 13629 13936 +307
===================================================
+ Hits 10182 10374 +192
- Misses 3437 3552 +115
Partials 10 10
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| "notification_processor": "node extensions/notification/queueProcessor/task.js", | ||
| "notification_delivery_worker": "node extensions/notification/deliveryWorker/task.js", | ||
| "notification_delivery_replay": "node bin/notificationDeliveryReplay.js" | ||
| }, |
There was a problem hiding this comment.
This script references bin/notificationDeliveryReplay.js which does not exist in the repo or in this PR. Running npm run notification_delivery_replay will crash with MODULE_NOT_FOUND.
If the replay entry point is coming in a follow-up PR, remove this line until then to avoid a broken script in the shipped package.json.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Intent: why does this change exist?
The delivery pool needs something to read the addressed records and dispatch them, so that one shared consumer group serves every destination instead of the current one group per destination.
System impact: what's affected, including downstream?
Everything is new and lives under
extensions/notification/deliveryWorker/, plus two package.json scripts. Nothing invokes it unless it is deployed, and it only has work to do once the populator's delivery pool flag is on. Thenotification_delivery_replayscript points at the replay tool from its own PR, so it is inert until that lands. The existing notification queue processor is not touched.Preserved behavior: what explicitly stays the same?
Additive only. The per-destination consumer path, the destination config shape and the message format the customer receives are all unchanged.
Intended change: what's different after this PR?
Adds a worker that consumes the shared delivery topic under a single consumer group, keeps a pool of producers keyed per destination, and holds the consumer offset until each send reaches a terminal outcome, so an offset never advances past a record still in flight. Terminal means either delivered or dropped: a send that fails or exceeds
deliveryTimeoutMsis counted indropped_totalwith a reason and the offset then advances, so a stalled destination costs bounded, counted loss rather than an unbounded backlog. There is no retry or dead letter path here. Each worker honours a probe port handed to it by the deployment, since one host runs several of them.Verification: how do we know this worked, or how would we know if it didn't?
Unit tests cover the worker and the producer pool: the offset is held until the delivery report, each drop reason is counted, ordering is derived per destination and object with the payload parsed once, and for the pool, single-flight creation, idle reaping that spares in-flight producers, capacity eviction and the message timeout reaching the producer topic config. 32 tests in the two new spec files, 192 passing across tests/unit/notification, eslint clean. This PR itself ships no functional tests; that suite lands separately (#2802), where this worker was driven against a live broker, including an end to end run from the real populator through this worker to the destination topics with correct per-destination routing.