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 59e6751d44 TIKA-4815: route tika-grpc through the PipesParser pool
(#3022)
59e6751d44 is described below
commit 59e6751d441b5aa03dd956b3d6a1802348cd6c73
Author: Davide Polato <[email protected]>
AuthorDate: Fri Aug 14 19:58:37 2026 +0200
TIKA-4815: route tika-grpc through the PipesParser pool (#3022)
---
CHANGES.txt | 9 +
tika-grpc/README.md | 8 +
.../apache/tika/pipes/grpc/TikaGrpcServerImpl.java | 24 +-
tika-grpc/src/main/proto/tika.proto | 3 +-
.../tika/pipes/grpc/TikaGrpcConcurrencyTest.java | 273 ++++++++++++++++++++
.../tika/pipes/core/PerClientServerManager.java | 6 +
.../org/apache/tika/pipes/core/PipesClient.java | 12 +
.../org/apache/tika/pipes/core/ServerManager.java | 14 ++
.../tika/pipes/core/PipesClientInterruptTest.java | 280 +++++++++++++++++++++
9 files changed, 617 insertions(+), 12 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 848567bb09..26dadf72e3 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -130,6 +130,15 @@ Release 4.0.0 - ???
after delivering its reply, instead of leaving the client waiting
for a terminal signal that never came (TIKA-4804).
+ * tika-grpc now routes fetchAndParse through the PipesParser client
+ pool instead of one shared single-threaded PipesClient: concurrent
+ calls no longer crash the worker, pipes.numClients and (for the
+ first time) pipes.useSharedServer take effect, up to numClients
+ forked worker JVMs instead of one, and pool saturation surfaces
+ in-band as CLIENT_UNAVAILABLE_WITHIN_MS. An interrupted call
+ closes its connection and recycles the per-client worker, so a
+ pooled client cannot go back to the queue dirty (TIKA-4815).
+
Release 4.0.0-beta-1 - 6/29/2026
diff --git a/tika-grpc/README.md b/tika-grpc/README.md
index 2ebea44c00..00b865ab75 100644
--- a/tika-grpc/README.md
+++ b/tika-grpc/README.md
@@ -37,6 +37,14 @@ to avoid uploading hundreds of megabytes of native libraries
and plugin bundles
```
This produces `tika-grpc/target/tika-grpc-<version>.zip` but does **not**
deploy it to Nexus.
+## Concurrency
+
+`fetchAndParse` runs on a pool of forked worker JVMs sized by
`pipes.numClients`
+(when unset, derived from host cores, at most 4). A call that cannot get a
+worker within `pipes.maxWaitForClientMillis` (default 60s) returns the in-band
+status `CLIENT_UNAVAILABLE_WITHIN_MS`: the server is at capacity, not failing.
+With `pipes.useSharedServer: true` the workers share one JVM instead.
+
## Quick Start - Development Mode
The fastest way to run tika-grpc in development mode with plugin hot-reloading:
diff --git
a/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java
b/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java
index 8f63ee4767..367d6333b4 100644
--- a/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java
+++ b/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java
@@ -68,8 +68,9 @@ import org.apache.tika.pipes.api.fetcher.FetchKey;
import org.apache.tika.pipes.api.fetcher.Fetcher;
import org.apache.tika.pipes.api.fetcher.FetcherFactory;
import org.apache.tika.pipes.api.pipesiterator.PipesIteratorFactory;
-import org.apache.tika.pipes.core.PipesClient;
import org.apache.tika.pipes.core.PipesConfig;
+import org.apache.tika.pipes.core.PipesException;
+import org.apache.tika.pipes.core.PipesParser;
import org.apache.tika.pipes.core.config.ConfigStore;
import org.apache.tika.pipes.core.config.ConfigStoreFactory;
import org.apache.tika.pipes.core.fetcher.FetcherManager;
@@ -88,7 +89,7 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
PipesConfig pipesConfig;
TikaGrpcConfig tikaGrpcConfig;
- PipesClient pipesClient;
+ PipesParser pipesParser;
FetcherManager fetcherManager;
ConfigStore configStore;
Path tikaConfigPath;
@@ -120,7 +121,8 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
// modifications) are off unless explicitly enabled in the "grpc"
section.
tikaGrpcConfig = TikaGrpcConfig.load(tikaJsonConfig);
- pipesClient = new PipesClient(pipesConfig, configPath);
+ // PipesClient is single-threaded; the pool admits pipes.numClients at
a time.
+ pipesParser = PipesParser.load(tikaJsonConfig, pipesConfig,
configPath);
try {
if (pluginRootsOverride != null &&
!pluginRootsOverride.trim().isEmpty()) {
@@ -304,7 +306,7 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
contextNode.fields().forEachRemaining(entry ->
parseContext.setJsonConfig(entry.getKey(),
entry.getValue().toString()));
}
- PipesResult pipesResult = pipesClient.process(new
FetchEmitTuple(request.getFetchKey(), new
FetchKey(fetcher.getExtensionConfig().id(), request.getFetchKey()),
+ PipesResult pipesResult = pipesParser.parse(new
FetchEmitTuple(request.getFetchKey(), new
FetchKey(fetcher.getExtensionConfig().id(), request.getFetchKey()),
new EmitKey(), tikaMetadata, parseContext,
FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
FetchAndParseReply.Builder fetchReplyBuilder =
FetchAndParseReply.newBuilder()
@@ -324,7 +326,7 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
}
}
responseObserver.onNext(fetchReplyBuilder.build());
- } catch (IOException e) {
+ } catch (IOException | PipesException e) {
throw new RuntimeException(e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -630,17 +632,17 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
}
/**
- * Close the pipe client, to be called after TikaGrpcServer has shut down.
+ * Close the pipes parser, to be called after TikaGrpcServer has shut down.
*/
void postShutdown() {
- if (pipesClient != null) {
- LOG.info("Shutting down the pipes client");
+ if (pipesParser != null) {
+ LOG.info("Shutting down the pipes parser");
try {
- pipesClient.close();
+ pipesParser.close();
} catch (IOException e) {
- LOG.error("Error closing the pipes client", e);
+ LOG.error("Error closing the pipes parser", e);
} finally {
- pipesClient = null;
+ pipesParser = null;
}
}
}
diff --git a/tika-grpc/src/main/proto/tika.proto
b/tika-grpc/src/main/proto/tika.proto
index 7d365f95d1..64d3f1e67b 100644
--- a/tika-grpc/src/main/proto/tika.proto
+++ b/tika-grpc/src/main/proto/tika.proto
@@ -113,7 +113,8 @@ message FetchAndParseReply {
string fetch_key = 1;
// Metadata fields from the parse output.
map<string, string> fields = 2;
- // The status from the message. See javadoc for
org.apache.tika.pipes.PipesResult.STATUS for the list of status.
+ // The status from the message. See javadoc for
+ // org.apache.tika.pipes.api.PipesResult.RESULT_STATUS for the list of
statuses.
string status = 3;
// If there was an error, this will contain the error message.
string error_message = 4;
diff --git
a/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcConcurrencyTest.java
b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcConcurrencyTest.java
new file mode 100644
index 0000000000..d8064ae833
--- /dev/null
+++
b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcConcurrencyTest.java
@@ -0,0 +1,273 @@
+/*
+ * 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.grpc;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import com.asarkar.grpc.test.GrpcCleanupExtension;
+import com.asarkar.grpc.test.Resources;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import org.apache.commons.io.FileUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import org.apache.tika.FetchAndParseReply;
+import org.apache.tika.FetchAndParseRequest;
+import org.apache.tika.TikaGrpc;
+import org.apache.tika.pipes.api.PipesResult;
+import org.apache.tika.pipes.fetcher.fs.FileSystemFetcher;
+import org.apache.tika.serialization.config.JsonConfigHelper;
+
+/**
+ * Concurrent fetchAndParse against a server built WITHOUT directExecutor(),
+ * like the production server: each call runs on its own handler thread, so
+ * these tests exercise the pipes layer under real handler concurrency.
+ */
+@ExtendWith(GrpcCleanupExtension.class)
+public class TikaGrpcConcurrencyTest {
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ // The fetcher must come from the config file: one saved at runtime through
+ // saveFetcher is not visible to the forked worker.
+ private static final String FETCHER_ID = "nick1:is:cool:super/" +
FileSystemFetcher.class;
+
+ /**
+ * All concurrent calls must parse, and each reply must carry its own
+ * document. The barrier guarantees the calls overlap; the per-request
+ * marker catches replies wired to the wrong request even when every
+ * status says success.
+ */
+ @Test
+ public void concurrentCallsAllParseTheirOwnDocument(Resources resources)
throws Exception {
+ runConcurrentBurst(resources, writeConfig(null, null, null));
+ }
+
+ /**
+ * The same burst with pipes.useSharedServer=true. This change makes shared
+ * mode reachable from tika-grpc for the first time (the old single-client
+ * constructor always forced per-client mode), so prove the wiring end to
+ * end: one shared worker JVM, two connections, four calls.
+ */
+ @Test
+ public void sharedServerModeParsesConcurrently(Resources resources) throws
Exception {
+ runConcurrentBurst(resources, writeConfig(null, null, Boolean.TRUE));
+ }
+
+ private void runConcurrentBurst(Resources resources, Path config) throws
Exception {
+ int concurrency = 4;
+ TikaGrpcServerImpl service = new
TikaGrpcServerImpl(config.toAbsolutePath().toString());
+ List<File> testFiles = new ArrayList<>();
+ ExecutorService pool = Executors.newFixedThreadPool(concurrency);
+ try {
+ TikaGrpc.TikaBlockingStub stub = startServer(resources, service);
+ warmUp(stub, testFiles);
+
+ CyclicBarrier barrier = new CyclicBarrier(concurrency);
+ List<Callable<FetchAndParseReply>> calls = new ArrayList<>();
+ List<String> fetchKeys = new ArrayList<>();
+ List<String> markers = new ArrayList<>();
+ for (int i = 0; i < concurrency; i++) {
+ String marker = "tika4815-marker-" + i + "-" +
UUID.randomUUID();
+ String fetchKey = "tika4815-doc-" + i + "-" +
UUID.randomUUID() + ".html";
+ writeDoc(testFiles, fetchKey, marker);
+ fetchKeys.add(fetchKey);
+ markers.add(marker);
+ calls.add(() -> {
+ barrier.await(30, TimeUnit.SECONDS);
+ return stub.fetchAndParse(FetchAndParseRequest.newBuilder()
+ .setFetcherId(FETCHER_ID)
+ .setFetchKey(fetchKey)
+ .build());
+ });
+ }
+ List<Future<FetchAndParseReply>> futures =
+ pool.invokeAll(calls, 120, TimeUnit.SECONDS);
+ for (int i = 0; i < concurrency; i++) {
+ Future<FetchAndParseReply> future = futures.get(i);
+ assertFalse(future.isCancelled(),
+ "call " + i + " did not finish within the time
budget");
+ FetchAndParseReply reply = future.get();
+ assertEquals(fetchKeys.get(i), reply.getFetchKey());
+ assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS.name(),
reply.getStatus(),
+ "call " + i + " must parse; error: " +
reply.getErrorMessage());
+ String marker = markers.get(i);
+ assertTrue(reply.getFieldsMap().values().stream()
+ .anyMatch(v -> v.contains(marker)),
+ "call " + i + " must carry its own document, not
another call's");
+ }
+ } finally {
+ pool.shutdownNow();
+ service.postShutdown();
+ cleanUp(config, testFiles);
+ }
+ }
+
+ /**
+ * With one client and a zero wait, two overlapping calls must split into
+ * one parse and one in-band CLIENT_UNAVAILABLE_WITHIN_MS, the same way
+ * every other worker outcome already reaches the caller.
+ * <p>
+ * Deliberately not warmed up: the winning call holds the only client for
+ * the whole worker fork, seconds against the loser's zero-wait admission
+ * check. A warm worker would shrink that window to one small parse.
+ */
+ @Test
+ public void saturationSurfacesInBand(Resources resources) throws Exception
{
+ Path config = writeConfig(1, 0L, null);
+ TikaGrpcServerImpl service = new
TikaGrpcServerImpl(config.toAbsolutePath().toString());
+ List<File> testFiles = new ArrayList<>();
+ ExecutorService pool = Executors.newFixedThreadPool(2);
+ try {
+ TikaGrpc.TikaBlockingStub stub = startServer(resources, service);
+
+ CyclicBarrier barrier = new CyclicBarrier(2);
+ List<Callable<FetchAndParseReply>> calls = new ArrayList<>();
+ for (int i = 0; i < 2; i++) {
+ String fetchKey = "tika4815-sat-" + i + "-" +
UUID.randomUUID() + ".html";
+ writeDoc(testFiles, fetchKey, "saturation " + i);
+ calls.add(() -> {
+ barrier.await(30, TimeUnit.SECONDS);
+ return stub.fetchAndParse(FetchAndParseRequest.newBuilder()
+ .setFetcherId(FETCHER_ID)
+ .setFetchKey(fetchKey)
+ .build());
+ });
+ }
+ List<String> statuses = new ArrayList<>();
+ List<Future<FetchAndParseReply>> futures =
+ pool.invokeAll(calls, 120, TimeUnit.SECONDS);
+ for (int i = 0; i < futures.size(); i++) {
+ Future<FetchAndParseReply> future = futures.get(i);
+ assertFalse(future.isCancelled(),
+ "call " + i + " did not finish within the time
budget");
+ statuses.add(future.get().getStatus());
+ }
+ Collections.sort(statuses);
+ assertEquals(List.of(
+
PipesResult.RESULT_STATUS.CLIENT_UNAVAILABLE_WITHIN_MS.name(),
+ PipesResult.RESULT_STATUS.PARSE_SUCCESS.name()),
+ statuses);
+ } finally {
+ pool.shutdownNow();
+ service.postShutdown();
+ cleanUp(config, testFiles);
+ }
+ }
+
+ private static TikaGrpc.TikaBlockingStub startServer(Resources resources,
+ TikaGrpcServerImpl service) throws Exception {
+ String serverName = InProcessServerBuilder.generateName();
+ // NOTE: no directExecutor() anywhere -- the production server
+ // (Grpc.newServerBuilderForPort) also dispatches on a thread pool.
+ Server server = InProcessServerBuilder.forName(serverName)
+ .addService(service)
+ .build()
+ .start();
+ resources.register(server, Duration.ofSeconds(30));
+ ManagedChannel channel =
InProcessChannelBuilder.forName(serverName).build();
+ resources.register(channel, Duration.ofSeconds(30));
+ return TikaGrpc.newBlockingStub(channel);
+ }
+
+ /**
+ * One sequential call first, so the worker is already up and the burst
+ * cannot be blamed on cold start.
+ */
+ private static void warmUp(TikaGrpc.TikaBlockingStub stub, List<File>
testFiles)
+ throws Exception {
+ String fetchKey = "tika4815-warmup-" + UUID.randomUUID() + ".html";
+ writeDoc(testFiles, fetchKey, "warmup");
+ FetchAndParseReply reply =
stub.fetchAndParse(FetchAndParseRequest.newBuilder()
+ .setFetcherId(FETCHER_ID)
+ .setFetchKey(fetchKey)
+ .build());
+ assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS.name(),
reply.getStatus(),
+ "the warmup fixture must parse, or this test proves nothing");
+ }
+
+ private static void writeDoc(List<File> testFiles, String fetchKey, String
marker)
+ throws Exception {
+ File doc = new File("target", fetchKey);
+ synchronized (testFiles) {
+ testFiles.add(doc);
+ }
+ FileUtils.writeStringToFile(doc,
+ "<html><head><title>" + marker + "</title></head><body>" +
marker
+ + "</body></html>", StandardCharsets.UTF_8);
+ }
+
+ private static Path writeConfig(Integer numClients, Long
maxWaitForClientMillis,
+ Boolean useSharedServer) throws Exception {
+ Path config = Paths.get("target", "tika4815-config-" +
UUID.randomUUID() + ".json");
+ Map<String, Object> replacements = new HashMap<>();
+ replacements.put("JAVA_PATH",
Paths.get(System.getProperty("java.home"), "bin", "java"));
+ replacements.put("FETCHER_BASE_PATH",
Paths.get("target").toAbsolutePath());
+ replacements.put("PLUGIN_ROOTS",
Paths.get("target").toAbsolutePath().resolve("plugins"));
+
JsonConfigHelper.writeConfigFromResource("/tika-pipes-test-config.json",
+ TikaGrpcConcurrencyTest.class, replacements, config);
+ if (numClients != null || maxWaitForClientMillis != null ||
useSharedServer != null) {
+ ObjectNode root = (ObjectNode)
OBJECT_MAPPER.readTree(config.toFile());
+ ObjectNode pipes = (ObjectNode) root.get("pipes");
+ if (numClients != null) {
+ pipes.put("numClients", numClients);
+ }
+ if (maxWaitForClientMillis != null) {
+ pipes.put("maxWaitForClientMillis", maxWaitForClientMillis);
+ }
+ if (useSharedServer != null) {
+ pipes.put("useSharedServer", useSharedServer);
+ }
+ Files.writeString(config,
OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
+ .writeValueAsString(root), StandardCharsets.UTF_8);
+ }
+ return config;
+ }
+
+ private static void cleanUp(Path config, List<File> testFiles) throws
Exception {
+ Files.deleteIfExists(config);
+ for (File f : testFiles) {
+ FileUtils.deleteQuietly(f);
+ }
+ }
+}
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 c4077a18b2..024afd292e 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
@@ -288,6 +288,12 @@ public class PerClientServerManager implements
ServerManager {
pendingRestart = true;
}
+ @Override
+ public void connectionAbandoned() {
+ LOG.info("clientId={}: connection abandoned mid-request, recycling the
worker", clientId);
+ pendingRestart = true;
+ }
+
@Override
public int handleCrashAndGetExitCode() {
pendingRestart = true;
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java
index 7fb00cb9f3..d0db707baf 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java
@@ -208,6 +208,13 @@ public class PipesClient implements Closeable {
PipesResult result = null;
try {
maybeInit();
+ } catch (InterruptedException e) {
+ // Same invariant as the in-flight path below: an abandoned
connection,
+ // here possibly half-established, must not be re-queued, and an
+ // abandoned per-client worker never dials back.
+ serverManager.connectionAbandoned();
+ closeConnection();
+ throw e;
} catch (ServerInitializationException e) {
LOG.error("server initialization failed: {} ", t.getId(), e);
closeConnection();
@@ -227,6 +234,11 @@ public class PipesClient implements Closeable {
// Update server manager's file counter for
maxFilesProcessedPerProcess tracking
serverManager.incrementFilesProcessed(pipesConfig.getMaxFilesProcessedPerProcess());
} catch (InterruptedException | SecurityException e) {
+ // A pooled client is re-queued right after this throw; the next
borrower
+ // must not inherit a connection with a request still in flight,
nor a
+ // per-client worker that will never dial back for a fresh connect.
+ serverManager.connectionAbandoned();
+ closeConnection();
throw e;
} catch (Exception e) {
LOG.error("exception waiting for server to complete task: {} ",
t.getId(), e);
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java
index e84bfbd754..3a663193e6 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java
@@ -110,6 +110,20 @@ public interface ServerManager extends Closeable {
// Default no-op for backward compatibility
}
+ /**
+ * Signals that the client walked away from its connection, whether
+ * mid-handshake or with a request still in flight.
+ * <p>
+ * A per-client worker dials the parent once and never dials back, so an
+ * abandoned worker must be recycled or the next {@link #connect(int)}
waits
+ * out the full accept timeout against a process that will never call. A
+ * shared server outlives its clients and needs nothing here; its
+ * connection handlers already detect the dead socket themselves.
+ */
+ default void connectionAbandoned() {
+ // Default no-op: only per-client mode must recycle the worker
+ }
+
/**
* Increments the count of files processed and marks for restart if limit
reached.
* <p>
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientInterruptTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientInterruptTest.java
new file mode 100644
index 0000000000..f6b478fd73
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientInterruptTest.java
@@ -0,0 +1,280 @@
+/*
+ * 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.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+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.emitter.EmitKey;
+import org.apache.tika.pipes.api.fetcher.FetchKey;
+import org.apache.tika.pipes.core.protocol.PipesMessage;
+import org.apache.tika.pipes.core.protocol.PipesMessageType;
+
+/**
+ * A scripted stand-in for the forked server proves what happens to the
+ * connection when the thread inside {@link PipesClient#process} is
+ * interrupted. The client rethrows InterruptedException; a pooled client
+ * then goes back to the queue, so the connection must not stay open with a
+ * request in flight -- the next borrower's ping would hang on it until
+ * the socket timeout.
+ */
+public class PipesClientInterruptTest {
+
+ /**
+ * Interrupting an in-flight process() must close the connection: the
+ * scripted server sees SHUT_DOWN or EOF instead of a socket that stays
+ * open with an abandoned request on it.
+ */
+ @Test
+ @Timeout(30)
+ public void interruptClosesTheConnection() throws Exception {
+ try (ServerSocket serverSocket = new ServerSocket(0)) {
+ CountDownLatch heartbeatStarted = new CountDownLatch(1);
+ CountDownLatch connectionClosed = new CountDownLatch(1);
+ Thread sentinel = new Thread(() ->
+ runScriptedServer(serverSocket, heartbeatStarted,
connectionClosed));
+ sentinel.setDaemon(true);
+ sentinel.start();
+
+ PipesConfig pipesConfig = new PipesConfig();
+ SentinelServerManager manager = new
SentinelServerManager(serverSocket.getLocalPort());
+ PipesClient client = new PipesClient(pipesConfig, manager);
+
+ AtomicReference<Throwable> fromProcess = new AtomicReference<>();
+ CountDownLatch processReturned = new CountDownLatch(1);
+ Thread worker = new Thread(() -> {
+ try {
+ client.process(new FetchEmitTuple("interrupt-test",
+ new FetchKey("fetcher", "key"), new EmitKey(), new
Metadata(),
+ new ParseContext(),
FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
+ } catch (Throwable t) {
+ fromProcess.set(t);
+ } finally {
+ processReturned.countDown();
+ }
+ });
+ worker.start();
+
+ assertTrue(heartbeatStarted.await(15, TimeUnit.SECONDS),
+ "the scripted server never got the request; the test
proves nothing");
+ worker.interrupt();
+
+ assertTrue(processReturned.await(15, TimeUnit.SECONDS),
+ "process() must return after the interrupt");
+ assertTrue(fromProcess.get() instanceof InterruptedException,
+ "process() must rethrow the interrupt, got: " +
fromProcess.get());
+ assertTrue(connectionClosed.await(5, TimeUnit.SECONDS),
+ "the interrupted client left its connection open with a
request in flight");
+ assertTrue(manager.abandoned,
+ "the manager was not told; a per-client worker never dials
back, so the "
+ + "next connect() would wait out the accept
timeout for nothing");
+ client.close();
+ }
+ }
+
+ /**
+ * Interrupting during startup retry must leave the same state as
+ * interrupting mid-parse: connection closed, manager told. The scripted
+ * server breaks the handshake (a valid frame of the wrong type instead of
+ * READY), which lands the client in the retry backoff where the interrupt
+ * is delivered.
+ */
+ @Test
+ @Timeout(30)
+ public void interruptDuringStartupBackoffAbandonsTheConnection() throws
Exception {
+ try (ServerSocket serverSocket = new ServerSocket(0)) {
+ CountDownLatch badHandshakeSent = new CountDownLatch(1);
+ CountDownLatch connectionClosed = new CountDownLatch(1);
+ Thread sentinel = new Thread(() ->
+ runBadHandshakeServer(serverSocket, badHandshakeSent,
connectionClosed));
+ sentinel.setDaemon(true);
+ sentinel.start();
+
+ PipesConfig pipesConfig = new PipesConfig();
+ SentinelServerManager manager = new
SentinelServerManager(serverSocket.getLocalPort());
+ PipesClient client = new PipesClient(pipesConfig, manager);
+
+ AtomicReference<Throwable> fromProcess = new AtomicReference<>();
+ CountDownLatch processReturned = new CountDownLatch(1);
+ Thread worker = new Thread(() -> {
+ try {
+ client.process(new FetchEmitTuple("interrupt-startup-test",
+ new FetchKey("fetcher", "key"), new EmitKey(), new
Metadata(),
+ new ParseContext(),
FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
+ } catch (Throwable t) {
+ fromProcess.set(t);
+ } finally {
+ processReturned.countDown();
+ }
+ });
+ worker.start();
+
+ assertTrue(badHandshakeSent.await(15, TimeUnit.SECONDS),
+ "the scripted server never got a connection; the test
proves nothing");
+ worker.interrupt();
+
+ assertTrue(processReturned.await(15, TimeUnit.SECONDS),
+ "process() must return after the interrupt");
+ assertTrue(fromProcess.get() instanceof InterruptedException,
+ "process() must rethrow the interrupt, got: " +
fromProcess.get());
+ assertTrue(connectionClosed.await(5, TimeUnit.SECONDS),
+ "the interrupted client left its half-established
connection open");
+ assertTrue(manager.abandoned,
+ "the manager was not told; an abandoned per-client worker
never "
+ + "dials back, mid-handshake or not");
+ client.close();
+ }
+ }
+
+ /**
+ * Accepts one connection and answers the handshake with a valid frame of
+ * the wrong type, sending the client into its reconnect backoff. Releases
+ * connectionClosed when the client sends SHUT_DOWN or the socket reaches
+ * EOF.
+ */
+ private static void runBadHandshakeServer(ServerSocket serverSocket,
+ CountDownLatch badHandshakeSent, CountDownLatch connectionClosed) {
+ try (Socket socket = serverSocket.accept();
+ DataInputStream in = new
DataInputStream(socket.getInputStream());
+ DataOutputStream out = new
DataOutputStream(socket.getOutputStream())) {
+ PipesMessage.ack().write(out);
+ badHandshakeSent.countDown();
+ while (true) {
+ PipesMessage message = PipesMessage.read(in);
+ if (message.type() == PipesMessageType.SHUT_DOWN) {
+ break;
+ }
+ }
+ connectionClosed.countDown();
+ } catch (IOException e) {
+ // EOF or reset: the connection is gone either way
+ connectionClosed.countDown();
+ }
+ }
+
+ /**
+ * Accepts one connection and speaks just enough protocol: READY, consume
+ * the NEW_REQUEST, then WORKING heartbeats -- each one wakes the client's
+ * read loop so the interrupt check runs. Releases connectionClosed when
+ * the client sends SHUT_DOWN or the socket reaches EOF.
+ */
+ private static void runScriptedServer(ServerSocket serverSocket,
+ CountDownLatch heartbeatStarted, CountDownLatch connectionClosed) {
+ try (Socket socket = serverSocket.accept();
+ DataInputStream in = new
DataInputStream(socket.getInputStream());
+ DataOutputStream out = new
DataOutputStream(socket.getOutputStream())) {
+ PipesMessage.ready().write(out);
+ PipesMessage.read(in); // NEW_REQUEST
+
+ Thread heartbeat = new Thread(() -> {
+ try {
+ while (true) {
+ PipesMessage.working().write(out);
+ heartbeatStarted.countDown();
+ Thread.sleep(100);
+ }
+ } catch (IOException | InterruptedException e) {
+ // socket closed under us, or test over -- either way, done
+ }
+ });
+ heartbeat.setDaemon(true);
+ heartbeat.start();
+
+ while (true) {
+ PipesMessage message = PipesMessage.read(in);
+ if (message.type() == PipesMessageType.SHUT_DOWN) {
+ break;
+ }
+ }
+ connectionClosed.countDown();
+ } catch (EOFException e) {
+ // close without SHUT_DOWN still counts: the connection is gone
+ connectionClosed.countDown();
+ } catch (IOException e) {
+ connectionClosed.countDown();
+ }
+ }
+
+ /**
+ * Points the client at the scripted server; no forked process anywhere.
+ */
+ private static final class SentinelServerManager implements ServerManager {
+ private final int port;
+ private volatile boolean abandoned;
+
+ private SentinelServerManager(int port) {
+ this.port = port;
+ }
+
+ @Override
+ public void connectionAbandoned() {
+ abandoned = true;
+ }
+
+ @Override
+ public int getPort() {
+ return port;
+ }
+
+ @Override
+ public void ensureRunning() {
+ // the scripted server is already listening
+ }
+
+ @Override
+ public Socket connect(int socketTimeoutMs) throws IOException {
+ Socket socket = new Socket("localhost", port);
+ socket.setSoTimeout(socketTimeoutMs);
+ return socket;
+ }
+
+ @Override
+ public void shutdown() {
+ // nothing to shut down
+ }
+
+ @Override
+ public boolean isRunning() {
+ return true;
+ }
+
+ @Override
+ public java.nio.file.Path getTempDirectory() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+ // nothing to close
+ }
+ }
+}
