TIKA-4839 - simplify signature (#3081)

Author: tballison

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java

@@ -179,6 +179,7 @@ private static long totalMemorySize() {
     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 boolean needsRestart() {
         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 void connectionAbandoned() {
     }
 
     @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 @@ private synchronized void startServer() throws IOException, InterruptedException
 
         try {
             process = pb.start();
+            generation++;
         } catch (Exception e) {
             deleteDir(tmpDir);
             tmpDir = null;
@@ -531,11 +546,17 @@ private void teardown() throws InterruptedException {
     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;
         }
     }
 

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java

@@ -235,6 +235,16 @@ public PipesResult process(FetchEmitTuple t) throws IOException, InterruptedExce
             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 {

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 @@ default boolean needsRestart() {
      *
      * @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);
 
 }

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java

@@ -169,19 +169,6 @@ public void ensureRunning() throws IOException, InterruptedException, TimeoutExc
      * 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 long getRestartCount(RestartReason reason) {
     }
 
     /** 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) {

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

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 void interruptClosesTheConnection() throws Exception {
                 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");
             }
         }
     }

tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientPayloadLimitTest.java

@@ -18,6 +18,7 @@
 
 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 void oversizedRequestFailsFastWithoutSending() throws Exception {
                 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");
             }

tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/SentinelServerManager.java

@@ -26,6 +26,9 @@
 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 @@ public int getPort() {
 
     @Override
     public void ensureRunning() {
+        if (closed) {
+            throw new IllegalStateException("sentinel server manager is closed");
+        }
         // the scripted server is already listening
     }
 
@@ -68,8 +74,23 @@ public Path getTempDirectory() {
         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;
     }
 }