sink/periodic-jobs: deliver and report one item at a time (DI-3015)

docs/chunk-scheduling-redesign.md

@@ -1723,6 +1723,26 @@ What it does **not** skip is result reporting: every item is still reported indi
 because the DELIVERING phase counters and the per-job gate are driven by those reports,
 and a job whose items are never reported never completes.
 
+##### Job-end work runs against complete data by construction
+
+An aggregating sink's job-end work — the `PeriodicJobs*FinalizerBean`s, marcconv's
+`ConversionFinalizer` — reads what every preceding item of the job persisted, so it is
+only correct if all of those writes are committed before it starts. The per-item protocol
+gives that for free, from the order in which one item is handled: `deliverItem` commits
+its own transaction and returns, and `SinkMessageConsumerAdapter` reports the result only
+afterwards. A reported item is therefore an item whose writes are durable, and the
+termination chunk is released only once every data chunk of the job has reported (by
+`waitingOn` today, by `gate_open` once the dispatch filter lands — both driven by
+`chunkDeliveringDone`, which fires when a chunk's last item result is committed).
+
+The chunk protocol had this the other way round: each sink called
+`sendResultToJobStore(result)` *before* committing its own transaction, so job-store could
+see a chunk as delivered while that chunk's data was still uncommitted, and the
+termination chunk could be released against an incomplete set. `periodic-jobs` covered
+that window with a fixed five second sleep before finalizing, removed in DI-3015 along
+with the ordering problem it guessed at. `marcconv` has the same shape and the same
+argument applies to it.
+
 ### Watermark calls (`job-store-service-connector`)
 
 An earlier version of this document specified a separate `WatermarkServiceConnector`.

sink/CLAUDE.md

@@ -26,15 +26,35 @@ mvn package -pl marc-client -P generate-source
 
 ## Architecture
 
-All sinks follow the same structural pattern built on the `jse-artemis` framework:
+All sinks share the same structure, built on the `jse-artemis` framework:
 
 1. **`*SinkApp`** — entry point, extends `MessageConsumerApp`. Creates a `ServiceHub` and a `Supplier<MessageConsumer>`, then calls `go(serviceHub, messageConsumer)`. Database-backed sinks also call `JPAHelper.migrate()` and `JPAHelper.makeEntityManagerFactory()` here.
 
-2. **`*MessageConsumer`** — extends `MessageConsumerAdapter`. Implements `handleConsumedMessage(ConsumedMessage)`, which is the main processing loop. The standard flow is:
-   - `unmarshallPayload(consumedMessage)` → `Chunk` of type PROCESSED
-   - Iterate over `ChunkItem`s; handle SUCCESS / FAILURE / IGNORE status
-   - Build a new `Chunk` of type DELIVERED
-   - `sendResultToJobStore(deliveredChunk)`
+2. **`*MessageConsumer`** — the consumer, extending `SinkMessageConsumerAdapter` (epic DI-2946, see `docs/chunk-scheduling-redesign.md`). job-store dispatches one JMS message per item with `payload = ITEM_PAYLOAD_TYPE` and a single `ChunkItem` body, and the sink implements one method:
+
+   ```java
+   protected ItemDeliveryResult deliverItem(ConsumedMessage message, ChunkItem item)
+   ```
+
+   returning `ItemDeliveryResult.of(status, outcomeItem)`. `handleConsumedMessage` is `final` in the base class. The framework — not the sink — owns the header reads, the delivery watermark check, reporting the result to job-store, the `DBCTrackedLogContext` tracking-id scope, and the `dataio_item_delivery_count` metric. In particular, never read `JMSHeader.recordKey` or re-derive a watermark key from record content.
+
+   Four verdicts, three of them the sink's to return:
+
+   | Verdict | Meaning | DELIVERING counter | Returned by |
+   |---|---|---|---|
+   | `DELIVERED` | sent to the target | succeeded | the sink |
+   | `IGNORED` | not sent, nothing to send | ignored | the sink |
+   | `FAILED` | attempted, rejected in a way retrying will not fix | failed | the sink |
+   | `SUPERSEDED` | a newer version of the record was already delivered | ignored | the framework only |
+
+   Rules that are easy to get wrong:
+   - **Throwing means "retry"** — it rolls the JMS session back and the item is redelivered until the broker gives up. A terminal failure must be returned as `FAILED`, not thrown.
+   - **A processing outcome passed through without being sent is `IGNORED`, not `DELIVERED` with an `IGNORE` item.** `DELIVERED` is the only verdict that advances the watermark, so using it for an unsent item both overstates the succeeded count and makes a false claim about what is at the target.
+   - **Commit your own writes before returning.** The framework reports the result after `deliverItem` returns, so a reported item is an item whose writes are durable — which is what lets an aggregating sink's job-end work run against complete data.
+   - Sinks that aggregate a whole job before delivering anything (`periodic-jobs`, `marcconv`) override `usesDeliveryWatermark()` to `false`. They still report every item individually: the phase counters and the per-job gate are driven by those reports.
+   - The job termination item arrives as an ordinary item message carrying `ChunkItem.Type.JOB_END`; sinks needing job-end work branch on that.
+
+   `dlq-errorhandler` and `job-processor2` are not sinks in this sense and stay on the chunk-level `MessageConsumerAdapter` by design: they implement `handleConsumedMessage(ConsumedMessage)` themselves and report whole `Chunk`s via `sendResultToJobStore`.
 
 3. **`SinkConfig`** — enum implementing `EnvConfig`. Each constant maps to an environment variable. Values are read at startup; default values can be provided in the constructor.
 
@@ -44,7 +64,12 @@ Configuration that varies per job (e.g. endpoint, credentials) comes from the **
 
 Files named `*IT.java` are integration tests run by `maven-failsafe-plugin` during `verify`. They use **Testcontainers** (PostgreSQL) via `PostgresContainerJPAUtils`. Unit tests (`*Test.java`) use Mockito and run with `maven-surefire-plugin` during `test`.
 
-The `testutil` module provides `ObjectFactory.createConsumedMessage(Chunk)` — the standard way to build a `ConsumedMessage` in tests.
+There is no shared helper for building a per-item `ConsumedMessage`: build it from a header map (`JMSHeader.payload` = `ITEM_PAYLOAD_TYPE`, plus `jobId`, `chunkId`, `itemId`, `sinkId` and, unless the sink opted out, `recordKey`) and a `JSONBContext`-marshalled `ChunkItem` body. `DummyMessageConsumerTest` and `PeriodicJobsMessageConsumerTest` are the models. Construct the consumer with `new ServiceHub.Builder().withJobStoreServiceConnector(mock).test()` — `test()` rather than `build()`, so no HTTP service is started.
+
+Two things worth knowing before writing such a test:
+
+- **A test that drives `handleConsumedMessage` and then asserts the reported result is re-testing the framework.** Header reading, the watermark comparison and result reporting all live in `SinkMessageConsumerAdapter` and are covered by `SinkMessageConsumerAdapterTest`. A sink's own surface is `deliverItem` plus its `usesDeliveryWatermark()` choice — call `deliverItem` directly. The one exception is asserting the watermark opt-out, which is observable only as the lookup being (or not being) made.
+- **A unit test that constructs a consumer needs `APP_NAME`** — `UserAgent.forInternalRequests()` reads it while the consumer builds its connectors, and without it the test fails with "APP_NAME environment variable has not been set". Several modules set it only for failsafe; add the same `<environmentVariables>` block to `maven-surefire-plugin` (see `dpf` and `periodic-jobs`).
 
 ## Notable modules
 

sink/periodic-jobs/pom.xml

@@ -146,6 +146,17 @@
                 <groupId>org.codehaus.mojo</groupId>
                 <artifactId>exec-maven-plugin</artifactId>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <!-- UserAgent.forInternalRequests(), called when the message consumer
+                         builds its connectors, reads APP_NAME -->
+                    <environmentVariables>
+                        <APP_NAME>${app-name}</APP_NAME>
+                    </environmentVariables>
+                </configuration>
+            </plugin>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-failsafe-plugin</artifactId>

sink/periodic-jobs/src/main/java/dk/dbc/dataio/sink/periodicjobs/PeriodicJobsConfigurationBean.java

@@ -2,7 +2,6 @@
 
 import dk.dbc.dataio.common.utils.flowstore.FlowStoreServiceConnector;
 import dk.dbc.dataio.common.utils.flowstore.FlowStoreServiceConnectorException;
-import dk.dbc.dataio.commons.types.Chunk;
 import dk.dbc.dataio.commons.types.HarvesterToken;
 import dk.dbc.dataio.commons.utils.jobstore.JobStoreServiceConnector;
 import dk.dbc.dataio.commons.utils.jobstore.JobStoreServiceConnectorException;
@@ -25,13 +24,13 @@ public class PeriodicJobsConfigurationBean {
     JobStoreServiceConnector jobStoreServiceConnector;
 
     /**
-     * Returns delivery configuration for given chunk
+     * Returns delivery configuration for given job
      *
-     * @param chunk {@link Chunk} for which get delivery configuration
+     * @param chunkId id of the chunk the lookup is made for, which gates whether the
+     *                delivery entity is persisted
      * @return delivery configuration as {@link PeriodicJobsDelivery}
      */
-    public PeriodicJobsDelivery getDelivery(Chunk chunk, EntityManager entityManager) {
-        Integer jobId = Math.toIntExact(chunk.getJobId());
+    public PeriodicJobsDelivery getDelivery(int jobId, int chunkId, EntityManager entityManager) {
         PeriodicJobsDelivery periodicJobsDelivery = deliveryCache.getIfPresent(jobId);
         if (periodicJobsDelivery != null) {
             // Return delivery entity from local bean cache.
@@ -41,11 +40,11 @@ public PeriodicJobsDelivery getDelivery(Chunk chunk, EntityManager entityManager
         if (periodicJobsDelivery == null) {
             // Retrieve harvester config from flow-store and create new
             // delivery entity.
-            PeriodicJobsHarvesterConfig periodicJobsHarvesterConfig = getHarvesterConfig(chunk);
-            periodicJobsDelivery = new PeriodicJobsDelivery(Math.toIntExact(chunk.getJobId()));
+            PeriodicJobsHarvesterConfig periodicJobsHarvesterConfig = getHarvesterConfig(jobId);
+            periodicJobsDelivery = new PeriodicJobsDelivery(jobId);
             periodicJobsDelivery.setConfig(periodicJobsHarvesterConfig);
         }
-        if (chunk.getChunkId() == 0) {
+        if (chunkId == 0) {
             // Only allow the first chunk to persist the delivery entity
             entityManager.persist(periodicJobsDelivery);
         }
@@ -54,8 +53,8 @@ public PeriodicJobsDelivery getDelivery(Chunk chunk, EntityManager entityManager
         return periodicJobsDelivery;
     }
 
-    private PeriodicJobsHarvesterConfig getHarvesterConfig(Chunk chunk) {
-        HarvesterToken harvesterToken = getHarvesterToken(chunk);
+    private PeriodicJobsHarvesterConfig getHarvesterConfig(int jobId) {
+        HarvesterToken harvesterToken = getHarvesterToken(jobId);
         try {
             return flowStoreServiceConnector
                     .getHarvesterConfig(harvesterToken.getId(), PeriodicJobsHarvesterConfig.class);
@@ -65,16 +64,16 @@ private PeriodicJobsHarvesterConfig getHarvesterConfig(Chunk chunk) {
         }
     }
 
-    private HarvesterToken getHarvesterToken(Chunk chunk)  {
+    private HarvesterToken getHarvesterToken(int jobId)  {
         try {
             JobListCriteria findJobCriteria = new JobListCriteria()
                     .where(new ListFilter<>(JobListCriteria.Field.JOB_ID,
-                            ListFilter.Op.EQUAL, chunk.getJobId()));
+                            ListFilter.Op.EQUAL, jobId));
             JobInfoSnapshot jobInfoSnapshot = jobStoreServiceConnector.listJobs(findJobCriteria).get(0);
             return HarvesterToken.of(jobInfoSnapshot.getSpecification().getAncestry().getHarvesterToken());
         } catch (RuntimeException | JobStoreServiceConnectorException e) {
             throw new RuntimeException(
-                    String.format("Failed to find job %d", chunk.getJobId()), e);
+                    String.format("Failed to find job %d", jobId), e);
         }
     }
 

sink/periodic-jobs/src/main/java/dk/dbc/dataio/sink/periodicjobs/PeriodicJobsFinalizerBean.java

@@ -1,6 +1,5 @@
 package dk.dbc.dataio.sink.periodicjobs;
 
-import dk.dbc.dataio.commons.types.Chunk;
 import dk.dbc.dataio.commons.types.ChunkItem;
 import dk.dbc.dataio.commons.types.exceptions.InvalidMessageException;
 import dk.dbc.dataio.harvester.types.FtpPickup;
@@ -28,23 +27,32 @@ public class PeriodicJobsFinalizerBean {
     PeriodicJobsFtpFinalizerBean periodicJobsFtpFinalizerBean;
     PeriodicJobsSFtpFinalizerBean periodicJobsSFtpFinalizerBean;
 
-    public Chunk handleTerminationChunk(Chunk chunk, EntityManager entityManager) throws InvalidMessageException {
-        LOGGER.info("Finalizing periodic job {}", chunk.getJobId());
+    /**
+     * Delivers the job's accumulated datablocks to its pickup destination
+     *
+     * @param chunkId id of the job's termination chunk, needed only to tell an empty job
+     *                from one that has data, see
+     *                {@link dk.dbc.dataio.sink.periodicjobs.pickup.PeriodicJobsPickupFinalizer#isEmptyJob(int, int)}
+     * @return the job's delivering outcome, as the JOB_END item reported for its
+     * termination item
+     */
+    public ChunkItem finalizeJob(int jobId, int chunkId, EntityManager entityManager) throws InvalidMessageException {
+        LOGGER.info("Finalizing periodic job {}", jobId);
 
-        PeriodicJobsDelivery delivery = periodicJobsConfigurationBean.getDelivery(chunk, entityManager);
+        PeriodicJobsDelivery delivery = periodicJobsConfigurationBean.getDelivery(jobId, chunkId, entityManager);
         Pickup pickup = delivery.getConfig().getContent().getPickup();
-        Chunk result;
+        ChunkItem result;
 
         if (pickup instanceof HttpPickup) {
-            result = periodicJobsHttpFinalizerBean.deliver(chunk, delivery, entityManager);
+            result = periodicJobsHttpFinalizerBean.deliver(jobId, chunkId, delivery, entityManager);
         } else if (pickup instanceof MailPickup) {
-            result = periodicJobsMailFinalizerBean.deliver(chunk, delivery, entityManager);
+            result = periodicJobsMailFinalizerBean.deliver(jobId, chunkId, delivery, entityManager);
         } else if (pickup instanceof FtpPickup) {
-            result = periodicJobsFtpFinalizerBean.deliver(chunk, delivery, entityManager);
+            result = periodicJobsFtpFinalizerBean.deliver(jobId, chunkId, delivery, entityManager);
         } else if (pickup instanceof SFtpPickup) {
-            result = periodicJobsSFtpFinalizerBean.deliver(chunk, delivery, entityManager);
+            result = periodicJobsSFtpFinalizerBean.deliver(jobId, chunkId, delivery, entityManager);
         } else {
-            result = getUnhandledPickupTypeResult(chunk, pickup);
+            result = unhandledPickupTypeResult(pickup);
         }
 
         LOGGER.info("Deleted {} data blocks for job {}",
@@ -68,13 +76,10 @@ public int deleteDelivery(Integer jobId, EntityManager entityManager) {
                 .executeUpdate();
     }
 
-    private Chunk getUnhandledPickupTypeResult(Chunk chunk, Pickup pickupType) {
-        final Chunk result = new Chunk(chunk.getJobId(), chunk.getChunkId(), Chunk.Type.DELIVERED);
-        result.insertItem(
-                ChunkItem.failedChunkItem()
-                        .withType(ChunkItem.Type.JOB_END)
-                        .withData("Unhandled pickup type: " + pickupType));
-        return result;
+    private ChunkItem unhandledPickupTypeResult(Pickup pickupType) {
+        return ChunkItem.failedChunkItem()
+                .withType(ChunkItem.Type.JOB_END)
+                .withData("Unhandled pickup type: " + pickupType);
     }
     public PeriodicJobsFinalizerBean withPeriodicJobsConfigurationBean(PeriodicJobsConfigurationBean periodicJobsConfigurationBean) {
         this.periodicJobsConfigurationBean = periodicJobsConfigurationBean;