[WIP] DBCPR-328-dataio-chunk-scheduling-redesign - #299
Draft
jbndbc wants to merge 46 commits into
Draft
Conversation
Implements the correlationKey field/derivation part of the chunk-scheduling redesign's Phase 2 (docs/chunk-scheduling-redesign.md). A computed, non-persisted key used to serialise deliveries of a record via the broker's JMSXGroupID. A standalone record keys against older/newer versions of itself, while every head/section/volume record instead shares one constant key, grouping the whole hierarchy into a single delivery group.
Contributor
Author
|
Ændringer i design dokumentet er medtaget i dette PR, da rettelserne opstod som en del af løsningen af denne story. |
atmdbc
approved these changes
Aug 14, 2026
…retention pruning (DI-2995) Adds the Flyway-migrated watermark table from the chunk-scheduling redesign plus a nightly purge job so the table doesn't grow unbounded.
JNA native-library loading uses tmp files in locations that breaks when running inside sandboxed environments, so we use the project's pwn filesystem instead.
…exercise run() (DI-2995) run() guards on Hazelcast.isSlave(), which dereferences the static Hazelcast.INSTANCE. In a plain unit test that's null, so isSlave() threw NPE, the catch-all swallowed it, and purgeStaleWatermarks() was never reached — both tests passed even with the try/catch deleted. Mocks a HazelcastInstance/Cluster/Member so isSlave() resolves deterministically to false, and verifies purgeStaleWatermarks() was actually called. Hazelcast.testInstance(null) in @AfterEach avoids leaking the mock into other tests sharing the same static state.
The doc claimed last_modified is set on every upsert, but the ON CONFLICT DO UPDATE branch is gated by the same version-advance WHERE clause as the rest of the row, so it only refreshes when the watermark actually advances — not on exact retransmits or SKIPPED older versions. Reframes the Retention Policy trade-off around "no watermark advance" rather than "no delivery", and re-justifies why it's still acceptable: normal records stop generating traffic once current, so the two only diverge under redelivery/retry scenarios that resolve within the same incident, not months later. Calls out the residual gap (a chunk stuck retrying for the full retention window) as a known limitation instead of glossing over it.
ja7
approved these changes
Aug 14, 2026
DI-2995 jobstore watermark table
jbndbc
changed the title
jbndbc
changed the title
jbndbc
marked this pull request as draft
August 14, 2026 12:07
…2996) Lets a sink look up the last delivered (jobId, chunkId, itemId) for a given (sinkId, recordKey) before delivering, per docs/chunk-scheduling- redesign.md's watermark protocol. - recordKey is a query parameter, not a path segment: it is an opaque, agency-qualified token built from arbitrary harvested record ids, which may contain '/' or '%' and would otherwise silently 404 or get mangled by container-level path normalisation. - sinkId is accepted as long at the REST boundary, matching the specced WatermarkServiceConnector signature and the existing SINK_STATUS endpoint, narrowed to int only when building WatermarkEntity.Key. - WatermarkEntity is marked @Cacheable(false): its only writer is a native upsert EclipseLink can't see, so the shared L2 cache would otherwise serve stale watermarks to em.find().
…point job-store-service: Add GET endpoint for sink delivery watermarks (DI-…
…livered Only DELIVERED advances the watermark (Open Question #3 resolved for DI-2997's write endpoint) - the FAILED-advances-it alternative remains defensible but is not adopted now, leaving the priority-inversion case a known limitation. Also rename the per-item result-reporting endpoint from the original .../delivering to .../items/{itemId}/delivered, reusing the existing GET path that already reads ItemEntity.deliveringOutcome back, and matching the existing bulk .../chunks/{chunkId}/delivered endpoint's naming convention.
Sinks report the outcome of a single item's delivery via
POST /jobs/{jobId}/chunks/{chunkId}/items/{itemId}/delivered. One transaction
writes ItemEntity.deliveringOutcome, advances the DELIVERING-phase counters on
the item, chunk and job, conditionally advances sink_record_delivery_watermark,
and completes the job when this item is the one that completes it.
…dpoint DI-2997 jobstore item delivered endpoint
atmdbc
approved these changes
Aug 19, 2026
…rce-class-from-rs-to-ejb-package job-store-service/war: move WatermarksBean from rs to ejb package
Add getWatermark(int, String) and addItemDelivered(...) to
JobStoreServiceConnector.
- Promote the GET /sinks/{sinkId}/watermarks response envelope to a shared
WatermarkResponse record in job-store-service/types, used by both
WatermarksBean and the connector instead of two independent definitions
- Add a real end-to-end IT (JobsIT) exercising both methods against a live
job-store-service container: watermark absent before delivery, correct
after, job not completed until every chunk is delivered
job-store-service-connector: Add watermark connector methods (DI-2998)
…ion records (DI-2994) Chunks whose items include a non-deleted MARC head or section record are scheduled at Priority.HIGH regardless of the job's own priority. The chunk entity now tracks this via a new containsliveheadorsectionrecord column, computed once at chunk creation time.
…-head-section job-store-service/war: prioritize chunks containing live head or sect…
sink::<name> is Artemis FQQN syntax, not a queue name containing '::'. It means queue <name> on address 'sink', so every sink queue shares the single address 'sink' and one exact-match address-setting covers all of them. The doc's original per-queue <address> block, and the wildcard scheme that replaced it, were both built on the wrong reading. The Artemis Broker Configuration section is rewritten to record what was measured against a real 2.42.0 broker rather than what the design assumed: - Sink queues are auto-created from each sink's QUEUE variable and are never declared in broker.xml, so the settings cannot hang off <queue> elements - Per-group serialisation comes from JMSXGroupID alone and is on by default. The four settings do not create the ordering guarantee, they keep it true across scaling and restarts and bound its memory cost - group-first-key is set but nothing in dataio reads the resulting header, so Phase 7 does not need to expose it through MessageConsumerAdapter - default-* settings apply only at queue creation, and group-rebalance-pause-dispatch cannot be set on an existing queue by any management operation, so the existing sink queues have to be dropped and re-created while drained. Phase 5 is therefore not configuration-only The integration-test Artemis image moves from 2_30_0-0 to 2_42_0-2, the tag built from the matching artemis-setup change, so tests run the same broker version and the same broker.xml as staging and production.
Per-item messages need two headers that do not exist today. itemId is the third element of the (jobId, chunkId, itemId) version tuple compared against the delivery watermark, and recordKey is the agency-qualified watermark key <agencyId>:RecordInfo.getId(). recordKey is carried on the message rather than re-derived per sink because the RecordInfo constructor whitespace-normalises the id, so a key re-derived from raw record bytes can differ from the one job-store compares against and stale delivery detection then fails silently. The javadoc on the constant states that invariant, since getRecordKey lives on MessageConsumerAdapter and this is the only place a sink implementor sees it.
Correct Phase 5 broker config for FQQN addressing (DI-2999)
…d id (DI-3059)
The termination item was constructed as new RecordInfo("End Item"), and that is
the only place in the code that creates it. RecordInfo.getCorrelationKey()
returns id, so every job's termination item shared the literal correlationKey
"End Item". Under per-item dispatch that breaks two things. The broker
serialises all termination items across all jobs into one group instead of
leaving them unkeyed. And the watermark key becomes <agencyId>:End Item, shared
by every job from the same submitter, so delivering job 1000's termination item
would make job 999's look stale and get it reported SKIPPED. Not a theoretical
risk, since marcconv's ConversionFinalizer and PeriodicJobs*FinalizerBean do
real delivery work on job end.
A null id yields correlationKey == null on its own, with no new RecordInfo
subclass, no @Class discriminator change and no migration. The watermark half
follows once JMSHeader.recordKey lands (DI-3000, DI-3001), which composes the
key only for items that have a record id, so no sink-side change is needed.
- Guard both JobRerunnerBean harvester rerun paths on a non-null record id.
They checked only recordInfo != null, so the termination item was already
submitted to the harvester as a record reference for the fictional id
EndItem, and would otherwise become a null-id reference. Neither harvester
crashes on a null id, so this is hygiene rather than a fix
- The human-readable "End Item" label disappears wherever the item's
RecordInfo.getId() is surfaced today, including the GUI item view via
ItemInfoSnapshot. The item stays identifiable by its ChunkItem type JOB_END,
data "Job termination item" and trackingId <jobId>.JOB_END
- Existing rows keep the id "End Item". No migration, the change applies only
to items created after deployment
…on-item job-store-service/war: construct job termination item with null recor…
commons/types: Add itemId and recordKey JMS headers (DI-3000)
…-3001) SinkMessageProducerBean now iterates the chunk's ItemEntity rows and sends one message per item, carrying JMSXGroupID from RecordInfo.getCorrelationKey() so the broker serializes deliveries of the same record across consumer threads and pods. JMSHeader.recordKey is composed here, in this one place, as <agencyId>:RecordInfo.getId() with agencyId read from job.getSpecification().getSubmitterId(). The key is opaque to sinks. The RecordInfo constructor whitespace-normalises the id, so a sink re-deriving it from delivered content would compose a key that never matches the one job-store compares against, and stale delivery detection would then fail silently. Item messages get their own payload type, JMSHeader.ITEM_PAYLOAD_TYPE "Item" (commons/types).
… (DI-3001) addChunkProcessed called chunkProcessingDone before addChunk had written the processing outcomes. That was harmless while delivery dispatched the Chunk object received over REST, because nothing read the database. Per item dispatch reads the chunk's ItemEntity rows instead, and at that point every row still carries a null processing outcome. The producer's verifyDeliverable guard correctly refused the chunk, but JobStoreException is @applicationexception(rollback = true), so the container marked the transaction rollback-only. submitToDelivering's catch block then called setStatus before logging, that call failed with TransactionRolledbackLocalException "Client's transaction aborted", and the replacement exception erased the real cause. The result was a 500 from addChunkProcessed, nothing on the sink queue, and a container log with no trace of why. Persisting first is also the order addChunkDelivered already uses, and it removes a second hazard that predates per-item dispatch: scheduling first left dependency tracking advanced, and possibly the chunk dispatched, for a chunk whose write then rolled back, since dependency tracking is not enrolled in the transaction.
Di-3001 jobstore per-item dispatch
SinkMessageConsumerAdapter handles the delivery watermark check and the per item result reporting around deliverItem, so a sink implements nothing else. It reads the watermark before every delivery and reports the result before the JMS session commits, which leaves a crash between the two harmless: the message is redelivered and the idempotency guard on the report endpoint absorbs the duplicate. getRecordKey is final and reads JMSHeader.recordKey off the message, since the RecordInfo constructor whitespace-normalises the id and a key re-derived from delivered content would never match the one job-store compares against. Sinks with no per record supersession to enforce opt out with usesDeliveryWatermark, and still report a result per item, since the phase counters and the per job gate are driven by those reports. The protocol lands on a new class rather than on MessageConsumerAdapter because two of that class's subclasses consume whole chunks permanently. job-processor2 is still sent CHUNK_PAYLOAD_TYPE, and dlq-errorhandler has to read both payload types. Neither will ever implement deliverItem, so putting the item protocol on the shared base would leave both inheriting a method that is meaningless for them and that could never be made abstract. dlq-errorhandler now fails a dead item per item instead of unmarshalling every body as a Chunk. Nothing else reports an item the broker has given up on, so without that branch its delivering outcome stays null, the phase counters never reach their total, and the job hangs unfinished with nothing saying why. No sink is changed here, so the reactor keeps building until each sink moves in its own change.
handleConsumedMessage now reads the watermark key itself, with usesDeliveryWatermark in the conditional producing it, so a sink opting out of the watermark is visible where the delivery decision is made instead of behind a lookup that returned null both for that case and for an item with no record identity. The comparison lives in deliverUnlessSuperseded, and the key is attached once at the single report site.
commons/artemis-jse-app: deliver and report one item at a time (DI-3002)
Replace existing JobProcessorException exception since it was simply misleading as part the JMS message consumer adapter framework.
…tion-refactoring Add better naming for exceptions thrown by JMS message consumers
…encytrack-table job-store-service/war: Add flyway migrations for termination chunk ga…
Split ownership of is_termination and gate_open while dependencytracking is still a write-behind projection: the MapStore owns row lifecycle, job-store writes both columns synchronously and performs the termination row's own insert. Replace the row-presence cross-job barrier check with job.termination_barrier_lifted, nullable, and cover the abort path. Describe the lost-wakeup race on termination insert, the (sinkid, submitter) advisory lock, and the job-row-before-advisory-lock ordering. Keep the per-job gate in job-store when the scheduler is extracted: re-anchor the delivery-side evaluation onto the JobsBean callback, leave the dependencytracking DELETE with the counter increment, and leave the recheckBlocks gate sweep behind.
…-3019) A termination chunk must not be dispatched until every data chunk of its own job has been delivered, and until no earlier job with the same submitter on the same sink still holds a barrier. Both halves are now maintained, alongside the waitingOn mechanism they will replace. - JobGateBean advances gate state on delivery: it counts a data chunk against its job, or lifts the job's barrier and re-evaluates the jobs queued behind it when the chunk was the job's own termination chunk - JobGateRepository holds the gate's queries. is_termination and gate_open are written only here, never through the MapStore's upsert, which is documented as a standing constraint on DependencyTrackingStore
…n-chunks job-store-service/war: add the per-job gate on termination chunks (DI-3019)
…e (DI-3020) Delivery candidates are taken from dependencytracking in SQL rather than from the Hazelcast map, because the ordering keys and the gate columns live on the table and no predicate over the map can see the gate. - Every sink's delivery dispatch is now ordered by (priority DESC, jobId ASC, chunkId ASC), where it previously followed map iteration order. Sinks without termination chunks get this and nothing else from the change: their gates are never closed, so the gate predicate and the two hasClosedGate calls always pass - DeliveryDispatchRepository holds the two queries the map cannot answer: the ordered candidate query, whose sorting is index-backed by dependencytracking_delivery_order_index, and the per-chunk gate check - Both dispatch paths refuse a chunk whose gate is closed and park it in SCHEDULED_FOR_DELIVERY, so the once-a-second bulk sweep releases it when the gate opens instead of the five minute stale sweep - submitToDelivering is the choke point every dispatch path reaches, which makes "no chunk with a closed gate leaves the scheduler" true by construction rather than by enumerating callers - The cascade in chunkDeliveringDone applies the same order, since each unblocked chunk gets its own transaction and can take the last queue slot - While the table is a write-behind projection of the map, the sweep re-checks each candidate against the map and over-fetches past the free queue slots. Those two compensations and DeliveryDispatchStaleStatusIT, which is the only cover for them, are deleted with the MapStore - ChunkSchedulingStatus documents the three shapes every phase has, what parks a chunk in each state and what moves it on, that BLOCKED is the exception held by other chunks rather than a queue, and why the persisted values are not in lifecycle order
…temDeliveryResult (DI-3017) The DELIVERING counters used to be read from each item's ChunkItem.Status, where the per-item protocol reads them from the sink's verdict instead. A sink passing a failed or ignored processing outcome through therefore had to report DELIVERED, which counts the item as succeeded and advances the record's watermark, claiming a delivery that never happened. IGNORED is the verdict for an item that was not sent. - deliveryStatusAsStateChange counts SUPERSEDED and IGNORED as ignored, so an item the sink did not send keeps the count the chunk-level path gave it - The watermark upsert stays conditioned on DELIVERED alone. Advancing the row for an item that never reached the target would have a genuinely older version of the same record judged stale and superseded afterwards, leaving the target with neither - SKIPPED is renamed SUPERSEDED, because IGNORED beside it carried no distinction. The two are indistinguishable to job-store, which counts both as ignored and advances neither watermark, so their names are the only thing separating them, and SUPERSEDED is already the framework's word for the condition in deliverUnlessSuperseded and supersededItem - Only SinkMessageConsumerAdapter returns SUPERSEDED, documented rather than enforced: a sink returning it wrongly produces the same counter and the same watermark behaviour as IGNORED
The sink consumes one message per item, as job-store dispatches them, and implements deliverItem on SinkMessageConsumerAdapter. The framework owns the watermark check, the result reporting and the tracking-id scope around the delivery, leaving the sink one method that decides an outcome per item. A successfully processed item is reported DELIVERED, anything else IGNORED with an IGNORE outcome item, which keeps the second group counted as ignored in the DELIVERING phase and advances no watermark for records the sink never sent.
…very-dispatch job-store-service: order delivery dispatch and honour the per-job gate (DI-3020)
… (DI-3076) A sink whose job-end work is not scoped to its own job cannot be ordered by the termination chunk alone. Tickle is that shape: createBatch marks and closeBatch sweeps a whole dataset rather than one batch, so a later job's first item opens a second batch over the same records the earlier job is still sweeping, and the dataset ends silently short with no error anywhere. Marcconv and periodic-jobs finalize by job id, so for them the narrow gate is enough. Width therefore follows the sink type. REQUIRES_FULL_WIDTH_BARRIER holds TICKLE alone and is a true subset of REQUIRES_TERMINATION_CHUNK. It lives next to that set rather than on SinkContent, because width follows from what the sink does at job end rather than from an operator choice, and SinkContent is cached per job
job-store-service/war: make barrier width a property of the sink type (DI-3076)
DI-3017 dummy sink migration
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
3 participants
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.
Implements the correlationKey field/derivation part of the chunk-scheduling redesign's Phase 2 (docs/chunk-scheduling-redesign.md). A computed, non-persisted key used to serialise deliveries of a record via the broker's JMSXGroupID.
A standalone record keys against older/newer versions of itself, while every head/section/volume record instead shares one constant key, grouping the whole hierarchy into a single delivery group.