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 339693b6cf TIKA-5040: kill descendant processes with the forked server
and in ProcessUtils.execute (#3532)
339693b6cf is described below
commit 339693b6cf22ddcc868294d0f597b946c967b67c
Author: Davide Polato <[email protected]>
AuthorDate: Thu Nov 5 15:26:06 2026 +0100
TIKA-5040: kill descendant processes with the forked server and in
ProcessUtils.execute (#3532)
* TIKA-5040: kill descendant processes with the forked server and in
ProcessUtils.execute
Signed-off-by: Davide Polato <[email protected]>
---
CHANGES.txt | 5 +
.../java/org/apache/tika/utils/ProcessUtils.java | 43 +++++-
.../org/apache/tika/utils/ProcessTestHelper.java | 70 +++++++++
.../org/apache/tika/utils/ProcessUtilsTest.java | 59 ++++++++
.../tika/pipes/core/PerClientServerManager.java | 3 +-
.../tika/pipes/core/SharedServerManager.java | 3 +-
.../pipes/core/ExternalProcessCleanupTest.java | 157 +++++++++++++++++++++
.../apache/tika/pipes/core/PluginsTestHelper.java | 11 +-
.../configs/tika-config-external-sleep.json | 54 +++++++
9 files changed, 396 insertions(+), 9 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 53a1b25680..dc08815664 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -15,6 +15,11 @@ Release 4.3.0 - ???
SERVER, PIPES AND OPERATIONS
+ * When pipes kills a forked server, it now kills the server's descendant
processes too,
+ such as ExternalParser commands and their children; they kept running. On
a timeout,
+ an interrupt or JVM shutdown, ProcessUtils.execute also kills the
descendants of the
+ process it started (TIKA-5040).
+
* Fetchers and plugins built on HttpClientFactory (HTTP, Atlassian JWT, ES,
OpenSearch,
Solr) no longer send Authorization or Cookie headers, or answer a 401
with the
configured password, after a redirect to another host, port or scheme.
That includes
diff --git a/tika-core/src/main/java/org/apache/tika/utils/ProcessUtils.java
b/tika-core/src/main/java/org/apache/tika/utils/ProcessUtils.java
index 244742cb74..36da26225e 100644
--- a/tika-core/src/main/java/org/apache/tika/utils/ProcessUtils.java
+++ b/tika-core/src/main/java/org/apache/tika/utils/ProcessUtils.java
@@ -52,10 +52,41 @@ public class ProcessUtils {
static {
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
- PROCESS_MAP.forEachValue(1, Process::destroyForcibly);
+ PROCESS_MAP.forEachValue(1,
ProcessUtils::destroyForciblyWithDescendants);
}));
}
+ /**
+ * Forcibly destroys {@code process} and the descendants it has at the
time of the call.
+ *
+ * @param process the process to destroy
+ */
+ public static void destroyForciblyWithDescendants(Process process) {
+ List<ProcessHandle> descendants = listDescendants(process);
+ process.destroyForcibly();
+ descendants.forEach(ProcessHandle::destroyForcibly);
+ }
+
+ private static List<ProcessHandle> listDescendants(Process process) {
+ ProcessHandle handle;
+ try {
+ handle = process.toHandle();
+ } catch (UnsupportedOperationException e) {
+ LOG.debug("cannot list the descendants of {}; destroying only the
process", process, e);
+ return List.of();
+ }
+ if (!handle.isAlive()) {
+ return List.of();
+ }
+ // the JDK signals a handle only if its pid still has the handle's
start time, and it
+ // skips that check when the start time is unknown
+ List<ProcessHandle> descendants = handle.descendants()
+ .filter(h -> h.info().startInstant().isPresent())
+ .toList();
+ // descendants() finds the parent by pid; isAlive() checks its start
time
+ return handle.isAlive() ? descendants : List.of();
+ }
+
private static String register(Process p) {
String id = UUID.randomUUID().toString();
PROCESS_MAP.put(id, p);
@@ -214,7 +245,7 @@ public class ProcessUtils {
outThread.join(1000);
errThread.join(1000);
} else {
- p.destroyForcibly();
+ destroyForciblyWithDescendants(p);
outThread.join(1000);
errThread.join(1000);
boolean completed = p.waitFor(500, TimeUnit.MILLISECONDS);
@@ -249,7 +280,7 @@ public class ProcessUtils {
return result;
} finally {
if (p != null) {
- p.destroyForcibly();
+ destroyForciblyWithDescendants(p);
}
if (id != null) {
release(id);
@@ -332,7 +363,7 @@ public class ProcessUtils {
exitValue = p.exitValue();
errThread.join(1000);
} else {
- p.destroyForcibly();
+ destroyForciblyWithDescendants(p);
errThread.join(1000);
}
} catch (InterruptedException e) {
@@ -354,7 +385,7 @@ public class ProcessUtils {
return result;
} finally {
if (p != null) {
- p.destroyForcibly();
+ destroyForciblyWithDescendants(p);
}
if (id != null) {
release(id);
@@ -455,7 +486,7 @@ public class ProcessUtils {
throw err;
} finally {
if (process != null) {
- process.destroyForcibly();
+ destroyForciblyWithDescendants(process);
}
}
}
diff --git
a/tika-core/src/test/java/org/apache/tika/utils/ProcessTestHelper.java
b/tika-core/src/test/java/org/apache/tika/utils/ProcessTestHelper.java
new file mode 100644
index 0000000000..491d37458d
--- /dev/null
+++ b/tika-core/src/test/java/org/apache/tika/utils/ProcessTestHelper.java
@@ -0,0 +1,70 @@
+/*
+ * 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.utils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Finds the processes a test started by a marker string that appears in their
command lines.
+ */
+public final class ProcessTestHelper {
+
+ private static final long POLL_MILLIS = 20;
+
+ private ProcessTestHelper() {
+ }
+
+ /** Live processes whose command line contains {@code marker}; a zombie
has no command line. */
+ public static List<ProcessHandle> processesWith(String marker) {
+ List<ProcessHandle> found = new ArrayList<>();
+ ProcessHandle.allProcesses().forEach(ph -> {
+ if (ph.info().commandLine().orElse("").contains(marker) &&
ph.isAlive()) {
+ found.add(ph);
+ }
+ });
+ return found;
+ }
+
+ /**
+ * Polls {@link #processesWith(String)} until it returns {@code count}
processes or
+ * {@code timeoutMillis} passes.
+ *
+ * @return the processes found by the last poll
+ */
+ public static List<ProcessHandle> awaitProcessesWith(String marker, int
count, long timeoutMillis)
+ throws InterruptedException {
+ long deadline = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(timeoutMillis);
+ List<ProcessHandle> found = processesWith(marker);
+ while (found.size() != count && System.nanoTime() < deadline) {
+ Thread.sleep(POLL_MILLIS);
+ found = processesWith(marker);
+ }
+ return found;
+ }
+
+ /** One {@code pid:ppid=N:commandLine} entry per process, for assertion
messages. */
+ public static List<String> describe(List<ProcessHandle> processes) {
+ List<String> described = new ArrayList<>();
+ for (ProcessHandle ph : processes) {
+ described.add(ph.pid() + ":ppid=" +
ph.parent().map(ProcessHandle::pid).orElse(-1L)
+ + ":" + ph.info().commandLine().orElse("?"));
+ }
+ return described;
+ }
+}
diff --git
a/tika-core/src/test/java/org/apache/tika/utils/ProcessUtilsTest.java
b/tika-core/src/test/java/org/apache/tika/utils/ProcessUtilsTest.java
index 9b920a4aa7..6da933256a 100644
--- a/tika-core/src/test/java/org/apache/tika/utils/ProcessUtilsTest.java
+++ b/tika-core/src/test/java/org/apache/tika/utils/ProcessUtilsTest.java
@@ -16,6 +16,9 @@
*/
package org.apache.tika.utils;
+import static org.apache.tika.utils.ProcessTestHelper.awaitProcessesWith;
+import static org.apache.tika.utils.ProcessTestHelper.describe;
+import static org.apache.tika.utils.ProcessTestHelper.processesWith;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -24,9 +27,16 @@ import static org.junit.jupiter.api.Assumptions.assumeFalse;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.List;
+import java.util.Locale;
+import java.util.concurrent.ThreadLocalRandom;
import org.jspecify.annotations.NonNull;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.apache.tika.config.ParseTimeout;
import org.apache.tika.config.TimeoutLimits;
@@ -169,6 +179,55 @@ public class ProcessUtilsTest {
assertSubprocessDidNotStart(result, elapsed);
}
+ /**
+ * When execute() stops waiting (timeout or interrupt), {@code sh} and its
child {@code sleep}
+ * end, and a process started outside execute() is left alone.
+ */
+ @ParameterizedTest(name = "redirect={0} interrupt={1}")
+ @CsvSource({"false, false", "true, false", "false, true", "true, true"})
+ @EnabledOnOs(OS.LINUX)
+ public void testStoppedWaitDestroysTheProcessesItStarted(boolean redirect,
boolean interrupt)
+ throws Exception {
+ // doubles as the sleep duration, so both command lines contain it;
random per run
+ int suffix = ThreadLocalRandom.current().nextInt(1_000_000);
+ String marker = String.format(Locale.ROOT, "301.%06d", suffix);
+ ProcessBuilder pb = new ProcessBuilder("sh", "-c", "sleep " + marker +
"; true");
+ long timeoutMillis = interrupt ? 60_000 : 2000;
+ Process bystander = new ProcessBuilder("sleep",
String.format(Locale.ROOT, "302.%06d", suffix))
+ .start();
+ Path stdoutRedirect = redirect ? createTmpRedirect() : null;
+ Thread runner = new Thread(() -> {
+ try {
+ if (redirect) {
+ ProcessUtils.execute(pb, null, timeoutMillis,
stdoutRedirect, 1000);
+ } else {
+ ProcessUtils.execute(pb, null, timeoutMillis, 1000, 1000);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ try {
+ runner.start();
+ List<ProcessHandle> running = awaitProcessesWith(marker, 2,
10_000);
+ assertEquals(2, running.size(), "sh and its sleep should be
running: " + describe(running));
+ if (interrupt) {
+ runner.interrupt();
+ }
+ runner.join(10_000);
+ assertFalse(runner.isAlive(), "execute() should have returned");
+
+ List<ProcessHandle> survivors = awaitProcessesWith(marker, 0,
2_000);
+ assertTrue(survivors.isEmpty(), "redirect=" + redirect + "
interrupt=" + interrupt
+ + ": processes outlived execute(): " +
describe(survivors));
+ assertTrue(bystander.isAlive(), "a process execute() did not start
was destroyed");
+ } finally {
+ runner.interrupt();
+ processesWith(marker).forEach(ProcessHandle::destroyForcibly);
+ bystander.destroyForcibly();
+ }
+ }
+
private static @NonNull Path createTmpRedirect() throws IOException {
Path tmpDir = Path.of(System.getProperty("java.io.tmpdir"));
Files.createDirectories(tmpDir);
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java
index f6a7b17a45..1bec6bfd70 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java
@@ -606,7 +606,8 @@ public class PerClientServerManager implements
ServerManager {
private void destroyProcess() throws InterruptedException {
if (process != null) {
- process.destroyForcibly();
+ // a forced kill skips the server's shutdown hooks, which end its
external processes
+ ProcessUtils.destroyForciblyWithDescendants(process);
try {
process.waitFor(WAIT_ON_DESTROY_MS, TimeUnit.MILLISECONDS);
if (process.isAlive()) {
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java
index f39383cb69..76f8ee7a97 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java
@@ -462,7 +462,8 @@ public class SharedServerManager implements ServerManager {
private void destroyProcessUnsafe() throws InterruptedException {
if (process != null) {
- process.destroyForcibly();
+ // a forced kill skips the server's shutdown hooks, which end its
external processes
+ ProcessUtils.destroyForciblyWithDescendants(process);
try {
process.waitFor(WAIT_ON_DESTROY_MS, TimeUnit.MILLISECONDS);
if (process.isAlive()) {
diff --git
a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/ExternalProcessCleanupTest.java
b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/ExternalProcessCleanupTest.java
new file mode 100644
index 0000000000..98e593df39
--- /dev/null
+++
b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/ExternalProcessCleanupTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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;
+
+import static org.apache.tika.utils.ProcessTestHelper.awaitProcessesWith;
+import static org.apache.tika.utils.ProcessTestHelper.describe;
+import static org.apache.tika.utils.ProcessTestHelper.processesWith;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+
+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.config.loader.TikaJsonConfig;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.parser.ParseContext;
+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;
+
+/**
+ * A forked server's external processes and their children end with the
server. The parser runs
+ * {@code sh -c "sleep N; true"}, so {@code sleep} is a child of the process
Tika started.
+ */
+@EnabledOnOs(OS.LINUX)
+public class ExternalProcessCleanupTest {
+
+ private static final String DOC = "doc.txt";
+
+ enum ServerMode {
+ PER_CLIENT,
+ SHARED
+ }
+
+ enum ServerEnd {
+ /** PipesParser.close() mid-parse: the parent destroys the server. */
+ PARENT_KILLS_SERVER,
+ /** The calling thread is interrupted; the per-client server loses its
connection and exits. */
+ SERVER_EXITS_ITSELF
+ }
+
+ @Test
+ public void perClientServerKilledByParent(@TempDir Path tmp) throws
Exception {
+ run(tmp, ServerMode.PER_CLIENT, ServerEnd.PARENT_KILLS_SERVER);
+ }
+
+ @Test
+ public void perClientServerExitsItself(@TempDir Path tmp) throws Exception
{
+ run(tmp, ServerMode.PER_CLIENT, ServerEnd.SERVER_EXITS_ITSELF);
+ }
+
+ @Test
+ public void sharedServerKilledByParent(@TempDir Path tmp) throws Exception
{
+ run(tmp, ServerMode.SHARED, ServerEnd.PARENT_KILLS_SERVER);
+ }
+
+ private void run(Path tmp, ServerMode mode, ServerEnd end) throws
Exception {
+ // doubles as the sleep duration, so both command lines contain it;
random per run
+ String marker = String.format(Locale.ROOT, "301.%06d",
+ ThreadLocalRandom.current().nextInt(1_000_000));
+ Path inputDir = tmp.resolve("input");
+ Files.createDirectories(inputDir);
+ Files.writeString(inputDir.resolve(DOC), "hello",
StandardCharsets.UTF_8);
+ Path tikaConfigPath = PluginsTestHelper.getFileSystemFetcherConfig(
+ "tika-config-external-sleep.json", tmp, inputDir,
tmp.resolve("output"), false,
+ Map.of("USE_SHARED_SERVER", mode == ServerMode.SHARED,
+ "COMMAND_LINE", List.of("sh", "-c", "sleep " + marker
+ "; true")));
+ TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(tikaConfigPath);
+ PipesParser pipesParser = PipesParser.load(tikaJsonConfig,
PipesConfig.load(tikaJsonConfig));
+
+ AtomicReference<String> parseOutcome = new AtomicReference<>();
+ Thread parseThread = new Thread(() -> {
+ try {
+ PipesResult r = pipesParser.parse(new FetchEmitTuple(DOC, new
FetchKey("fsf", DOC),
+ new EmitKey(), new Metadata(), new ParseContext(),
+ FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
+ parseOutcome.set(r.status() + " " + r.message());
+ } catch (Throwable t) {
+ parseOutcome.set(t.getClass().getSimpleName());
+ }
+ }, "external-process-cleanup-parse");
+ try {
+ parseThread.start();
+ List<ProcessHandle> running = awaitProcessesWith(marker, 2,
30_000);
+ assertEquals(2, running.size(), "sh and its sleep should be
running: " + describe(running)
+ + " parseOutcome=" + parseOutcome.get());
+ ProcessHandle server = serverOf(running);
+
+ if (end == ServerEnd.PARENT_KILLS_SERVER) {
+ pipesParser.close();
+ } else {
+ parseThread.interrupt();
+ }
+ boolean serverExited = awaitExit(server, 30_000);
+ List<ProcessHandle> survivors = awaitProcessesWith(marker, 0,
3_000);
+ String report = String.format(Locale.ROOT,
+ "mode=%s end=%s server=%d serverExited=%s parseOutcome=%s
before=%s survivors=%s",
+ mode, end, server.pid(), serverExited, parseOutcome.get(),
describe(running),
+ describe(survivors));
+ assertTrue(serverExited, "the server did not exit: " + report);
+ assertTrue(survivors.isEmpty(), "processes outlived the server: "
+ report);
+ } finally {
+ parseThread.interrupt();
+ pipesParser.close();
+ processesWith(marker).forEach(ProcessHandle::destroyForcibly);
+ }
+ }
+
+ /** The forked server: the parent of the process it started, and a child
of this JVM. */
+ private static ProcessHandle serverOf(List<ProcessHandle> running) {
+ long thisJvm = ProcessHandle.current().pid();
+ for (ProcessHandle ph : running) {
+ ProcessHandle parent = ph.parent().orElse(null);
+ if (parent != null &&
parent.parent().map(ProcessHandle::pid).orElse(-1L) == thisJvm) {
+ return parent;
+ }
+ }
+ throw new IllegalStateException("no process under a server of this
JVM: " + describe(running));
+ }
+
+ private static boolean awaitExit(ProcessHandle ph, long millis) throws
Exception {
+ try {
+ ph.onExit().get(millis, TimeUnit.MILLISECONDS);
+ return true;
+ } catch (TimeoutException e) {
+ return false;
+ }
+ }
+}
diff --git
a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PluginsTestHelper.java
b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PluginsTestHelper.java
index 1add238190..cc1928e7ff 100644
---
a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PluginsTestHelper.java
+++
b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PluginsTestHelper.java
@@ -50,9 +50,18 @@ public class PluginsTestHelper {
}
public static Path getFileSystemFetcherConfig(String templateName, Path
configBase, Path fetcherBase, Path emitterBase, boolean
emitIntermediateResults) throws Exception {
+ return getFileSystemFetcherConfig(templateName, configBase,
fetcherBase, emitterBase, emitIntermediateResults, Map.of());
+ }
+
+ /**
+ * @param extraReplacements template placeholders beyond the standard
ones, mapped to their
+ * values (see {@link JsonConfigHelper} for the
supported value types)
+ */
+ public static Path getFileSystemFetcherConfig(String templateName, Path
configBase, Path fetcherBase, Path emitterBase,
+ boolean
emitIntermediateResults, Map<String, Object> extraReplacements) throws
Exception {
Path pipesConfig = configBase.resolve("pipes-config.json");
- Map<String, Object> replacements = new HashMap<>();
+ Map<String, Object> replacements = new HashMap<>(extraReplacements);
replacements.put("FETCHER_BASE_PATH", fetcherBase);
if (emitterBase != null) {
diff --git
a/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-external-sleep.json
b/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-external-sleep.json
new file mode 100644
index 0000000000..2b1e330ca9
--- /dev/null
+++
b/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-external-sleep.json
@@ -0,0 +1,54 @@
+{
+ "content-handler-factory": {
+ "basic-content-handler-factory": {
+ "type": "TEXT",
+ "writeLimit": -1,
+ "throwOnWriteLimitReached": true
+ }
+ },
+ "fetchers": {
+ "fsf": {
+ "file-system-fetcher": {
+ "basePath": "FETCHER_BASE_PATH",
+ "extractFileSystemMetadata": false
+ }
+ }
+ },
+ "emitters": {
+ "fse": {
+ "file-system-emitter": {
+ "basePath": "EMITTER_BASE_PATH",
+ "fileExtension": "json",
+ "onExists": "EXCEPTION"
+ }
+ }
+ },
+ "pipes": {
+ "parseMode": "RMETA",
+ "onParseException": "EMIT",
+ "numClients": 1,
+ "useSharedServer": "USE_SHARED_SERVER",
+ "emitIntermediateResults": "EMIT_INTERMEDIATE_RESULTS",
+ "heartbeatIntervalMillis": 200,
+ "forkedJvmArgs": ["-Xmx256m", "-XX:TieredStopAtLevel=1"],
+ "emitStrategy": {
+ "type": "DYNAMIC",
+ "thresholdBytes": 1000000
+ }
+ },
+ "parsers": [
+ {
+ "external-parser": {
+ "_mime-include": ["text/plain"],
+ "commandLine": "COMMAND_LINE",
+ "timeoutMillis": 300000
+ }
+ }
+ ],
+ "parse-context": {
+ "timeout-limits": {
+ "progressTimeoutMillis": 30000
+ }
+ },
+ "plugin-roots": "PLUGINS_PATHS"
+}
