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 cbfeb5141c TIKA-4829 -- handle byte passing more cleanly (#3043)
cbfeb5141c is described below
commit cbfeb5141c73c2a86341779a7c9b86e10858b8f1
Author: Tim Allison <[email protected]>
AuthorDate: Fri Aug 21 21:24:27 2026 -0400
TIKA-4829 -- handle byte passing more cleanly (#3043)
---
CHANGES.txt | 9 +-
docs/modules/ROOT/pages/pipes/fetchers.adoc | 2 +-
.../ROOT/pages/using-tika/server/index.adoc | 6 +-
.../org/apache/tika/pipes/core/PipesClient.java | 18 ++-
.../org/apache/tika/pipes/core/PipesConfig.java | 2 +-
.../tika/pipes/core/fetcher/BytesFetcher.java | 3 +-
.../tika/pipes/core/fetcher/InlineBytes.java | 11 +-
.../protocol/PayloadLimitExceededException.java | 13 +-
.../serialization/FetchEmitTupleDeserializer.java | 20 ++-
.../serialization/FetchEmitTupleSerializer.java | 15 +-
.../pipes/core/serialization/JsonPipesIpc.java | 11 +-
.../pipes/core/serialization/PipesRequest.java | 81 ++++++++++
.../serialization/PipesRequestDeserializer.java | 43 +++++
.../core/serialization/PipesRequestSerializer.java | 42 +++++
.../tika/pipes/core/server/ConnectionHandler.java | 10 +-
.../apache/tika/pipes/core/server/PipesServer.java | 10 +-
.../tika/pipes/core/PipesClientInterruptTest.java | 53 -------
.../pipes/core/PipesClientPayloadLimitTest.java | 94 +++++++++++
.../tika/pipes/core/SentinelServerManager.java | 75 +++++++++
.../core/serialization/InlineBytesWireTest.java | 173 +++++++++++++++++++++
.../serialization/SystemComponentIdWireTest.java | 9 +-
.../WireRestrictedFetchEmitTupleTest.java | 9 +-
.../pipes/core/serialization/WireTestUtil.java} | 26 ++--
.../tika/pipes/fork/PipesForkParserTest.java | 2 +
24 files changed, 627 insertions(+), 110 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 5dde33b1b4..1b59350a8c 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,10 @@
-Release 4.1.0 - ???
+Release 4.1.0 - unreleased
+
+ * Pipes IPC: carry inline document bytes as a raw binary field beside the
+ tuple in the request envelope -- never inside the tuple or its
+ ParseContext -- and disable Smile's 7-bit binary encoding. Tuple JSON
+ serialized by 4.0.0 with an "inline-bytes" parse-context entry no longer
+ loads; it is rejected with a tailored message (TIKA-4829).
* Digesting embedded documents no longer buffers each embedded object to a
temp file. Zip entries are re-read from the parent archive on rewind, and
@@ -34,7 +40,6 @@ Release 4.1.0 - ???
detection but also gains the attachments. Disable via
"raw-tiff-parser": {"extractPreviews": false} (TIKA-4824).
-
Release 4.0.0 - 8/18/2026
This section is the complete delta from 3.x. It includes everything first
diff --git a/docs/modules/ROOT/pages/pipes/fetchers.adoc
b/docs/modules/ROOT/pages/pipes/fetchers.adoc
index 64608a8a48..8ec15ae40f 100644
--- a/docs/modules/ROOT/pages/pipes/fetchers.adoc
+++ b/docs/modules/ROOT/pages/pipes/fetchers.adoc
@@ -70,7 +70,7 @@ IDs you configure yourself may contain only letters, digits,
`.`, `_` and `-`, a
A host that already holds a document -- tika-server serving `/tika`, or an
application calling `PipesForkParser` with an in-memory stream -- does not have
to write it to disk for the forked worker to read. Content at or below
xref:pipes/configuration.adoc#payload-limits[`maxInlineBytes`] travels inside
the request and is served in the worker by the built-in `\_\_bytes` fetcher;
larger content is written once to a file instead. A stream that is already
backed by a file always keeps its file.
-This is automatic and needs no configuration. `\_\_bytes` is not declarable in
a config file and cannot be named by a request.
+This is automatic and needs no configuration. `\_\_bytes` is not declarable in
a config file and cannot be named by a request. The payload itself has no
request-suppliable form either: it travels beside the tuple in the host's
parent-to-worker request envelope, never as a tuple field (`inlineBytes` in a
request tuple is rejected). The IPC wire format is internal and
same-version-only; parent and worker run from the same classpath by default,
and a version-skewed worker classpath (a `-cp` [...]
[#plugins]
== Available Fetchers
diff --git a/docs/modules/ROOT/pages/using-tika/server/index.adoc
b/docs/modules/ROOT/pages/using-tika/server/index.adoc
index f7c8f6c024..a156fa187f 100644
--- a/docs/modules/ROOT/pages/using-tika/server/index.adoc
+++ b/docs/modules/ROOT/pages/using-tika/server/index.adoc
@@ -156,7 +156,11 @@ curl -X POST http://localhost:9998/async -H "Content-Type:
application/json" \
A tuple's fields are `id`, `fetcher`, `fetchKey`, `emitter`, `emitKey`, and
optionally
`fetchRangeStart`, `fetchRangeEnd`, `metadata`, `parse-context` and
`onParseException`. Any other
-field is a `400` — there is no silent tolerance for a typo. The
`{"tuples":[...]}` envelope is
+field is a `400` — there is no silent tolerance for a typo. There is no field
for content:
+`inlineBytes` draws a tailored `400`, because inline content is
+xref:pipes/fetchers.adoc#reserved-ids[how the server feeds its own workers],
not something a
+tuple can carry. To parse content you already hold, PUT it to `/tika` or
`/rmeta`, which inline
+it for you. The `{"tuples":[...]}` envelope is
required on `/async`; a bare array is rejected.
=== Best practices
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 d0db707baf..7b2b34a974 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
@@ -49,6 +49,7 @@ 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;
+import org.apache.tika.pipes.core.serialization.PipesRequest;
import org.apache.tika.pipes.core.server.IntermediateResult;
import org.apache.tika.utils.ExceptionUtils;
import org.apache.tika.utils.StringUtils;
@@ -240,6 +241,14 @@ public class PipesClient implements Closeable {
serverManager.connectionAbandoned();
closeConnection();
throw e;
+ } catch (PayloadLimitExceededException e) {
+ // Only writeTask's pre-send check throws this here (waitForServer
handles its
+ // own); nothing was written, so the connection stays in sync --
keep it.
+ LOG.warn("clientId={}: request too large for id={}: {}",
pipesClientId, t.getId(),
+ e.getMessage());
+ return buildFatalResult(t.getId(), t.getEmitKey(),
+ PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
+ intermediateResult.get(), e.getMessage());
} catch (Exception e) {
LOG.error("exception waiting for server to complete task: {} ",
t.getId(), e);
closeConnection();
@@ -340,7 +349,14 @@ public class PipesClient implements Closeable {
throw new IOException("connection closed");
}
LOG.debug("pipesClientId={}: sending NEW_REQUEST for id={}",
pipesClientId, t.getId());
- byte[] bytes = JsonPipesIpc.toBytes(t);
+ byte[] bytes = JsonPipesIpc.toBytes(PipesRequest.of(t));
+ // Fail fast before sending: the server would refuse the frame anyway,
but only by
+ // dying or dropping the connection, misreported as a crash.
+ if (bytes.length > maxIpcPayloadBytes) {
+ throw new PayloadLimitExceededException("serialized request for
id=" + t.getId()
+ + " is " + bytes.length + " bytes, over
maxIpcPayloadBytes="
+ + maxIpcPayloadBytes + "; raise maxIpcPayloadBytes or
shrink the request");
+ }
PipesMessage.newRequest(bytes).write(tuple.output);
}
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 c33512fc77..a694d04b44 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
@@ -601,7 +601,7 @@ public class PipesConfig {
* (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.
+ * client-side before sending, reported as {@code PAYLOAD_LIMIT_EXCEEDED}.
* <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
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/BytesFetcher.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/BytesFetcher.java
index 4a85f80210..7d86b0a314 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/BytesFetcher.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/BytesFetcher.java
@@ -29,7 +29,8 @@ import org.apache.tika.plugins.ExtensionConfig;
import org.apache.tika.utils.StringUtils;
/**
- * Serves the bytes a caller put in the {@link InlineBytes} parse-context
entry, so a host that
+ * Serves the bytes in the {@link InlineBytes} parse-context entry -- set by
the caller
+ * in-process, or planted by the server from the {@code PipesRequest} envelope
-- so a host that
* already holds the content does not have to spool it to disk purely to hand
it across the
* process boundary.
* <p>
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/InlineBytes.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/InlineBytes.java
index 2981847f49..de834d5139 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/InlineBytes.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/InlineBytes.java
@@ -19,18 +19,19 @@ package org.apache.tika.pipes.core.fetcher;
import java.io.Serializable;
import java.util.Arrays;
-import org.apache.tika.annotation.TikaComponent;
-
/**
* Document bytes carried in the {@code ParseContext} instead of fetched from
a source, for
* callers that already hold the content and would otherwise have to spool it
to disk just to
* hand it to the forked worker.
* <p>
* Read by {@link BytesFetcher}, which the tuple selects with fetcher id
- * {@link BytesFetcher#FETCHER_ID}. The IPC is Smile, so this rides as native
binary rather than
- * base64; it counts against {@code maxIpcPayloadBytes} like any other part of
the request.
+ * {@link BytesFetcher#FETCHER_ID}. In-process only: deliberately not a
registered component,
+ * so no serialized form of it exists and serialization refuses loudly. On the
IPC wire the
+ * payload travels beside
+ * the tuple in {@code PipesRequest} (which lifts it out on the parent and
plants it back into
+ * the worker's context on the child); it counts against {@code
maxIpcPayloadBytes} like any
+ * other part of the request. A request can supply it in no form at all.
*/
-@TikaComponent(name = "inline-bytes", spi = false)
public class InlineBytes implements Serializable {
private static final long serialVersionUID = 1L;
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
index 739cf51921..b3fc6e0bfd 100644
---
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
@@ -19,13 +19,14 @@ package org.apache.tika.pipes.core.protocol;
import java.io.IOException;
/**
- * Thrown when an incoming IPC payload's declared length exceeds the
configured limit
+ * Thrown when an IPC payload 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.
+ * default {@link PipesMessage#MAX_PAYLOAD_BYTES}). On the read side 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. On the send side ({@code PipesClient} pre-send
check) nothing
+ * was written and the connection stays usable.
*/
public class PayloadLimitExceededException extends IOException {
public PayloadLimitExceededException(String message) {
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleDeserializer.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleDeserializer.java
index 07bc0c6f09..0b656dda94 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleDeserializer.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleDeserializer.java
@@ -32,6 +32,7 @@ import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
+import java.util.TreeSet;
import com.fasterxml.jackson.core.JacksonException;
import com.fasterxml.jackson.core.JsonParser;
@@ -49,6 +50,9 @@ import
org.apache.tika.serialization.serdes.ParseContextDeserializer;
public class FetchEmitTupleDeserializer extends
JsonDeserializer<FetchEmitTuple> {
+ /** The parse-context name InlineBytes was registered under in 4.0.0. */
+ static final String LEGACY_INLINE_BYTES_ENTRY = "inline-bytes";
+
private static final Set<String> KNOWN_KEYS = Set.of(
ID, FETCHER, FETCH_KEY, EMITTER, EMIT_KEY, FETCH_RANGE_START,
FETCH_RANGE_END,
METADATA_KEY, PARSE_CONTEXT, ON_PARSE_EXCEPTION);
@@ -80,6 +84,20 @@ public class FetchEmitTupleDeserializer extends
JsonDeserializer<FetchEmitTuple>
@Override
public FetchEmitTuple deserialize(JsonParser jsonParser,
DeserializationContext deserializationContext) throws IOException,
JacksonException {
JsonNode root = jsonParser.readValueAsTree();
+ // Both checked before rejectUnknownKeys so they get tailored messages.
+ if (root.has(PipesRequest.INLINE_BYTES)) {
+ throw new IOException("'" + PipesRequest.INLINE_BYTES
+ + "' is not a FetchEmitTuple field; content travels
outside the tuple, and"
+ + " only on the host's internal IPC. For tika-server, PUT
content you"
+ + " already hold to /tika or /rmeta, which inline it for
you.");
+ }
+ if (root.path(PARSE_CONTEXT).has(LEGACY_INLINE_BYTES_ENTRY)) {
+ // 4.0.0 serialized this entry; the generic "check for a typo"
would mislead upgraders.
+ throw new IOException("'" + LEGACY_INLINE_BYTES_ENTRY + "' is no
longer a serializable"
+ + " parse-context entry (4.0.0 wrote it as base64):
content travels outside"
+ + " the tuple, and only on the host's internal IPC. For
tika-server, PUT"
+ + " content you already hold to /tika or /rmeta, which
inline it for you.");
+ }
rejectUnknownKeys(root);
String id = readVal(ID, root, null, true);
@@ -124,7 +142,7 @@ public class FetchEmitTupleDeserializer extends
JsonDeserializer<FetchEmitTuple>
String name = it.next();
if (!KNOWN_KEYS.contains(name)) {
throw new IOException("Unrecognized FetchEmitTuple field '" +
name
- + "'. Check for a typo; known fields are " +
KNOWN_KEYS + ".");
+ + "'. Check for a typo; known fields are " + new
TreeSet<>(KNOWN_KEYS) + ".");
}
}
}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleSerializer.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleSerializer.java
index 9dc06b560c..608ab15f37 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleSerializer.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleSerializer.java
@@ -25,7 +25,9 @@ import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.JsonSerializer;
import com.fasterxml.jackson.databind.SerializerProvider;
+import org.apache.tika.parser.ParseContext;
import org.apache.tika.pipes.api.FetchEmitTuple;
+import org.apache.tika.pipes.core.fetcher.InlineBytes;
import org.apache.tika.utils.StringUtils;
public class FetchEmitTupleSerializer extends JsonSerializer<FetchEmitTuple> {
@@ -57,8 +59,17 @@ public class FetchEmitTupleSerializer extends
JsonSerializer<FetchEmitTuple> {
jsonGenerator.writeObjectField(METADATA_KEY, t.getMetadata());
}
jsonGenerator.writeStringField(ON_PARSE_EXCEPTION,
t.getOnParseException().name().toLowerCase(Locale.US));
- if (!t.getParseContext().isEmpty()) {
- jsonGenerator.writeObjectField(PARSE_CONTEXT, t.getParseContext());
+ ParseContext parseContext = t.getParseContext();
+ // Tailored: ParseContextSerializer's generic refusal suggests
registering the
+ // component -- for InlineBytes, exactly the forbidden fix.
+ if (parseContext.get(InlineBytes.class) != null) {
+ throw new IOException("A FetchEmitTuple whose ParseContext holds
InlineBytes has no"
+ + " serialized form: inline content is in-process only
and, on the pipes IPC,"
+ + " travels beside the tuple in the request envelope.
Remove the InlineBytes"
+ + " entry before serializing.");
+ }
+ if (!parseContext.isEmpty()) {
+ jsonGenerator.writeObjectField(PARSE_CONTEXT, parseContext);
}
jsonGenerator.writeEndObject();
}
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 d35f45a66d..6f5cfbe074 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
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.core.StreamReadConstraints;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.fasterxml.jackson.dataformat.smile.SmileFactory;
+import com.fasterxml.jackson.dataformat.smile.SmileGenerator;
import org.apache.tika.config.loader.TikaObjectMapperFactory;
import org.apache.tika.pipes.api.FetchEmitTuple;
@@ -42,8 +43,12 @@ public class JsonPipesIpc {
private static final ObjectMapper OBJECT_MAPPER;
static {
- // Use SmileFactory for binary format - more compact and faster than
text JSON
- SmileFactory smileFactory = new SmileFactory();
+ // Use SmileFactory for binary format - more compact and faster than
text JSON.
+ // 7-bit binary encoding (the Smile default) costs +14% size and a
transcode pass
+ // on every binary payload; this is a private same-version channel, so
write raw.
+ SmileFactory smileFactory = SmileFactory.builder()
+ .disable(SmileGenerator.Feature.ENCODE_BINARY_AS_7BIT)
+ .build();
// Configure stream constraints for large content (e.g., 30MB+
documents)
// Default Jackson limit is 20MB which is too small for IPC with large
documents
@@ -60,6 +65,8 @@ public class JsonPipesIpc {
pipesModule.addSerializer(FetchEmitTuple.class, new
FetchEmitTupleSerializer());
// Parent-to-child IPC: the host builds these tuples itself and they
name __ components.
pipesModule.addDeserializer(FetchEmitTuple.class,
FetchEmitTupleDeserializer.internal());
+ pipesModule.addSerializer(PipesRequest.class, new
PipesRequestSerializer());
+ pipesModule.addDeserializer(PipesRequest.class, new
PipesRequestDeserializer());
pipesModule.addSerializer(EmitData.class, new EmitDataSerializer());
pipesModule.addDeserializer(EmitDataImpl.class, new
EmitDataDeserializer());
pipesModule.addSerializer(PipesResult.class, new
PipesResultSerializer());
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesRequest.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesRequest.java
new file mode 100644
index 0000000000..15745cb41a
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesRequest.java
@@ -0,0 +1,81 @@
+/*
+ * 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.serialization;
+
+import org.apache.tika.parser.ParseContext;
+import org.apache.tika.pipes.api.FetchEmitTuple;
+import org.apache.tika.pipes.core.fetcher.InlineBytes;
+
+/**
+ * The NEW_REQUEST envelope on the parent-to-child IPC: the tuple, plus the
optional inline
+ * document payload beside it. Bytes are data, not tuple state -- a {@link
FetchEmitTuple} and
+ * its ParseContext are never serialized with content in them ({@code
FetchEmitTupleSerializer}
+ * refuses loudly if an {@link InlineBytes} slips through).
+ * <p>
+ * {@link #of} lifts the payload out of the caller's context on the parent;
+ * {@link #applyTo} plants it into the worker's merged context on the child,
where
+ * {@link org.apache.tika.pipes.core.fetcher.BytesFetcher} reads it.
+ */
+public final class PipesRequest {
+
+ /** Wire name of the payload field; not a FetchEmitTuple field. */
+ static final String INLINE_BYTES = "inlineBytes";
+
+ static final String TUPLE = "tuple";
+
+ private final FetchEmitTuple tuple;
+ private final byte[] inlineBytes;
+
+ PipesRequest(FetchEmitTuple tuple, byte[] inlineBytes) {
+ this.tuple = tuple;
+ this.inlineBytes = inlineBytes;
+ }
+
+ /**
+ * Wraps {@code t} for the wire, lifting any {@link InlineBytes} out of
its ParseContext.
+ * The caller's live context is never mutated; the stripped copy exists
only for
+ * serialization.
+ */
+ public static PipesRequest of(FetchEmitTuple t) {
+ ParseContext ctx = t.getParseContext();
+ InlineBytes inline = ctx == null ? null : ctx.get(InlineBytes.class);
+ if (inline == null) {
+ return new PipesRequest(t, null);
+ }
+ ParseContext copy = new ParseContext();
+ copy.copyFrom(ctx);
+ copy.set(InlineBytes.class, null);
+ FetchEmitTuple stripped = new FetchEmitTuple(t.getId(),
t.getFetchKey(), t.getEmitKey(),
+ t.getMetadata(), copy, t.getOnParseException());
+ return new PipesRequest(stripped, inline.getBytes());
+ }
+
+ /** Plants the payload into the worker's context for BytesFetcher; no-op
without one. */
+ public void applyTo(ParseContext mergedContext) {
+ if (inlineBytes != null) {
+ mergedContext.set(InlineBytes.class, new InlineBytes(inlineBytes));
+ }
+ }
+
+ public FetchEmitTuple getTuple() {
+ return tuple;
+ }
+
+ public byte[] getInlineBytes() {
+ return inlineBytes;
+ }
+}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesRequestDeserializer.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesRequestDeserializer.java
new file mode 100644
index 0000000000..a6970814dd
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesRequestDeserializer.java
@@ -0,0 +1,43 @@
+/*
+ * 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.serialization;
+
+import java.io.IOException;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
+
+import org.apache.tika.pipes.api.FetchEmitTuple;
+
+class PipesRequestDeserializer extends JsonDeserializer<PipesRequest> {
+
+ @Override
+ public PipesRequest deserialize(JsonParser p, DeserializationContext ctxt)
+ throws IOException {
+ JsonNode root = p.readValueAsTree();
+ JsonNode tupleNode = root.get(PipesRequest.TUPLE);
+ if (tupleNode == null) {
+ throw new IOException("PipesRequest is missing its '" +
PipesRequest.TUPLE + "' field");
+ }
+ FetchEmitTuple tuple = p.getCodec().treeToValue(tupleNode,
FetchEmitTuple.class);
+ JsonNode bytesNode = root.get(PipesRequest.INLINE_BYTES);
+ byte[] inlineBytes = bytesNode == null ? null :
bytesNode.binaryValue();
+ return new PipesRequest(tuple, inlineBytes);
+ }
+}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesRequestSerializer.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesRequestSerializer.java
new file mode 100644
index 0000000000..b350770073
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/PipesRequestSerializer.java
@@ -0,0 +1,42 @@
+/*
+ * 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.serialization;
+
+import java.io.IOException;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+
+/**
+ * Payload as a raw top-level binary field beside the tuple: on the Smile IPC
this avoids both
+ * base64 (+33%) and 7-bit (+14%) inflation, and keeps content out of the
tuple's serialized form
+ * entirely.
+ */
+class PipesRequestSerializer extends JsonSerializer<PipesRequest> {
+
+ @Override
+ public void serialize(PipesRequest r, JsonGenerator gen,
SerializerProvider provider)
+ throws IOException {
+ gen.writeStartObject();
+ gen.writeObjectField(PipesRequest.TUPLE, r.getTuple());
+ if (r.getInlineBytes() != null) {
+ gen.writeBinaryField(PipesRequest.INLINE_BYTES,
r.getInlineBytes());
+ }
+ gen.writeEndObject();
+ }
+}
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 19808fce9f..b1b7c1861e 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
@@ -52,6 +52,7 @@ import org.apache.tika.pipes.core.PipesConfig;
import org.apache.tika.pipes.core.protocol.PipesMessage;
import org.apache.tika.pipes.core.protocol.PipesMessageType;
import org.apache.tika.pipes.core.serialization.JsonPipesIpc;
+import org.apache.tika.pipes.core.serialization.PipesRequest;
import org.apache.tika.serialization.ParseContextUtils;
/**
@@ -151,11 +152,13 @@ public class ConnectionHandler implements Runnable,
Closeable {
intermediateResult.clear();
CountDownLatch countDownLatch = new CountDownLatch(1);
+ PipesRequest pipesRequest;
FetchEmitTuple fetchEmitTuple;
try {
- fetchEmitTuple =
JsonPipesIpc.fromBytes(msg.payload(), FetchEmitTuple.class);
+ pipesRequest =
JsonPipesIpc.fromBytes(msg.payload(), PipesRequest.class);
+ fetchEmitTuple = pipesRequest.getTuple();
} catch (IOException e) {
- LOG.error("handlerId={}: problem deserializing
FetchEmitTuple", handlerId, e);
+ LOG.error("handlerId={}: problem deserializing
PipesRequest", handlerId, e);
handleCrash(PipesMessageType.UNSPECIFIED_CRASH,
"unknown", e);
return; // connection is unsalvageable after
deserialization failure
}
@@ -167,6 +170,9 @@ public class ConnectionHandler implements Runnable,
Closeable {
ServerProtocolIO.clampRequestTimeoutLimits(
fetchEmitTuple.getParseContext(),
mergedContext,
pipesConfig.getMaxTotalTaskTimeoutMillis());
+ // After resolveAll: the payload is typed runtime
state for
+ // BytesFetcher, never a resolvable config entry.
+ pipesRequest.applyTo(mergedContext);
// Installed here, before submit, so the worker
thread's own
// ParseTimeout.getOrCreate(mergedContext) call
(inside CompositeParser)
// sees this instance rather than racing to
install its own.
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 c359f89227..ee14e103f5 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
@@ -73,6 +73,7 @@ import org.apache.tika.pipes.core.protocol.PipesMessage;
import org.apache.tika.pipes.core.protocol.PipesMessageType;
import org.apache.tika.pipes.core.protocol.ShutDownReceivedException;
import org.apache.tika.pipes.core.serialization.JsonPipesIpc;
+import org.apache.tika.pipes.core.serialization.PipesRequest;
import org.apache.tika.plugins.ExtensionConfig;
import org.apache.tika.plugins.TikaPluginManager;
import org.apache.tika.sax.ContentHandlerFactory;
@@ -414,11 +415,13 @@ public class PipesServer implements AutoCloseable {
intermediateResult.clear();
CountDownLatch countDownLatch = new CountDownLatch(1);
+ PipesRequest pipesRequest;
FetchEmitTuple fetchEmitTuple;
try {
- fetchEmitTuple =
JsonPipesIpc.fromBytes(msg.payload(), FetchEmitTuple.class);
+ pipesRequest =
JsonPipesIpc.fromBytes(msg.payload(), PipesRequest.class);
+ fetchEmitTuple = pipesRequest.getTuple();
} catch (IOException e) {
- LOG.error("problem deserializing FetchEmitTuple",
e);
+ LOG.error("problem deserializing PipesRequest", e);
handleCrash(PipesMessageType.UNSPECIFIED_CRASH,
"unknown", e);
break; // unreachable after handleCrash/exit, but
needed for compilation
}
@@ -431,6 +434,9 @@ public class PipesServer implements AutoCloseable {
ServerProtocolIO.clampRequestTimeoutLimits(
fetchEmitTuple.getParseContext(),
mergedContext,
pipesConfig.getMaxTotalTaskTimeoutMillis());
+ // After resolveAll: the payload is typed runtime
state for
+ // BytesFetcher, never a resolvable config entry.
+ pipesRequest.applyTo(mergedContext);
// Installed here, before submit, so the worker
thread's own
// ParseTimeout.getOrCreate(mergedContext) call
(inside CompositeParser)
// sees this instance rather than racing to
install its own.
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientInterruptTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientInterruptTest.java
index a7a3e1be53..48a7336934 100644
---
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientInterruptTest.java
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientInterruptTest.java
@@ -224,57 +224,4 @@ public class PipesClientInterruptTest {
}
}
- /**
- * Points the client at the scripted server; no forked process anywhere.
- */
- private static final class SentinelServerManager implements ServerManager {
- private final int port;
- private volatile boolean abandoned;
-
- private SentinelServerManager(int port) {
- this.port = port;
- }
-
- @Override
- public void connectionAbandoned() {
- abandoned = true;
- }
-
- @Override
- public int getPort() {
- return port;
- }
-
- @Override
- public void ensureRunning() {
- // the scripted server is already listening
- }
-
- @Override
- public Socket connect(int socketTimeoutMs) throws IOException {
- Socket socket = new Socket("localhost", port);
- socket.setSoTimeout(socketTimeoutMs);
- return socket;
- }
-
- @Override
- public void shutdown() {
- // nothing to shut down
- }
-
- @Override
- public boolean isRunning() {
- return true;
- }
-
- @Override
- public java.nio.file.Path getTempDirectory() {
- return null;
- }
-
- @Override
- public void close() {
- // nothing to close
- }
- }
}
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientPayloadLimitTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientPayloadLimitTest.java
new file mode 100644
index 0000000000..67b62a50f3
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientPayloadLimitTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.parser.ParseContext;
+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.api.fetcher.FetchKey;
+import org.apache.tika.pipes.core.protocol.PipesMessage;
+import org.apache.tika.pipes.core.server.ServerProtocolIO;
+
+public class PipesClientPayloadLimitTest {
+
+ /**
+ * A request whose serialized form exceeds maxIpcPayloadBytes must be
refused before
+ * sending -- a clean PAYLOAD_LIMIT_EXCEEDED, not a worker death
misreported as a crash --
+ * and must leave the connection usable.
+ */
+ @Test
+ @Timeout(45)
+ public void oversizedRequestFailsFastWithoutSending() throws Exception {
+ try (ServerSocket serverSocket = new ServerSocket(0)) {
+ CountDownLatch connectionClosed = new CountDownLatch(1);
+ Thread sentinel = new Thread(() ->
runReadyOnlyServer(serverSocket, connectionClosed));
+ sentinel.setDaemon(true);
+ sentinel.start();
+
+ PipesConfig pipesConfig = new PipesConfig();
+
pipesConfig.setMaxIpcPayloadBytes(ServerProtocolIO.MIN_FALLBACK_PAYLOAD_BYTES);
+ SentinelServerManager manager = new
SentinelServerManager(serverSocket.getLocalPort());
+ try (PipesClient client = new PipesClient(pipesConfig, manager)) {
+ Metadata metadata = new Metadata();
+ metadata.set("oversized", "x".repeat(10_000));
+ PipesResult result = client.process(new
FetchEmitTuple("payload-limit-test",
+ new FetchKey("fetcher", "key"), new EmitKey(),
metadata,
+ new ParseContext(),
FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
+
+ assertEquals(PipesResult.RESULT_STATUS.PAYLOAD_LIMIT_EXCEEDED,
result.status(),
+ "expected client-side refusal, got: " + result.status()
+ + " / " + result.message());
+ assertTrue(result.message().contains("maxIpcPayloadBytes"),
+ "message should name the limit, got: " +
result.message());
+ assertFalse(manager.abandoned, "nothing was sent; no reason to
abandon");
+ assertFalse(connectionClosed.await(300, TimeUnit.MILLISECONDS),
+ "nothing was sent; the connection must stay usable");
+ }
+ }
+ }
+
+ /** Accepts one connection, sends READY, then just holds the socket open.
*/
+ private static void runReadyOnlyServer(ServerSocket serverSocket,
+ CountDownLatch connectionClosed) {
+ try (Socket socket = serverSocket.accept();
+ DataInputStream in = new
DataInputStream(socket.getInputStream());
+ DataOutputStream out = new
DataOutputStream(socket.getOutputStream())) {
+ PipesMessage.ready().write(out);
+ PipesMessage.read(in);
+ } catch (IOException e) {
+ // EOF or reset: the connection is gone
+ }
+ connectionClosed.countDown();
+ }
+}
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/SentinelServerManager.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/SentinelServerManager.java
new file mode 100644
index 0000000000..7007072b51
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/SentinelServerManager.java
@@ -0,0 +1,75 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.nio.file.Path;
+
+/**
+ * Points a {@link PipesClient} at a scripted in-test server; no forked
process anywhere.
+ */
+final class SentinelServerManager implements ServerManager {
+ private final int port;
+ volatile boolean abandoned;
+
+ SentinelServerManager(int port) {
+ this.port = port;
+ }
+
+ @Override
+ public void connectionAbandoned() {
+ abandoned = true;
+ }
+
+ @Override
+ public int getPort() {
+ return port;
+ }
+
+ @Override
+ public void ensureRunning() {
+ // the scripted server is already listening
+ }
+
+ @Override
+ public Socket connect(int socketTimeoutMs) throws IOException {
+ Socket socket = new Socket("localhost", port);
+ socket.setSoTimeout(socketTimeoutMs);
+ return socket;
+ }
+
+ @Override
+ public void shutdown() {
+ // nothing to shut down
+ }
+
+ @Override
+ public boolean isRunning() {
+ return true;
+ }
+
+ @Override
+ public Path getTempDirectory() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+ // nothing to close
+ }
+}
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/InlineBytesWireTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/InlineBytesWireTest.java
new file mode 100644
index 0000000000..4d88c6a4b0
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/InlineBytesWireTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.serialization;
+
+import static org.apache.tika.pipes.core.serialization.WireTestUtil.root;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.Random;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+
+import org.apache.tika.io.TikaInputStream;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.TikaCoreProperties;
+import org.apache.tika.parser.ParseContext;
+import org.apache.tika.pipes.api.FetchEmitTuple;
+import org.apache.tika.pipes.api.ParseMode;
+import org.apache.tika.pipes.api.emitter.EmitKey;
+import org.apache.tika.pipes.api.fetcher.FetchKey;
+import org.apache.tika.pipes.core.fetcher.BytesFetcher;
+import org.apache.tika.pipes.core.fetcher.InlineBytes;
+import org.apache.tika.serialization.ParseContextUtils;
+import org.apache.tika.serialization.serdes.ParseContextDeserializer;
+
+/**
+ * The inline payload travels beside the tuple in the {@link PipesRequest}
envelope -- never
+ * inside the tuple or its serialized ParseContext, in any format.
+ */
+public class InlineBytesWireTest {
+
+ private static byte[] payload(int size) {
+ byte[] b = new byte[size];
+ new Random(17).nextBytes(b);
+ return b;
+ }
+
+ private static FetchEmitTuple tuple(byte[] payload) {
+ ParseContext ctx = new ParseContext();
+ ctx.set(InlineBytes.class, new InlineBytes(payload));
+ ctx.set(ParseMode.class, ParseMode.RMETA);
+ return new FetchEmitTuple("t", new FetchKey(BytesFetcher.FETCHER_ID,
"doc.bin"),
+ EmitKey.NO_EMIT, new Metadata(), ctx);
+ }
+
+ @Test
+ public void ipcRoundTripPreservesPayload() throws Exception {
+ byte[] payload = payload(100_000);
+ FetchEmitTuple t = tuple(payload);
+ PipesRequest request = PipesRequest.of(t);
+ // lifting the payload into the envelope must not mutate the caller's
live context
+ assertNotNull(t.getParseContext().get(InlineBytes.class));
+
+ byte[] wire = JsonPipesIpc.toBytes(request);
+ PipesRequest back = JsonPipesIpc.fromBytes(wire, PipesRequest.class);
+ // the child's path: merge, resolve, plant, fetch
+ ParseContext merged = new ParseContext();
+ merged.copyFrom(back.getTuple().getParseContext());
+ ParseContextUtils.resolveAll(merged, getClass().getClassLoader());
+ back.applyTo(merged);
+ assertEquals(ParseMode.RMETA, merged.get(ParseMode.class));
+
+ Metadata metadata = new Metadata();
+ try (TikaInputStream tis = new BytesFetcher().fetch("doc.bin",
metadata, merged)) {
+ assertArrayEquals(payload, tis.readAllBytes());
+ }
+ assertEquals("doc.bin",
metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY));
+ }
+
+ @Test
+ public void wireCarriesRawBinary() throws Exception {
+ byte[] payload = payload(1_000_000);
+ byte[] wire = JsonPipesIpc.toBytes(PipesRequest.of(tuple(payload)));
+ // raw binary: no base64 (+33%) and no Smile 7-bit encoding (+14%)
+ assertTrue(wire.length >= payload.length, "wire shorter than
payload?");
+ assertTrue(wire.length < payload.length + 1024,
+ "payload not raw on the wire: " + wire.length + " bytes for "
+ payload.length);
+ }
+
+ @Test
+ public void payloadStaysOutOfTheTuple() throws Exception {
+ byte[] wire =
JsonPipesIpc.toBytes(PipesRequest.of(tuple(payload(1000))));
+ PipesRequest back = JsonPipesIpc.fromBytes(wire, PipesRequest.class);
+
assertFalse(back.getTuple().getParseContext().hasJsonConfig("inline-bytes"),
+ "payload leaked into the lazy-config path");
+ assertNull(back.getTuple().getParseContext().get(InlineBytes.class),
+ "payload leaked into the tuple's context");
+ assertNotNull(back.getInlineBytes());
+ }
+
+ @Test
+ public void serializingATupleStillCarryingPayloadFailsLoudly() {
+ // A tuple whose context still holds InlineBytes has no serialized
form;
+ // PipesRequest.of is the only way onto the wire.
+ Exception e = assertThrows(Exception.class, () ->
JsonPipesIpc.toBytes(tuple(payload(10))));
+ assertTrue(root(e).contains("no serialized form"),
+ "expected loud refusal, got: " + root(e));
+ Exception text = assertThrows(Exception.class,
+ () -> JsonFetchEmitTuple.toJson(tuple(payload(10))));
+ assertTrue(root(text).contains("no serialized form"),
+ "expected loud refusal, got: " + root(text));
+ }
+
+ @Test
+ public void requestBodyRejectsInlineBytes() {
+ String json = "{\"id\":\"t\",\"fetcher\":\"f\",\"fetchKey\":\"k\"," +
+ "\"emitter\":\"e\",\"inlineBytes\":\"QUJD\"}";
+ Exception e = assertThrows(IOException.class,
+ () -> JsonFetchEmitTuple.fromJson(new StringReader(json)));
+ assertTrue(root(e).contains("not a FetchEmitTuple field"),
+ "expected inlineBytes rejection, got: " + root(e));
+ }
+
+ @Test
+ public void requestBodyRejectsLegacyParseContextSpelling() {
+ // 4.0.0 serialized InlineBytes under this parse-context name;
upgraders get a
+ // tailored message, not the generic "check for a typo".
+ String json =
"{\"id\":\"t\",\"fetcher\":\"f\",\"fetchKey\":\"k\",\"emitter\":\"e\"," +
+ "\"parse-context\":{\"inline-bytes\":{\"bytes\":\"QUJD\"}}}";
+ Exception e = assertThrows(IOException.class,
+ () -> JsonFetchEmitTuple.fromJson(new StringReader(json)));
+ assertTrue(root(e).contains("no longer a serializable parse-context
entry"),
+ "expected legacy-spelling rejection, got: " + root(e));
+ }
+
+ @Test
+ public void requestParseContextFormCannotBind() throws Exception {
+ // Defense in depth below the tuple gate: even if the entry reaches a
ParseContext
+ // (it is admitted as an inert config at most), resolution fails
closed.
+ JsonNode node = new ObjectMapper()
+ .readTree("{\"inline-bytes\":{\"bytes\":\"QUJD\"}}");
+ ParseContext ctx = ParseContextDeserializer.readParseContext(node,
true);
+ Exception e = assertThrows(Exception.class,
+ () -> ParseContextUtils.resolveAll(ctx,
getClass().getClassLoader()));
+ assertTrue(root(e).contains("Unrecognized parse-context entry"),
+ "expected fail-closed resolution, got: " + root(e));
+ assertNull(ctx.get(InlineBytes.class));
+ }
+
+ @Test
+ public void unknownFieldErrorDoesNotAdvertiseInlineBytes() {
+ String json =
"{\"id\":\"t\",\"fetcher\":\"f\",\"fetchKey\":\"k\",\"emitter\":\"e\"," +
+ "\"fetchKye\":\"typo\"}";
+ Exception e = assertThrows(IOException.class,
+ () -> JsonFetchEmitTuple.fromJson(new StringReader(json)));
+ assertTrue(root(e).contains("Unrecognized"), "expected unknown-field
error, got: " + root(e));
+ assertFalse(root(e).contains("inlineBytes"),
+ "IPC-only field advertised to requests: " + root(e));
+ }
+}
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/SystemComponentIdWireTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/SystemComponentIdWireTest.java
index 574bd351b8..7c70728dba 100644
---
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/SystemComponentIdWireTest.java
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/SystemComponentIdWireTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.tika.pipes.core.serialization;
+import static org.apache.tika.pipes.core.serialization.WireTestUtil.root;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -118,12 +119,4 @@ public class SystemComponentIdWireTest {
() -> JsonPipesIpc.fromBytes(smile, FetchEmitTuple.class));
assertTrue(root(e).contains("Illegal"), "expected charset rejection,
got: " + root(e));
}
-
- private static String root(Throwable t) {
- StringBuilder sb = new StringBuilder();
- for (Throwable c = t; c != null; c = c.getCause()) {
- sb.append(c.getMessage()).append(' ');
- }
- return sb.toString();
- }
}
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/WireRestrictedFetchEmitTupleTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/WireRestrictedFetchEmitTupleTest.java
index 9815f63695..143bc9b5dd 100644
---
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/WireRestrictedFetchEmitTupleTest.java
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/WireRestrictedFetchEmitTupleTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.tika.pipes.core.serialization;
+import static org.apache.tika.pipes.core.serialization.WireTestUtil.root;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -98,12 +99,4 @@ public class WireRestrictedFetchEmitTupleTest {
assertTrue(root(e).contains("may not be supplied via a request
parseContext"),
"expected wire-blocked rejection at fork IPC, got: " +
root(e));
}
-
- private static String root(Throwable t) {
- Throwable r = t;
- while (r.getCause() != null && r.getCause() != r) {
- r = r.getCause();
- }
- return String.valueOf(r.getMessage());
- }
}
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/test/java/org/apache/tika/pipes/core/serialization/WireTestUtil.java
similarity index 51%
copy from
tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PayloadLimitExceededException.java
copy to
tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/WireTestUtil.java
index 739cf51921..fa82b44095 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PayloadLimitExceededException.java
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/WireTestUtil.java
@@ -14,21 +14,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.pipes.core.protocol;
+package org.apache.tika.pipes.core.serialization;
-import java.io.IOException;
+final class WireTestUtil {
-/**
- * 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);
+ private WireTestUtil() {
+ }
+
+ /** Root-cause message, for asserting on rejections Jackson may wrap. */
+ static String root(Throwable t) {
+ Throwable r = t;
+ while (r.getCause() != null && r.getCause() != r) {
+ r = r.getCause();
+ }
+ return String.valueOf(r.getMessage());
}
}
diff --git
a/tika-pipes/tika-pipes-fork-parser/src/test/java/org/apache/tika/pipes/fork/PipesForkParserTest.java
b/tika-pipes/tika-pipes-fork-parser/src/test/java/org/apache/tika/pipes/fork/PipesForkParserTest.java
index 725c446062..832255c917 100644
---
a/tika-pipes/tika-pipes-fork-parser/src/test/java/org/apache/tika/pipes/fork/PipesForkParserTest.java
+++
b/tika-pipes/tika-pipes-fork-parser/src/test/java/org/apache/tika/pipes/fork/PipesForkParserTest.java
@@ -126,6 +126,8 @@ public class PipesForkParserTest {
PipesForkResult result = parser.parse(tis, new Metadata(),
parseContext);
assertTrue(result.isSuccess(), "Parse should succeed. Status: " +
result.getStatus()
+ ", message: " + result.getMessage());
+ assertTrue(result.getContent().contains("inline body"),
+ "worker did not parse the inline payload; content: " +
result.getContent());
assertNull(parseContext.get(InlineBytes.class),
"inline payload must not outlive its request in the
caller's context");
}
