TIKA-4793: make Pipes IPC payload limit configurable (#2962)

* TIKA-4793: make Pipes IPC payload limit configurable

The hard-coded 100 MB ceiling in PipesMessage was not operator-tunable.
Add PipesConfig.maxIpcPayloadBytes (default 100 MB) and thread it through
PipesClient, PipesServer, ConnectionHandler, and ServerProtocolIO so the
limit is applied on every read() call. The write path is unchanged.
Includes unit tests for default value, JSON loading, and validation.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* TIKA-4793: make Pipes IPC payload limit configurable

Remove final from PipesMessage.MAX_PAYLOAD_BYTES (long) so it can be set
at runtime. PipesConfig.setMaxIpcPayloadBytes() updates the static whenever
the limit is changed via JSON config or programmatically. No changes to
call sites — all existing PipesMessage.read() callers pick up the new value
automatically through the shared static.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* TIKA-4793: make Pipes IPC payload limit configurable

Remove final from PipesMessage.MAX_PAYLOAD_BYTES so it can be updated
at runtime. Add maxIpcPayloadBytes to PipesConfig (int, default 100 MB)
with a setter that updates PipesMessage.MAX_PAYLOAD_BYTES as a side
effect. Both client and server JVMs load from the same tika-config.json
so setting it once covers both ends automatically. No changes to call
sites — all existing PipesMessage.read() callers pick up the value
through the shared static.

Configurable via tika-config.json:
  {"pipes": {"maxIpcPayloadBytes": 209715200}}

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* TIKA-4793: address reviewer feedback -- threaded limit, PAYLOAD_LIMIT_EXCEEDED status

- Restore MAX_PAYLOAD_BYTES to final; add read(DataInputStream, int) overload
  so callers can pass a per-connection limit without mutating shared state
- PipesConfig setter no longer has the global side-effect; PipesClient captures
  maxIpcPayloadBytes at construction and passes it to read() in waitForServer()
- Add PAYLOAD_LIMIT_EXCEEDED(TASK_EXCEPTION) to RESULT_STATUS so oversized
  responses are treated as per-document errors, not process crashes
- Introduce PayloadLimitExceededException (IOException subtype) and catch it
  specifically in PipesClient: close the desynchronized connection but do not
  restart the healthy server
- Add JSON zero-value rejection test through the deserialization path

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* TIKA-4793: fix javadocs and add PAYLOAD_LIMIT_EXCEEDED to server HTTP mapping

- Correct PipesConfig.setMaxIpcPayloadBytes javadoc: limit applies to
  client-side reads of server responses, not both ends
- Correct PayloadLimitExceededException javadoc: limit is the configured
  per-read value; server exit behaviour differs between shared and per-client modes
- Add PAYLOAD_LIMIT_EXCEEDED to PipesParsingHelper.mapStatusToHttpResponse
  (INTERNAL_SERVER_ERROR arm) — exhaustive switch otherwise fails to compile

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* TIKA-4793: thread limit through all remaining PipesMessage.read() call sites

Apply reviewer patch: pass maxIpcPayloadBytes to read() consistently across
PipesClient (ping + waitForStartup), ConnectionHandler main loop, and PipesServer
main loop. The startup-failure error path in PipesServer intentionally keeps
the default since pipesConfig may not have loaded when that path is reached.
Add two PipesMessageTest cases proving a caller-supplied limit below
MAX_PAYLOAD_BYTES is enforced independently.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: srujana-kuntumalla

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;