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 433f576407 TIKA-4815: trivial follow-ups from the PR #3022 review
(#3026)
433f576407 is described below
commit 433f57640712648698ff58ed159ad4c0e60b7168
Author: Tim Allison <[email protected]>
AuthorDate: Fri Aug 14 15:40:46 2026 -0400
TIKA-4815: trivial follow-ups from the PR #3022 review (#3026)
---
docs/modules/ROOT/pages/using-tika/grpc/index.adoc | 8 ++
.../tika/pipes/grpc/TikaGrpcConcurrencyTest.java | 10 +-
.../tika/pipes/core/PerClientServerManager.java | 2 +-
.../tika/pipes/core/PipesClientInterruptTest.java | 120 ++++++++++-----------
4 files changed, 76 insertions(+), 64 deletions(-)
diff --git a/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
b/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
index ab58eec64d..e3b803b506 100644
--- a/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
+++ b/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
@@ -29,6 +29,14 @@ register a fetcher (`SaveFetcher`) and then submit
`FetchAndParseRequest`
messages, each of which returns a `FetchAndParseReply` with extracted
metadata and content.
+== Concurrency
+
+`FetchAndParse` and its streaming variants run on a pool of forked worker JVMs
+sized by `pipes.numClients` (default: derived from host cores, at most 4). A
+call that cannot get a worker within `pipes.maxWaitForClientMillis` (default
+60s) returns the in-band reply status `CLIENT_UNAVAILABLE_WITHIN_MS` -- at
+capacity, not failing. `pipes.useSharedServer: true` shares one JVM instead.
+
== Security
[WARNING]
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
index d8064ae833..edbf52f834 100644
---
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
@@ -80,7 +80,7 @@ public class TikaGrpcConcurrencyTest {
*/
@Test
public void concurrentCallsAllParseTheirOwnDocument(Resources resources)
throws Exception {
- runConcurrentBurst(resources, writeConfig(null, null, null));
+ runConcurrentBurst(resources, writeConfig(null, null, null), false);
}
/**
@@ -91,12 +91,16 @@ public class TikaGrpcConcurrencyTest {
*/
@Test
public void sharedServerModeParsesConcurrently(Resources resources) throws
Exception {
- runConcurrentBurst(resources, writeConfig(null, null, Boolean.TRUE));
+ runConcurrentBurst(resources, writeConfig(null, null, Boolean.TRUE),
true);
}
- private void runConcurrentBurst(Resources resources, Path config) throws
Exception {
+ private void runConcurrentBurst(Resources resources, Path config, boolean
expectSharedMode)
+ throws Exception {
int concurrency = 4;
TikaGrpcServerImpl service = new
TikaGrpcServerImpl(config.toAbsolutePath().toString());
+ // the burst alone can't tell the modes apart
+ assertEquals(expectSharedMode, service.pipesParser.isSharedMode(),
+ "pipes.useSharedServer did not take effect");
List<File> testFiles = new ArrayList<>();
ExecutorService pool = Executors.newFixedThreadPool(concurrency);
try {
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 024afd292e..69da42d536 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
@@ -290,7 +290,7 @@ public class PerClientServerManager implements
ServerManager {
@Override
public void connectionAbandoned() {
- LOG.info("clientId={}: connection abandoned mid-request, recycling the
worker", clientId);
+ LOG.info("clientId={}: connection abandoned, worker will be recycled
on next use", clientId);
pendingRestart = true;
}
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
index f6b478fd73..a7a3e1be53 100644
---
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
@@ -55,7 +55,7 @@ public class PipesClientInterruptTest {
* open with an abandoned request on it.
*/
@Test
- @Timeout(30)
+ @Timeout(45)
public void interruptClosesTheConnection() throws Exception {
try (ServerSocket serverSocket = new ServerSocket(0)) {
CountDownLatch heartbeatStarted = new CountDownLatch(1);
@@ -67,37 +67,36 @@ public class PipesClientInterruptTest {
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();
+ try (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(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();
+ 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");
+ }
}
}
@@ -109,7 +108,7 @@ public class PipesClientInterruptTest {
* is delivered.
*/
@Test
- @Timeout(30)
+ @Timeout(45)
public void interruptDuringStartupBackoffAbandonsTheConnection() throws
Exception {
try (ServerSocket serverSocket = new ServerSocket(0)) {
CountDownLatch badHandshakeSent = new CountDownLatch(1);
@@ -120,38 +119,39 @@ public class PipesClientInterruptTest {
sentinel.start();
PipesConfig pipesConfig = new PipesConfig();
+ // handshake reads aren't interrupt-responsive; a late interrupt
waits this out
+ pipesConfig.setStartupTimeoutMillis(1000);
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();
+ try (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(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();
+ 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");
+ }
}
}
