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 7a79fd2549 TIKA-4793: make the Pipes IPC payload limit configurable
(#3009)
7a79fd2549 is described below
commit 7a79fd2549c69b6588c2277816e62869fa7ab694
Author: Srinivasarao Daruna <[email protected]>
AuthorDate: Fri Aug 14 06:42:07 2026 -0400
TIKA-4793: make the Pipes IPC payload limit configurable (#3009)
Adds maxIpcPayloadBytes (default 100 MiB) and a server-side
BoundedOutputStream guard so an oversized result fails as
PAYLOAD_LIMIT_EXCEEDED instead of OOMing the worker. Preserves
already-emitted statuses on overflow and fixes archive sizing.
Closes #3009
Co-authored-by: Tim Allison <[email protected]>
---
CHANGES.txt | 5 +
docs/modules/ROOT/pages/pipes/configuration.adoc | 11 +
.../org/apache/tika/pipes/core/PipesClient.java | 10 +
.../org/apache/tika/pipes/core/PipesConfig.java | 25 +-
.../tika/pipes/core/emitter/EmitDataImpl.java | 12 +-
.../pipes/core/serialization/JsonPipesIpc.java | 11 +
.../tika/pipes/core/server/ConnectionHandler.java | 16 +-
.../apache/tika/pipes/core/server/PipesServer.java | 8 +-
.../tika/pipes/core/server/ServerProtocolIO.java | 180 +++++++++++-
.../tika/pipes/core/TikaPipesConfigTest.java | 13 +-
.../pipes/core/server/ServerProtocolIOTest.java | 303 ++++++++++++++++++++-
11 files changed, 563 insertions(+), 31 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index d2f1ae5bcf..848567bb09 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -111,6 +111,11 @@ Release 4.0.0 - ???
OTHER CHANGES
+ * PipesClient/PipesServer IPC now enforces a configurable payload limit
+ (pipes.maxIpcPayloadBytes, default 100 MB) in both directions. Results
+ that exceed the limit return PAYLOAD_LIMIT_EXCEEDED instead of causing
+ heap exhaustion; crash messages are also size-capped (TIKA-4793).
+
* MagicDetector now compiles its regular expression once, in the
constructor, instead of recompiling it on every match (TIKA-4796).
diff --git a/docs/modules/ROOT/pages/pipes/configuration.adoc
b/docs/modules/ROOT/pages/pipes/configuration.adoc
index 29528b011d..9a75633aa7 100644
--- a/docs/modules/ROOT/pages/pipes/configuration.adoc
+++ b/docs/modules/ROOT/pages/pipes/configuration.adoc
@@ -138,6 +138,17 @@ These settings control how parsed results are batched
before sending to emitters
|When `false`, only successfully-parsed tuples reach the emitter — files that
crash, time out, or otherwise fail are dropped from the output. When `true`,
every tuple is emitted, including failures (the metadata carries the
exception). Turn this on if you need a complete record of what was attempted
(audit, retry logic, chaos-monkey tests).
|===
+== IPC Payload Limit
+
+[cols="1,1,3"]
+|===
+|Field |Default |Description
+
+|`maxIpcPayloadBytes`
+|`104857600` (100 MB)
+|Maximum size in bytes of a single IPC message between the client and the
forked server. This limit is *bidirectional*: it applies both to parse results
returned from the server (FINISHED) and to requests sent from the client
(NEW_REQUEST). Raising it lets very large documents pass over IPC; set the
forked JVM `-Xmx` to at least approximately 3× this value to keep heap usage
under control. Setting it too small (below the size of a typical
`FetchEmitTuple`) will cause requests to be rejec [...]
+|===
+
== Emit Strategy
`emitStrategy` controls whether parsed extracts are emitted directly from the
forked PipesServer or passed back to the parent process first. The default is
balanced for typical workloads — tune only if you have a memory or throughput
problem.
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 1ded2996d9..7fb00cb9f3 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
@@ -421,6 +421,16 @@ public class PipesClient implements Closeable {
if (result.emitData() instanceof EmitDataImpl
emitDataImpl) {
emitDataImpl.setParseContext(t.getParseContext());
}
+ // The server's static PAYLOAD_LIMIT_EXCEEDED fallback
frame carries null
+ // emitData/emitKey. AsyncEmitter silently skips
null-emitData results, so
+ // the document would disappear from the audit trail.
Rebuild with the
+ // original emit key using what partial metadata we
have.
+ if (result.emitData() == null
+ && result.status() ==
PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED) {
+ return buildFatalResult(t.getId(), t.getEmitKey(),
+
PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
+ intermediateResult.get());
+ }
return result;
default:
throw new IOException("Unexpected message type from
server: " + msg.type());
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 ac266b7253..8f2907fc0a 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
@@ -30,6 +30,7 @@ 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;
+import org.apache.tika.pipes.core.server.ServerProtocolIO;
public class PipesConfig {
@@ -546,18 +547,26 @@ public class PipesConfig {
}
/**
- * 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.
+ * Sets the maximum IPC payload size in bytes. This limit is
<em>bidirectional</em>:
+ * it controls both the largest result the client will accept back from
the forked server
+ * (the FINISHED payload) and the largest request the server will accept
from the client
+ * (the NEW_REQUEST payload). Lowering this value below the size of a
typical
+ * {@link org.apache.tika.pipes.api.FetchEmitTuple} will cause requests to
be rejected
+ * on the server side and reported as undiagnosable {@code
UNSPECIFIED_CRASH} errors.
+ * <p>
+ * The value must be at least {@link
org.apache.tika.pipes.core.server.ServerProtocolIO#MIN_FALLBACK_PAYLOAD_BYTES}
+ * so that the server can always write a {@code PAYLOAD_LIMIT_EXCEEDED}
response
+ * that the client will accept.
*
- * @param maxIpcPayloadBytes positive payload limit in bytes
- * @throws IllegalArgumentException if the value is not positive
+ * @param maxIpcPayloadBytes payload limit in bytes (must be ≥ {@code
ServerProtocolIO.MIN_FALLBACK_PAYLOAD_BYTES})
+ * @throws IllegalArgumentException if the value is below the minimum
*/
public void setMaxIpcPayloadBytes(int maxIpcPayloadBytes) {
- if (maxIpcPayloadBytes <= 0) {
+ if (maxIpcPayloadBytes < ServerProtocolIO.MIN_FALLBACK_PAYLOAD_BYTES) {
throw new IllegalArgumentException(
- "maxIpcPayloadBytes must be positive, got: " +
maxIpcPayloadBytes);
+ "maxIpcPayloadBytes must be at least " +
+ ServerProtocolIO.MIN_FALLBACK_PAYLOAD_BYTES +
+ " (minimum to carry a PAYLOAD_LIMIT_EXCEEDED response),
got: " + maxIpcPayloadBytes);
}
this.maxIpcPayloadBytes = maxIpcPayloadBytes;
}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitDataImpl.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitDataImpl.java
index 930d594918..5fec421209 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitDataImpl.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitDataImpl.java
@@ -77,13 +77,17 @@ public class EmitDataImpl implements EmitData {
private static long estimateSizeInBytes(String id, List<Metadata>
metadataList,
String containerStackTrace) {
- long sz = 36 + id.length() * 2;
- sz += 36 + containerStackTrace.length() * 2;
+ // Estimates Java heap cost (UTF-16: 2 bytes/char + object overhead).
+ // Used by the DYNAMIC emit strategy to decide passback vs.
direct-emit; it is not
+ // used to enforce the IPC payload limit (that is handled by
BoundedOutputStream in
+ // ServerProtocolIO, which measures actual wire bytes during
serialization).
+ long sz = 36 + id.length() * 2L;
+ sz += 36 + containerStackTrace.length() * 2L;
for (Metadata m : metadataList) {
for (String n : m.names()) {
- sz += 36 + n.length() * 2;
+ sz += 36 + n.length() * 2L;
for (String v : m.getValues(n)) {
- sz += 36 + v.length() * 2;
+ sz += 36 + v.length() * 2L;
}
}
}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/JsonPipesIpc.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/JsonPipesIpc.java
index 30cb4c6b6a..1de3451cc2 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/JsonPipesIpc.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/JsonPipesIpc.java
@@ -17,6 +17,7 @@
package org.apache.tika.pipes.core.serialization;
import java.io.IOException;
+import java.io.OutputStream;
import com.fasterxml.jackson.core.StreamReadConstraints;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -72,6 +73,16 @@ public class JsonPipesIpc {
return OBJECT_MAPPER.writeValueAsBytes(obj);
}
+ /**
+ * Serialize an object to Smile binary format, writing directly into
{@code out}.
+ * Any {@link IOException} thrown by {@code out} (e.g. from a size-capped
stream)
+ * propagates unchanged, letting callers distinguish payload-limit aborts
from
+ * genuine I/O errors.
+ */
+ public static void toStream(Object obj, OutputStream out) throws
IOException {
+ OBJECT_MAPPER.writeValue(out, obj);
+ }
+
/**
* Deserialize Smile binary format bytes to an object.
*/
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 f00fc7012a..19808fce9f 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
@@ -104,7 +104,7 @@ public class ConnectionHandler implements Runnable,
Closeable {
this.resources = resources;
this.pipesConfig = pipesConfig;
this.heartbeatIntervalMillis =
pipesConfig.getHeartbeatIntervalMillis();
- this.protocolIO = new ServerProtocolIO(input, output);
+ this.protocolIO = new ServerProtocolIO(input, output,
pipesConfig.getMaxIpcPayloadBytes());
}
@Override
@@ -181,6 +181,20 @@ public class ConnectionHandler implements Runnable,
Closeable {
LOG.error("handlerId={}: config error processing
request", handlerId, e);
handleCrash(PipesMessageType.UNSPECIFIED_CRASH,
fetchEmitTuple.getId(), e);
} catch (Throwable t) {
+ if (t instanceof Error) {
+ // OOM or other JVM-level error: don't trust
the heap; exit
+ // immediately. Everything before the exit is
best-effort and
+ // inside the try -- a secondary OOM in
logging or writeCrash
+ // must not escape and leave this shared JVM
alive post-Error.
+ try {
+ LOG.error("handlerId={}: fatal JVM error;
exiting", handlerId, t);
+
protocolIO.writeCrash(PipesMessageType.OOM, t);
+ } catch (Throwable ignored) {
+ //swallow
+ } finally {
+
System.exit(PipesMessageType.OOM.getExitCode().orElse(18));
+ }
+ }
// respond, or the client blocks until socket
timeout and
// restarts a healthy server
LOG.error("handlerId={}: error processing
request", handlerId, t);
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 1a2e87aba2..e0602b1123 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
@@ -189,7 +189,7 @@ public class PipesServer implements AutoCloseable {
validateHeartbeatInterval(pipesConfig);
emitStrategy = pipesConfig.getEmitStrategy().getType();
- this.protocolIO = new ServerProtocolIO(input, output);
+ this.protocolIO = new ServerProtocolIO(input, output,
pipesConfig.getMaxIpcPayloadBytes());
}
@@ -402,6 +402,12 @@ public class PipesServer implements AutoCloseable {
try {
loopUntilDone(fetchEmitTuple, mergedContext,
executorCompletionService, intermediateResult, countDownLatch, parseTimeout);
} catch (Throwable t) {
+ if (t instanceof Error) {
+ // OOM or other JVM-level error: exit rather
than continue in a
+ // possibly corrupt heap state.
+ handleCrash(PipesMessageType.OOM,
fetchEmitTuple.getId(), t);
+ return; // handleCrash calls exit();
unreachable
+ }
LOG.error("Serious problem processing request", t);
}
break;
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ServerProtocolIO.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ServerProtocolIO.java
index 54ee7de184..e94407d1c5 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ServerProtocolIO.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ServerProtocolIO.java
@@ -16,9 +16,12 @@
*/
package org.apache.tika.pipes.core.server;
+import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Locale;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,48 +53,163 @@ public class ServerProtocolIO {
private static final Logger LOG =
LoggerFactory.getLogger(ServerProtocolIO.class);
+ /**
+ * Pre-serialized fallback payload (Smile-encoded {@code
PAYLOAD_LIMIT_EXCEEDED} result).
+ * Private to prevent external mutation of the array contents — {@code
static final}
+ * prevents reference reassignment but not element writes.
+ */
+ private static final byte[] FALLBACK_PAYLOAD_BYTES;
+
+ /**
+ * The minimum value accepted for {@code maxPayloadBytes} in the
constructor and in
+ * {@link
org.apache.tika.pipes.core.PipesConfig#setMaxIpcPayloadBytes(int)}: the
+ * serialized byte length of {@link #FALLBACK_PAYLOAD_BYTES}.
+ * Any configured limit smaller than this cannot carry even the fallback
frame.
+ */
+ public static final int MIN_FALLBACK_PAYLOAD_BYTES;
+
+ static {
+ try {
+ FALLBACK_PAYLOAD_BYTES = JsonPipesIpc.toBytes(
+ new
PipesResult(PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
+ "payload_limit_exceeded"));
+ MIN_FALLBACK_PAYLOAD_BYTES = FALLBACK_PAYLOAD_BYTES.length;
+ } catch (IOException e) {
+ throw new ExceptionInInitializerError(e);
+ }
+ }
+
private final DataInputStream input;
private final DataOutputStream output;
+ private final int maxIpcPayloadBytes;
- public ServerProtocolIO(DataInputStream input, DataOutputStream output) {
+ public ServerProtocolIO(DataInputStream input, DataOutputStream output,
int maxIpcPayloadBytes) {
+ if (maxIpcPayloadBytes < MIN_FALLBACK_PAYLOAD_BYTES) {
+ throw new IllegalArgumentException(String.format(Locale.ROOT,
+ "maxIpcPayloadBytes %d is below the minimum %d required to
carry a PAYLOAD_LIMIT_EXCEEDED response",
+ maxIpcPayloadBytes, MIN_FALLBACK_PAYLOAD_BYTES));
+ }
this.input = input;
this.output = output;
+ this.maxIpcPayloadBytes = maxIpcPayloadBytes;
}
/**
* Writes a FINISHED message with the serialized result and waits for ACK.
+ * <p>
+ * Serialization is streamed into a {@link BoundedOutputStream} capped at
+ * {@code maxIpcPayloadBytes}. If the payload overflows the cap, the
stream aborts
+ * before any bytes are sent to the client and a pre-computed
+ * {@code PAYLOAD_LIMIT_EXCEEDED} frame is sent instead. This keeps the
original
+ * result status intact when the payload fits, avoids unbounded heap
allocation,
+ * and prevents wire desynchronization on the client side.
*
* @throws ShutDownReceivedException if SHUT_DOWN is received instead of
ACK
* @throws IOException on serialization or I/O errors
*/
public void writeFinished(PipesResult pipesResult) throws IOException {
- byte[] bytes = JsonPipesIpc.toBytes(pipesResult);
- PipesMessage.finished(bytes).write(output);
+ BoundedOutputStream bos = new BoundedOutputStream(maxIpcPayloadBytes);
+ try {
+ JsonPipesIpc.toStream(pipesResult, bos);
+ } catch (IOException e) {
+ if (!bos.overflowed()) {
+ throw e;
+ }
+ LOG.warn("Payload exceeded maxIpcPayloadBytes {}; returning
PAYLOAD_LIMIT_EXCEEDED",
+ maxIpcPayloadBytes);
+ // If content was already emitted server-side, preserve that
status so the
+ // client does not duplicate the emission on the passback path.
The fixed
+ // message replaces the original, which may itself be the overflow
source
+ // (an accumulated parse-exception stack).
+ if (alreadyEmitted(pipesResult.status())) {
+ BoundedOutputStream fallbackBos = new
BoundedOutputStream(maxIpcPayloadBytes);
+ try {
+ JsonPipesIpc.toStream(
+ new PipesResult(pipesResult.status(),
"payload_limit_exceeded"),
+ fallbackBos);
+
PipesMessage.finished(fallbackBos.toByteArray()).write(output);
+ awaitAck();
+ return;
+ } catch (IOException fallbackE) {
+ if (!fallbackBos.overflowed()) {
+ throw fallbackE;
+ }
+ // Even the status-only result overflows — fall through to
the
+ // guaranteed-fit static fallback.
+ }
+ }
+ doWritePayloadLimitExceeded();
+ return;
+ }
+ PipesMessage.finished(bos.toByteArray()).write(output);
+ awaitAck();
+ }
+
+ /**
+ * True for statuses whose content the server already emitted. Replacing
one of these
+ * with a failure status makes the client treat an emitted document as
failed, so a
+ * retry emits it a second time.
+ */
+ private static boolean alreadyEmitted(PipesResult.RESULT_STATUS status) {
+ return status == PipesResult.RESULT_STATUS.EMIT_SUCCESS ||
+ status == PipesResult.RESULT_STATUS.EMIT_SUCCESS_PASSBACK ||
+ status ==
PipesResult.RESULT_STATUS.EMIT_SUCCESS_PARSE_EXCEPTION;
+ }
+
+ private void doWritePayloadLimitExceeded() throws IOException {
+ // FALLBACK_PAYLOAD_BYTES is pre-computed at class load and guaranteed
to be smaller
+ // than maxIpcPayloadBytes (enforced by the constructor), so the
client always accepts it.
+ PipesMessage.finished(FALLBACK_PAYLOAD_BYTES).write(output);
awaitAck();
}
/**
* Writes an INTERMEDIATE_RESULT message with the serialized metadata and
waits for ACK.
+ * If the metadata exceeds {@code maxIpcPayloadBytes}, the intermediate is
silently skipped
+ * (the FINISHED message will still follow).
*
* @throws ShutDownReceivedException if SHUT_DOWN is received instead of
ACK
* @throws IOException on serialization or I/O errors
*/
public void writeIntermediate(Metadata metadata) throws IOException {
- byte[] bytes = JsonPipesIpc.toBytes(metadata);
- PipesMessage.intermediateResult(bytes).write(output);
+ BoundedOutputStream bos = new BoundedOutputStream(maxIpcPayloadBytes);
+ try {
+ JsonPipesIpc.toStream(metadata, bos);
+ } catch (IOException e) {
+ if (bos.overflowed()) {
+ LOG.warn("Intermediate result payload exceeded
maxIpcPayloadBytes {}; skipping intermediate",
+ maxIpcPayloadBytes);
+ return;
+ }
+ throw e;
+ }
+ PipesMessage.intermediateResult(bos.toByteArray()).write(output);
awaitAck();
}
/**
* Writes a crash message (OOM, TIMEOUT, or UNSPECIFIED_CRASH) with the
- * serialized stack trace and waits for ACK.
+ * serialized stack trace and waits for ACK. Serialization is streamed into
+ * a {@link BoundedOutputStream} capped at {@code maxIpcPayloadBytes}. If
+ * the stack trace overflows the cap, an empty payload is sent instead.
*
* @throws IOException on serialization, I/O, or unexpected ACK response
*/
public void writeCrash(PipesMessageType crashType, Throwable t) throws
IOException {
String msg = (t != null) ? ExceptionUtils.getStackTrace(t) : "";
- byte[] bytes = JsonPipesIpc.toBytes(msg);
- PipesMessage.crash(crashType, bytes).write(output);
+ BoundedOutputStream bos = new BoundedOutputStream(maxIpcPayloadBytes);
+ try {
+ JsonPipesIpc.toStream(msg, bos);
+ } catch (IOException e) {
+ if (!bos.overflowed()) {
+ throw e;
+ }
+ // Stack trace overflows limit (e.g., CJK chars encode at 3
bytes/char in Smile).
+ // Fall back to an empty payload, guaranteed to fit within any
valid limit.
+ bos = new BoundedOutputStream(maxIpcPayloadBytes);
+ JsonPipesIpc.toStream("", bos);
+ }
+ PipesMessage.crash(crashType, bos.toByteArray()).write(output);
awaitAck();
}
@@ -102,7 +220,7 @@ public class ServerProtocolIO {
* @throws IOException if the message is any other non-ACK type, or on I/O
error
*/
public void awaitAck() throws IOException {
- PipesMessage msg = PipesMessage.read(input);
+ PipesMessage msg = PipesMessage.read(input, maxIpcPayloadBytes);
if (msg.type() == PipesMessageType.ACK) {
return;
}
@@ -165,4 +283,48 @@ public class ServerProtocolIO {
mergedContext.set(TimeoutLimits.class, clamped);
}
}
+
+ /**
+ * An {@link OutputStream} backed by a {@link ByteArrayOutputStream} that
aborts
+ * with an {@link IOException} the moment accumulated bytes would exceed
{@code limit}.
+ * The caller distinguishes an overflow abort from genuine I/O errors via
+ * {@link #overflowed()}.
+ */
+ private static final class BoundedOutputStream extends OutputStream {
+
+ private final int limit;
+ private final ByteArrayOutputStream buf;
+ private boolean overflowed = false;
+
+ BoundedOutputStream(int limit) {
+ this.limit = limit;
+ this.buf = new ByteArrayOutputStream(Math.min(limit, 8192));
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ if (buf.size() >= limit) {
+ overflowed = true;
+ throw new IOException("payload_overflow");
+ }
+ buf.write(b);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ if ((long) buf.size() + len > limit) {
+ overflowed = true;
+ throw new IOException("payload_overflow");
+ }
+ buf.write(b, off, len);
+ }
+
+ boolean overflowed() {
+ return overflowed;
+ }
+
+ byte[] toByteArray() {
+ return buf.toByteArray();
+ }
+ }
}
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 4e10fe1da6..ffa24c5879 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
@@ -27,6 +27,7 @@ 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;
+import org.apache.tika.pipes.core.server.ServerProtocolIO;
public class TikaPipesConfigTest extends TikaTest {
@@ -63,10 +64,20 @@ public class TikaPipesConfigTest extends TikaTest {
}
@Test
- void testMaxIpcPayloadBytesRejectsNonPositive() {
+ void testMaxIpcPayloadBytesRejectsTooSmall() {
PipesConfig config = new PipesConfig();
+ // 0 and -1 are rejected (below MIN_FALLBACK_PAYLOAD_BYTES)
assertThrows(IllegalArgumentException.class, () ->
config.setMaxIpcPayloadBytes(0));
assertThrows(IllegalArgumentException.class, () ->
config.setMaxIpcPayloadBytes(-1));
+ // A small-but-positive value below the minimum is also rejected
+ int belowMin = ServerProtocolIO.MIN_FALLBACK_PAYLOAD_BYTES - 1;
+ if (belowMin > 0) {
+ assertThrows(IllegalArgumentException.class, () ->
config.setMaxIpcPayloadBytes(belowMin));
+ }
+ // A value at or above the minimum is accepted
+ int atMin = ServerProtocolIO.MIN_FALLBACK_PAYLOAD_BYTES;
+ config.setMaxIpcPayloadBytes(atMin);
+ assertEquals(atMin, config.getMaxIpcPayloadBytes());
}
@Test
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/ServerProtocolIOTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/ServerProtocolIOTest.java
index cc90f7765f..016f630ee4 100644
---
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/ServerProtocolIOTest.java
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/ServerProtocolIOTest.java
@@ -17,16 +17,37 @@
package org.apache.tika.pipes.core.server;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.Test;
import org.apache.tika.config.TimeoutLimits;
+import org.apache.tika.metadata.Metadata;
import org.apache.tika.parser.ParseContext;
+import org.apache.tika.pipes.api.PipesResult;
+import org.apache.tika.pipes.core.emitter.EmitDataImpl;
+import org.apache.tika.pipes.core.protocol.PipesMessage;
+import org.apache.tika.pipes.core.protocol.PipesMessageType;
+import org.apache.tika.pipes.core.serialization.JsonPipesIpc;
-public class ServerProtocolIOTest {
+/**
+ * Unit tests for payload-size protection in {@link
ServerProtocolIO#writeFinished}
+ * and timeout-limit clamping in {@link
ServerProtocolIO#clampRequestTimeoutLimits}.
+ */
+class ServerProtocolIOTest {
private static final long MAX = 3_600_000L;
+ // ---- timeout clamping tests ----
+
@Test
public void testRequestLimitsOverCapAreClamped() {
ParseContext request = new ParseContext();
@@ -43,13 +64,9 @@ public class ServerProtocolIOTest {
@Test
public void testUnresolvedJsonRequestLimitsAlsoTriggerClamp() {
- // A request can carry timeout-limits as an unresolved JSON config;
after
- // resolveAll the value lives in the merged context's typed slot --
the clamp
- // must key off the request's json entry, not only its typed one.
ParseContext request = new ParseContext();
request.setJsonConfig("timeout-limits", "{\"totalTaskTimeoutMillis\":
9999999999}");
ParseContext merged = new ParseContext();
- // simulate post-resolveAll state
merged.set(TimeoutLimits.class, new TimeoutLimits(9_999_999_999L,
120_000));
ServerProtocolIO.clampRequestTimeoutLimits(request, merged, MAX);
@@ -59,8 +76,6 @@ public class ServerProtocolIOTest {
@Test
public void testServerConfigLimitsAreTrustedAndNeverClamped() {
- // Operator raised the total in the server's own tika-config; the
request carries
- // no limits, so the cap must not apply.
ParseContext request = new ParseContext();
ParseContext merged = new ParseContext();
merged.set(TimeoutLimits.class, new TimeoutLimits(7_200_000L,
120_000));
@@ -69,4 +84,278 @@ public class ServerProtocolIOTest {
assertEquals(7_200_000L,
merged.get(TimeoutLimits.class).getTotalTaskTimeoutMillis());
}
+
+ // ---- payload guard tests ----
+
+ /**
+ * Runs a writeFinished() call through a pair of piped streams, acting as
the
+ * "client" in a background thread: reads the FINISHED message using the
same
+ * {@code maxPayloadBytes} limit the server uses, sends ACK, and returns
the
+ * deserialized PipesResult.
+ */
+ private PipesResult exchange(PipesResult toWrite, int maxPayloadBytes)
throws Exception {
+ PipedOutputStream serverOutPipe = new PipedOutputStream();
+ PipedInputStream clientInPipe = new PipedInputStream(serverOutPipe,
1024 * 1024);
+ PipedOutputStream clientOutPipe = new PipedOutputStream();
+ PipedInputStream serverInPipe = new PipedInputStream(clientOutPipe,
1024);
+
+ AtomicReference<PipesResult> received = new AtomicReference<>();
+ AtomicReference<Exception> clientError = new AtomicReference<>();
+
+ Thread clientThread = new Thread(() -> {
+ try {
+ DataInputStream clientDis = new DataInputStream(clientInPipe);
+ DataOutputStream clientDos = new
DataOutputStream(clientOutPipe);
+
+ // Read with the same limit the server uses — mirrors
production PipesClient behaviour.
+ PipesMessage msg = PipesMessage.read(clientDis,
maxPayloadBytes);
+ assertEquals(PipesMessageType.FINISHED, msg.type());
+ received.set(JsonPipesIpc.fromBytes(msg.payload(),
PipesResult.class));
+ PipesMessage.ack().write(clientDos);
+ } catch (Exception e) {
+ clientError.set(e);
+ }
+ });
+ clientThread.setDaemon(true);
+ clientThread.start();
+
+ ServerProtocolIO io = new ServerProtocolIO(
+ new DataInputStream(serverInPipe),
+ new DataOutputStream(serverOutPipe),
+ maxPayloadBytes);
+ io.writeFinished(toWrite);
+
+ clientThread.join(5000);
+
+ if (clientError.get() != null) {
+ throw clientError.get();
+ }
+ assertNotNull(received.get(), "client never received a FINISHED
message");
+ return received.get();
+ }
+
+ /**
+ * A result whose serialized size is under the configured limit passes
through
+ * unchanged — original status is preserved.
+ */
+ @Test
+ void testSmallResultPassesThrough() throws Exception {
+ PipesResult original = new
PipesResult(PipesResult.RESULT_STATUS.PARSE_SUCCESS,
+ new EmitDataImpl("key", List.of(new Metadata())));
+
+ PipesResult returned = exchange(original,
PipesMessage.MAX_PAYLOAD_BYTES);
+
+ assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS,
returned.status());
+ }
+
+ /**
+ * When the serialized payload is one byte over the configured limit the
+ * BoundedOutputStream aborts serialization mid-stream and the server
returns
+ * PAYLOAD_LIMIT_EXCEEDED instead of writing the oversized frame.
+ */
+ @Test
+ void testPayloadOneByteTooLargeReturnPayloadLimitExceeded() throws
Exception {
+ Metadata m = new Metadata();
+ m.add("content", "a".repeat(500));
+ PipesResult big = new
PipesResult(PipesResult.RESULT_STATUS.PARSE_SUCCESS,
+ new EmitDataImpl("mykey", List.of(m)));
+
+ byte[] serialized = JsonPipesIpc.toBytes(big);
+ int tinyLimit = serialized.length - 1;
+
+ PipesResult returned = exchange(big, tinyLimit);
+
+ assertEquals(PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
returned.status());
+ }
+
+ /**
+ * A result whose serialized size clearly exceeds the configured limit
triggers
+ * PAYLOAD_LIMIT_EXCEEDED. Uses a fixed small limit so the test is
independent of
+ * the estimate formula.
+ */
+ @Test
+ void testLargePayloadReturnPayloadLimitExceeded() throws Exception {
+ Metadata m = new Metadata();
+ m.add("content", "x".repeat(10_000));
+ PipesResult result = new
PipesResult(PipesResult.RESULT_STATUS.PARSE_SUCCESS,
+ new EmitDataImpl("key", List.of(m)));
+
+ // The 10,000-char content serializes to ~10 KB; pick a limit well
below that.
+ int limit = 512;
+ assertTrue(JsonPipesIpc.toBytes(result).length > limit,
+ "test setup: serialized content must exceed limit");
+
+ PipesResult returned = exchange(result, limit);
+
+ assertEquals(PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
returned.status());
+ }
+
+ /**
+ * Status-only results (no emitData) always pass through unchanged.
+ */
+ @Test
+ void testStatusOnlyResultPassesThrough() throws Exception {
+ PipesResult statusOnly = new
PipesResult(PipesResult.RESULT_STATUS.FETCH_EXCEPTION,
+ "something went wrong");
+
+ PipesResult returned = exchange(statusOnly, 1024);
+
+ assertEquals(PipesResult.RESULT_STATUS.FETCH_EXCEPTION,
returned.status());
+ }
+
+ /**
+ * Regression for the "fallback-too-big" bug: when the configured limit
equals
+ * MIN_FALLBACK_PAYLOAD_BYTES (the tightest limit the constructor
accepts), the
+ * fallback PAYLOAD_LIMIT_EXCEEDED frame must still fit within that limit
so the
+ * client can read it with the same configured limit.
+ */
+ @Test
+ void testFallbackFitsWithinMinimumConfiguredLimit() throws Exception {
+ int limit = ServerProtocolIO.MIN_FALLBACK_PAYLOAD_BYTES;
+
+ Metadata m = new Metadata();
+ m.add("content", "x".repeat(10_000));
+ PipesResult result = new
PipesResult(PipesResult.RESULT_STATUS.PARSE_SUCCESS,
+ new EmitDataImpl("key", List.of(m)));
+
+ PipesResult returned = exchange(result, limit);
+
+ assertEquals(PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
returned.status());
+ }
+
+ /**
+ * When an EMIT_SUCCESS_PASSBACK result overflows the limit, the server
preserves
+ * the EMIT_SUCCESS_PASSBACK status rather than replacing it with
PAYLOAD_LIMIT_EXCEEDED,
+ * so the client does not re-emit content that was already emitted
server-side.
+ */
+ @Test
+ void testEmitSuccessPassbackStatusPreservedOnOverflow() throws Exception {
+ Metadata m = new Metadata();
+ m.add("content", "x".repeat(10_000));
+ PipesResult result = new
PipesResult(PipesResult.RESULT_STATUS.EMIT_SUCCESS_PASSBACK,
+ new EmitDataImpl("key", List.of(m)));
+
+ int limit = 512;
+ assertTrue(JsonPipesIpc.toBytes(result).length > limit,
+ "test setup: serialized content must exceed limit");
+
+ PipesResult returned = exchange(result, limit);
+
+ assertEquals(PipesResult.RESULT_STATUS.EMIT_SUCCESS_PASSBACK,
returned.status());
+ }
+
+ /**
+ * EMIT_SUCCESS_PARSE_EXCEPTION also means the content was already emitted
server-side,
+ * so an overflow must not downgrade it to a failure status -- that would
make the
+ * client record an emitted document as failed and a retry would emit it
twice.
+ */
+ @Test
+ void testEmitSuccessParseExceptionStatusPreservedOnOverflow() throws
Exception {
+ Metadata m = new Metadata();
+ m.add("content", "x".repeat(10_000));
+ PipesResult result =
+ new
PipesResult(PipesResult.RESULT_STATUS.EMIT_SUCCESS_PARSE_EXCEPTION,
+ new EmitDataImpl("key", List.of(m)));
+
+ int limit = 512;
+ assertTrue(JsonPipesIpc.toBytes(result).length > limit,
+ "test setup: serialized content must exceed limit");
+
+ PipesResult returned = exchange(result, limit);
+
+ assertEquals(PipesResult.RESULT_STATUS.EMIT_SUCCESS_PARSE_EXCEPTION,
returned.status());
+ }
+
+ /**
+ * The overflow can be the message itself -- EmitHandler accumulates
parse-exception
+ * stacks into it -- so the status-preserving retry must replace the
message, not
+ * carry it over.
+ */
+ @Test
+ void testOversizedMessageOnAlreadyEmittedStatusPreservesStatus() throws
Exception {
+ PipesResult result =
+ new
PipesResult(PipesResult.RESULT_STATUS.EMIT_SUCCESS_PARSE_EXCEPTION,
+ "stack".repeat(5_000));
+
+ int limit = 512;
+ assertTrue(JsonPipesIpc.toBytes(result).length > limit,
+ "test setup: serialized message must exceed limit");
+
+ PipesResult returned = exchange(result, limit);
+
+ assertEquals(PipesResult.RESULT_STATUS.EMIT_SUCCESS_PARSE_EXCEPTION,
returned.status());
+ }
+
+ // ---- writeCrash tests ----
+
+ /**
+ * Helper: calls writeCrash on the server side, reads the crash message
from the client side,
+ * sends ACK, and returns the deserialized stack trace string.
+ */
+ private String exchangeCrash(Throwable t, int maxPayloadBytes) throws
Exception {
+ PipedOutputStream serverOutPipe = new PipedOutputStream();
+ PipedInputStream clientInPipe = new PipedInputStream(serverOutPipe,
1024 * 1024);
+ PipedOutputStream clientOutPipe = new PipedOutputStream();
+ PipedInputStream serverInPipe = new PipedInputStream(clientOutPipe,
1024);
+
+ AtomicReference<String> received = new AtomicReference<>();
+ AtomicReference<Exception> clientError = new AtomicReference<>();
+
+ Thread clientThread = new Thread(() -> {
+ try {
+ DataInputStream clientDis = new DataInputStream(clientInPipe);
+ DataOutputStream clientDos = new
DataOutputStream(clientOutPipe);
+
+ PipesMessage msg = PipesMessage.read(clientDis,
maxPayloadBytes);
+ assertEquals(PipesMessageType.UNSPECIFIED_CRASH, msg.type());
+ received.set(JsonPipesIpc.fromBytes(msg.payload(),
String.class));
+ PipesMessage.ack().write(clientDos);
+ } catch (Exception e) {
+ clientError.set(e);
+ }
+ });
+ clientThread.setDaemon(true);
+ clientThread.start();
+
+ ServerProtocolIO io = new ServerProtocolIO(
+ new DataInputStream(serverInPipe),
+ new DataOutputStream(serverOutPipe),
+ maxPayloadBytes);
+ io.writeCrash(PipesMessageType.UNSPECIFIED_CRASH, t);
+
+ clientThread.join(5000);
+ if (clientError.get() != null) {
+ throw clientError.get();
+ }
+ assertNotNull(received.get(), "client never received a CRASH message");
+ return received.get();
+ }
+
+ /**
+ * A crash with a small stack trace passes through unchanged and the
client reads
+ * it within the configured limit.
+ */
+ @Test
+ void testCrashSmallPayloadPassesThrough() throws Exception {
+ RuntimeException ex = new RuntimeException("something failed");
+ String returned = exchangeCrash(ex, PipesMessage.MAX_PAYLOAD_BYTES);
+ assertTrue(returned.contains("something failed"));
+ }
+
+ /**
+ * A crash whose stack trace would overflow the limit falls back to an
empty string,
+ * not a truncated string that could still exceed the limit due to
multi-byte encoding.
+ * The client must be able to read the response with the same configured
limit.
+ */
+ @Test
+ void testCrashOversizedPayloadFallsBackToEmptyString() throws Exception {
+ // Build a large exception message that will serialize beyond a small
limit.
+ RuntimeException ex = new RuntimeException("x".repeat(10_000));
+ int limit = 512;
+
+ String returned = exchangeCrash(ex, limit);
+
+ // Empty string fallback: the trace was too large, we get an empty
payload.
+ assertEquals("", returned);
+ }
}
