diff --git a/insight-plugin/README.md b/insight-plugin/README.md index f626ad830..ced08e49b 100644 --- a/insight-plugin/README.md +++ b/insight-plugin/README.md @@ -28,11 +28,13 @@ DurableConfig config = DurableConfig.builder() Exporters: `LambdaLogExporter` (default; writes the `operationsByName` map to stdout → CloudWatch), `S3Exporter` (canonical `operations` array, one object per execution), -`CloudWatchLogsExporter` (PutLogEvents to a specific log group, `operationsByName` map). Implement -`InsightExporter` for custom sinks. +`CloudWatchLogsExporter` (PutLogEvents to a specific log group, `operationsByName` map), +`FileExporter` (writes to the local filesystem — an EFS mount, an S3 File Gateway path, or `/tmp`). +Implement `InsightExporter` for custom sinks. -`LambdaLogExporter` needs no extra dependency. The AWS SDK service modules used by the remote -exporters are optional so applications that use only Lambda logs do not package them. Add the module +`LambdaLogExporter` and `FileExporter` need no extra dependency (`FileExporter` uses only +`java.nio.file`). The AWS SDK service modules used by the remote exporters are optional so +applications that use only Lambda logs or the filesystem do not package them. Add the module for each remote exporter you configure, using the AWS SDK for Java 2.x version managed by your application: @@ -52,6 +54,44 @@ application: ``` +### FileExporter + +Writes records to any writable directory — a Lambda [EFS mount](https://docs.aws.amazon.com/lambda/latest/dg/services-efs.html), +an S3 File Gateway path, or `/tmp` for local testing. Two modes: + +```java +// NDJSON (default): append one compact JSON line per emission to a date-partitioned file, +// {directory}/{YYYY-MM-DD}.ndjson (date is the record's emittedAt day, UTC) +.addExporter(FileExporter.builder() + .directory("/mnt/efs/workflow-insight") + .build()) + +// JSON: one pretty-printed file per execution, overwritten on each update, +// {directory}/{executionName}.json +.addExporter(FileExporter.builder() + .directory("/mnt/efs/workflow-insight") + .mode(FileExporter.Mode.JSON) + .operationsFormat(FileExporter.OperationsFormat.BOTH) // ARRAY (default) | BY_NAME | BOTH + .build()) +``` + +- **`directory`** is required; the exporter creates it (recursively) on first write. +- **File names are deterministic and safe.** NDJSON files are named only by UTC date; JSON files use + the execution name (falling back to the ARN), with every character outside `[a-zA-Z0-9._-]` + replaced by `_`, so a name can never contain a path separator or `..`. +- **`maxRecordSizeBytes`** has no default — the filesystem has no practical per-record limit. Set it + only if you want smaller files; the plugin then truncates each record (oldest results first) to fit + before it is written. +- On `/tmp` (ephemeral, per-container) files do not survive a cold start; use an EFS mount to persist + across invocations and containers. +- **Concurrent NDJSON append is not atomic on shared filesystems.** NDJSON mode uses `Files.write(..., APPEND)`. + A single writer on a local disk (`/tmp`) appends one whole line at a time, so lines never interleave. On a + **shared NFS/EFS mount written by more than one Lambda environment at once**, the append is not guaranteed + atomic: concurrent writers can interleave partial lines or overwrite each other, producing malformed or lost + records. If multiple execution environments may write to the same directory, prefer **JSON mode** (one file per + execution, keyed by name/ARN — no shared append) or ensure a **single writer** per NDJSON file (for example, a + per-environment subdirectory). + ## Design - **Snapshot-based, not accumulated.** Each record is built directly from the current-invocation diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/Json.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/Json.java index 352bb132d..c6a241a15 100644 --- a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/Json.java +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/Json.java @@ -3,6 +3,8 @@ package software.amazon.lambda.durable.insight; import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.util.DefaultIndenter; +import com.fasterxml.jackson.core.util.DefaultPrettyPrinter; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; @@ -23,6 +25,18 @@ public final class Json { .registerModule(new JavaTimeModule()) .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + // 2-space indentation with "\n" line breaks, matching JS JSON.stringify(x, null, 2) (LF, no CR). The + // DefaultPrettyPrinter is stateful, so build a fresh instance per write via createInstance(). + private static final DefaultPrettyPrinter PRETTY_PRINTER = buildPrettyPrinter(); + + private static DefaultPrettyPrinter buildPrettyPrinter() { + DefaultIndenter indenter = new DefaultIndenter(" ", "\n"); + DefaultPrettyPrinter pp = new DefaultPrettyPrinter(); + pp.indentObjectsWith(indenter); + pp.indentArraysWith(indenter); + return pp; + } + private Json() {} public static String stringify(Object value) { @@ -33,6 +47,15 @@ public static String stringify(Object value) { } } + /** Pretty-prints with 2-space indentation (matches the JS {@code JSON.stringify(x, null, 2)} file output). */ + public static String stringifyPretty(Object value) { + try { + return MAPPER.writer(PRETTY_PRINTER).writeValueAsString(value); + } catch (JsonProcessingException e) { + throw new IllegalStateException("failed to serialize insight record", e); + } + } + /** UTF-8 byte length of the value's JSON, or {@code null} if it can't be serialized. */ public static Integer byteSize(Object value) { try { diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsightRecord.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsightRecord.java index f2a1b1060..b85bfa18f 100644 --- a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsightRecord.java +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsightRecord.java @@ -49,6 +49,10 @@ void addOperation(OperationRecord operation) { operations.add(operation); } + public String emittedAt() { + return emittedAt; + } + public String executionArn() { return executionArn; } @@ -204,4 +208,22 @@ public Map toByNameWireMap() { putTruncationMarkers(data); return data; } + + /** + * Combined wire map mirroring the JS {@code applyOperationsFormat(record, "both")} shape: the canonical + * {@code operations} array plus an added {@code operationsByName} map. JS spreads the whole record and then adds + * the key ({@code {...record, operationsByName}}), so {@code operationsByName} is the LAST key — after every record + * field and any truncation markers. This method preserves that order by starting from the canonical map (which + * already ends with the truncation markers) and appending {@code operationsByName} last. + */ + public Map toBothWireMap() { + Map data = toWireMap(); + Map byName = new LinkedHashMap<>(); + for (Map.Entry e : + OperationsIndex.buildOperationsByName(operations).entrySet()) { + byName.put(e.getKey(), e.getValue().toWireMap()); + } + data.put("operationsByName", byName); + return data; + } } diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/FileExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/FileExporter.java new file mode 100644 index 000000000..155ee42cc --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/FileExporter.java @@ -0,0 +1,228 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.concurrent.atomic.AtomicBoolean; +import software.amazon.lambda.durable.annotations.Experimental; +import software.amazon.lambda.durable.insight.InsightExporter; +import software.amazon.lambda.durable.insight.Json; +import software.amazon.lambda.durable.insight.WorkflowInsightRecord; + +/** + * Exports workflow insight records to the local filesystem (an EFS mount, an S3 File Gateway path, {@code /tmp} for + * testing, or any writable directory). Mirrors the JS {@code FileExporter}. + * + *

