This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4839-tweaks in repository https://gitbox.apache.org/repos/asf/tika.git
commit 2cfc49aa3383a36b4faa5846c9e59a93a68a5ee4 Author: tallison <[email protected]> AuthorDate: Thu Aug 27 06:44:45 2026 -0400 TIKA-4839 - simplify signature --- CHANGES.txt | 9 ++ .../tika/pipes/core/PerClientServerManager.java | 34 +++-- .../org/apache/tika/pipes/core/PipesClient.java | 9 ++ .../org/apache/tika/pipes/core/ServerManager.java | 64 ++------- .../tika/pipes/core/SharedServerManager.java | 22 --- .../tika/pipes/core/PipesClientInterruptTest.java | 6 + .../pipes/core/PipesClientPayloadLimitTest.java | 4 + .../tika/pipes/core/SentinelServerManager.java | 16 +++ .../pipes/core/ServerManagerMarkContractTest.java | 150 --------------------- .../apache/tika/pipes/core/PipesClientTest.java | 20 +++ 10 files changed, 102 insertions(+), 232 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index d517042ccb..eb409c8654 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,14 @@ Release 4.1.0 - unreleased + * tika-pipes: a parse that raced PipesParser.close()/AsyncProcessor.close() + threw IllegalStateException out of PipesParser.parse() in per-client mode; + both modes now return a FAILED_TO_INITIALIZE result, which a caller can act + on. ServerManager's restart-reporting surface is reduced to one spelling -- + markServerForRestart(RestartReason, long) and handleCrashAndGetExitCode(long), + both abstract. The previous no-arg and reasonless forms defaulted to one + another, so an implementation that overrode only one left the others + silently inert (TIKA-4839). + * Add Micrometer reporting and opt-in endpoint for tika-server (TIKA-4839). * Improve spooling/decrease number of spills to disk (TIKA-4835). 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 b381251d8c..32a19d0cf0 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 @@ -179,6 +179,7 @@ public class PerClientServerManager implements ServerManager { private volatile Path tmpDir; private volatile int port = -1; private long filesProcessed = 0; + private volatile long generation; private volatile boolean pendingRestart = false; private final RestartCounter restarts = new RestartCounter(); // Set once by shutdown()/close(); guards a request thread from starting a fresh @@ -291,15 +292,21 @@ public class PerClientServerManager implements ServerManager { return pendingRestart; } + /** + * One client owns one manager here, so {@code generation} carries no information a sibling + * could invalidate and is accepted only to satisfy the single {@link ServerManager} spelling. + * Shared mode is where staleness is real. + */ @Override - public void markServerForRestart() { - markServerForRestart(RestartReason.CRASH); + public void markServerForRestart(RestartReason reason, long generation) { + LOG.info("clientId={}: marking server for restart ({})", clientId, reason); + markForRestart(reason); } + /** Counts forks so {@code PipesParser.getGeneration()} is meaningful in per-client mode too. */ @Override - public void markServerForRestart(RestartReason reason) { - LOG.info("clientId={}: marking server for restart ({})", clientId, reason); - markForRestart(reason); + public long getGeneration() { + return generation; } private void markForRestart(RestartReason reason) { @@ -319,7 +326,7 @@ public class PerClientServerManager implements ServerManager { } @Override - public int handleCrashAndGetExitCode() { + public int handleCrashAndGetExitCode(long generation) { // Not marked: RestartCounter attributes by exit code; the caller refines OOM/TIMEOUT. pendingRestart = true; if (process != null) { @@ -475,6 +482,7 @@ public class PerClientServerManager implements ServerManager { try { process = pb.start(); + generation++; } catch (Exception e) { deleteDir(tmpDir); tmpDir = null; @@ -531,11 +539,17 @@ public class PerClientServerManager implements ServerManager { 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); + try { + process.waitFor(WAIT_ON_DESTROY_MS, TimeUnit.MILLISECONDS); + if (process.isAlive()) { + LOG.error("clientId={}: process still alive after {}ms", clientId, WAIT_ON_DESTROY_MS); + } + } finally { + // An interrupt here must not leave the field pointing at a SIGKILLed process: + // ensureRunning would then see process == previous and skip counting the restart, + // startServer() would try to reap it again, and tmpDir would never be deleted. + process = null; } - process = null; } } 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 77339145f5..f7f7a85396 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 @@ -235,6 +235,15 @@ public class PipesClient implements Closeable { closeConnection(); return buildFatalResult(t.getId(), t.getEmitKey(), PipesResult.RESULT_STATUS.FAILED_TO_INITIALIZE, intermediateResult.get()); + } catch (IllegalStateException e) { + // The manager was closed underneath us: a request thread racing PipesParser.close() + // or AsyncProcessor.close(), which interrupts workers without awaiting them. Nothing + // to restart and nothing to recover -- but report it rather than letting an unchecked + // exception escape PipesParser.parse() to a caller that cannot act on it. + LOG.warn("clientId={}: server manager closed while initializing {}", pipesClientId, t.getId()); + closeConnection(); + return buildFatalResult(t.getId(), t.getEmitKey(), PipesResult.RESULT_STATUS.FAILED_TO_INITIALIZE, + intermediateResult.get(), e.getMessage()); } try { 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 f620cda3c0..faa0138807 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 @@ -95,61 +95,28 @@ public interface ServerManager extends Closeable { */ java.nio.file.Path getTempDirectory(); - /** - * Marks the server for restart due to a fatal error (OOM, timeout, etc.). - * <p> - * This is called by clients when they receive a fatal error status from the server. - * It signals that the server process is stopping, even if {@link #isRunning()} - * might still return true briefly. The next call to {@link #ensureRunning()} will - * wait for the process to fully exit and then restart. - * <p> - * The reason form below defaults to this one, so this must NOT default to the reason form: - * an implementation overriding neither would recurse until the stack blew. Concrete managers - * in tika-pipes override both, so callers of either spelling reach a real implementation. - */ - default void markServerForRestart() { - // Default no-op: preserves implementations written before RestartReason existed. - } - - /** As {@link #markServerForRestart()}, attributing the restart to {@code reason}. Override this one. */ - default void markServerForRestart(RestartReason reason) { - markServerForRestart(); - } - /** * The generation of the currently running process: a counter incremented every time this * manager forks a replacement. A client captures it when it connects and hands it back with * every report, so a report about a process that has already been replaced can be recognised * and dropped rather than being applied to its healthy successor. */ - default long getGeneration() { - return 0; - } - - /** - * As {@link #markServerForRestart(RestartReason)}, but only if {@code generation} is still - * current. Reports about a superseded process are dropped. - */ - default void markServerForRestart(RestartReason reason, long generation) { - markServerForRestart(reason); - } + long getGeneration(); /** - * The reasonless spelling of the above, kept for callers that cannot attribute the failure. - * Routed through the reason form rather than the bare no-arg default: that default exists - * only to keep pre-RestartReason implementations working, and delegating here would leave - * this silently inert for any implementation that overrides only the reason form. - */ - default void markServerForRestart(long generation) { - markServerForRestart(RestartReason.CRASH, generation); - } - - /** - * As {@link #handleCrashAndGetExitCode()}, but only if {@code generation} is still current. + * Marks the server for restart due to a fatal error, attributed to {@code reason}, but only + * if {@code generation} is still current -- reports about a superseded process are dropped. + * <p> + * Called by a client that received a fatal status: the process is stopping even if + * {@link #isRunning()} still says otherwise, and the next {@link #ensureRunning()} waits for + * it to exit and restarts it. + * <p> + * Deliberately the only spelling, and deliberately abstract. Earlier revisions offered a + * no-arg and a reasonless form defaulting to one another; an implementation that overrode + * only one left the others silently inert, which is how a worker known to be poisoned kept + * being handed documents. */ - default int handleCrashAndGetExitCode(long generation) { - return handleCrashAndGetExitCode(); - } + void markServerForRestart(RestartReason reason, long generation); /** Restarts performed so far for {@code reason}; monotonic, never reset. */ default long getRestartCount(RestartReason reason) { @@ -205,9 +172,6 @@ public interface ServerManager extends Closeable { * * @return the exit code if available, or -1 if the process is still running or unavailable */ - default int handleCrashAndGetExitCode() { - markServerForRestart(RestartReason.CRASH); - return -1; - } + int handleCrashAndGetExitCode(long generation); } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java index 17cdf4f6ff..a4a0478ae9 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java @@ -169,19 +169,6 @@ public class SharedServerManager implements ServerManager { * Called by a client that received OOM or TIMEOUT: the process is exiting even if * isRunning() still says otherwise; the next ensureRunning() restarts it. */ - @Override - public void markServerForRestart() { - markServerForRestart(RestartReason.CRASH); - } - - @Override - public void markServerForRestart(RestartReason reason) { - synchronized (lock) { - LOG.debug("Server marked for restart ({}) - will restart on next ensureRunning()", reason); - markForRestart(reason); - } - } - @Override public void markServerForRestart(RestartReason reason, long generation) { synchronized (lock) { @@ -228,15 +215,6 @@ public class SharedServerManager implements ServerManager { } /** Another client may already have attributed this crash (OOM/TIMEOUT); don't overwrite it. */ - @Override - public int handleCrashAndGetExitCode() { - synchronized (lock) { - restarts.markIfUnmarked(RestartReason.CRASH); - pendingRestart = true; - } - return -1; - } - @Override public int handleCrashAndGetExitCode(long generation) { synchronized (lock) { 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 48a7336934..46fc2a1210 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 @@ -16,6 +16,7 @@ */ package org.apache.tika.pipes.core; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.DataInputStream; @@ -96,6 +97,11 @@ public class PipesClientInterruptTest { 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"); + // Recycling on an abandoned connection travels connectionAbandoned(), which the + // real managers attribute to CONNECTION_ABANDONED. Marking here too would double + // count the restart and overwrite that reason with a less specific one. + assertNull(manager.marked, + "an interrupt must recycle via connectionAbandoned(), not by marking"); } } } diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientPayloadLimitTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientPayloadLimitTest.java index 67b62a50f3..6440a4a087 100644 --- a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientPayloadLimitTest.java +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientPayloadLimitTest.java @@ -18,6 +18,7 @@ package org.apache.tika.pipes.core; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.DataInputStream; @@ -72,6 +73,9 @@ public class PipesClientPayloadLimitTest { assertTrue(result.message().contains("maxIpcPayloadBytes"), "message should name the limit, got: " + result.message()); assertFalse(manager.abandoned, "nothing was sent; no reason to abandon"); + assertNull(manager.marked, + "the request was refused before anything was written; the worker is " + + "healthy and must not be recycled"); assertFalse(connectionClosed.await(300, TimeUnit.MILLISECONDS), "nothing was sent; the connection must stay usable"); } diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/SentinelServerManager.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/SentinelServerManager.java index 7007072b51..885f195102 100644 --- a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/SentinelServerManager.java +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/SentinelServerManager.java @@ -26,6 +26,7 @@ import java.nio.file.Path; final class SentinelServerManager implements ServerManager { private final int port; volatile boolean abandoned; + volatile RestartReason marked; SentinelServerManager(int port) { this.port = port; @@ -68,6 +69,21 @@ final class SentinelServerManager implements ServerManager { return null; } + @Override + public long getGeneration() { + return 0; + } + + @Override + public void markServerForRestart(RestartReason reason, long generation) { + marked = reason; + } + + @Override + public int handleCrashAndGetExitCode(long generation) { + return -1; + } + @Override public void close() { // nothing to close diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/ServerManagerMarkContractTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/ServerManagerMarkContractTest.java deleted file mode 100644 index 72bd63e7df..0000000000 --- a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/ServerManagerMarkContractTest.java +++ /dev/null @@ -1,150 +0,0 @@ -/* - * 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.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; - -import java.net.Socket; -import java.nio.file.Path; - -import org.junit.jupiter.api.Test; - -/** - * {@link ServerManager} offers two spellings of "recycle this worker" and every in-repo caller - * uses the reason form, so a regression in the no-arg form is invisible to every other test. - * These pin the contract from the <em>caller's</em> side: whichever spelling a downstream - * integration picked up, the worker must actually be marked. - */ -public class ServerManagerMarkContractTest { - - /** Overrides only the no-arg form -- what a pre-RestartReason implementation would have. */ - private static class NoArgOnly implements ServerManager { - private int marks; - - @Override - public void markServerForRestart() { - marks++; - } - - @Override - public int getPort() { - return -1; - } - - @Override - public void ensureRunning() { - } - - @Override - public Socket connect(int socketTimeoutMillis) { - return null; - } - - @Override - public void shutdown() { - } - - @Override - public boolean isRunning() { - return false; - } - - @Override - public Path getTempDirectory() { - return null; - } - - @Override - public void close() { - } - } - - private static PipesConfig config() { - return new PipesConfig(); - } - - @Test - public void testPerClientHonoursBothSpellings() { - PerClientServerManager sm = new PerClientServerManager(config(), null, 0); - sm.markServerForRestart(); - assertTrue(sm.needsRestart(), "no-arg markServerForRestart() must recycle the worker"); - - PerClientServerManager other = new PerClientServerManager(config(), null, 1); - other.markServerForRestart(RestartReason.OOM); - assertTrue(other.needsRestart(), "reason form must recycle the worker"); - } - - @Test - public void testSharedHonoursBothSpellings() { - SharedServerManager sm = new SharedServerManager(config(), null, 2); - sm.markServerForRestart(); - assertTrue(sm.needsRestart(), "no-arg markServerForRestart() must recycle the worker"); - - SharedServerManager other = new SharedServerManager(config(), null, 2); - other.markServerForRestart(RestartReason.OOM); - assertTrue(other.needsRestart(), "reason form must recycle the worker"); - } - - @Test - public void testReasonFormReachesANoArgOnlyImplementation() { - NoArgOnly sm = new NoArgOnly(); - sm.markServerForRestart(RestartReason.OOM); - assertEquals(1, sm.marks, "reason form must fall back to an older no-arg implementation"); - } - - @Test - public void testDefaultsDoNotRecurse() { - // markServerForRestart(reason) defaults to the no-arg form, so the no-arg form must not - // default back to it: an implementation overriding neither would blow the stack. - ServerManager sm = new ServerManager() { - @Override - public int getPort() { - return -1; - } - - @Override - public void ensureRunning() { - } - - @Override - public Socket connect(int socketTimeoutMillis) { - return null; - } - - @Override - public void shutdown() { - } - - @Override - public boolean isRunning() { - return false; - } - - @Override - public Path getTempDirectory() { - return null; - } - - @Override - public void close() { - } - }; - sm.markServerForRestart(); - sm.markServerForRestart(RestartReason.CRASH); - } -} diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java index ad3e7e2840..e343efd060 100644 --- a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java +++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java @@ -966,6 +966,10 @@ public class PipesClientTest { assertTrue(elapsed < 10000, "client backstop should fire within a few seconds of its ~1800ms " + "deadline, not wait for the 60s default SO_TIMEOUT (took " + elapsed + "ms)"); + // Returning TIMEOUT is only half the job: the worker is still wedged on this + // request, so it must also be marked, or the pool hands it the next document. + assertEquals(RestartReason.TIMEOUT, fakeServerManager.marked, + "a worker that blew the client backstop must be marked for restart"); } } @@ -979,6 +983,7 @@ public class PipesClientTest { private static class ChattyNeverFinishingServerManager implements ServerManager { private final ServerSocket serverSocket; private volatile boolean running = true; + volatile RestartReason marked; ChattyNeverFinishingServerManager() throws IOException { serverSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); @@ -1048,6 +1053,21 @@ public class PipesClientTest { return null; } + @Override + public long getGeneration() { + return 0; + } + + @Override + public void markServerForRestart(RestartReason reason, long generation) { + marked = reason; + } + + @Override + public int handleCrashAndGetExitCode(long generation) { + return -1; + } + @Override public void close() throws IOException { running = false;
