PerClientServerManager.java
/*
* 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 java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.tika.config.TikaExtras;
import org.apache.tika.pipes.core.server.PipesServer;
import org.apache.tika.utils.ProcessUtils;
/**
* Manages a dedicated PipesServer process for a single PipesClient.
* <p>
* This is the default mode where each PipesClient has its own server process,
* providing isolation at the cost of memory overhead.
* <p>
* Connection model: The client creates a ServerSocket and the server connects TO it.
* This is the reverse of typical client-server patterns but allows the client to
* control the port assignment.
*/
public class PerClientServerManager implements ServerManager {
private static final Logger LOG = LoggerFactory.getLogger(PerClientServerManager.class);
private static final long WAIT_ON_DESTROY_MS = 10000;
public static final int SOCKET_CONNECT_TIMEOUT_MS = 60000;
/** Cores reserved for the parent JVM when auto-sizing forked JVMs'
* -XX:ActiveProcessorCount. The parent has client-side serialization,
* response deserialization, and heartbeat bookkeeping; if it's CPU-starved
* small operations like socket flush show pathological tail latency. */
private static final int PARENT_RESERVED_CORES = 2;
/** Don't auto-cap below this many CPUs per fork. At cap=1 the fork's only
* CPU is fully consumed by parsing, so its socket-reader thread can't run
* and the parent's writes block on receiver-side back-pressure -- worse
* than no cap at all. This guard matters for small k8s pods where the
* formula could otherwise produce slice=1. */
private static final int MIN_AUTO_CAP_SLICE = 2;
/** Share of host/container memory the forks may collectively claim; the remainder is
* left for the parent JVM, the OS, and page cache for spooled input. */
private static final int FORK_HEAP_BUDGET_PERCENT = 75;
private static boolean userSetHeap(List<String> args) {
return args.stream().anyMatch(a -> a.startsWith("-Xmx")
|| a.startsWith("-XX:MaxRAMPercentage")
|| a.startsWith("-XX:MaxRAMFraction"));
}
private static int forkHeapPercentage(int numClients) {
return Math.max(1, FORK_HEAP_BUDGET_PERCENT / numClients);
}
/**
* Explicit heap settings suppress auto-division, so nothing else checks that
* numClients forks at the user's size can actually coexist -- the mistake only
* surfaces later as fork OOMs or the host OOM-killer. Returns a warning when
* the combined explicit ceiling exceeds the fork budget, else null.
* Package-private for tests.
*/
static String heapOvercommitWarning(List<String> forkedJvmArgs, int numClients,
long totalMemBytes) {
long xmxBytes = -1;
double maxRamPct = -1;
// last occurrence wins, same as the JVM
for (String arg : forkedJvmArgs) {
if (arg.startsWith("-Xmx")) {
xmxBytes = parseJvmMemArg(arg.substring("-Xmx".length()));
} else if (arg.startsWith("-XX:MaxRAMPercentage=")) {
try {
maxRamPct = Double.parseDouble(
arg.substring("-XX:MaxRAMPercentage=".length()));
} catch (NumberFormatException e) {
maxRamPct = -1;
}
}
}
// -Xmx beats MaxRAMPercentage in the JVM, so only judge the percentage alone
if (xmxBytes <= 0 && maxRamPct > 0 && numClients * maxRamPct > FORK_HEAP_BUDGET_PERCENT) {
return String.format(java.util.Locale.ROOT,
"numClients=%d x -XX:MaxRAMPercentage=%s commits %.0f%% of memory; " +
"the parent JVM and OS need the remainder (fork budget: %d%%). " +
"Lower numClients or the percentage.",
numClients, maxRamPct, numClients * maxRamPct, FORK_HEAP_BUDGET_PERCENT);
}
if (xmxBytes > 0 && totalMemBytes > 0
&& numClients * (double) xmxBytes > totalMemBytes * FORK_HEAP_BUDGET_PERCENT / 100.0) {
return String.format(java.util.Locale.ROOT,
"numClients=%d x -Xmx (%,dMB each) commits %,dMB of %,dMB total memory, " +
"over the %d%% fork budget; the parent JVM and OS need the remainder. " +
"Lower numClients or -Xmx.",
numClients, xmxBytes / MB, numClients * xmxBytes / MB, totalMemBytes / MB,
FORK_HEAP_BUDGET_PERCENT);
}
return null;
}
private static final long MB = 1024 * 1024;
// "-Xmx" style value: digits with optional k/m/g/t suffix; -1 if unparseable
static long parseJvmMemArg(String value) {
if (value == null || value.isEmpty()) {
return -1;
}
long multiplier = switch (Character.toLowerCase(value.charAt(value.length() - 1))) {
case 'k' -> 1024L;
case 'm' -> 1024L * 1024;
case 'g' -> 1024L * 1024 * 1024;
case 't' -> 1024L * 1024 * 1024 * 1024;
default -> 1;
};
String digits = multiplier == 1 ? value : value.substring(0, value.length() - 1);
try {
long n = Long.parseLong(digits);
return n <= 0 ? -1 : Math.multiplyExact(n, multiplier);
} catch (NumberFormatException | ArithmeticException e) {
return -1;
}
}
// Container-aware on JDK 17 (honors cgroup memory limits); -1 if unavailable.
// Read via JMX attribute name: the typed accessor lives on a com.sun
// interface that forbiddenapis bans as non-portable.
private static long totalMemorySize() {
try {
Object value = java.lang.management.ManagementFactory.getPlatformMBeanServer()
.getAttribute(new javax.management.ObjectName("java.lang:type=OperatingSystem"),
"TotalMemorySize");
return value instanceof Number n ? n.longValue() : -1;
} catch (Exception e) {
return -1;
}
}
private final PipesConfig pipesConfig;
private final Path tikaConfigPath;
private final int clientId;
private volatile Process process;
private volatile ServerSocket serverSocket;
private volatile Path tmpDir;
private volatile int port = -1;
private long filesProcessed = 0;
private volatile boolean pendingRestart = false;
// Set once by shutdown()/close(); guards a request thread from starting a fresh
// process after the manager has been torn down (which would leak the child).
private volatile boolean closed = false;
public PerClientServerManager(PipesConfig pipesConfig, Path tikaConfigPath, int clientId) {
this.pipesConfig = pipesConfig;
this.tikaConfigPath = tikaConfigPath;
this.clientId = clientId;
// Emit CPU-sizing diagnostics once per PipesParser (only on the first client).
if (clientId == 0) {
logCpuSizing();
}
}
/**
* Emits a one-shot summary of how the auto-cap will behave for this PipesParser,
* plus warnings for clearly-pathological provisioning. Grep for "pipes-cpu-sizing"
* in logs to see the decision the JVM made.
*/
private void logCpuSizing() {
int hostCores = Runtime.getRuntime().availableProcessors();
int numClients = pipesConfig.getNumClients();
boolean userSetCap = pipesConfig.getForkedJvmArgs().stream()
.anyMatch(a -> a.startsWith("-XX:ActiveProcessorCount="));
// Hostile environment: fewer than 2 cores means the parser thread, GC, JIT,
// and protocol heartbeat all share one CPU. Pipes will run but tail latency
// will be poor regardless of numClients.
if (hostCores < 2) {
LOG.warn("pipes-cpu-sizing: hostCores={} is below the practical minimum. " +
"Each fork JVM needs roughly 2 CPUs (1 for parsing, 1 for GC/JIT/" +
"protocol heartbeat); on a single-CPU host these contend with each " +
"other and performance will be poor.", hostCores);
}
// Over-provisioned: numClients packed too tightly given the host's cores.
// Triggers earlier than the slice<MIN guard so the user is warned even
// when they explicitly set -XX:ActiveProcessorCount themselves.
if (numClients > 1 && numClients * MIN_AUTO_CAP_SLICE + PARENT_RESERVED_CORES > hostCores) {
int recommendedMax = Math.max(1,
(hostCores - PARENT_RESERVED_CORES) / MIN_AUTO_CAP_SLICE);
LOG.warn("pipes-cpu-sizing: numClients={} is over-provisioned for {}-core " +
"host. Recommended max for this host: numClients={}. Forks need at " +
"least {} CPUs each plus {} reserved for the parent JVM; otherwise " +
"GC/JIT/protocol threads contend with parser threads across forks.",
numClients, hostCores, recommendedMax,
MIN_AUTO_CAP_SLICE, PARENT_RESERVED_CORES);
}
// Always-on summary so ops can see what was decided. Grep for "pipes-cpu-sizing".
String capDecision;
if (userSetCap) {
capDecision = "user-set in forkedJvmArgs";
} else {
int budget = Math.max(1, hostCores - PARENT_RESERVED_CORES);
int slice = budget / numClients;
capDecision = (slice >= MIN_AUTO_CAP_SLICE)
? "slice=" + slice
: "skipped (slice<" + MIN_AUTO_CAP_SLICE + ")";
}
String heapDecision;
if (userSetHeap(pipesConfig.getForkedJvmArgs())) {
heapDecision = "user-set in forkedJvmArgs";
} else {
heapDecision = "MaxRAMPercentage=" + forkHeapPercentage(numClients);
}
LOG.info("pipes-cpu-sizing: hostCores={}, numClients={}, parentReserved={}, " +
"autoCap={}, heap={}", hostCores, numClients, PARENT_RESERVED_CORES,
capDecision, heapDecision);
String overcommit = heapOvercommitWarning(pipesConfig.getForkedJvmArgs(),
numClients, totalMemorySize());
if (overcommit != null) {
LOG.warn("pipes-cpu-sizing: {}", overcommit);
}
}
@Override
public int getPort() {
return port;
}
@Override
public Path getTempDirectory() {
return tmpDir;
}
@Override
public boolean isRunning() {
return process != null && process.isAlive();
}
@Override
public void incrementFilesProcessed(long maxFilesPerProcess) {
if (maxFilesPerProcess <= 0) {
return;
}
filesProcessed++;
if (filesProcessed >= maxFilesPerProcess) {
LOG.info("clientId={}: reached max files limit ({}/{}), marking for restart",
clientId, filesProcessed, maxFilesPerProcess);
pendingRestart = true;
}
}
@Override
public boolean needsRestart() {
return pendingRestart;
}
@Override
public void markServerForRestart() {
LOG.info("clientId={}: marking server for restart", clientId);
pendingRestart = true;
}
@Override
public void connectionAbandoned() {
LOG.info("clientId={}: connection abandoned, worker will be recycled on next use", clientId);
pendingRestart = true;
}
@Override
public int handleCrashAndGetExitCode() {
pendingRestart = true;
if (process != null) {
try {
process.waitFor(1, TimeUnit.SECONDS);
if (!process.isAlive()) {
int exitValue = process.exitValue();
if (exitValue == 0) {
LOG.info("clientId={}: process exited cleanly", clientId);
} else {
LOG.warn("clientId={}: process exited with code {}", clientId, exitValue);
}
return exitValue;
} else {
LOG.warn("clientId={}: process still running after crash", clientId);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
return -1;
}
@Override
public synchronized void ensureRunning() throws IOException, InterruptedException, TimeoutException, ServerInitializationException {
if (closed) {
throw new IllegalStateException("PerClientServerManager is closed");
}
// Check if server is running AND not marked for restart
if (isRunning() && !pendingRestart) {
return;
}
startServer();
}
@Override
public Socket connect(int socketTimeoutMillis) throws IOException, ServerInitializationException {
// Capture the socket up front: shutdown() may null the field concurrently, but this
// request keeps using (and detects the close on) the instance it started with.
ServerSocket ss = serverSocket;
if (ss == null) {
throw new IllegalStateException("Server not started. Call ensureRunning() first.");
}
// Accept incoming connection from the server process
ss.setSoTimeout(1000); // 1 second timeout for each poll
long startTime = System.currentTimeMillis();
while (true) {
try {
Socket socket = ss.accept();
socket.setSoTimeout(socketTimeoutMillis);
socket.setTcpNoDelay(true);
LOG.debug("clientId={}: accepted connection from server", clientId);
return socket;
} catch (SocketTimeoutException e) {
// Check if the process died (or the manager was shut down) before connecting.
Process p = process;
if (p == null) {
throw new IOException("Server manager was shut down while connecting");
}
if (!p.isAlive()) {
int exitValue = p.exitValue();
LOG.error("clientId={}: Process exited with code {} before connecting to socket",
clientId, exitValue);
ServerProcessIO.surfaceCrashDiagnostics(LOG, "clientId=" + clientId, tmpDir);
// Always treat pre-connect death as retryable.
// The only non-retryable paths are:
// 1. pb.start() fails (can't launch process) - handled in startServer()
// 2. Server explicitly reports bad config via protocol - handled in waitForStartup()
// 3. Exhausted all retry attempts - handled in maybeInit()
pendingRestart = true;
throw new IOException(
"Server process died before connecting (exit code " + exitValue + ") - will retry");
}
// Check if we've exceeded the overall timeout
long elapsed = System.currentTimeMillis() - startTime;
if (elapsed > SOCKET_CONNECT_TIMEOUT_MS) {
LOG.error("clientId={}: Timed out waiting for server to connect after {}ms", clientId, elapsed);
ServerProcessIO.surfaceCrashDiagnostics(LOG, "clientId=" + clientId, tmpDir);
throw new ServerInitializationException(
"Server did not connect within " + SOCKET_CONNECT_TIMEOUT_MS + "ms");
}
// Continue polling
}
}
}
private synchronized void startServer() throws IOException, InterruptedException, TimeoutException, ServerInitializationException {
if (closed) {
throw new IllegalStateException("PerClientServerManager is closed");
}
// Clean up any previous server (restart) -- teardown, not shutdown, so we do not
// mark the manager closed.
if (process != null || serverSocket != null || tmpDir != null) {
teardown();
}
// Create new server socket to get a free port
serverSocket = new ServerSocket();
serverSocket.setReuseAddress(true);
serverSocket.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 50);
port = serverSocket.getLocalPort();
LOG.trace("clientId={}: starting server on port={}", clientId, port);
tmpDir = Files.createTempDirectory("pipes-server-" + clientId + "-");
ProcessBuilder pb = new ProcessBuilder(getCommandline(tmpDir));
// Tell the child our PID so it can watch ProcessHandle.onExit() and
// self-terminate promptly if we die. Without this, an orphan child
// can only notice via socket-read timeout (default 60s) and can't
// notice at all while it is mid-parse -- the leak that TIKA-4740
// surfaced via @TempDir cleanup failures.
pb.environment().put(PipesServer.PARENT_PID_ENV,
Long.toString(ProcessHandle.current().pid()));
// Default: inherit stdio so the pipes-server's log records show up
// in the parent's stdio stream (the production case is Docker/K8s
// where container stdio is picked up by log aggregators -- writing
// to files in a container is anti-pattern). Set
// -Dtika.pipes.server.stdio=discard on the parent to suppress.
//
// The Windows surefire hang that previously made inheritIO() risky
// is mitigated by PipesServer.watchParentProcess(): when the parent
// exits, the child detects it via ProcessHandle.onExit() within
// milliseconds and System.exit()s, releasing its inherited stderr
// handle so upstream pipe readers see EOF promptly.
//
// hs_err crash logs are pointed at tmpDir via -XX:ErrorFile in
// getCommandline() and surfaced via SLF4J on abnormal exit.
if (ServerProcessIO.inheritStdio()) {
pb.inheritIO();
} else {
pb.redirectOutput(ProcessBuilder.Redirect.DISCARD);
pb.redirectError(ProcessBuilder.Redirect.DISCARD);
}
try {
process = pb.start();
} catch (Exception e) {
deleteDir(tmpDir);
tmpDir = null;
LOG.error("clientId={}: failed to start server", clientId, e);
String msg = "Failed to start server process";
if (e.getMessage() != null) {
msg += ": " + e.getMessage();
}
throw new ServerInitializationException(msg, e);
}
// Server is started, but we don't wait for connection here.
// The connection is established in connect() method.
LOG.debug("clientId={}: server process started, waiting for connection in connect()", clientId);
// Reset counters after successful start
filesProcessed = 0;
pendingRestart = false;
}
@Override
public synchronized void shutdown() throws InterruptedException {
closed = true;
teardown();
}
/**
* Tears down the current server process, socket, and temp dir without marking the manager
* closed -- shared by the final {@link #shutdown()} and by {@link #startServer()} on restart.
* Callers hold the monitor.
*/
private void teardown() throws InterruptedException {
LOG.debug("clientId={}: tearing down server", clientId);
if (serverSocket != null) {
try {
serverSocket.close();
} catch (IOException e) {
LOG.debug("Error closing server socket", e);
}
serverSocket = null;
}
destroyProcess();
if (tmpDir != null) {
deleteDir(tmpDir);
tmpDir = null;
}
port = -1;
}
private void destroyProcess() throws InterruptedException {
if (process != null) {
process.destroyForcibly();
process.waitFor(WAIT_ON_DESTROY_MS, TimeUnit.MILLISECONDS);
if (process.isAlive()) {
LOG.error("clientId={}: process still alive after {}ms", clientId, WAIT_ON_DESTROY_MS);
}
process = null;
}
}
@Override
public void close() throws IOException {
try {
shutdown();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException("Interrupted during shutdown", e);
}
}
/**
* Returns the ServerSocket for accepting client connections.
* Used by PipesClient to accept the server's incoming connection.
*
* @return the server socket
*/
public ServerSocket getServerSocket() {
return serverSocket;
}
/**
* Returns the server process.
*
* @return the process, or null if not started
*/
public Process getProcess() {
return process;
}
private void deleteDir(Path dir) {
if (dir == null) {
return;
}
try {
FileUtils.deleteDirectory(dir.toFile());
} catch (IOException e) {
LOG.warn("couldn't delete tmp dir {}", dir);
}
}
// package-private and tmpDir passed in (rather than read from the field set by
// startServer) so tests can exercise arg injection without forking a JVM
String[] getCommandline(Path tmpDir) throws IOException {
List<String> configArgs = new ArrayList<>(pipesConfig.getForkedJvmArgs());
boolean hasClassPath = false;
boolean hasHeadless = false;
boolean hasExitOnOOM = false;
boolean hasLog4j = false;
boolean hasActiveProcessorCount = false;
boolean hasErrorFile = false;
String origGCString = null;
String newGCLogString = null;
for (String arg : configArgs) {
if (arg.startsWith("-Djava.awt.headless")) {
hasHeadless = true;
}
if (arg.equals("-cp") || arg.equals("--classpath")) {
hasClassPath = true;
}
if (arg.equals("-XX:+ExitOnOutOfMemoryError") || arg.equals("-XX:+CrashOnOutOfMemoryError")) {
hasExitOnOOM = true;
}
if (arg.startsWith("-Dlog4j.configuration") || arg.startsWith("-Dlog4j2.configuration")) {
hasLog4j = true;
}
if (arg.startsWith("-XX:ActiveProcessorCount=")) {
hasActiveProcessorCount = true;
}
if (arg.startsWith("-XX:ErrorFile=")) {
hasErrorFile = true;
}
if (arg.startsWith("-Xloggc:")) {
origGCString = arg;
newGCLogString = arg.replace("${pipesClientId}", "id-" + clientId);
}
}
// Direct native-crash dumps (hs_err_pid<N>.log) into tmpDir so
// ServerProcessIO.surfaceCrashDiagnostics() can find and emit them on
// abnormal exit. The child JVM inherits the parent's CWD (we do NOT
// call pb.directory()), so without this the JVM would write hs_err
// wherever the parent was launched -- typically lost.
if (!hasErrorFile) {
configArgs.add("-XX:ErrorFile=" + tmpDir.resolve("hs_err_pid%p.log")
.toAbsolutePath());
}
// Heap gets the same treatment as CPU. Left alone, every fork independently
// takes the JVM's own default max heap (a fixed fraction of host/container
// memory), so numClients forks have a combined ceiling well above what the
// host has -- the same "each JVM thinks it owns the machine" problem the
// ActiveProcessorCount cap solves. Give each fork a slice of a fixed budget
// instead, leaving the remainder for the parent and the OS.
if (!userSetHeap(configArgs)) {
int pct = forkHeapPercentage(pipesConfig.getNumClients());
configArgs.add("-XX:MaxRAMPercentage=" + pct);
LOG.debug("clientId={}: auto-injected -XX:MaxRAMPercentage={} (numClients={})",
clientId, pct, pipesConfig.getNumClients());
}
// If the user hasn't explicitly set -XX:ActiveProcessorCount, size each
// forked JVM's view of CPUs to a fair slice of the host. Otherwise each
// JVM defaults its GC, JIT, and common ForkJoinPool to "all cores", which
// means N forked JVMs collectively spawn N x cores GC threads etc. and
// fight each other. We also reserve PARENT_RESERVED_CORES so the parent
// JVM (which serializes requests, deserializes responses, runs heartbeat
// bookkeeping) isn't starved for CPU.
// Skip the auto-cap when the computed slice would drop below
// MIN_AUTO_CAP_SLICE -- below that, the fork can't keep its socket
// reader responsive and back-pressures the parent.
if (!hasActiveProcessorCount) {
int hostCores = Runtime.getRuntime().availableProcessors();
int forkBudget = Math.max(1, hostCores - PARENT_RESERVED_CORES);
int slice = forkBudget / pipesConfig.getNumClients();
if (slice >= MIN_AUTO_CAP_SLICE) {
configArgs.add("-XX:ActiveProcessorCount=" + slice);
LOG.debug("clientId={}: auto-injected -XX:ActiveProcessorCount={} " +
"(hostCores={}, parentReserved={}, numClients={})",
clientId, slice, hostCores, PARENT_RESERVED_CORES,
pipesConfig.getNumClients());
} else {
LOG.info("clientId={}: skipping -XX:ActiveProcessorCount auto-cap " +
"(would yield slice={} < MIN_AUTO_CAP_SLICE={}; " +
"hostCores={}, parentReserved={}, numClients={}).{}",
clientId, slice, MIN_AUTO_CAP_SLICE, hostCores,
PARENT_RESERVED_CORES, pipesConfig.getNumClients(),
pipesConfig.getNumClients() > 1
? " Consider lowering numClients on this host." : "");
}
}
if (origGCString != null && newGCLogString != null) {
configArgs.remove(origGCString);
configArgs.add(newGCLogString);
}
List<String> commandLine = new ArrayList<>();
String javaPath = pipesConfig.getJavaPath();
commandLine.add(ProcessUtils.escapeCommandLine(javaPath));
if (!hasClassPath) {
Path argFile = writeArgFile(tmpDir);
commandLine.add("@" + argFile.toAbsolutePath());
}
if (!hasHeadless) {
commandLine.add("-Djava.awt.headless=true");
}
if (hasExitOnOOM) {
LOG.info("I notice that you have a jdk setting to exit/crash on OOM. If you run heavy external processes " +
"like tesseract, this setting may result in orphaned processes which could be disastrous for performance.");
}
if (!hasLog4j) {
commandLine.add("-Dlog4j.configurationFile=classpath:pipes-fork-server-default-log4j2.xml");
}
commandLine.add("-DpipesClientId=" + clientId);
commandLine.addAll(configArgs);
commandLine.add("-Djava.io.tmpdir=" + tmpDir.toAbsolutePath());
commandLine.add("org.apache.tika.pipes.core.server.PipesServer");
commandLine.add(Integer.toString(port));
commandLine.add(tikaConfigPath.toAbsolutePath().toString());
LOG.debug("clientId={}: commandline: {}", clientId, commandLine);
return commandLine.toArray(new String[0]);
}
private Path writeArgFile(Path tmpDir) throws IOException {
Path argFile = tmpDir.resolve("jvm-args.txt");
// forward any tika.extras.dir jars to the forked PipesServer
String classpath = TikaExtras.appendJarsToClasspath(System.getProperty("java.class.path"));
String normalizedClasspath = classpath.replace("\\", "/");
String content = "-cp\n\"" + normalizedClasspath + "\"\n";
Files.writeString(argFile, content, StandardCharsets.UTF_8);
LOG.debug("clientId={}: wrote argfile with classpath ({} chars) to {}, content starts with: {}",
clientId, classpath.length(), argFile, content.substring(0, Math.min(100, content.length())));
return argFile;
}
}