Two output modes: + * + *

    + *
  • {@link Mode#NDJSON} (default) — every record is appended as one compact JSON line to a date-partitioned file + * {@code {directory}/{YYYY-MM-DD}.ndjson}, where the date is the record's {@code emittedAt} day (UTC). One line + * per emission, so replay/suspend emissions accumulate. + *
  • {@link Mode#JSON} — each execution gets its own pretty-printed (2-space) file + * {@code {directory}/{executionName}.json}, overwritten on each update so the file always holds the latest view. + *
+ * + *

File names are derived only from the record's {@code emittedAt} day (NDJSON) or a sanitized execution name/ARN + * (JSON): every character outside {@code [a-zA-Z0-9._-]} is replaced with {@code _}, so the name is deterministic and + * cannot escape {@code directory} via path separators or {@code ..}. The resolved path is additionally normalized and + * verified to stay within {@code directory} as defense in depth. + * + *

The plugin applies {@link #maxRecordSizeBytes()} truncation against {@link #render(WorkflowInsightRecord)} before + * {@link #export(WorkflowInsightRecord)} is called; the filesystem has no practical per-record limit, so there is no + * default cap. Writes are immediate, so {@link #flush()} is a no-op. + * + * @see InsightExporter + */ +@Experimental +public final class FileExporter implements InsightExporter { + + /** File output mode. */ + @Experimental + public enum Mode { + /** Append every record as one compact JSON line to a date-partitioned {@code .ndjson} file (default). */ + NDJSON, + /** Write one pretty-printed JSON file per execution, overwriting on update. */ + JSON + } + + /** How operations are rendered in the written record. */ + @Experimental + public enum OperationsFormat { + /** The canonical {@code operations} array (default). */ + ARRAY, + /** The {@code operationsByName} map, replacing the array. */ + BY_NAME, + /** Both the {@code operations} array and an added {@code operationsByName} map. */ + BOTH + } + + private final Path directory; + private final Mode mode; + private final OperationsFormat operationsFormat; + private final Integer maxRecordSizeBytes; + private final AtomicBoolean dirCreated = new AtomicBoolean(false); + + private FileExporter(Builder b) { + this.directory = b.directory; + this.mode = b.mode != null ? b.mode : Mode.NDJSON; + this.operationsFormat = b.operationsFormat != null ? b.operationsFormat : OperationsFormat.ARRAY; + this.maxRecordSizeBytes = b.maxRecordSizeBytes; + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public Object render(WorkflowInsightRecord record) { + switch (operationsFormat) { + case BY_NAME: + return record.toByNameWireMap(); + case BOTH: + return record.toBothWireMap(); + case ARRAY: + default: + return record.toWireMap(); + } + } + + @Override + public void export(WorkflowInsightRecord record) { + ensureDir(); + Object formatted = render(record); + try { + if (mode == Mode.NDJSON) { + Path file = resolveChild(datePartition(record) + ".ndjson"); + Files.write( + file, + (Json.stringify(formatted) + "\n").getBytes(StandardCharsets.UTF_8), + StandardOpenOption.CREATE, + StandardOpenOption.APPEND); + } else { + String name = record.executionName() != null ? record.executionName() : record.executionArn(); + Path file = resolveChild(sanitize(name) + ".json"); + Files.write( + file, + Json.stringifyPretty(formatted).getBytes(StandardCharsets.UTF_8), + StandardOpenOption.CREATE, + StandardOpenOption.WRITE, + StandardOpenOption.TRUNCATE_EXISTING); + } + } catch (IOException e) { + throw new UncheckedIOException("failed to write insight record to " + directory, e); + } + } + + /** Writes are immediate; nothing is buffered. */ + @Override + public void flush() { + // no-op + } + + /** UTC day of the record's {@code emittedAt}, or of "now" when the record carries no {@code emittedAt}. */ + private static String datePartition(WorkflowInsightRecord record) { + String emittedAt = record.emittedAt(); + if (emittedAt != null && emittedAt.length() >= 10) { + return emittedAt.substring(0, 10); + } + return java.time.Instant.now() + .atZone(java.time.ZoneOffset.UTC) + .toLocalDate() + .toString(); + } + + private void ensureDir() { + if (dirCreated.get()) { + return; + } + try { + Files.createDirectories(directory); + dirCreated.set(true); + } catch (IOException e) { + throw new UncheckedIOException("failed to create insight directory " + directory, e); + } + } + + /** + * Resolves a sanitized child file name against {@code directory} and verifies the normalized result stays inside + * {@code directory}. The name is already sanitized to {@code [a-zA-Z0-9._-]}, so this is defense in depth against + * any future change in the naming rule. + */ + private Path resolveChild(String fileName) { + Path base = directory.toAbsolutePath().normalize(); + Path resolved = base.resolve(fileName).normalize(); + if (!resolved.startsWith(base)) { + throw new IllegalStateException("resolved insight file path escapes the configured directory: " + fileName); + } + return resolved; + } + + /** Replaces every character outside {@code [a-zA-Z0-9._-]} with {@code _} (matches the JS {@code sanitize}). */ + private static String sanitize(String value) { + return value.replaceAll("[^a-zA-Z0-9._-]", "_"); + } + + /** Builder for {@link FileExporter}. */ + public static final class Builder { + private Path directory; + private Mode mode; + private OperationsFormat operationsFormat; + private Integer maxRecordSizeBytes; + + /** Base directory to write files to, e.g. {@code /mnt/efs/workflow-insight} or {@code /tmp/insight}. */ + public Builder directory(String directory) { + this.directory = directory != null ? Path.of(directory) : null; + return this; + } + + /** Base directory to write files to. */ + public Builder directory(Path directory) { + this.directory = directory; + return this; + } + + /** File output mode; defaults to {@link Mode#NDJSON}. */ + public Builder mode(Mode mode) { + this.mode = mode; + return this; + } + + /** How operations are rendered; defaults to {@link OperationsFormat#ARRAY}. */ + public Builder operationsFormat(OperationsFormat operationsFormat) { + this.operationsFormat = operationsFormat; + return this; + } + + /** + * Max serialized record size before truncation; no default (the filesystem has no practical per-record cap). + */ + public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + return this; + } + + public FileExporter build() { + if (directory == null) { + throw new IllegalArgumentException("directory is required"); + } + if (directory.toString().isBlank()) { + throw new IllegalArgumentException("directory must not be blank"); + } + if (maxRecordSizeBytes != null && maxRecordSizeBytes <= 0) { + throw new IllegalArgumentException("maxRecordSizeBytes must be positive when set"); + } + return new FileExporter(this); + } + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/FileExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/FileExporterTest.java new file mode 100644 index 000000000..e6d7363d3 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/FileExporterTest.java @@ -0,0 +1,252 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +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 com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import software.amazon.lambda.durable.insight.exporters.FileExporter; + +/** Exercises {@link FileExporter} against a real filesystem (no mocked IO). */ +class FileExporterTest { + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.executionArn = "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1"; + r.executionName = "exec-1"; + r.functionName = "f"; + r.status = "SUCCEEDED"; + r.emittedAt = "2026-08-05T12:34:56Z"; + r.startTime = "2026-08-05T00:00:00Z"; + r.addOperation(new OperationRecord() + .id("op-1") + .name("greet") + .type("STEP") + .subType("Step") + .status("SUCCEEDED")); + return r; + } + + @Test + void ndjsonAppendsDatePartitionedCompactLine(@TempDir Path dir) throws Exception { + FileExporter exporter = FileExporter.builder().directory(dir).build(); + + exporter.export(sampleRecord()); + exporter.export(sampleRecord()); + + Path file = dir.resolve("2026-08-05.ndjson"); + assertTrue(Files.exists(file), "date-partitioned ndjson file should exist"); + List lines = Files.readAllLines(file); + assertEquals(2, lines.size(), "each export appends one line"); + for (String line : lines) { + assertFalse(line.isBlank()); + assertFalse(line.contains("\n "), "ndjson lines are compact, not pretty-printed"); + assertTrue(line.contains("\"operations\""), "array format is the default"); + assertFalse(line.contains("operationsByName")); + assertTrue(line.contains("\"greet\"")); + } + } + + @Test + void jsonModeWritesOnePrettyFilePerExecutionAndOverwrites(@TempDir Path dir) throws Exception { + FileExporter exporter = FileExporter.builder() + .directory(dir) + .mode(FileExporter.Mode.JSON) + .build(); + + exporter.export(sampleRecord()); + exporter.export(sampleRecord()); // second update overwrites, not appends + + Path file = dir.resolve("exec-1.json"); + assertTrue(Files.exists(file)); + String content = Files.readString(file); + assertTrue(content.contains("\n "), "json mode is pretty-printed with 2-space indent"); + assertTrue(content.contains("\"executionName\" : \"exec-1\"")); + // Overwrite (not append): the file parses as a single JSON object. + assertEquals('{', content.trim().charAt(0)); + assertEquals('}', content.trim().charAt(content.trim().length() - 1)); + } + + @Test + void jsonModeFallsBackToArnWhenExecutionNameNull(@TempDir Path dir) throws Exception { + WorkflowInsightRecord r = sampleRecord(); + r.executionName = null; + FileExporter exporter = FileExporter.builder() + .directory(dir) + .mode(FileExporter.Mode.JSON) + .build(); + + exporter.export(r); + + // The ARN contains ':' '/' '$' — all sanitized to '_'. + String expected = "arn_aws_lambda_us-west-2_1_function_f__LATEST_durable-execution_exec-1_invocation-1.json"; + assertTrue(Files.exists(dir.resolve(expected)), "unsafe ARN chars are sanitized into the file name"); + } + + @Test + void byNameFormatReplacesArrayWithMap(@TempDir Path dir) throws Exception { + FileExporter exporter = FileExporter.builder() + .directory(dir) + .operationsFormat(FileExporter.OperationsFormat.BY_NAME) + .build(); + + exporter.export(sampleRecord()); + + String line = Files.readAllLines(dir.resolve("2026-08-05.ndjson")).get(0); + assertTrue(line.contains("operationsByName")); + assertFalse(line.contains("\"operations\":"), "by-name replaces the array"); + assertFalse(line.contains("\"operations\" :")); + } + + @Test + void bothFormatEmitsArrayAndByNameMap(@TempDir Path dir) throws Exception { + FileExporter exporter = FileExporter.builder() + .directory(dir) + .operationsFormat(FileExporter.OperationsFormat.BOTH) + .build(); + + exporter.export(sampleRecord()); + + String line = Files.readAllLines(dir.resolve("2026-08-05.ndjson")).get(0); + assertTrue(line.contains("\"operations\"")); + assertTrue(line.contains("operationsByName")); + } + + @Test + void createsNestedDirectoryWhenMissing(@TempDir Path dir) throws Exception { + Path nested = dir.resolve("a/b/c"); + FileExporter exporter = FileExporter.builder().directory(nested).build(); + + exporter.export(sampleRecord()); + + assertTrue(Files.exists(nested.resolve("2026-08-05.ndjson"))); + } + + @Test + void missingEmittedAtFallsBackToTodayPartition(@TempDir Path dir) throws Exception { + WorkflowInsightRecord r = sampleRecord(); + r.emittedAt = null; + FileExporter exporter = FileExporter.builder().directory(dir).build(); + + exporter.export(r); + + String today = java.time.Instant.now() + .atZone(java.time.ZoneOffset.UTC) + .toLocalDate() + .toString(); + assertTrue(Files.exists(dir.resolve(today + ".ndjson"))); + } + + @Test + void maxRecordSizeBytesDefaultsToNull(@TempDir Path dir) { + FileExporter exporter = FileExporter.builder().directory(dir).build(); + assertEquals(null, exporter.maxRecordSizeBytes()); + } + + @Test + void maxRecordSizeBytesIsReportedWhenSet(@TempDir Path dir) { + FileExporter exporter = + FileExporter.builder().directory(dir).maxRecordSizeBytes(512).build(); + assertEquals(512, exporter.maxRecordSizeBytes()); + } + + @Test + void builderRejectsMissingDirectory() { + assertThrows( + IllegalArgumentException.class, () -> FileExporter.builder().build()); + } + + @Test + void builderRejectsBlankDirectory() { + assertThrows( + IllegalArgumentException.class, + () -> FileExporter.builder().directory(" ").build()); + } + + @Test + void builderRejectsNonPositiveMaxRecordSize(@TempDir Path dir) { + assertThrows(IllegalArgumentException.class, () -> FileExporter.builder() + .directory(dir) + .maxRecordSizeBytes(0) + .build()); + } + + @Test + void flushIsNoOpAndDoesNotThrow(@TempDir Path dir) { + FileExporter exporter = FileExporter.builder().directory(dir).build(); + exporter.flush(); + } + + @Test + void traversalShapedEmittedAtIsRejectedByResolveChild(@TempDir Path dir) { + // The NDJSON partition is emittedAt.substring(0, 10), so a traversal-shaped emittedAt drives a file name + // containing path separators and "..". resolveChild must reject it rather than write outside `directory`. + WorkflowInsightRecord r = sampleRecord(); + r.emittedAt = "../../etc/passwd"; + FileExporter exporter = FileExporter.builder().directory(dir).build(); + + IllegalStateException ex = assertThrows(IllegalStateException.class, () -> exporter.export(r)); + assertTrue( + ex.getMessage().contains("escapes the configured directory"), + "resolveChild should reject an escape attempt"); + // Nothing was written outside the configured directory: substring(0,10) of the emittedAt is "../../etc/", + // so the attempted target would be dir/../../etc/.ndjson. + assertFalse(Files.exists(dir.getParent().getParent().resolve("etc/.ndjson"))); + } + + @Test + void concurrentNdjsonAppendsWriteCompleteParseableLines(@TempDir Path dir) throws Exception { + // Real filesystem, real threads: each export must land as exactly one complete, parseable NDJSON line. + FileExporter exporter = FileExporter.builder().directory(dir).build(); + int threads = 8; + int perThread = 25; + int total = threads * perThread; + + ExecutorService pool = Executors.newFixedThreadPool(threads); + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + futures.add(pool.submit(() -> { + start.await(); + for (int i = 0; i < perThread; i++) { + exporter.export(sampleRecord()); + } + return null; + })); + } + start.countDown(); + for (Future f : futures) { + f.get(); + } + pool.shutdown(); + assertTrue(pool.awaitTermination(30, TimeUnit.SECONDS)); + + Path file = dir.resolve("2026-08-05.ndjson"); + assertTrue(Files.exists(file)); + List lines = Files.readAllLines(file); + assertEquals(total, lines.size(), "each export appends exactly one line"); + ObjectMapper mapper = new ObjectMapper(); + for (String line : lines) { + assertFalse(line.isBlank(), "no blank/partial lines"); + // Each line parses as a complete JSON object (no interleaving/truncation). + JsonNode node = mapper.readTree(line); + assertTrue(node.isObject(), "each line is a complete JSON object"); + assertEquals("WorkflowInsight", node.get("recordType").asText()); + } + } +}