TIKA-4846 -- add jsonl reporter

CHANGES.txt

@@ -1,5 +1,7 @@
 Release 4.1.0 - unreleased
 
+   * New file-system-jsonl-reporter pipes reporter (TIKA-4846).
+
    * Stop spooling OLE2 objects whose header over-reserves BAT capacity
      (TIKA-4845).
 

docs/modules/ROOT/examples/pipes-fs-jsonl-reporter.json

@@ -0,0 +1 @@
+../../../../tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/resources/config-examples/file-system-jsonl-reporter.json

docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc

@@ -40,6 +40,10 @@ The File System plugin (`tika-pipes-file-system`) is the most common starting po
 |Reporter
 |`file-system-reporter`
 |`FileSystemStatusReporter`
+
+|Reporter
+|`file-system-jsonl-reporter`
+|`FileSystemJsonlReporter`
 |===
 
 == Complete Pipeline Example
@@ -252,6 +256,63 @@ Tradeoffs:
 * The reporter thread sleeps between writes, so the worst-case staleness of the file is `reportUpdateMs` milliseconds plus serialization time.
 * Per-record `report()` calls are cheap (counter increment only). The cost of "watching" is bounded by the periodic write, not by document throughput.
 
+[#file-system-jsonl-reporter]
+== File System JSONL Reporter (`file-system-jsonl-reporter`)
+
+Append-only per-document audit log: one JSON object per line for every result that passes the `includes`/`excludes` filter. Where the status reporter above summarizes counts, this one records *which* documents ended in which state, so a downstream process (a dead-letter queue, `tika-eval`) can act on them.
+
+With the default `emitIntermediateResults: false` (see xref:pipes/configuration.adoc[]), a crash result (`OOM`, `TIMEOUT`, `UNSPECIFIED_CRASH`) leaves nothing in the emitter's output; this file is then the only record that the document was attempted. The driver process writes it, so it survives the forked worker's death.
+
+[source,json]
+----
+include::example$pipes-fs-jsonl-reporter.json[]
+----
+
+=== Line format
+
+[source,json]
+----
+{"id":"reports/q3.pdf","status":"OOM","category":"PROCESS_CRASH","message":"...","elapsedMs":4120,"timestamp":"2026-08-27T14:02:11.482Z"}
+----
+
+* `id` — `FetchEmitTuple.getId()` verbatim. For the file system iterator that is the path relative to `basePath`, so it matches the emitted file name minus the emitter's `fileExtension`.
+* `status` — `PipesResult.RESULT_STATUS` name.
+* `category` — the status's `PipesResult.CATEGORY` (`PROCESS_CRASH`, `TASK_EXCEPTION`, ...), so consumers can group without tracking every status.
+* `message` — the result's message when the worker managed to send one (a stack trace for crashes); `null` when it did not. Capped at `maxMessageLength`.
+* `elapsedMs` — wall-clock time the driver spent on this document: fetch, parse, any emit done inside the worker (`EMIT_SUCCESS*` statuses), and any wait for the driver's emit queue. Emits batched by the driver (`PARSE_SUCCESS*`) happen later and are not included.
+* `timestamp` — ISO-8601 UTC timestamp of the report.
+
+Each line is flushed to the OS before `report` returns, so it survives the driver process dying; there is no fsync, so a host crash can lose the last lines. A write failure (disk full, etc.) stops the whole run rather than continuing without a record.
+
+If the pipeline dies, a final line of a different shape, `{"error":"<stack trace>","timestamp":"..."}`, is written and the file is closed. Readers should dispatch on the presence of `id` vs `error`.
+
+Fields may be added in later releases; existing fields will not be renamed or removed.
+
+=== Configuration
+
+[cols="1,1,3"]
+|===
+|Field |Default |Description
+
+|`path`
+|_required_
+|Path of the JSONL file, resolved against the driver's working directory if relative. Missing parent directories are created at startup.
+
+|`onExists`
+|`EXCEPTION`
+|What to do when `path` already exists at startup: `EXCEPTION` refuses to start, `APPEND` continues the existing file (terminating a partial last line left by a killed run, with a warning), `REPLACE` truncates it. There is no `SKIP`, unlike the emitter's `onExists`; `tika-async-cli --on-exists` applies to this reporter too, mapping `skip` to `APPEND`. The default is deliberately strict — appending a new run onto an old ledger is the mistake this reporter exists to prevent.
+
+|`includes` / `excludes`
+|_all statuses_
+|Mutually exclusive sets of `RESULT_STATUS` names. For a crash ledger, `includes` the crash and exception statuses; leave both unset for a full per-document audit trail.
+
+|`maxMessageLength`
+|`10000`
+|Characters of `message` (and of the final `error` line) to keep; longer messages are truncated with a suffix stating how many characters were dropped. `0` means the default; negative values are rejected.
+|===
+
+Each line is written and flushed before `report()` returns, so the ordering across documents is the order the driver finished them, not the iterator's order. If a write fails (disk full), that `report()` and every later one throw, which aborts the pipeline rather than dropping lines silently.
+
 [#security-notes]
 == Security Notes
 

docs/modules/ROOT/pages/pipes/reporters.adoc

@@ -47,6 +47,8 @@ Reporters live under the plural top-level `pipes-reporters` key. The keys inside
 
 Each entry's outer key is the reporter's component name — there is no separate ID layer because reporters do not get referenced by other components.
 
+Every configured reporter receives every call even if another reporter throws; the first exception is rethrown afterward and stops the run.
+
 [#plugins]
 == Available Reporters
 
@@ -58,6 +60,10 @@ Each entry's outer key is the reporter's component name — there is no separate
 |`file-system-reporter`
 |Writes a JSON status file periodically. Pair with an external watcher — see xref:pipes/plugins/filesystem.adoc#watching[Live status for watching applications].
 
+|xref:pipes/plugins/filesystem.adoc#file-system-jsonl-reporter[File System]
+|`file-system-jsonl-reporter`
+|Appends one JSON line per document to a file. The record of documents whose worker crashed (by default nothing reaches the emitter for those).
+
 |xref:pipes/plugins/jdbc.adoc[JDBC]
 |`jdbc-reporter`
 |Writes per-doc status rows to a SQL table.

docs/modules/ROOT/pages/using-tika/cli/index.adoc

@@ -322,7 +322,7 @@ as usual.
 |File list, one path per line, relative to `--inputDir` or absolute.
 
 |`--on-exists=<mode>`
-|Behavior when an output file already exists: `exception` (default), `replace`, `skip`.
+|Behavior when an output file already exists: `exception` (default), `replace`, `skip`. Also applied to a configured `file-system-jsonl-reporter` (`skip` becomes `APPEND` there).
 |===
 
 === Output formatting

tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/PluginsWriter.java

@@ -162,6 +162,7 @@ void write(Path output) throws IOException {
             if (!StringUtils.isBlank(simpleAsyncConfig.getOnExists())) {
                 patchFileSystemField(root, "emitters", "file-system-emitter",
                         "onExists", simpleAsyncConfig.getOnExists());
+                patchJsonlReporterOnExists(root, simpleAsyncConfig.getOnExists());
             }
 
             // merge, don't replace: other configured timeout-limits fields must survive
@@ -215,6 +216,16 @@ private static void patchFileSystemField(ObjectNode root, String section,
         }
     }
 
+    // the jsonl reporter has no SKIP; a rerun that keeps old outputs should keep the old ledger too
+    private static void patchJsonlReporterOnExists(ObjectNode root, String emitterOnExists) {
+        JsonNode reporters = root.get("pipes-reporters");
+        if (reporters == null || !reporters.isObject() || !reporters.has("file-system-jsonl-reporter")) {
+            return;
+        }
+        String mapped = "SKIP".equalsIgnoreCase(emitterOnExists) ? "APPEND" : emitterOnExists;
+        ((ObjectNode) reporters.get("file-system-jsonl-reporter")).put("onExists", mapped);
+    }
+
     /**
      * Sets {@code basePath} on a singleton section ({@code pipes-iterator})
      * whose wrapper type matches {@code typeName}.

tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/TikaConfigAsyncWriterTest.java

@@ -83,4 +83,28 @@ public void testTimeoutMillisMapsToTotalAndMerges(@TempDir Path dir) throws Exce
         assertEquals(60000L, timeouts.path("progressTimeoutMillis").asLong());
         assertTrue(timeouts.path("throwOnDeadline").asBoolean());
     }
+
+    @Test
+    public void testOnExistsReachesJsonlReporter(@TempDir Path dir) throws Exception {
+        Path config = dir.resolve("config.json");
+        Files.writeString(config, """
+                {
+                  "pipes-reporters": {
+                    "file-system-jsonl-reporter": { "path": "audit.jsonl" }
+                  }
+                }
+                """);
+        for (String[] pair : new String[][]{{"REPLACE", "REPLACE"}, {"SKIP", "APPEND"}, {"EXCEPTION", "EXCEPTION"}}) {
+            SimpleAsyncConfig simpleAsyncConfig = new SimpleAsyncConfig("input", "output", 4,
+                    null, null, null, config.toAbsolutePath().toString().replace("\\", "/"),
+                    BasicContentHandlerFactory.HANDLER_TYPE.TEXT,
+                    SimpleAsyncConfig.ExtractBytesMode.NONE, null);
+            simpleAsyncConfig.setOnExists(pair[0]);
+            Path tmp = Files.createTempFile(dir, "plugins-", ".json");
+            new PluginsWriter(simpleAsyncConfig, null).write(tmp);
+            JsonNode root = new ObjectMapper().readTree(tmp.toFile());
+            assertEquals(pair[1], root.path("pipes-reporters").path("file-system-jsonl-reporter")
+                    .path("onExists").asText(), pair[0]);
+        }
+    }
 }

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java

@@ -31,6 +31,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -79,6 +80,8 @@ public class AsyncProcessor implements Closeable {
     private final List<ServerManager> serverManagers = new ArrayList<>();
     private final AtomicLong totalProcessed = new AtomicLong(0);
     private final AtomicBoolean applicationErrorOccurred = new AtomicBoolean(false);
+    // first worker/emitter failure; reported to the reporter exactly once
+    private final AtomicReference<ExecutionException> failure = new AtomicReference<>();
     private static long MAX_OFFER_WAIT_MS = 120000;
     private volatile int numParserThreadsFinished = 0;
     private volatile int numEmitterThreadsFinished = 0;
@@ -142,6 +145,12 @@ private AsyncProcessor(Path tikaConfigPath, PipesIterator pipesIterator,
                         checkActive();
                     } catch (InterruptedException e) {
                         return WATCHER_FUTURE_CODE;
+                    } catch (RuntimeException e) {
+                        if (failure.get() == null) {
+                            throw e;
+                        }
+                        // already latched in failure; rethrowing would make this future a second one
+                        return WATCHER_FUTURE_CODE;
                     }
                 }
             });
@@ -322,7 +331,9 @@ public void finished() throws InterruptedException {
     }
 
     public synchronized boolean checkActive() throws InterruptedException {
-
+        if (failure.get() != null) {
+            throw new RuntimeException(failure.get());
+        }
         Future<Integer> future = executorCompletionService.poll();
         if (future != null) {
             try {
@@ -343,8 +354,15 @@ public synchronized boolean checkActive() throws InterruptedException {
                         throw new IllegalArgumentException("Don't recognize this future code: " + i);
                 }
             } catch (ExecutionException e) {
-                LOG.error("execution exception", e);
-                this.pipesReporter.error(e);
+                if (failure.compareAndSet(null, e)) {
+                    LOG.error("execution exception", e);
+                    try {
+                        this.pipesReporter.error(e);
+                    } catch (RuntimeException re) {
+                        // the worker failure is the primary; don't let the reporter mask it
+                        e.addSuppressed(re);
+                    }
+                }
                 throw new RuntimeException(e);
             }
         }
@@ -448,10 +466,16 @@ public Integer call() throws Exception {
                                     describeStopReason(result),
                                     result.status());
                             applicationErrorOccurred.set(true);
-                            pipesReporter.report(t, result, System.currentTimeMillis() - start);
-                            throw new PipesException(describeStopReason(result) + ": " +
+                            PipesException stop = new PipesException(describeStopReason(result) + ": " +
                                     result.status() +
                                     (result.message() != null ? " - " + result.message() : ""));
+                            try {
+                                pipesReporter.report(t, result, System.currentTimeMillis() - start);
+                            } catch (RuntimeException e) {
+                                // the stop reason is the primary failure; don't let the reporter mask it
+                                stop.addSuppressed(e);
+                            }
+                            throw stop;
                         }
                         if (LOG.isTraceEnabled()) {
                             LOG.trace("timer -- pipes client process: {} ms",
@@ -474,13 +498,25 @@ public Integer call() throws Exception {
                                     System.currentTimeMillis() - offerStart);
                         }
                         long elapsed = System.currentTimeMillis() - start;
-                        pipesReporter.report(t, result, elapsed);
+                        report(t, result, elapsed);
                         totalProcessed.incrementAndGet();
                     }
                 }
             }
         }
 
+        // a reporter that throws must stop every worker, or the other workers keep
+        // emitting documents that never get an audit line
+        private void report(FetchEmitTuple t, PipesResult result, long elapsed) throws PipesException {
+            try {
+                pipesReporter.report(t, result, elapsed);
+            } catch (RuntimeException e) {
+                LOG.error("reporter failed; stopping all processing", e);
+                applicationErrorOccurred.set(true);
+                throw new PipesException("reporter failed", e);
+            }
+        }
+
         private boolean shouldEmit(PipesResult result) {
 
             // emitData is null for SUCCESS statuses where the server already emitted

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/reporter/CompositePipesReporter.java

@@ -33,12 +33,31 @@ public CompositePipesReporter(List<PipesReporter> pipesReporterList) {
         pipesReporters = pipesReporterList;
     }
 
+    /**
+     * Every reporter sees every call, even if an earlier one throws; the first
+     * exception is rethrown after the loop with the rest suppressed.
+     */
     @Override
     public void report(FetchEmitTuple t, PipesResult result, long elapsed) {
+        RuntimeException first = null;
         for (PipesReporter reporter : pipesReporters) {
-            reporter.report(t, result, elapsed);
+            try {
+                reporter.report(t, result, elapsed);
+            } catch (RuntimeException e) {
+                first = collect(first, e);
+            }
         }
+        if (first != null) {
+            throw first;
+        }
+    }
 
+    private static RuntimeException collect(RuntimeException first, RuntimeException e) {
+        if (first == null) {
+            return e;
+        }
+        first.addSuppressed(e);
+        return first;
     }
 
     @Override
@@ -60,15 +79,31 @@ public boolean supportsTotalCount() {
 
     @Override
     public void error(Throwable t) {
+        RuntimeException first = null;
         for (PipesReporter reporter : pipesReporters) {
-            reporter.error(t);
+            try {
+                reporter.error(t);
+            } catch (RuntimeException e) {
+                first = collect(first, e);
+            }
+        }
+        if (first != null) {
+            throw first;
         }
     }
 
     @Override
     public void error(String msg) {
+        RuntimeException first = null;
         for (PipesReporter reporter : pipesReporters) {
-            reporter.error(msg);
+            try {
+                reporter.error(msg);
+            } catch (RuntimeException e) {
+                first = collect(first, e);
+            }
+        }
+        if (first != null) {
+            throw first;
         }
     }