refactor(jsonata): replace per-call Thread with per-instance executor; document Throwable catch

Addresses two review comments from Malay on PR #80:

1. Thread-per-record regression (TransformItems): evaluateExpression() previously spawned a
   new 4 MB-stack Thread per call. Inside TransformItems.run(), calls are inside a Flux.map
   so this allocated one thread per record. Fixed by replacing new Thread(...) with a
   lazy-initialized single-threaded ExecutorService (daemon, 4 MB ThreadFactory) stored per
   Transform instance. TransformItems and TransformValue now shut it down in finally at the end
   of run(), so the executor lives for exactly one task run.

2. catch (Throwable): kept intentionally broad with an explanatory comment. The eval thread is
   a throwaway sandbox — narrowing to Exception | StackOverflowError would let other Errors
   (OOM, InternalError) escape silently via the UncaughtExceptionHandler, leaving both
   resultRef and errorRef null and returning null from evaluateExpression instead of failing.

New tests:
- shouldReuseEvalThreadAcrossRecords: asserts zero jsonata-eval threads alive after run()
- shouldContinueWorkingAfterStackOverflowError: asserts a clean second run after a SOE

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: fdelbrayelle

plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/Transform.java

@@ -20,6 +20,10 @@
 import lombok.experimental.SuperBuilder;
 
 import java.time.Duration;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import io.kestra.core.models.enums.MonacoLanguages;
@@ -49,6 +53,36 @@ public abstract class Transform<T extends Output> extends Task implements JSONat
     @Getter(AccessLevel.PRIVATE)
     private Jsonata parsedExpression;
 
