Skip to content

TIKA-4815: trivial follow-ups from the PR #3022 review #3026

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tballison merged 1 commit into from
Aug 14, 2026
Merged

TIKA-4815: trivial follow-ups from the PR #3022 review #3026

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions docs/modules/ROOT/pages/using-tika/grpc/index.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down
10 changes: 7 additions & 3 deletions tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcConcurrencyTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}

/**
Expand All @@ -91,12 +91,16 @@ public void concurrentCallsAllParseTheirOwnDocument(Resources resources) throws
*/
@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 {
Expand Down
2 changes: 1 addition & 1 deletion ...ipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,7 +290,7 @@ public void markServerForRestart() {

@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;
}

Expand Down
120 changes: 60 additions & 60 deletions ...es/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientInterruptTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -67,37 +67,36 @@ public void interruptClosesTheConnection() throws Exception {

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");
}
}
}

Expand All @@ -109,7 +108,7 @@ public void interruptClosesTheConnection() throws Exception {
* is delivered.
*/
@Test
@Timeout(30)
@Timeout(45)
public void interruptDuringStartupBackoffAbandonsTheConnection() throws Exception {
try (ServerSocket serverSocket = new ServerSocket(0)) {
CountDownLatch badHandshakeSent = new CountDownLatch(1);
Expand All @@ -120,38 +119,39 @@ public void interruptDuringStartupBackoffAbandonsTheConnection() throws Exceptio
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");
}
}
}

Expand Down
Loading