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 ff2d687d69 TIKA-4839 - simplify signature (#3081)
ff2d687d69 is described below
commit ff2d687d692c311dfeac88ebc8b034c6ce52da65
Author: Tim Allison <[email protected]>
AuthorDate: Thu Aug 27 14:42:48 2026 -0400
TIKA-4839 - simplify signature (#3081)
---
.../tika/pipes/core/PerClientServerManager.java | 43 ++++--
.../org/apache/tika/pipes/core/PipesClient.java | 10 ++
.../org/apache/tika/pipes/core/ServerManager.java | 64 ++-------
.../tika/pipes/core/SharedServerManager.java | 22 ---
.../pipes/core/PipesClientClosedManagerTest.java | 64 +++++++++
.../tika/pipes/core/PipesClientInterruptTest.java | 6 +
.../pipes/core/PipesClientPayloadLimitTest.java | 4 +
.../tika/pipes/core/SentinelServerManager.java | 23 +++-
.../pipes/core/ServerManagerMarkContractTest.java | 150 ---------------------
.../apache/tika/pipes/core/PipesClientTest.java | 20 +++
10 files changed, 172 insertions(+), 234 deletions(-)
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..103cb397a3 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,18 +292,31 @@ 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
ignoredGeneration) {
+ 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) {
+ /**
+ * Takes the same monitor as {@link #ensureRunning()}, which consumes the
mark: recording the
+ * reason and raising the flag must not straddle a restart, or a reason
lands against a
+ * restart that has already been counted. Correctness here previously
rested on an
+ * undocumented one-client-per-manager invariant; shared mode, where
siblings are the norm,
+ * already locked for this and TIKA-4844 is what a stale mark costs.
+ */
+ private synchronized void markForRestart(RestartReason reason) {
restarts.mark(reason);
pendingRestart = true;
}
@@ -319,7 +333,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 +489,7 @@ public class PerClientServerManager implements
ServerManager {
try {
process = pb.start();
+ generation++;
} catch (Exception e) {
deleteDir(tmpDir);
tmpDir = null;
@@ -531,11 +546,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..340fbf25ef 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,16 @@ public class PipesClient implements Closeable {
closeConnection();
return buildFatalResult(t.getId(), t.getEmitKey(),
PipesResult.RESULT_STATUS.FAILED_TO_INITIALIZE,
intermediateResult.get());
+ } catch (IllegalStateException e) {
+ // Typically 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 rejected initialization of
{}", pipesClientId,
+ t.getId(), e);
+ 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/PipesClientClosedManagerTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientClosedManagerTest.java
new file mode 100644
index 0000000000..61802b5aa9
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientClosedManagerTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.ServerSocket;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.parser.ParseContext;
+import org.apache.tika.pipes.api.FetchEmitTuple;
+import org.apache.tika.pipes.api.PipesResult;
+import org.apache.tika.pipes.api.emitter.EmitKey;
+import org.apache.tika.pipes.api.fetcher.FetchKey;
+
+public class PipesClientClosedManagerTest {
+
+ /**
+ * A request that reaches initialization after its manager was closed (a
parse racing
+ * PipesParser.close()/AsyncProcessor.close()) must come back as
FAILED_TO_INITIALIZE
+ * rather than escaping as an unchecked IllegalStateException -- and must
not mark a
+ * worker for restart, since there is nothing left to restart.
+ */
+ @Test
+ @Timeout(30)
+ public void closedManagerDuringInitReturnsFailedToInitialize() throws
Exception {
+ try (ServerSocket serverSocket = new ServerSocket(0)) {
+ SentinelServerManager manager = new
SentinelServerManager(serverSocket.getLocalPort());
+ manager.closed = true;
+ try (PipesClient client = new PipesClient(new PipesConfig(),
manager)) {
+ PipesResult result = client.process(new
FetchEmitTuple("closed-manager-test",
+ new FetchKey("fetcher", "key"), new EmitKey(), new
Metadata(),
+ new ParseContext(),
FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
+
+ assertEquals(PipesResult.RESULT_STATUS.FAILED_TO_INITIALIZE,
result.status(),
+ "got: " + result.status() + " / " + result.message());
+ assertTrue(result.message().contains("closed"),
+ "message should carry the manager's reason, got: " +
result.message());
+ assertNull(manager.marked, "nothing to restart on a closed
manager");
+ assertFalse(manager.abandoned, "no connection was established
to abandon");
+ }
+ }
+ }
+}
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..49a6813ed6 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,9 @@ import java.nio.file.Path;
final class SentinelServerManager implements ServerManager {
private final int port;
volatile boolean abandoned;
+ volatile RestartReason marked;
+ /** When set, {@link #ensureRunning()} throws like a real manager that has
been closed. */
+ volatile boolean closed;
SentinelServerManager(int port) {
this.port = port;
@@ -43,6 +46,9 @@ final class SentinelServerManager implements ServerManager {
@Override
public void ensureRunning() {
+ if (closed) {
+ throw new IllegalStateException("sentinel server manager is
closed");
+ }
// the scripted server is already listening
}
@@ -68,8 +74,23 @@ 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
+ closed = true;
}
}
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;
