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 cbb8df22b8 TIKA-4793: make Pipes IPC payload limit configurable (#2962)
cbb8df22b8 is described below
commit cbb8df22b80406fad369359238f7f62761d52640
Author: srujana-kuntumalla
<[email protected]>
AuthorDate: Wed Jul 29 10:09:21 2026 -0400
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 <[email protected]>
* 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 <[email protected]>
* 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 <[email protected]>
* 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 <[email protected]>
* 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 <[email protected]>
* 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 <[email protected]>
---------
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
---
.../org/apache/tika/pipes/api/PipesResult.java | 1 +
.../org/apache/tika/pipes/core/PipesClient.java | 17 +++++--
.../org/apache/tika/pipes/core/PipesConfig.java | 32 +++++++++++++
.../protocol/PayloadLimitExceededException.java | 34 ++++++++++++++
.../tika/pipes/core/protocol/PipesMessage.java | 27 ++++++++---
.../tika/pipes/core/server/ConnectionHandler.java | 2 +-
.../apache/tika/pipes/core/server/PipesServer.java | 4 +-
.../tika/pipes/core/TikaPipesConfigTest.java | 52 ++++++++++++++++++++++
.../tika/pipes/core/protocol/PipesMessageTest.java | 33 +++++++++++++-
.../server/core/resource/PipesParsingHelper.java | 1 +
10 files changed, 191 insertions(+), 12 deletions(-)
diff --git
a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java
b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java
index 482e13e7f4..894125fe37 100644
---
a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java
+++
b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java
@@ -66,6 +66,7 @@ public record PipesResult(RESULT_STATUS status, EmitData
emitData, String messag
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),
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 1df7caec4b..e5702962e8 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
@@ -47,6 +47,7 @@ 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.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 class PipesClient implements Closeable {
*/
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 @@ public class PipesClient implements Closeable {
}
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 @@ public class PipesClient implements Closeable {
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 @@ public class PipesClient implements Closeable {
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 @@ public class PipesClient implements Closeable {
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) {
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java
index c0dd1247ea..b3a42a368b 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java
@@ -23,10 +23,13 @@ import org.apache.tika.config.loader.TikaJsonConfig;
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 class PipesConfig {
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;
+ }
}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PayloadLimitExceededException.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PayloadLimitExceededException.java
new file mode 100644
index 0000000000..739cf51921
--- /dev/null
+++
b/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);
+ }
+}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PipesMessage.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PipesMessage.java
index 5aecc5c90a..a418990592 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PipesMessage.java
+++
b/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 record PipesMessage(PipesMessageType type, byte[]
payload) {
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;
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java
index 5628cbde4d..ce91bbc923 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java
@@ -136,7 +136,7 @@ public class ConnectionHandler implements Runnable,
Closeable {
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",
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
index 29334b2fd2..5bd291cbba 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
@@ -160,6 +160,8 @@ public class PipesServer implements AutoCloseable {
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 class PipesServer implements AutoCloseable {
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.
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java
index 308207dc25..530ea1339e 100644
---
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java
+++
b/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
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/protocol/PipesMessageTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/protocol/PipesMessageTest.java
index 396ae61de7..34f8e44cb3 100644
---
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/protocol/PipesMessageTest.java
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/protocol/PipesMessageTest.java
@@ -127,10 +127,41 @@ class PipesMessageTest {
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());
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java
index 1fc81a557b..fa29824de4 100644
---
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java
@@ -282,6 +282,7 @@ public class PipesParsingHelper {
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;
