You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tika.apache.org by ta...@apache.org on 2026/08/27 10:45:05 UTC
(tika) branch TIKA-4839-tweaks created (now 2cfc49aa33)
This is an automated email from the ASF dual-hosted git repository.
tballison pushed a change to branch TIKA-4839-tweaks
in repository https://gitbox.apache.org/repos/asf/tika.git
at 2cfc49aa33 TIKA-4839 - simplify signature
This branch includes the following new commits:
new 2cfc49aa33 TIKA-4839 - simplify signature
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(tika) 01/01: TIKA-4839 - simplify signature
Posted by ta...@apache.org.
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 <ta...@apache.org>
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;