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 482e13e7f40..894125fe378 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 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), 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 1df7caec4be..e5702962e80 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.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) { 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 c0dd1247eaf..b3a42a368b4 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.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; + } } 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 00000000000..739cf519217 --- /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 5aecc5c90a5..a418990592e 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 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; 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 5628cbde4d8..ce91bbc923f 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 @@ 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", 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 29334b2fd2a..5bd291cbba5 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 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. 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 308207dc257..530ea1339e8 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 396ae61de7f..34f8e44cb37 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 @@ 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()); 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 1fc81a557b0..fa29824de49 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 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;