TIKA-4793: make Pipes IPC payload limit configurable

tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java

@@ -66,6 +66,7 @@ public enum RESULT_STATUS {
         EMIT_EXCEPTION(CATEGORY.TASK_EXCEPTION),
         FETCHER_NOT_FOUND(CATEGORY.TASK_EXCEPTION),
         EMITTER_NOT_FOUND(CATEGORY.TASK_EXCEPTION),
+        PAYLOAD_LIMIT_EXCEEDED(CATEGORY.TASK_EXCEPTION),
 
         // Process crashes - forked process died, auto-restart
         OOM(CATEGORY.PROCESS_CRASH),

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

@@ -47,6 +47,7 @@
 import org.apache.tika.pipes.api.PipesResult;
 import org.apache.tika.pipes.api.emitter.EmitKey;
 import org.apache.tika.pipes.core.emitter.EmitDataImpl;
+import org.apache.tika.pipes.core.protocol.PayloadLimitExceededException;
 import org.apache.tika.pipes.core.protocol.PipesMessage;
 import org.apache.tika.pipes.core.protocol.PipesMessageType;
 import org.apache.tika.pipes.core.serialization.JsonPipesIpc;
@@ -72,6 +73,7 @@ public class PipesClient implements Closeable {
     public static final int SOCKET_TIMEOUT_MS = 60000;
 
     private final PipesConfig pipesConfig;
+    private final int maxIpcPayloadBytes;
     private final ServerManager serverManager;
     private final boolean ownsServerManager;
     private final int pipesClientId;
@@ -95,6 +97,7 @@ public class PipesClient implements Closeable {
      */
     public PipesClient(PipesConfig pipesConfig, ServerManager serverManager) {
         this.pipesConfig = pipesConfig;
+        this.maxIpcPayloadBytes = pipesConfig.getMaxIpcPayloadBytes();
         this.serverManager = serverManager;
         this.ownsServerManager = false;
         this.pipesClientId = CLIENT_COUNTER.getAndIncrement();
@@ -112,6 +115,7 @@ public PipesClient(PipesConfig pipesConfig, ServerManager serverManager) {
      */
     public PipesClient(PipesConfig pipesConfig, java.nio.file.Path tikaConfigPath) {
         this.pipesConfig = pipesConfig;
+        this.maxIpcPayloadBytes = pipesConfig.getMaxIpcPayloadBytes();
         this.pipesClientId = CLIENT_COUNTER.getAndIncrement();
         this.serverManager = new PerClientServerManager(pipesConfig, tikaConfigPath, pipesClientId);
         this.ownsServerManager = true;
@@ -135,7 +139,7 @@ private boolean ping() {
         }
         try {
             PipesMessage.ping().write(tuple.output);
-            PipesMessage response = PipesMessage.read(tuple.input);
+            PipesMessage response = PipesMessage.read(tuple.input, maxIpcPayloadBytes);
             if (response.type() == PipesMessageType.PING) {
                 return true;
             }
@@ -369,7 +373,7 @@ private PipesResult waitForServer(FetchEmitTuple t, IntermediateResult intermedi
                         intermediateResult.get());
             }
             try {
-                PipesMessage msg = PipesMessage.read(tuple.input);
+                PipesMessage msg = PipesMessage.read(tuple.input, maxIpcPayloadBytes);
                 LOG.trace("clientId={}: received message type={} id={}", pipesClientId, msg.type(), t.getId());
 
                 // Send ACK only for messages that require it
@@ -420,6 +424,13 @@ private PipesResult waitForServer(FetchEmitTuple t, IntermediateResult intermedi
                 closeConnection();
                 return buildFatalResult(t.getId(), t.getEmitKey(), TIMEOUT, intermediateResult.get(),
                         ExceptionUtils.getStackTrace(e));
+            } catch (PayloadLimitExceededException e) {
+                // Stream is desynchronized (payload bytes were not consumed); close the connection.
+                LOG.warn("clientId={}: payload too large for id={}: {}", pipesClientId, t.getId(), e.getMessage());
+                closeConnection();
+                return buildFatalResult(t.getId(), t.getEmitKey(),
+                        PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
+                        intermediateResult.get(), e.getMessage());
             } catch (SecurityException e) {
                 throw e;
             } catch (Exception e) {
@@ -465,7 +476,7 @@ private void waitForStartup() throws IOException {
         if (tuple == null) {
             throw new IOException("connection closed");
         }
-        PipesMessage msg = PipesMessage.read(tuple.input);
+        PipesMessage msg = PipesMessage.read(tuple.input, maxIpcPayloadBytes);
         if (msg.type() == PipesMessageType.READY) {
             LOG.info("clientId={}: server successfully started", pipesClientId);
         } else if (msg.type() == PipesMessageType.STARTUP_FAILED) {

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

@@ -23,10 +23,13 @@
 import org.apache.tika.exception.TikaConfigException;
 import org.apache.tika.pipes.api.FetchEmitTuple;
 import org.apache.tika.pipes.api.ParseMode;
+import org.apache.tika.pipes.core.protocol.PipesMessage;
 
 public class PipesConfig {
 
 
+    public static final int DEFAULT_MAX_IPC_PAYLOAD_BYTES = PipesMessage.MAX_PAYLOAD_BYTES;
+
     public static final long DEFAULT_SHUTDOWN_CLIENT_AFTER_MILLS = 300000;
 
     public static final int DEFAULT_NUM_CLIENTS = 4;
@@ -56,6 +59,8 @@ public class PipesConfig {
      */
     private boolean useSharedServer = DEFAULT_USE_SHARED_SERVER;
 
+    private int maxIpcPayloadBytes = DEFAULT_MAX_IPC_PAYLOAD_BYTES;
+
     private long socketTimeoutMs = DEFAULT_SOCKET_TIMEOUT_MS;
     private long startupTimeoutMs = DEFAULT_STARTUP_TIMEOUT_MS;
     private long heartbeatIntervalMs = DEFAULT_HEARTBEAT_INTERVAL_MS;
@@ -480,4 +485,31 @@ public boolean isUseSharedServer() {
     public void setUseSharedServer(boolean useSharedServer) {
         this.useSharedServer = useSharedServer;
     }
+
+    /**
+     * Returns the maximum IPC payload size in bytes.
+     * Configurable via {@code maxIpcPayloadBytes} in the {@code pipes} section of tika-config.json.
+     *
+     * @return the maximum IPC payload size in bytes (default 100 MB)
+     */
+    public int getMaxIpcPayloadBytes() {
+        return maxIpcPayloadBytes;
+    }
+
+    /**
+     * Sets the maximum IPC payload size in bytes. Must be a positive value.
+     * This bounds the size of a message the client will accept back from the
+     * forked server (chiefly the FINISHED result). Request payloads
+     * (client to server) are small and use the built-in default.
+     *
+     * @param maxIpcPayloadBytes positive payload limit in bytes
+     * @throws IllegalArgumentException if the value is not positive
+     */
+    public void setMaxIpcPayloadBytes(int maxIpcPayloadBytes) {
+        if (maxIpcPayloadBytes <= 0) {
+            throw new IllegalArgumentException(
+                    "maxIpcPayloadBytes must be positive, got: " + maxIpcPayloadBytes);
+        }
+        this.maxIpcPayloadBytes = maxIpcPayloadBytes;
+    }
 }

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PayloadLimitExceededException.java

@@ -0,0 +1,34 @@
+/*
+ * 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.protocol;
+
+import java.io.IOException;
+
+/**
+ * Thrown when an incoming IPC payload's declared length exceeds the configured limit
+ * (see {@link org.apache.tika.pipes.core.PipesConfig#getMaxIpcPayloadBytes()};
+ * default {@link PipesMessage#MAX_PAYLOAD_BYTES}). The payload bytes were not consumed,
+ * so the stream is desynchronized and the connection must be closed. With a shared server
+ * the process keeps running (only this connection ends); with the default per-client forked
+ * server the process may still exit on the failed write, and the client reconnects on the
+ * next task.
+ */
+public class PayloadLimitExceededException extends IOException {
+    public PayloadLimitExceededException(String message) {
+        super(message);
+    }
+}

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PipesMessage.java

@@ -40,19 +40,34 @@ public record PipesMessage(PipesMessageType type, byte[] payload) {
     static final byte MAGIC_0 = 0x54; // 'T'
     static final byte MAGIC_1 = 0x4B; // 'K'
 
-    /** Maximum payload size: 100 MB (same as old MAX_FETCH_EMIT_TUPLE_BYTES). */
+    /** Default maximum payload size. Override per-read via {@link #read(DataInputStream, int)}. */
     public static final int MAX_PAYLOAD_BYTES = 100 * 1024 * 1024;
 
     private static final byte[] EMPTY = new byte[0];
 
     /**
-     * Reads one framed message from the stream.
+     * Reads one framed message from the stream, enforcing {@link #MAX_PAYLOAD_BYTES}.
      *
      * @throws ProtocolDesyncException if magic bytes don't match
      * @throws EOFException if the stream ends before a complete message
-     * @throws IOException on payload size violations or I/O errors
+     * @throws PayloadLimitExceededException if the payload length exceeds {@link #MAX_PAYLOAD_BYTES}
+     * @throws IOException on other I/O errors
      */
     public static PipesMessage read(DataInputStream in) throws IOException {
+        return read(in, MAX_PAYLOAD_BYTES);
+    }
+
+    /**
+     * Reads one framed message from the stream, enforcing the given payload limit.
+     * Use this overload when the caller has a per-connection limit from config.
+     *
+     * @param maxPayloadBytes maximum allowed payload size in bytes
+     * @throws ProtocolDesyncException if magic bytes don't match
+     * @throws EOFException if the stream ends before a complete message
+     * @throws PayloadLimitExceededException if the payload length exceeds {@code maxPayloadBytes}
+     * @throws IOException on other I/O errors
+     */
+    public static PipesMessage read(DataInputStream in, int maxPayloadBytes) throws IOException {
         int m0 = in.read();
         if (m0 == -1) {
             throw new EOFException("Stream closed before magic byte");
@@ -77,9 +92,9 @@ public static PipesMessage read(DataInputStream in) throws IOException {
         if (len < 0) {
             throw new IOException("Negative payload length: " + len);
         }
-        if (len > MAX_PAYLOAD_BYTES) {
-            throw new IOException("Payload length " + len +
-                    " exceeds maximum of " + MAX_PAYLOAD_BYTES + " bytes");
+        if (len > maxPayloadBytes) {
+            throw new PayloadLimitExceededException("Payload length " + len +
+                    " exceeds maximum of " + maxPayloadBytes + " bytes");
         }
 
         byte[] payload;

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java

@@ -136,7 +136,7 @@ private void mainLoop() {
             try {
                 PipesMessage msg;
                 try {
-                    msg = PipesMessage.read(input);
+                    msg = PipesMessage.read(input, pipesConfig.getMaxIpcPayloadBytes());
                 } catch (SocketTimeoutException e) {
                     // Socket timeout while idle is the normal inactivity shutdown path.
                     LOG.info("handlerId={}: socket timeout while waiting for task, closing connection",

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java

@@ -160,6 +160,8 @@ public static PipesServer load(int port, Path tikaConfigPath) throws Exception {
                 String msg = ExceptionUtils.getStackTrace(e);
                 byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
                 PipesMessage.startupFailed(bytes).write(dos);
+                // pipesConfig may not have loaded successfully (that may be why we're
+                // here); use the built-in default rather than an unreliable reference.
                 PipesMessage ackMsg = PipesMessage.read(dis);
                 if (ackMsg.type() != PipesMessageType.ACK) {
                     LOG.warn("Expected ACK but got: {}", ackMsg.type());
@@ -336,7 +338,7 @@ public void mainLoop() {
             while (true) {
                 PipesMessage msg;
                 try {
-                    msg = PipesMessage.read(input);
+                    msg = PipesMessage.read(input, pipesConfig.getMaxIpcPayloadBytes());
                 } catch (SocketTimeoutException e) {
                     // Socket timeout while idle is the normal inactivity shutdown path.
                     // Exit cleanly — PipesClient will restart the server if needed.

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

@@ -16,9 +16,61 @@
  */
 package org.apache.tika.pipes.core;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+
+import org.junit.jupiter.api.Test;
+
 import org.apache.tika.TikaTest;
+import org.apache.tika.config.loader.TikaJsonConfig;
+import org.apache.tika.pipes.core.protocol.PipesMessage;
 
 public class TikaPipesConfigTest extends TikaTest {
+
+    @Test
+    void testMaxIpcPayloadBytesDefault() {
+        PipesConfig config = new PipesConfig();
+        assertEquals(PipesConfig.DEFAULT_MAX_IPC_PAYLOAD_BYTES, config.getMaxIpcPayloadBytes());
+        assertEquals(100 * 1024 * 1024, config.getMaxIpcPayloadBytes());
+    }
+
+    @Test
+    void testMaxIpcPayloadBytesFromJson() throws Exception {
+        String json = """
+                {
+                  "pipes": {
+                    "maxIpcPayloadBytes": 209715200
+                  }
+                }
+                """;
+        TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(
+                new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8)));
+        PipesConfig config = PipesConfig.load(tikaJsonConfig);
+        assertEquals(209715200, config.getMaxIpcPayloadBytes());
+        // The global constant is unchanged — the configured limit is passed per-read
+        assertEquals(100 * 1024 * 1024, PipesMessage.MAX_PAYLOAD_BYTES);
+    }
+
+    @Test
+    void testMaxIpcPayloadBytesRejectsNonPositive() {
+        PipesConfig config = new PipesConfig();
+        assertThrows(IllegalArgumentException.class, () -> config.setMaxIpcPayloadBytes(0));
+        assertThrows(IllegalArgumentException.class, () -> config.setMaxIpcPayloadBytes(-1));
+    }
+
+    @Test
+    void testMaxIpcPayloadBytesFromJsonRejectsZero() throws Exception {
+        String json = """
+                {"pipes": {"maxIpcPayloadBytes": 0}}
+                """;
+        TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(
+                new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8)));
+        assertThrows(Exception.class, () -> PipesConfig.load(tikaJsonConfig));
+    }
+
     //this handles tests for the newer pipes type configs.
 /*
     TODO -- reimplent these with json

tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/protocol/PipesMessageTest.java

@@ -127,10 +127,41 @@ void testOversizedPayloadRejection() throws IOException {
         dos.writeInt(PipesMessage.MAX_PAYLOAD_BYTES + 1);
         dos.flush();
 
-        assertThrows(IOException.class, () ->
+        assertThrows(PayloadLimitExceededException.class, () ->
                 PipesMessage.read(new DataInputStream(new ByteArrayInputStream(baos.toByteArray()))));
     }
 
+    /**
+     * A caller-supplied limit well under {@link PipesMessage#MAX_PAYLOAD_BYTES} must be
+     * enforced on its own, not just the built-in default — this is what makes the limit
+     * actually configurable rather than a second name for the same constant.
+     */
+    @Test
+    void testCustomPayloadLimitRejectsAboveBound() throws IOException {
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(baos);
+        dos.write(PipesMessage.MAGIC_0);
+        dos.write(PipesMessage.MAGIC_1);
+        dos.write(PipesMessageType.FINISHED.getByte());
+        dos.writeInt(1000); // one byte over the 999-byte custom limit below
+        dos.flush();
+
+        assertThrows(PayloadLimitExceededException.class, () ->
+                PipesMessage.read(new DataInputStream(new ByteArrayInputStream(baos.toByteArray())), 999));
+    }
+
+    @Test
+    void testCustomPayloadLimitAcceptsAtBound() throws IOException {
+        byte[] payload = new byte[999];
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        PipesMessage.finished(payload).write(new DataOutputStream(baos));
+
+        PipesMessage roundTripped = PipesMessage.read(
+                new DataInputStream(new ByteArrayInputStream(baos.toByteArray())), 999);
+        assertEquals(PipesMessageType.FINISHED, roundTripped.type());
+        assertEquals(999, roundTripped.payload().length);
+    }
+
     @Test
     void testRequiresAckAssertions() {
         assertFalse(PipesMessageType.PING.requiresAck());

tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java

@@ -282,6 +282,7 @@ public static Response.Status mapStatusToHttpResponse(PipesResult.RESULT_STATUS
                     Response.Status.SERVICE_UNAVAILABLE;
             case FETCH_EXCEPTION, EMIT_EXCEPTION,
                  FETCHER_NOT_FOUND, EMITTER_NOT_FOUND,
+                 PAYLOAD_LIMIT_EXCEEDED,
                  FETCHER_INITIALIZATION_EXCEPTION, EMITTER_INITIALIZATION_EXCEPTION,
                  FAILED_TO_INITIALIZE ->
                     Response.Status.INTERNAL_SERVER_ERROR;