diff --git a/docs/modules/ROOT/pages/using-tika/grpc/index.adoc b/docs/modules/ROOT/pages/using-tika/grpc/index.adoc index ab58eec64df..e3b803b5061 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 d8064ae833c..edbf52f834b 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 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 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 024afd292ee..69da42d536d 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 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; } 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 f6b478fd739..a7a3e1be531 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 void interruptClosesTheConnection() throws Exception { PipesConfig pipesConfig = new PipesConfig(); SentinelServerManager manager = new SentinelServerManager(serverSocket.getLocalPort()); - PipesClient client = new PipesClient(pipesConfig, manager); - - AtomicReference 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 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 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); @@ -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 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 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"); + } } }