This is an automated email from the ASF dual-hosted git repository.
tballison pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tika.git
The following commit(s) were added to refs/heads/main by this push:
new ac31d0c94a TIKA-4846 -- add jsonl reporter (#3082)
ac31d0c94a is described below
commit ac31d0c94a3474bd103f44f74aca663c86d3ba9d
Author: Tim Allison <[email protected]>
AuthorDate: Thu Aug 27 16:46:03 2026 -0400
TIKA-4846 -- add jsonl reporter (#3082)
---
CHANGES.txt | 2 +
.../ROOT/examples/pipes-fs-jsonl-reporter.json | 1 +
.../ROOT/pages/pipes/plugins/filesystem.adoc | 61 +++++
docs/modules/ROOT/pages/pipes/reporters.adoc | 6 +
docs/modules/ROOT/pages/using-tika/cli/index.adoc | 2 +-
.../org/apache/tika/async/cli/PluginsWriter.java | 11 +
.../tika/async/cli/TikaConfigAsyncWriterTest.java | 24 ++
.../tika/pipes/core/async/AsyncProcessor.java | 48 +++-
.../core/reporter/CompositePipesReporter.java | 41 +++-
.../core/reporter/CompositePipesReporterTest.java | 101 ++++++++
.../tika-pipes-file-system/pom.xml | 5 +
.../pipes/reporter/fs/FileSystemJsonlReporter.java | 216 +++++++++++++++++
.../reporter/fs/FileSystemJsonlReporterConfig.java | 48 ++++
.../fs/FileSystemJsonlReporterFactory.java | 41 ++++
.../apache/tika/pipes/fs/ConfigExamplesTest.java | 42 +++-
.../reporter/fs/FileSystemJsonlReporterTest.java | 257 +++++++++++++++++++++
.../file-system-jsonl-reporter.json | 10 +
17 files changed, 905 insertions(+), 11 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 3b336d2e81..4d7573febc 100644
--- a/CHANGES.txt
+++ b/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).
diff --git a/docs/modules/ROOT/examples/pipes-fs-jsonl-reporter.json
b/docs/modules/ROOT/examples/pipes-fs-jsonl-reporter.json
new file mode 120000
index 0000000000..e83e6c241b
--- /dev/null
+++ b/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
\ No newline at end of file
diff --git a/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc
b/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc
index 6fc9333f20..266186421f 100644
--- a/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc
+++ b/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
diff --git a/docs/modules/ROOT/pages/pipes/reporters.adoc
b/docs/modules/ROOT/pages/pipes/reporters.adoc
index 01bc05e604..25f704d2cc 100644
--- a/docs/modules/ROOT/pages/pipes/reporters.adoc
+++ b/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.
diff --git a/docs/modules/ROOT/pages/using-tika/cli/index.adoc
b/docs/modules/ROOT/pages/using-tika/cli/index.adoc
index f4979f4181..c02c2f8771 100644
--- a/docs/modules/ROOT/pages/using-tika/cli/index.adoc
+++ b/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
diff --git
a/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/PluginsWriter.java
b/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/PluginsWriter.java
index fb7ae28278..23d0547ca0 100644
---
a/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/PluginsWriter.java
+++
b/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/PluginsWriter.java
@@ -162,6 +162,7 @@ public class PluginsWriter {
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 @@ public class PluginsWriter {
}
}
+ // 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}.
diff --git
a/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/TikaConfigAsyncWriterTest.java
b/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/TikaConfigAsyncWriterTest.java
index 2f908aa005..fa6c69fb73 100644
---
a/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/TikaConfigAsyncWriterTest.java
+++
b/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/TikaConfigAsyncWriterTest.java
@@ -83,4 +83,28 @@ public class TikaConfigAsyncWriterTest {
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]);
+ }
+ }
}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java
index dbb8336e71..70849fa8c7 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java
@@ -31,6 +31,7 @@ import java.util.concurrent.Future;
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 @@ public class AsyncProcessor implements Closeable {
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 class AsyncProcessor implements Closeable {
}
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 class AsyncProcessor implements Closeable {
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 class AsyncProcessor implements Closeable {
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 class AsyncProcessor implements Closeable {
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
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/reporter/CompositePipesReporter.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/reporter/CompositePipesReporter.java
index f625c67230..de885b6664 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/reporter/CompositePipesReporter.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/reporter/CompositePipesReporter.java
@@ -33,12 +33,31 @@ public class CompositePipesReporter implements
PipesReporter {
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 class CompositePipesReporter implements
PipesReporter {
@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;
}
}
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/reporter/CompositePipesReporterTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/reporter/CompositePipesReporterTest.java
new file mode 100644
index 0000000000..44f45040e7
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/reporter/CompositePipesReporterTest.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.pipes.core.reporter;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+
+import org.apache.tika.pipes.api.FetchEmitTuple;
+import org.apache.tika.pipes.api.PipesResult;
+import org.apache.tika.pipes.api.emitter.EmitKey;
+import org.apache.tika.pipes.api.fetcher.FetchKey;
+import org.apache.tika.pipes.api.pipesiterator.TotalCountResult;
+import org.apache.tika.pipes.api.reporter.PipesReporter;
+import org.apache.tika.plugins.ExtensionConfig;
+
+public class CompositePipesReporterTest {
+
+ private static class Recording implements PipesReporter {
+ final List<String> seen = new ArrayList<>();
+ final boolean throwing;
+
+ Recording(boolean throwing) {
+ this.throwing = throwing;
+ }
+
+ @Override
+ public void report(FetchEmitTuple t, PipesResult result, long elapsed)
{
+ seen.add(t.getId());
+ if (throwing) {
+ throw new IllegalStateException("boom " + t.getId());
+ }
+ }
+
+ @Override
+ public void report(TotalCountResult totalCountResult) {
+ }
+
+ @Override
+ public boolean supportsTotalCount() {
+ return false;
+ }
+
+ @Override
+ public void error(Throwable t) {
+ seen.add("error");
+ if (throwing) {
+ throw new IllegalStateException("boom error");
+ }
+ }
+
+ @Override
+ public void error(String msg) {
+ error(new RuntimeException(msg));
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public ExtensionConfig getExtensionConfig() {
+ return null;
+ }
+ }
+
+ @Test
+ public void testThrowingReporterDoesNotStarveSiblings() {
+ Recording first = new Recording(true);
+ Recording second = new Recording(false);
+ CompositePipesReporter composite = new
CompositePipesReporter(List.of(first, second));
+ FetchEmitTuple t = new FetchEmitTuple("a", new FetchKey("f", "a"), new
EmitKey("e", "a"));
+ PipesResult result = new PipesResult(PipesResult.RESULT_STATUS.OOM);
+
+ IllegalStateException e = assertThrows(IllegalStateException.class,
+ () -> composite.report(t, result, 1));
+ assertEquals("boom a", e.getMessage());
+ assertEquals(List.of("a"), second.seen);
+
+ assertThrows(IllegalStateException.class, () ->
composite.error("dead"));
+ assertEquals(List.of("a", "error"), second.seen);
+ }
+}
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/pom.xml
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/pom.xml
index 06d1602e00..d6ce406eeb 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/pom.xml
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/pom.xml
@@ -41,6 +41,11 @@
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tika-pipes-reporter-commons</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jsr310</artifactId>
diff --git
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporter.java
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporter.java
new file mode 100644
index 0000000000..0d74ce5445
--- /dev/null
+++
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporter.java
@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.pipes.reporter.fs;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.charset.CodingErrorAction;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.time.Instant;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.pipes.api.FetchEmitTuple;
+import org.apache.tika.pipes.api.PipesResult;
+import org.apache.tika.pipes.api.pipesiterator.TotalCountResult;
+import org.apache.tika.pipes.reporters.PipesReporterBase;
+import org.apache.tika.plugins.ExtensionConfig;
+import org.apache.tika.utils.ExceptionUtils;
+
+/**
+ * Append-only per-document audit log: one JSON object per line for every
result
+ * accepted by the includes/excludes filter. The line's {@code id} is the
+ * {@link FetchEmitTuple#getId()} verbatim, so consumers join on it.
+ * <p>
+ * Each line is written and flushed to the OS synchronously in {@link
#report}, so
+ * it survives the driver process dying (not a host crash; there is no fsync).
A
+ * write failure (disk full, etc.) throws from that {@link #report} and every
later
+ * one rather than dropping lines silently.
+ */
+public class FileSystemJsonlReporter extends PipesReporterBase {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(FileSystemJsonlReporter.class);
+
+ private record Line(String id, String status, String category, String
message, long elapsedMs, String timestamp) {
+ }
+
+ private record ErrorLine(String error, String timestamp) {
+ }
+
+ public static FileSystemJsonlReporter build(ExtensionConfig pluginConfig)
throws TikaConfigException, IOException {
+ FileSystemJsonlReporterConfig config =
FileSystemJsonlReporterConfig.load(pluginConfig.json());
+ return new FileSystemJsonlReporter(pluginConfig, config);
+ }
+
+ private final FileSystemJsonlReporterConfig config;
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final BufferedWriter writer;
+ private IOException writerFailure;
+ private boolean closed;
+
+ public FileSystemJsonlReporter(ExtensionConfig pluginConfig,
FileSystemJsonlReporterConfig config) throws TikaConfigException, IOException {
+ super(pluginConfig, config.includes(), config.excludes());
+ this.config = config;
+ if (config.path() == null) {
+ throw new TikaConfigException("must initialize 'path'");
+ }
+ this.writer = open(config);
+ }
+
+ private static BufferedWriter open(FileSystemJsonlReporterConfig config)
throws TikaConfigException, IOException {
+ Path path = config.path();
+ if (Files.isDirectory(path)) {
+ throw new TikaConfigException("'" + path + "' is a directory;
'path' must be a file");
+ }
+ if (path.getParent() != null) {
+ Files.createDirectories(path.getParent());
+ }
+ StandardOpenOption[] options = switch (config.onExists()) {
+ case EXCEPTION -> new
StandardOpenOption[]{StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE};
+ case APPEND -> new StandardOpenOption[]{StandardOpenOption.CREATE,
StandardOpenOption.WRITE, StandardOpenOption.APPEND};
+ case REPLACE -> new
StandardOpenOption[]{StandardOpenOption.CREATE, StandardOpenOption.WRITE,
StandardOpenOption.TRUNCATE_EXISTING};
+ };
+ boolean needsNewline = config.onExists() ==
FileSystemJsonlReporterConfig.ON_EXISTS.APPEND && lacksTrailingNewline(path);
+ if (needsNewline) {
+ LOG.warn("'{}' ends in a partial line (a previous run died
mid-write); terminating it at offset {}", path, Files.size(path));
+ }
+ try {
+ // lone surrogates in ids/messages must not kill the log; default
encoder would throw
+ BufferedWriter writer = new BufferedWriter(new
OutputStreamWriter(Files.newOutputStream(path, options),
+ StandardCharsets.UTF_8.newEncoder()
+ .onMalformedInput(CodingErrorAction.REPLACE)
+
.onUnmappableCharacter(CodingErrorAction.REPLACE)));
+ if (needsNewline) {
+ writer.write('\n');
+ }
+ return writer;
+ } catch (FileAlreadyExistsException e) {
+ throw new TikaConfigException("'" + path + "' already exists; set
onExists to APPEND or REPLACE to reuse it", e);
+ }
+ }
+
+ private static boolean lacksTrailingNewline(Path path) throws IOException {
+ if (!Files.isRegularFile(path) || Files.size(path) == 0) {
+ return false;
+ }
+ try (SeekableByteChannel ch = Files.newByteChannel(path,
StandardOpenOption.READ)) {
+ ByteBuffer last = ByteBuffer.allocate(1);
+ ch.position(ch.size() - 1);
+ ch.read(last);
+ return last.get(0) != '\n';
+ } catch (AccessDeniedException e) {
+ // write-only file: can't inspect, so don't require read
permission just for this
+ LOG.warn("can't read '{}' to check for a partial last line;
appending as-is", path);
+ return false;
+ }
+ }
+
+ @Override
+ public void report(FetchEmitTuple t, PipesResult result, long elapsed) {
+ if (!accept(result.status())) {
+ return;
+ }
+ write(new Line(t.getId(), result.status().name(),
result.status().getCategory().name(),
+ truncate(result.message()), elapsed,
Instant.now().toString()), t.getId());
+ }
+
+ private String truncate(String msg) {
+ int max = config.maxMessageLength();
+ if (msg == null || msg.length() <= max) {
+ return msg;
+ }
+ int cut = Character.isHighSurrogate(msg.charAt(max - 1)) ? max - 1 :
max;
+ return msg.substring(0, cut) + "...[truncated " + (msg.length() - cut)
+ " chars]";
+ }
+
+ private synchronized void write(Object line, String id) {
+ if (writerFailure != null) {
+ throw new IllegalStateException("jsonl reporter writer failed
earlier; refusing to drop lines silently", writerFailure);
+ }
+ if (closed) {
+ LOG.warn("jsonl reporter already closed; dropping report for {}",
id);
+ return;
+ }
+ try {
+ // always \n, never the platform separator: jsonl is \n-delimited
+ writer.write(mapper.writeValueAsString(line));
+ writer.write('\n');
+ writer.flush();
+ } catch (IOException e) {
+ LOG.error("jsonl reporter failed writing {}", config.path(), e);
+ writerFailure = e;
+ throw new IllegalStateException("jsonl reporter failed writing " +
config.path(), e);
+ }
+ }
+
+ @Override
+ public void report(TotalCountResult totalCountResult) {
+ //no-op
+ }
+
+ @Override
+ public boolean supportsTotalCount() {
+ return false;
+ }
+
+ @Override
+ public void error(Throwable t) {
+ error(ExceptionUtils.getStackTrace(t));
+ }
+
+ @Override
+ public synchronized void error(String msg) {
+ // close() may never be called after this; get the line on disk now
+ try {
+ write(new ErrorLine(truncate(msg), Instant.now().toString()),
"<error>");
+ } catch (IllegalStateException e) {
+ LOG.warn("couldn't record error in jsonl reporter", e);
+ }
+ try {
+ closeWriter();
+ } catch (IOException e) {
+ LOG.warn("problem closing {}", config.path(), e);
+ }
+ }
+
+ @Override
+ public synchronized void close() throws IOException {
+ closeWriter();
+ if (writerFailure != null) {
+ throw writerFailure;
+ }
+ }
+
+ private void closeWriter() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ writer.close();
+ }
+}
diff --git
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterConfig.java
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterConfig.java
new file mode 100644
index 0000000000..2837e0b307
--- /dev/null
+++
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterConfig.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.pipes.reporter.fs;
+
+import java.nio.file.Path;
+import java.util.Set;
+
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.plugins.PluginJson;
+
+public record FileSystemJsonlReporterConfig(Path path, Set<String> includes,
Set<String> excludes, ON_EXISTS onExists, int maxMessageLength) {
+
+ public enum ON_EXISTS {
+ EXCEPTION, APPEND, REPLACE
+ }
+
+ public static final int DEFAULT_MAX_MESSAGE_LENGTH = 10_000;
+
+ public FileSystemJsonlReporterConfig {
+ if (onExists == null) {
+ onExists = ON_EXISTS.EXCEPTION;
+ }
+ if (maxMessageLength < 0) {
+ throw new IllegalArgumentException("maxMessageLength must be >= 0;
0 means the default (" + DEFAULT_MAX_MESSAGE_LENGTH + ")");
+ }
+ if (maxMessageLength == 0) {
+ maxMessageLength = DEFAULT_MAX_MESSAGE_LENGTH;
+ }
+ }
+
+ public static FileSystemJsonlReporterConfig load(final String json) throws
TikaConfigException {
+ return PluginJson.read(json, FileSystemJsonlReporterConfig.class);
+ }
+}
diff --git
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterFactory.java
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterFactory.java
new file mode 100644
index 0000000000..8d80a67166
--- /dev/null
+++
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterFactory.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.pipes.reporter.fs;
+
+import java.io.IOException;
+
+import org.pf4j.Extension;
+
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.pipes.api.reporter.PipesReporterFactory;
+import org.apache.tika.plugins.ExtensionConfig;
+
+@Extension
+public class FileSystemJsonlReporterFactory implements PipesReporterFactory {
+
+ public static final String NAME = "file-system-jsonl-reporter";
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public FileSystemJsonlReporter buildExtension(ExtensionConfig
extensionConfig) throws IOException, TikaConfigException {
+ return FileSystemJsonlReporter.build(extensionConfig);
+ }
+}
diff --git
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/fs/ConfigExamplesTest.java
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/fs/ConfigExamplesTest.java
index 041e079e15..5754091fb0 100644
---
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/fs/ConfigExamplesTest.java
+++
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/fs/ConfigExamplesTest.java
@@ -16,12 +16,25 @@
*/
package org.apache.tika.pipes.fs;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.InputStream;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Paths;
+import java.util.Collections;
+
+import com.fasterxml.jackson.databind.JsonNode;
import org.junit.jupiter.api.Test;
import org.apache.tika.pipes.core.testutil.AbstractConfigExamplesTest;
+import org.apache.tika.pipes.reporter.fs.FileSystemJsonlReporterConfig;
+import org.apache.tika.pipes.reporter.fs.FileSystemJsonlReporterFactory;
/**
- * Validates file system fetcher/emitter configuration examples used in
documentation.
+ * Validates file system plugin configuration examples used in documentation.
* <p>
* The JSON configuration examples are stored in {@code
src/test/resources/config-examples/}
* and are included directly in the AsciiDoc documentation via the {@code
include::} directive.
@@ -42,4 +55,31 @@ public class ConfigExamplesTest extends
AbstractConfigExamplesTest {
public void testFileSystemPipelineConfig() throws Exception {
loadAndValidate("file-system-pipeline.json");
}
+
+ @Test
+ public void testFileSystemJsonlReporterConfig() throws Exception {
+ loadAndValidate("file-system-jsonl-reporter.json");
+
+ JsonNode inner =
innerComponent(readExample("file-system-jsonl-reporter.json"),
+ "pipes-reporters", null, "file-system-jsonl-reporter");
+ FileSystemJsonlReporterConfig config =
FileSystemJsonlReporterConfig.load(inner.toString());
+ assertEquals(Paths.get("/var/log/tika/pipes-audit.jsonl"),
config.path());
+ assertEquals(FileSystemJsonlReporterConfig.ON_EXISTS.EXCEPTION,
config.onExists());
+ assertEquals(10000, config.maxMessageLength());
+ assertTrue(config.includes().contains("OOM"));
+ assertNull(config.excludes());
+ }
+
+ // the plugin only resolves by name if the factory made it into pf4j's
index
+ @Test
+ public void testJsonlReporterFactoryIsRegistered() throws Exception {
+ // test-classes carries its own (empty) index that shadows main's, so
scan them all
+ StringBuilder all = new StringBuilder();
+ for (URL url :
Collections.list(getClass().getClassLoader().getResources("META-INF/extensions.idx")))
{
+ try (InputStream is = url.openStream()) {
+ all.append(new String(is.readAllBytes(),
StandardCharsets.UTF_8));
+ }
+ }
+
assertTrue(all.toString().contains(FileSystemJsonlReporterFactory.class.getName()),
all.toString());
+ }
}
diff --git
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterTest.java
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterTest.java
new file mode 100644
index 0000000000..a52621196c
--- /dev/null
+++
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/reporter/fs/FileSystemJsonlReporterTest.java
@@ -0,0 +1,257 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.pipes.reporter.fs;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.pipes.api.FetchEmitTuple;
+import org.apache.tika.pipes.api.PipesResult;
+import org.apache.tika.pipes.api.emitter.EmitKey;
+import org.apache.tika.pipes.api.fetcher.FetchKey;
+import org.apache.tika.plugins.ExtensionConfig;
+
+public class FileSystemJsonlReporterTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static FileSystemJsonlReporter build(Path path, Map<String,
Object> extra) throws Exception {
+ Map<String, Object> cfg = new LinkedHashMap<>();
+ cfg.put("path", path.toAbsolutePath().toString());
+ cfg.putAll(extra);
+ return build(MAPPER.writeValueAsString(cfg));
+ }
+
+ private static FileSystemJsonlReporter build(String json) throws Exception
{
+ return new FileSystemJsonlReporterFactory().buildExtension(new
ExtensionConfig("test", FileSystemJsonlReporterFactory.NAME, json));
+ }
+
+ private static void report(FileSystemJsonlReporter r, String id,
PipesResult.RESULT_STATUS status, String msg) {
+ r.report(new FetchEmitTuple(id, new FetchKey("f", id), new
EmitKey("e", id)), new PipesResult(status, msg), 7);
+ }
+
+ private static List<Map<String, Object>> lines(Path path) throws
IOException {
+ List<Map<String, Object>> ret = new ArrayList<>();
+ for (String line : Files.readAllLines(path, StandardCharsets.UTF_8)) {
+ ret.add(MAPPER.readValue(line, Map.class));
+ }
+ return ret;
+ }
+
+ private static List<Object> ids(Path path) throws IOException {
+ return lines(path).stream().map(m -> m.get("id")).toList();
+ }
+
+ @Test
+ public void testIncludesAndFields(@TempDir Path tmp) throws Exception {
+ Path path = tmp.resolve("audit.jsonl");
+ try (FileSystemJsonlReporter r = build(path, Map.of("includes",
List.of("OOM", "TIMEOUT")))) {
+ report(r, "a/b.pdf", PipesResult.RESULT_STATUS.OOM, "boom");
+ report(r, "c.doc", PipesResult.RESULT_STATUS.PARSE_SUCCESS, null);
+ report(r, "d.doc", PipesResult.RESULT_STATUS.TIMEOUT, null);
+ }
+ List<Map<String, Object>> lines = lines(path);
+ assertEquals(2, lines.size());
+ Map<String, Object> first = lines.get(0);
+ assertEquals("a/b.pdf", first.get("id"));
+ assertEquals("OOM", first.get("status"));
+ assertEquals("PROCESS_CRASH", first.get("category"));
+ assertEquals("boom", first.get("message"));
+ assertEquals(7, first.get("elapsedMs"));
+ assertTrue(first.get("timestamp").toString().endsWith("Z"));
+ assertEquals(Set.of("id", "status", "category", "message",
"elapsedMs", "timestamp"), first.keySet());
+ assertEquals("d.doc", lines.get(1).get("id"));
+ }
+
+ @Test
+ public void testExcludes(@TempDir Path tmp) throws Exception {
+ Path path = tmp.resolve("audit.jsonl");
+ try (FileSystemJsonlReporter r = build(path, Map.of("excludes",
List.of("PARSE_SUCCESS")))) {
+ report(r, "a", PipesResult.RESULT_STATUS.PARSE_SUCCESS, null);
+ report(r, "b", PipesResult.RESULT_STATUS.OOM, null);
+ }
+ assertEquals(List.of("b"), ids(path));
+ }
+
+ @Test
+ public void testConcurrentReportsAllLand(@TempDir Path tmp) throws
Exception {
+ Path path = tmp.resolve("audit.jsonl");
+ int threads = 8;
+ int perThread = 500;
+ Set<String> expected = new HashSet<>();
+ try (FileSystemJsonlReporter r = build(path, Map.of())) {
+ ExecutorService ex = Executors.newFixedThreadPool(threads);
+ List<Future<?>> futures = new ArrayList<>();
+ for (int t = 0; t < threads; t++) {
+ final int tid = t;
+ futures.add(ex.submit(() -> {
+ for (int i = 0; i < perThread; i++) {
+ report(r, tid + "/" + i,
PipesResult.RESULT_STATUS.PARSE_SUCCESS, null);
+ }
+ }));
+ for (int i = 0; i < perThread; i++) {
+ expected.add(tid + "/" + i);
+ }
+ }
+ for (Future<?> f : futures) {
+ f.get();
+ }
+ ex.shutdown();
+ }
+ List<Object> ids = ids(path);
+ assertEquals(threads * perThread, ids.size());
+ assertEquals(expected, new HashSet<>(ids));
+ }
+
+ @Test
+ public void testOnExists(@TempDir Path tmp) throws Exception {
+ Path path = tmp.resolve("audit.jsonl");
+ try (FileSystemJsonlReporter r = build(path, Map.of())) {
+ report(r, "first", PipesResult.RESULT_STATUS.OOM, null);
+ }
+ assertThrows(TikaConfigException.class, () -> build(path, Map.of()));
+ assertThrows(TikaConfigException.class, () -> build(path,
Map.of("onExists", "EXCEPTION")));
+ assertEquals(List.of("first"), ids(path));
+
+ try (FileSystemJsonlReporter r = build(path, Map.of("onExists",
"APPEND"))) {
+ report(r, "second", PipesResult.RESULT_STATUS.OOM, null);
+ }
+ assertEquals(List.of("first", "second"), ids(path));
+
+ try (FileSystemJsonlReporter r = build(path, Map.of("onExists",
"REPLACE"))) {
+ report(r, "third", PipesResult.RESULT_STATUS.OOM, null);
+ }
+ assertEquals(List.of("third"), ids(path));
+ }
+
+ @Test
+ public void testAppendAfterPartialLine(@TempDir Path tmp) throws Exception
{
+ Path path = tmp.resolve("audit.jsonl");
+ Files.writeString(path, "{\"id\":\"cut", StandardCharsets.UTF_8);
+ try (FileSystemJsonlReporter r = build(path, Map.of("onExists",
"APPEND"))) {
+ report(r, "next", PipesResult.RESULT_STATUS.OOM, null);
+ }
+ List<String> raw = Files.readAllLines(path, StandardCharsets.UTF_8);
+ assertEquals(2, raw.size());
+ assertEquals("next", MAPPER.readValue(raw.get(1),
Map.class).get("id"));
+ }
+
+ @Test
+ public void testMessageCap(@TempDir Path tmp) throws Exception {
+ Path path = tmp.resolve("audit.jsonl");
+ try (FileSystemJsonlReporter r = build(path,
Map.of("maxMessageLength", 10))) {
+ report(r, "x", PipesResult.RESULT_STATUS.OOM, "0123456789abcdef");
+ report(r, "y", PipesResult.RESULT_STATUS.OOM, "012345678😀ab");
+ }
+ List<Map<String, Object>> lines = lines(path);
+ assertEquals("0123456789...[truncated 6 chars]",
lines.get(0).get("message"));
+ // cut lands on a surrogate pair: back off one so the pair isn't split
+ assertEquals("012345678...[truncated 4 chars]",
lines.get(1).get("message"));
+ }
+
+ @Test
+ public void testLoneSurrogateDoesNotKillWriter(@TempDir Path tmp) throws
Exception {
+ Path path = tmp.resolve("audit.jsonl");
+ try (FileSystemJsonlReporter r = build(path, Map.of())) {
+ report(r, "bad" + (char) 0xD83D, PipesResult.RESULT_STATUS.OOM,
(char) 0xDE00 + " lone low");
+ report(r, "after", PipesResult.RESULT_STATUS.OOM, null);
+ }
+ List<Map<String, Object>> lines = lines(path);
+ assertEquals(2, lines.size());
+ assertEquals("after", lines.get(1).get("id"));
+ }
+
+ @Test
+ public void testMultilineMessageStaysOneLine(@TempDir Path tmp) throws
Exception {
+ Path path = tmp.resolve("audit.jsonl");
+ try (FileSystemJsonlReporter r = build(path, Map.of())) {
+ report(r, "x", PipesResult.RESULT_STATUS.OOM, "line1\nline2\r\n
line3");
+ }
+ assertEquals(1, Files.readAllLines(path).size());
+ assertEquals("line1\nline2\r\n line3",
lines(path).get(0).get("message"));
+ }
+
+ @Test
+ public void testErrorFlushesWithoutClose(@TempDir Path tmp) throws
Exception {
+ Path path = tmp.resolve("audit.jsonl");
+ FileSystemJsonlReporter r = build(path, Map.of());
+ report(r, "x", PipesResult.RESULT_STATUS.OOM, null);
+ r.error(new RuntimeException("fatal"));
+ List<Map<String, Object>> lines = lines(path);
+ assertEquals(2, lines.size());
+ assertEquals(Set.of("error", "timestamp"), lines.get(1).keySet());
+ assertTrue(lines.get(1).get("error").toString().contains("fatal"));
+ // late reports after error/close are dropped, not thrown
+ assertDoesNotThrow(() -> report(r, "y", PipesResult.RESULT_STATUS.OOM,
null));
+ assertDoesNotThrow(r::close);
+ assertEquals(2, lines(path).size());
+ }
+
+ @Test
+ @EnabledOnOs(OS.LINUX)
+ public void testWriteFailureIsLoud() throws Exception {
+ Path devFull = Path.of("/dev/full");
+ try (FileSystemJsonlReporter r = build(devFull, Map.of("onExists",
"APPEND"))) {
+ assertThrows(IllegalStateException.class, () -> report(r, "x",
PipesResult.RESULT_STATUS.OOM, null));
+ assertThrows(IllegalStateException.class, () -> report(r, "y",
PipesResult.RESULT_STATUS.OOM, null));
+ assertThrows(IOException.class, r::close);
+ } catch (IOException expected) {
+ //try-with-resources close
+ }
+ }
+
+ @Test
+ public void testConfigErrors(@TempDir Path tmp) throws Exception {
+ assertThrows(TikaConfigException.class, () -> build("{}"));
+ assertThrows(TikaConfigException.class, () ->
build(tmp.resolve("a.jsonl"), Map.of("maxMessageLength", -1)));
+ assertThrows(TikaConfigException.class, () -> build(tmp, Map.of()));
+ assertFalse(Files.exists(tmp.resolve("a.jsonl")));
+ }
+
+ @Test
+ public void testCreatesParentDirs(@TempDir Path tmp) throws Exception {
+ Path path = tmp.resolve("a/b/audit.jsonl");
+ try (FileSystemJsonlReporter r = build(path, Map.of())) {
+ report(r, "x", PipesResult.RESULT_STATUS.OOM, null);
+ }
+ assertEquals(1, lines(path).size());
+ }
+}
diff --git
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/resources/config-examples/file-system-jsonl-reporter.json
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/resources/config-examples/file-system-jsonl-reporter.json
new file mode 100644
index 0000000000..b720101362
--- /dev/null
+++
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/resources/config-examples/file-system-jsonl-reporter.json
@@ -0,0 +1,10 @@
+{
+ "pipes-reporters": {
+ "file-system-jsonl-reporter": {
+ "path": "/var/log/tika/pipes-audit.jsonl",
+ "includes": ["OOM", "TIMEOUT", "UNSPECIFIED_CRASH",
"PAYLOAD_LIMIT_EXCEEDED", "EMIT_EXCEPTION", "FETCH_EXCEPTION"],
+ "onExists": "EXCEPTION",
+ "maxMessageLength": 10000
+ }
+ }
+}