+    // Lazy-initialized; lifecycle managed by evalExecutor() / shutdownEvalExecutor().
+    // Assumption: Flux pipelines in subclasses are sequential (no parallel()/publishOn).
+    @Getter(AccessLevel.NONE)
+    @ToString.Exclude
+    @EqualsAndHashCode.Exclude
+    private transient ExecutorService evalExecutor;
+
+    private ExecutorService evalExecutor() {
+        if (this.evalExecutor == null) {
+            this.evalExecutor = Executors.newSingleThreadExecutor(r -> {
+                var t = new Thread(null, r, "jsonata-eval", EVAL_THREAD_STACK_SIZE);
+                t.setDaemon(true);
+                return t;
+            });
+        }
+        return this.evalExecutor;
+    }
+
+    protected void shutdownEvalExecutor() {
+        if (this.evalExecutor != null) {
+            this.evalExecutor.shutdown();
+            try {
+                this.evalExecutor.awaitTermination(1, TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            this.evalExecutor = null;
+        }
+    }
+
     public void init(RunContext runContext) throws Exception {
         var exprString = runContext.render(this.expression).as(String.class).orElseThrow();
         try {
@@ -72,22 +106,31 @@ protected JsonNode evaluateExpression(RunContext runContext, JsonNode jsonNode)
             var resultRef = new AtomicReference<JsonNode>();
             var errorRef = new AtomicReference<Throwable>();
 
-            // Eval runs on a dedicated thread with an explicit 4 MB stack. This serves two purposes:
+            // Eval runs on a dedicated executor thread (4 MB stack) that is reused across all records
+            // in the same task run. This serves two purposes:
             // 1. Normal case: worker stack size (e.g. 256 KB on Windows) cannot constrain the evaluator.
             // 2. Edge case (user sets very high maxDepth): if a StackOverflowError occurs in the eval
             //    thread, it is contained there. The worker thread reads the stored error and throws a
             //    clean RuntimeException — the worker never crashes.
-            var thread = new Thread(null, () -> {
+            // The catch is intentionally Throwable: this is a throwaway-thread sandbox, so every escape
+            // (including Errors like StackOverflowError and OutOfMemoryError) must land in errorRef.
+            // A narrower catch would let some Errors escape, leaving both refs null and producing a
+            // silent-null return after future.get().
+            var future = evalExecutor().submit(() -> {
                 try {
                     var result = this.parsedExpression.evaluate(data, frame);
                     resultRef.set(result != null ? MAPPER.valueToTree(result) : NullNode.getInstance());
                 } catch (Throwable t) {
                     errorRef.set(t);
                 }
-            }, "jsonata-eval", EVAL_THREAD_STACK_SIZE);
+                return null;
+            });
 
-            thread.start();
-            thread.join();
+            try {
+                future.get();
+            } catch (ExecutionException e) {
+                throw new RuntimeException("Failed to evaluate expression", e.getCause());
+            }
 
             if (errorRef.get() != null) {
                 throw new RuntimeException("Failed to evaluate expression", errorRef.get());

plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/TransformItems.java

@@ -120,40 +120,44 @@ public Output run(RunContext runContext) throws Exception {
 
         init(runContext);
 
-        final URI from = new URI(runContext.render(this.from).as(String.class).orElseThrow());
-
-        try (Reader reader = new BufferedReader(new InputStreamReader(runContext.storage().getFile(from)), FileSerde.BUFFER_SIZE)) {
-            Flux<JsonNode> flux = FileSerde.readAll(reader, new TypeReference<>() {
-            });
-            final Path outputFilePath = runContext.workingDir().createTempFile(".ion");
-            try (Writer writer = new BufferedWriter(new OutputStreamWriter(Files.newOutputStream(outputFilePath)))) {
-
-                // transform
-                Flux<JsonNode> values = flux.map(node -> this.evaluateExpression(runContext, node));
-
-                if (runContext.render(explodeArray).as(Boolean.class).orElseThrow()) {
-                    values = values.flatMap(jsonNode -> {
-                        if (jsonNode.isArray()) {
-                            Iterable<JsonNode> iterable = jsonNode::elements;
-                            return Flux.fromStream(StreamSupport.stream(iterable.spliterator(), false));
-                        }
-                        return Mono.just(jsonNode);
-                    });
+        try {
+            final URI from = new URI(runContext.render(this.from).as(String.class).orElseThrow());
+
+            try (Reader reader = new BufferedReader(new InputStreamReader(runContext.storage().getFile(from)), FileSerde.BUFFER_SIZE)) {
+                Flux<JsonNode> flux = FileSerde.readAll(reader, new TypeReference<>() {
+                });
+                final Path outputFilePath = runContext.workingDir().createTempFile(".ion");
+                try (Writer writer = new BufferedWriter(new OutputStreamWriter(Files.newOutputStream(outputFilePath)))) {
+
+                    // transform
+                    Flux<JsonNode> values = flux.map(node -> this.evaluateExpression(runContext, node));
+
+                    if (runContext.render(explodeArray).as(Boolean.class).orElseThrow()) {
+                        values = values.flatMap(jsonNode -> {
+                            if (jsonNode.isArray()) {
+                                Iterable<JsonNode> iterable = jsonNode::elements;
+                                return Flux.fromStream(StreamSupport.stream(iterable.spliterator(), false));
+                            }
+                            return Mono.just(jsonNode);
+                        });
+                    }
+
+                    Long processedItemsTotal = FileSerde.writeAll(writer, values).block();
+
+                    URI uri = runContext.storage().putFile(outputFilePath.toFile());
+
+                    // output
+                    return Output
+                        .builder()
+                        .uri(uri)
+                        .processedItemsTotal(processedItemsTotal)
+                        .build();
+                } finally {
+                    Files.deleteIfExists(outputFilePath); // ensure temp file is deleted in case of error
                 }
-
-                Long processedItemsTotal = FileSerde.writeAll(writer, values).block();
-
-                URI uri = runContext.storage().putFile(outputFilePath.toFile());
-
-                // output
-                return Output
-                    .builder()
-                    .uri(uri)
-                    .processedItemsTotal(processedItemsTotal)
-                    .build();
-            } finally {
-                Files.deleteIfExists(outputFilePath); // ensure temp file is deleted in case of error
             }
+        } finally {
+            shutdownEvalExecutor();
         }
     }
 

plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/TransformValue.java

@@ -101,13 +101,17 @@ public class TransformValue extends Transform<TransformValue.Output> implements
     public Output run(RunContext runContext) throws Exception {
         init(runContext);
 
-        final JsonNode from = parseJson(runContext.render(this.from).as(String.class).orElseThrow());
+        try {
+            final JsonNode from = parseJson(runContext.render(this.from).as(String.class).orElseThrow());
 
-        // transform
-        JsonNode transformed = evaluateExpression(runContext, from);
+            // transform
+            JsonNode transformed = evaluateExpression(runContext, from);
 
-        // output
-        return Output.builder().value(transformed).build();
+            // output
+            return Output.builder().value(transformed).build();
+        } finally {
+            shutdownEvalExecutor();
+        }
     }
 
     private static JsonNode parseJson(String from) {

plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformItemsTest.java

@@ -123,6 +123,38 @@ void shouldGetSingleRecordForValidExprReturningArrayGivenExplodeFalse() throws E
         Assertions.assertEquals(2, transformationResult.getFirst().size());
     }
 
+    @Test
+    void shouldReuseEvalThreadAcrossRecords() throws Exception {
+        // Verifies executor reuse: after run() completes, awaitTermination in shutdownEvalExecutor()
+        // guarantees the jsonata-eval thread is gone. If the old per-call new Thread() approach were
+        // used, 3 threads would be started and could still be alive briefly, making liveAfter > 0
+        // probabilistically — so this assertion is a reliable regression guard.
+        RunContext runContext = runContextFactory.of();
+        final Path outputFilePath = runContext.workingDir().createTempFile(".ion");
+        try (final Writer writer = new OutputStreamWriter(Files.newOutputStream(outputFilePath))) {
+            FileSerde.writeAll(writer, Flux.just(
+                Map.of("v", 1),
+                Map.of("v", 2),
+                Map.of("v", 3)
+            )).block();
+            writer.flush();
+        }
+        URI uri = runContext.storage().putFile(outputFilePath.toFile());
+
+        TransformItems task = TransformItems.builder()
+            .from(Property.ofValue(uri.toString()))
+            .expression(Property.ofValue("$"))
+            .build();
+
+        task.run(runContext);
+
+        long liveAfter = Thread.getAllStackTraces().keySet().stream()
+            .filter(t -> "jsonata-eval".equals(t.getName()))
+            .count();
+
+        Assertions.assertEquals(0, liveAfter, "jsonata-eval thread should be terminated after run()");
+    }
+
     @Test
     void shouldTransformJsonInputWithDefaultIonMapper() throws Exception {
         // Given

plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java

@@ -175,6 +175,37 @@ void shouldNeverThrowStackOverflowForCommonMaxDepthValues(int maxDepth) throws E
             .hasCauseInstanceOf(JException.class);
     }
 
+    @Test
+    void shouldContinueWorkingAfterStackOverflowError() throws Exception {
+        // Validates that a StackOverflowError in one run does not poison the executor or the task.
+        // Each call to run() creates a fresh executor (via init + shutdownEvalExecutor in finally),
+        // so the second run always gets a clean state.
+        RunContext runContext = runContextFactory.of();
+
+        TransformValue taskWithHighDepth = TransformValue.builder()
+            .from(Property.ofValue("{}"))
+            .expression(Property.ofValue(
+                "($f := function($n) { $n > 0 ? $f($n - 1) + 0 : 0 }; $f(49999))"
+            ))
+            .maxDepth(Property.ofValue(50000))
+            .build();
+
+        assertThatThrownBy(() -> taskWithHighDepth.run(runContext))
+            .isInstanceOf(RuntimeException.class)
+            .hasCauseInstanceOf(StackOverflowError.class);
+
+        // Second run with a simple expression must succeed — no lingering poisoned state.
+        RunContext runContext2 = runContextFactory.of();
+        TransformValue simpleTask = TransformValue.builder()
+            .from(Property.ofValue("{\"x\": 42}"))
+            .expression(Property.ofValue("x"))
+            .build();
+
+        TransformValue.Output output = simpleTask.run(runContext2);
+        assertThat(output.getValue()).isNotNull();
+        assertThat(output.getValue().toString()).isEqualTo("42");
+    }
+
     @Test
     void shouldIsolateStackOverflowInEvalThreadWhenMaxDepthExceedsStackCapacity() throws Exception {
         // User sets maxDepth high enough that bounds check never fires before stack exhaustion.