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 333dc48e96 TIKA-4808 - improve in-memory pipes handling (#3028)
333dc48e96 is described below
commit 333dc48e96c6790dc433a11f38df84bf2769c670
Author: Tim Allison <[email protected]>
AuthorDate: Mon Aug 17 07:35:50 2026 -0400
TIKA-4808 - improve in-memory pipes handling (#3028)
---
.../advanced/integration-testing/tika-server.adoc | 6 +-
docs/modules/ROOT/pages/advanced/spooling.adoc | 22 ++
.../pages/migration-to-4x/migrating-to-4x.adoc | 17 ++
docs/modules/ROOT/pages/pipes/configuration.adoc | 7 +-
docs/modules/ROOT/pages/pipes/fetchers.adoc | 18 ++
.../ROOT/pages/using-tika/server/index.adoc | 2 +-
.../tika/pipes/grpc/TikaGrpcConcurrencyTest.java | 3 +-
.../apache/tika/pipes/grpc/TikaGrpcServerTest.java | 4 +-
.../src/test/resources/tika-pipes-test-config.json | 2 +-
.../org/apache/tika/pipes/api/ComponentIds.java | 101 +++++++++
.../tika/pipes/core/AbstractComponentManager.java | 57 +++++-
.../org/apache/tika/pipes/core/PipesConfig.java | 64 ++++++
.../tika/pipes/core/config/ConfigMerger.java | 46 ++++-
.../tika/pipes/core/fetcher/BytesFetcher.java | 67 ++++++
.../tika/pipes/core/fetcher/InlineBytes.java | 92 +++++++++
.../tika/pipes/core/fetcher/PayloadRouter.java | 155 ++++++++++++++
.../serialization/FetchEmitTupleDeserializer.java | 46 ++++-
.../pipes/core/serialization/JsonPipesIpc.java | 3 +-
.../apache/tika/pipes/core/server/EmitHandler.java | 4 +-
.../tika/pipes/core/server/FetchHandler.java | 11 +-
.../tika/pipes/core/TikaPipesConfigTest.java | 56 +++++
.../tika/pipes/core/config/ConfigMergerTest.java | 38 +++-
.../tika/pipes/core/fetcher/PayloadRouterTest.java | 228 +++++++++++++++++++++
.../serialization/SystemComponentIdWireTest.java | 129 ++++++++++++
.../apache/tika/pipes/fork/PipesForkParser.java | 40 +++-
.../tika/pipes/fork/PipesForkParserTest.java | 28 +++
.../server/core/resource/PipesParsingHelper.java | 115 ++++++-----
.../server/core/resource/UnpackerResource.java | 23 +--
.../tika/server/core/ForcedSpoolPathTest.java | 115 +++++++++++
.../core/resource/ReservedComponentIdTest.java | 14 +-
.../configs/cxf-unpack-test-template.json | 2 +-
31 files changed, 1414 insertions(+), 101 deletions(-)
diff --git
a/docs/modules/ROOT/pages/advanced/integration-testing/tika-server.adoc
b/docs/modules/ROOT/pages/advanced/integration-testing/tika-server.adoc
index ec3db034eb..b6d90f9272 100644
--- a/docs/modules/ROOT/pages/advanced/integration-testing/tika-server.adoc
+++ b/docs/modules/ROOT/pages/advanced/integration-testing/tika-server.adoc
@@ -495,12 +495,14 @@ curl -X PUT -T file.docx http://localhost:9998/unpack/all
-o output.zip
The server automatically configures the required fetcher and emitter for
pipes-based parsing:
-* **tika-server-fetcher**: A file-system-fetcher with `basePath` pointing to a
dedicated temp directory for input files. This enables the `/tika`, `/rmeta`,
and `/meta` endpoints to work with uploaded files.
+* **\_\_tika-server**: A file-system-fetcher with `basePath` pointing to a
dedicated temp directory for input files. This enables the `/tika`, `/rmeta`,
and `/meta` endpoints to work with uploaded files.
-* **unpack-emitter**: A file-system-emitter with `basePath` pointing to a
dedicated temp directory for unpacked files. This is only created when the
`/unpack` endpoint is enabled (default). This enables the `/unpack/all`
endpoint to return embedded files as a ZIP.
+* **\_\_unpack**: A file-system-emitter with `basePath` pointing to a
dedicated temp directory for unpacked files. This is only created when the
`/unpack` endpoint is enabled (default). This enables the `/unpack/all`
endpoint to return embedded files as a ZIP.
Both temp directories are cleaned up on server shutdown.
+The `\_\_` prefix is reserved. Component ids beginning with `\_\_` name
components the server wires up for itself: a `/pipes` or `/async` request that
names one is rejected, a config file that defines one fails at startup, and
they are omitted from the fetcher and emitter lists reported in error messages.
Ids you configure yourself may contain only letters, digits, `.`, `_` and `-`,
and are trimmed of surrounding whitespace when loaded.
+
If a user config file does not include `plugin-roots`, the server
automatically adds a default value pointing to a `plugins` directory in the
current working directory.
=== Security Boundary
diff --git a/docs/modules/ROOT/pages/advanced/spooling.adoc
b/docs/modules/ROOT/pages/advanced/spooling.adoc
index 29b30bf297..2883978457 100644
--- a/docs/modules/ROOT/pages/advanced/spooling.adoc
+++ b/docs/modules/ROOT/pages/advanced/spooling.adoc
@@ -82,6 +82,28 @@ file management. This means:
* The `rewind()` method efficiently resets the stream for re-reading.
* Memory-mapped and disk-backed strategies can be selected based on use case.
+[#passthrough]
+=== Nothing Spools Until Something Asks
+
+A `TikaInputStream` wrapping a plain `InputStream` starts in passthrough mode
and stays there.
+Reading through it -- at any size -- writes nothing to disk. Only two things
spool:
+
+* `getFile()` / `getPath()`, which materialise a file by definition;
+* `enableRewind()`, which starts caching and spills to disk past 1 MB.
+
+`mark()`/`reset()` in passthrough mode delegate to an in-memory buffer and do
not spool, which
+is why the usual detect-then-parse sequence (mark, peek, reset, `getFile()`)
still works.
+
+Two consequences worth knowing:
+
+* `enableRewind()` must be called at position 0, and `getPath()` throws if the
stream has already
+ been read past position 0. Decide up front whether a code path needs to
rewind.
+* `mark(readlimit)` grows a heap buffer up to `readlimit`. A large mark is a
memory cost, not a
+ disk one.
+
+This is what lets pipes hand small documents to a forked worker without
touching disk; see
+xref:pipes/fetchers.adoc#inline-bytes[Inline Bytes].
+
== User Guide
=== Default Behavior
diff --git a/docs/modules/ROOT/pages/migration-to-4x/migrating-to-4x.adoc
b/docs/modules/ROOT/pages/migration-to-4x/migrating-to-4x.adoc
index 6beb653386..1651617228 100644
--- a/docs/modules/ROOT/pages/migration-to-4x/migrating-to-4x.adoc
+++ b/docs/modules/ROOT/pages/migration-to-4x/migrating-to-4x.adoc
@@ -242,6 +242,23 @@ External parsers must now be explicitly configured via
JSON. See
xref:configuration/parsers/external-parser.adoc[External Parser Configuration]
for details.
+=== TikaInputStream no longer spools unless asked
+
+A `TikaInputStream` wrapping a plain `InputStream` reads straight through;
nothing is written to
+disk until `getFile()`/`getPath()` is called, or `enableRewind()` starts
caching.
+
+[WARNING]
+====
+`enableRewind()` must be called at position 0, and `getFile()`/`getPath()`
throw if the stream
+has already been read past position 0. A parser that reads part of the stream
and *then* asks for
+a file worked in 3.x and now fails. Call `enableRewind()` first, or take the
file before reading.
+
+`mark()`/`reset()` are unaffected -- they use an in-memory buffer in this mode
-- so the usual
+mark, peek, reset, `getFile()` sequence still works.
+====
+
+See xref:advanced/spooling.adoc#passthrough[Spooling] for detail.
+
=== EmbeddedDocumentExtractor is now stateless
[WARNING]
diff --git a/docs/modules/ROOT/pages/pipes/configuration.adoc
b/docs/modules/ROOT/pages/pipes/configuration.adoc
index ad3de231ff..165589556e 100644
--- a/docs/modules/ROOT/pages/pipes/configuration.adoc
+++ b/docs/modules/ROOT/pages/pipes/configuration.adoc
@@ -138,7 +138,8 @@ 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
+[#payload-limits]
+== IPC and Inline Payload Limits
[cols="1,1,3"]
|===
@@ -147,6 +148,10 @@ These settings control how parsed results are batched
before sending to emitters
|`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 [...]
+
+|`maxInlineBytes`
+|`10485760` (10 MB)
+|Largest document carried inline to the forked worker instead of being written
to a file first. A host that already holds the content (tika-server's `/tika`,
`/rmeta`, `/meta`, `/detect`, `/unpack`; `PipesForkParser` with a
non-file-backed stream) sends anything at or below this size inside the request
and touches no disk at all; anything larger is written once to the fetcher's
`basePath`. A stream already backed by a file always keeps its file, whatever
the size. Set to `0` to disable i [...]
|===
== Emit Strategy
diff --git a/docs/modules/ROOT/pages/pipes/fetchers.adoc
b/docs/modules/ROOT/pages/pipes/fetchers.adoc
index 96beaf4749..64608a8a48 100644
--- a/docs/modules/ROOT/pages/pipes/fetchers.adoc
+++ b/docs/modules/ROOT/pages/pipes/fetchers.adoc
@@ -54,6 +54,24 @@ Fetchers live under the top-level `fetchers` key. Each
fetcher gets an ID (the o
A single pipes config may declare multiple fetchers with different IDs and use
them in different iterators or pipelines.
+[#reserved-ids]
+== Reserved IDs
+
+IDs beginning with `\_\_` name components the host wires up for itself --
tika-server's upload fetcher and unpack emitter, `PipesForkParser`'s internal
fetcher, and the inline-bytes fetcher below. They are reserved:
+
+* a `/pipes` or `/async` request that names one is rejected;
+* a config file that defines one fails at startup;
+* they are omitted from the fetcher and emitter lists reported in "not found"
errors.
+
+IDs you configure yourself may contain only letters, digits, `.`, `_` and `-`,
and are trimmed of surrounding whitespace when loaded. Anything else is
rejected at config load time.
+
+[#inline-bytes]
+== Inline Bytes
+
+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.
+
[#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 7c674aa4eb..a5ea494007 100644
--- a/docs/modules/ROOT/pages/using-tika/server/index.adoc
+++ b/docs/modules/ROOT/pages/using-tika/server/index.adoc
@@ -362,7 +362,7 @@ Server behavior beyond host/port is controlled by a JSON
config file passed via
|`maxRequestSizeBytes`
|`1 GiB`
-|Maximum request body in bytes; larger requests are rejected with `413`.
Uploads are spooled to disk, so this bounds how much one request can write.
Raise it for larger documents, or set a negative value to disable the limit.
+|Maximum request body in bytes; larger requests are rejected with `413`.
Enforced on bytes actually read, so a chunked upload with no `Content-Length`
is bounded too. Bodies above
xref:pipes/configuration.adoc#payload-limits[`maxInlineBytes`] are spooled to
disk, so this also bounds how much one request can write. Raise it for larger
documents, or set a negative value to disable the limit.
|`maxQueuePauseMillis`
|`60000`
diff --git
a/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcConcurrencyTest.java
b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcConcurrencyTest.java
index edbf52f834..6a3dcc11f2 100644
---
a/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcConcurrencyTest.java
+++
b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcConcurrencyTest.java
@@ -55,7 +55,6 @@ import org.apache.tika.FetchAndParseReply;
import org.apache.tika.FetchAndParseRequest;
import org.apache.tika.TikaGrpc;
import org.apache.tika.pipes.api.PipesResult;
-import org.apache.tika.pipes.fetcher.fs.FileSystemFetcher;
import org.apache.tika.serialization.config.JsonConfigHelper;
/**
@@ -70,7 +69,7 @@ public class TikaGrpcConcurrencyTest {
// The fetcher must come from the config file: one saved at runtime through
// saveFetcher is not visible to the forked worker.
- private static final String FETCHER_ID = "nick1:is:cool:super/" +
FileSystemFetcher.class;
+ private static final String FETCHER_ID = "nick1.is.cool.super-fs";
/**
* All concurrent calls must parse, and each reply must carry its own
diff --git
a/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java
b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java
index 2709f6f070..891685083b 100644
--- a/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java
+++ b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java
@@ -78,7 +78,6 @@ import org.apache.tika.SavePipesIteratorReply;
import org.apache.tika.SavePipesIteratorRequest;
import org.apache.tika.TikaGrpc;
import org.apache.tika.pipes.api.PipesResult;
-import org.apache.tika.pipes.fetcher.fs.FileSystemFetcher;
import org.apache.tika.serialization.config.JsonConfigHelper;
@ExtendWith(GrpcCleanupExtension.class)
@@ -226,7 +225,8 @@ public class TikaGrpcServerTest {
@NotNull
private static String createFetcherId(int i) {
- return "nick" + i + ":is:cool:super/" + FileSystemFetcher.class;
+ // ComponentIds restricts ids to letters, digits, '.', '_' and '-'
+ return "nick" + i + ".is.cool.super-fs";
}
@Test
diff --git a/tika-grpc/src/test/resources/tika-pipes-test-config.json
b/tika-grpc/src/test/resources/tika-pipes-test-config.json
index dfcfc86a2d..a24f10a9ed 100644
--- a/tika-grpc/src/test/resources/tika-pipes-test-config.json
+++ b/tika-grpc/src/test/resources/tika-pipes-test-config.json
@@ -16,7 +16,7 @@
}
},
"fetchers": {
- "nick1:is:cool:super/class
org.apache.tika.pipes.fetcher.fs.FileSystemFetcher": {
+ "nick1.is.cool.super-fs": {
"file-system-fetcher": {
"basePath": "FETCHER_BASE_PATH",
"extractFileSystemMetadata": true
diff --git
a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/ComponentIds.java
b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/ComponentIds.java
new file mode 100644
index 0000000000..c8caae650c
--- /dev/null
+++
b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/ComponentIds.java
@@ -0,0 +1,101 @@
+/*
+ * 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.api;
+
+import java.util.regex.Pattern;
+
+/**
+ * Naming rules for pipes component instance ids (fetchers, emitters,
iterators).
+ * <p>
+ * Ids beginning with {@value #SYSTEM_PREFIX} are reserved for components the
host wires up for
+ * its own use -- tika-server's upload fetcher, for instance. Reserving a
namespace rather than
+ * enumerating names means a new internal component cannot be reached from a
request by someone
+ * forgetting to add it to a list.
+ * <p>
+ * Callers must {@link #normalize(String)} an id before comparing or storing
it, and must validate
+ * the normalized form. Validating the raw id and resolving the trimmed one
would let
+ * {@code " __foo"} slip past the prefix check and then bind to {@code __foo}.
+ */
+public final class ComponentIds {
+
+ /** Prefix reserved for host-wired components; not nameable from user
config or a request. */
+ public static final String SYSTEM_PREFIX = "__";
+
+ /**
+ * Ids are restricted to this set so a normalized id can be used verbatim
everywhere.
+ * Trimming alone leaves NBSP and zero-width characters, which produce ids
that look
+ * reserved but resolve to nothing; the allowlist refuses them outright.
It also keeps
+ * path separators and quoting characters out of anything derived from an
id.
+ * System ids satisfy it too -- {@code _} is a member.
+ */
+ public static final Pattern LEGAL_ID = Pattern.compile("[A-Za-z0-9._-]+");
+
+ private ComponentIds() {
+ }
+
+ /**
+ * Canonical form of an id: trimmed, or null if null.
+ */
+ public static String normalize(String id) {
+ return id == null ? null : id.trim();
+ }
+
+ /**
+ * True if the normalized id is in the reserved system namespace.
+ */
+ public static boolean isSystem(String id) {
+ String normalized = normalize(id);
+ return normalized != null && normalized.startsWith(SYSTEM_PREFIX);
+ }
+
+ /**
+ * Normalizes {@code id} and rejects it if it is blank, illegal, or
system-reserved.
+ *
+ * @param id the raw, caller-supplied id
+ * @param what component kind, for the message (e.g. "fetcher")
+ * @param where origin, for the message (e.g. "configuration" or "a
request")
+ * @return the normalized id
+ * @throws IllegalArgumentException if the id is blank, illegal, or
system-reserved
+ */
+ public static String requireUserId(String id, String what, String where) {
+ String normalized = requireLegalId(id, what, where);
+ if (normalized.startsWith(SYSTEM_PREFIX)) {
+ throw new IllegalArgumentException("'" + normalized + "' is
reserved: " + what
+ + " ids beginning with '" + SYSTEM_PREFIX + "' are for
internal use and may not"
+ + " be set from " + where + ".");
+ }
+ return normalized;
+ }
+
+ /**
+ * As {@link #requireUserId}, but permits the system namespace. For ids
the host itself
+ * supplies, where the charset rule still applies but {@code __} is
legitimate.
+ *
+ * @throws IllegalArgumentException if the id is blank or outside {@link
#LEGAL_ID}
+ */
+ public static String requireLegalId(String id, String what, String where) {
+ String normalized = normalize(id);
+ if (normalized == null || normalized.isEmpty()) {
+ throw new IllegalArgumentException("Blank " + what + " id in " +
where + ".");
+ }
+ if (!LEGAL_ID.matcher(normalized).matches()) {
+ throw new IllegalArgumentException("Illegal " + what + " id '" +
normalized + "' in "
+ + where + ". Ids may contain only letters, digits, '.',
'_' and '-'.");
+ }
+ return normalized;
+ }
+}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/AbstractComponentManager.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/AbstractComponentManager.java
index d6fbc04630..66da622823 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/AbstractComponentManager.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/AbstractComponentManager.java
@@ -19,10 +19,12 @@ package org.apache.tika.pipes.core;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.LinkedHashSet;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
@@ -33,6 +35,7 @@ import org.slf4j.LoggerFactory;
import org.apache.tika.config.loader.TikaObjectMapperFactory;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
+import org.apache.tika.pipes.api.ComponentIds;
import org.apache.tika.pipes.core.config.ConfigStore;
import org.apache.tika.pipes.core.config.InMemoryConfigStore;
import org.apache.tika.plugins.ExtensionConfig;
@@ -134,10 +137,20 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
Iterator<Map.Entry<String, JsonNode>> instanceFields =
configNode.fields();
while (instanceFields.hasNext()) {
Map.Entry<String, JsonNode> instanceEntry =
instanceFields.next();
- String instanceId = instanceEntry.getKey();
+ // Normalize before validating and storing: {"foo": ...} and
{"foo ": ...} are two
+ // distinct keys in valid JSON, and validating the raw id
while resolving the
+ // trimmed one would let " __foo" past the reserved-namespace
check.
+ // __ is permitted here -- ConfigMerger folds host-injected
overrides into the same
+ // JSON before this runs, and those legitimately use it.
+ String instanceId;
+ try {
+ instanceId =
ComponentIds.requireLegalId(instanceEntry.getKey(),
+ getComponentName(), "configuration");
+ } catch (IllegalArgumentException e) {
+ throw new TikaConfigException(e.getMessage(), e);
+ }
JsonNode typeNode = instanceEntry.getValue();
- // Check for duplicate IDs (should not happen due to JSON
parsing, but validate)
if (configs.containsKey(instanceId)) {
throw new TikaConfigException("Duplicate " +
getComponentName() +
" id: " + instanceId);
@@ -225,6 +238,10 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
* Gets a component by ID, lazily instantiating it if needed.
*/
public T getComponent(String id) throws IOException, TikaException {
+ // Canonicalize on the way in so lookup agrees with what load()
stored. __ is not refused
+ // here: the host's own components resolve through this method.
+ id = ComponentIds.normalize(id);
+
// Check cache first (fast path, no synchronization)
T component = componentCache.get(id);
if (component != null) {
@@ -236,7 +253,7 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
if (config == null) {
throw createNotFoundException(
"Can't find " + getComponentName() + " for id=" + id +
- ". Available: " + configStore.keySet());
+ ". Available: " + getSupported());
}
// Synchronized block to ensure only one thread builds the component
@@ -301,7 +318,16 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
throw new IllegalArgumentException("ExtensionConfig cannot be
null");
}
- String componentId = config.id();
+ // Runtime registration is remote input (grpc saveFetcher), so unlike
config load it must
+ // not be able to define or replace a host component.
+ String componentId;
+ try {
+ componentId = ComponentIds.requireUserId(config.id(),
getComponentName(),
+ "a runtime registration");
+ } catch (IllegalArgumentException e) {
+ throw new TikaConfigException(e.getMessage(), e);
+ }
+ config = new ExtensionConfig(componentId, config.name(),
config.json());
String typeName = config.name();
// Validate that factory exists for this type
@@ -342,6 +368,14 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
if (componentId == null) {
throw new IllegalArgumentException("Component ID cannot be null");
}
+ // As in saveComponent: a remote caller must not delete a host
component out from under
+ // an in-flight request.
+ try {
+ componentId = ComponentIds.requireUserId(componentId,
getComponentName(),
+ "a runtime deletion");
+ } catch (IllegalArgumentException e) {
+ throw new TikaConfigException(e.getMessage(), e);
+ }
if (!configStore.containsKey(componentId)) {
throw new TikaConfigException(
@@ -362,13 +396,24 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
* @return the component configuration, or null if not found
*/
public ExtensionConfig getComponentConfig(String componentId) {
- return configStore.get(componentId);
+ return configStore.get(ComponentIds.normalize(componentId));
}
/**
- * Returns the set of supported component IDs.
+ * Returns the component IDs a caller may name. Host components are
omitted: a request cannot
+ * bind them, so listing them in a "not found" message only advertises
internals.
*/
public Set<String> getSupported() {
+ return configStore.keySet().stream()
+ .filter(id -> !ComponentIds.isSystem(id))
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ }
+
+ /**
+ * Every configured id, host components included. For lifecycle and
existence checks, not for
+ * anything a caller sees.
+ */
+ public Set<String> getAllIds() {
return configStore.keySet();
}
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 82bbb80ab8..fc106b7f1f 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,6 +23,8 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.util.StdConverter;
import org.apache.tika.config.TimeoutLimits;
import org.apache.tika.config.loader.TikaJsonConfig;
@@ -32,11 +34,32 @@ import org.apache.tika.pipes.api.ParseMode;
import org.apache.tika.pipes.core.protocol.PipesMessage;
import org.apache.tika.pipes.core.server.ServerProtocolIO;
+// Cross-field limits are checked after binding, so JSON key order cannot
change the outcome.
+@JsonDeserialize(converter = PipesConfig.PostDeserializationCheck.class)
public class PipesConfig {
+ /** Runs {@link #checkPayloadLimits()} on every Jackson deserialization
path. */
+ public static class PostDeserializationCheck extends
StdConverter<PipesConfig, PipesConfig> {
+ @Override
+ public PipesConfig convert(PipesConfig config) {
+ config.checkPayloadLimits();
+ return config;
+ }
+ }
+
public static final int DEFAULT_MAX_IPC_PAYLOAD_BYTES =
PipesMessage.MAX_PAYLOAD_BYTES;
+ /**
+ * Largest request body carried inline to the forked worker rather than
spooled to disk.
+ * <p>
+ * Sized for the common case -- most documents are far smaller -- because
the cost is heap,
+ * not disk: the parent holds the payload and the Smile frame containing a
copy of it, and the
+ * child holds it again. Budget roughly {@code 2 * maxInlineBytes *
concurrent-requests} in the
+ * parent before raising this.
+ */
+ public static final int DEFAULT_MAX_INLINE_BYTES = 10 * 1024 * 1024;
+
public static final long DEFAULT_SHUTDOWN_CLIENT_AFTER_MILLIS = 300000;
/** Past this, worker count becomes a memory decision, and memory is not
visible here. */
@@ -83,6 +106,7 @@ public class PipesConfig {
private boolean useSharedServer = DEFAULT_USE_SHARED_SERVER;
private int maxIpcPayloadBytes = DEFAULT_MAX_IPC_PAYLOAD_BYTES;
+ private int maxInlineBytes = DEFAULT_MAX_INLINE_BYTES;
private long socketTimeoutMillis = DEFAULT_SOCKET_TIMEOUT_MILLIS;
private long startupTimeoutMillis = DEFAULT_STARTUP_TIMEOUT_MILLIS;
@@ -546,6 +570,30 @@ public class PipesConfig {
return maxIpcPayloadBytes;
}
+ /**
+ * @return largest request body sent inline instead of spooled; see
+ * {@link #DEFAULT_MAX_INLINE_BYTES}
+ */
+ public int getMaxInlineBytes() {
+ return maxInlineBytes;
+ }
+
+ /**
+ * Sets the inline-payload threshold. Must stay under {@code
maxIpcPayloadBytes}: the payload
+ * travels inside the NEW_REQUEST frame, so a threshold above that limit
would let the parent
+ * build requests the child refuses, surfacing as an undiagnosable crash
rather than a clean
+ * fallback to spooling. The pair is checked in {@link
#checkPayloadLimits()}, not here, so
+ * the two fields may be set in either order.
+ *
+ * @throws IllegalArgumentException if negative
+ */
+ public void setMaxInlineBytes(int maxInlineBytes) {
+ if (maxInlineBytes < 0) {
+ throw new IllegalArgumentException("maxInlineBytes must be >= 0,
got: " + maxInlineBytes);
+ }
+ this.maxInlineBytes = maxInlineBytes;
+ }
+
/**
* 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
@@ -570,4 +618,20 @@ public class PipesConfig {
}
this.maxIpcPayloadBytes = maxIpcPayloadBytes;
}
+
+ /**
+ * Checks that {@code maxInlineBytes} leaves headroom for the rest of the
tuple (metadata,
+ * parseContext) inside {@code maxIpcPayloadBytes}. Runs automatically
after Jackson
+ * deserialization; call it directly after configuring an instance through
setters.
+ *
+ * @throws IllegalArgumentException if the pair is inconsistent
+ */
+ public void checkPayloadLimits() {
+ long ceiling = maxIpcPayloadBytes - (maxIpcPayloadBytes / 10);
+ if (maxInlineBytes > ceiling) {
+ throw new IllegalArgumentException("maxInlineBytes (" +
maxInlineBytes +
+ ") must leave room for the rest of the request inside
maxIpcPayloadBytes (" +
+ maxIpcPayloadBytes + "); keep it at or below " + ceiling);
+ }
+ }
}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/ConfigMerger.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/ConfigMerger.java
index e4bf01846d..506e14577d 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/ConfigMerger.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/ConfigMerger.java
@@ -20,6 +20,7 @@ import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -33,6 +34,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.tika.config.TimeoutLimits;
+import org.apache.tika.pipes.api.ComponentIds;
/**
* Utility for merging configuration overrides with existing Tika JSON
configuration.
@@ -105,6 +107,12 @@ public class ConfigMerger {
LOG.debug("Creating new config (no existing config provided)");
}
+ // Refuse user-authored __ ids before any override is folded in. This
is the only point
+ // where the user's config and the host's overrides are still
distinguishable -- afterwards
+ // they are one JSON tree, and getOrCreateObject would silently merge
a user entry named
+ // __tika-server into the host's rather than failing.
+ rejectSystemIdsInUserConfig(root, existingConfig);
+
// Generate UUID for internal components
String uuid = UUID.randomUUID().toString();
@@ -118,7 +126,7 @@ public class ConfigMerger {
for (ConfigOverrides.FetcherOverride fetcher :
overrides.getFetchers()) {
String fetcherId = fetcher.getId();
if (fetcherId == null || fetcherId.isEmpty()) {
- fetcherId = "tika-internal-fetcher-" + uuid;
+ fetcherId = GENERATED_FETCHER_PREFIX + uuid;
}
generatedFetcherIds.add(fetcherId);
@@ -136,7 +144,7 @@ public class ConfigMerger {
for (ConfigOverrides.EmitterOverride emitter :
overrides.getEmitters()) {
String emitterId = emitter.getId();
if (emitterId == null || emitterId.isEmpty()) {
- emitterId = "tika-internal-emitter-" + uuid;
+ emitterId = GENERATED_EMITTER_PREFIX + uuid;
}
generatedEmitterIds.add(emitterId);
@@ -216,6 +224,40 @@ public class ConfigMerger {
/**
* Gets or creates an ObjectNode child of the parent.
*/
+ /**
+ * Ids for host components the caller did not name (PipesForkParser's
fetcher, for one). They
+ * live in the reserved namespace like every other host-wired component:
the UUID makes them
+ * unguessable, but that is not the same as being unnameable.
+ */
+ public static final String GENERATED_FETCHER_PREFIX =
+ ComponentIds.SYSTEM_PREFIX + "tika-internal-fetcher-";
+
+ public static final String GENERATED_EMITTER_PREFIX =
+ ComponentIds.SYSTEM_PREFIX + "tika-internal-emitter-";
+
+ /** Component sections whose keys are instance ids. */
+ private static final String[] ID_KEYED_SECTIONS = {"fetchers", "emitters",
"pipes-iterators"};
+
+ private static void rejectSystemIdsInUserConfig(ObjectNode root, Path
source)
+ throws IOException {
+ for (String section : ID_KEYED_SECTIONS) {
+ JsonNode node = root.get(section);
+ if (node == null || !node.isObject()) {
+ continue;
+ }
+ for (Iterator<String> it = node.fieldNames(); it.hasNext(); ) {
+ String id = it.next();
+ if (ComponentIds.isSystem(id)) {
+ throw new IOException("'" + id.trim() + "' in '" + section
+ "'"
+ + (source == null ? "" : " of " + source)
+ + " is reserved: ids beginning with '" +
ComponentIds.SYSTEM_PREFIX
+ + "' name components the host wires up for itself
and may not be"
+ + " defined in a config file.");
+ }
+ }
+ }
+ }
+
private static ObjectNode getOrCreateObject(ObjectMapper mapper,
ObjectNode parent, String key) {
if (parent.has(key) && parent.get(key).isObject()) {
return (ObjectNode) parent.get(key);
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
new file mode 100644
index 0000000000..4a85f80210
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/BytesFetcher.java
@@ -0,0 +1,67 @@
+/*
+ * 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.fetcher;
+
+import java.io.IOException;
+
+import org.apache.tika.exception.TikaException;
+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.ComponentIds;
+import org.apache.tika.pipes.api.fetcher.Fetcher;
+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
+ * already holds the content does not have to spool it to disk purely to hand
it across the
+ * process boundary.
+ * <p>
+ * The fetch key is not a location -- it carries the caller's filename, which
is why the parent
+ * no longer has to scrub a spool filename out of the returned metadata.
+ * <p>
+ * {@link org.apache.tika.pipes.core.fetcher.FetcherManager} hands out one
instance per id and
+ * requires thread safety, so the payload is read from the {@code
parseContext} argument on every
+ * call and never held as state.
+ */
+public class BytesFetcher implements Fetcher {
+
+ /** Reserved: a request cannot name this, so only the host can route a
parse through it. */
+ public static final String FETCHER_ID = ComponentIds.SYSTEM_PREFIX +
"bytes";
+
+ @Override
+ public TikaInputStream fetch(String fetchKey, Metadata metadata,
ParseContext parseContext)
+ throws TikaException, IOException {
+ InlineBytes inline = parseContext == null ? null :
parseContext.get(InlineBytes.class);
+ if (inline == null || inline.getBytes() == null) {
+ throw new IOException("no inline-bytes in the ParseContext; " +
FETCHER_ID +
+ " is only usable on a request that carries its content
inline");
+ }
+ if (!StringUtils.isBlank(fetchKey)
+ && metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY) == null)
{
+ metadata.set(TikaCoreProperties.RESOURCE_NAME_KEY, fetchKey);
+ }
+ return TikaInputStream.get(inline.getBytes(), metadata);
+ }
+
+ @Override
+ public ExtensionConfig getExtensionConfig() {
+ return new ExtensionConfig(FETCHER_ID, "bytes-fetcher", null);
+ }
+}
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
new file mode 100644
index 0000000000..2981847f49
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/InlineBytes.java
@@ -0,0 +1,92 @@
+/*
+ * 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.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.
+ */
+@TikaComponent(name = "inline-bytes", spi = false)
+public class InlineBytes implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private byte[] bytes;
+
+ public InlineBytes() {
+ }
+
+ public InlineBytes(byte[] bytes) {
+ this.bytes = bytes;
+ }
+
+ public byte[] getBytes() {
+ return bytes;
+ }
+
+ public void setBytes(byte[] bytes) {
+ this.bytes = bytes;
+ }
+
+ public int length() {
+ return bytes == null ? 0 : bytes.length;
+ }
+
+ /**
+ * Value equality on the payload. {@code Objects.equals} would compare
array identity here,
+ * which would silently make two tuples carrying identical content unequal.
+ */
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ return Arrays.equals(bytes, ((InlineBytes) o).bytes);
+ }
+
+ /**
+ * Length only. {@code FetchEmitTuple.hashCode()} hashes its ParseContext,
so hashing the
+ * payload itself would walk megabytes on every map insert;
unequal-hash-implies-unequal
+ * still holds, and collisions fall through to {@link #equals}.
+ */
+ @Override
+ public int hashCode() {
+ return length();
+ }
+
+ /**
+ * Length only -- {@code FetchEmitTuple.toString()} prints its
ParseContext, and a debug log
+ * of a multi-megabyte payload is a real operational hazard.
+ */
+ @Override
+ public String toString() {
+ return "InlineBytes{length=" + length() + "}";
+ }
+}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/PayloadRouter.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/PayloadRouter.java
new file mode 100644
index 0000000000..8a3edffe74
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/PayloadRouter.java
@@ -0,0 +1,155 @@
+/*
+ * 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.fetcher;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import org.apache.commons.io.IOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.tika.io.BoundedInputStream;
+import org.apache.tika.io.TikaInputStream;
+
+/**
+ * Decides how a document reaches the forked worker: as bytes on the wire, or
as a file the
+ * worker opens itself.
+ * <p>
+ * Every host that hands pipes an already-open stream needs this same
decision, so it lives here
+ * rather than in each of them -- three copies would be three thresholds that
drift apart.
+ */
+public final class PayloadRouter {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PayloadRouter.class);
+
+ public enum Route {
+ /** Already on disk; the worker opens that file and nothing is copied.
*/
+ EXISTING_FILE,
+ /** Small enough to travel inside the request. */
+ INLINE,
+ /** Too large to inline; written once to a file the worker can reach.
*/
+ SPOOLED
+ }
+
+ /** Creates the file a SPOOLED payload is written to. */
+ @FunctionalInterface
+ public interface SpoolTarget {
+ Path create() throws IOException;
+ }
+
+ private PayloadRouter() {
+ }
+
+ /**
+ * Routes {@code tis}, reading no more than it must to decide.
+ * <p>
+ * A stream that already has a file keeps it: reading an on-disk document
into heap to push it
+ * through a socket is strictly worse than letting the worker open the
file. Otherwise the
+ * decision is made from bytes actually read -- a declared length is
absent under chunked
+ * transfer encoding and client-supplied besides.
+ *
+ * @param tis the content; not closed here
+ * @param maxInlineBytes largest payload carried inline; 0 disables
inlining
+ * @param spoolTarget invoked only if the content exceeds {@code
maxInlineBytes}
+ */
+ public static Routed route(TikaInputStream tis, int maxInlineBytes,
SpoolTarget spoolTarget)
+ throws IOException {
+ if (tis.hasFile()) {
+ return new Routed(Route.EXISTING_FILE, null, tis.getPath(), false);
+ }
+ return route((InputStream) tis, maxInlineBytes, spoolTarget);
+ }
+
+ /**
+ * The outcome. Closing deletes the spool file if one was created; an
EXISTING_FILE path
+ * belongs to the caller's stream and is left alone.
+ */
+ public static final class Routed implements Closeable {
+
+ private final Route route;
+ private final InlineBytes inlineBytes;
+ private final Path path;
+ private final boolean ownsPath;
+
+ private Routed(Route route, InlineBytes inlineBytes, Path path,
boolean ownsPath) {
+ this.route = route;
+ this.inlineBytes = inlineBytes;
+ this.path = path;
+ this.ownsPath = ownsPath;
+ }
+
+ public Route route() {
+ return route;
+ }
+
+ public boolean isInline() {
+ return route == Route.INLINE;
+ }
+
+ /** Non-null exactly when {@link #isInline()}. */
+ public InlineBytes inlineBytes() {
+ return inlineBytes;
+ }
+
+ /** Non-null for EXISTING_FILE and SPOOLED. */
+ public Path path() {
+ return path;
+ }
+
+ @Override
+ public void close() {
+ if (!ownsPath || path == null) {
+ return;
+ }
+ try {
+ Files.deleteIfExists(path);
+ } catch (IOException e) {
+ LOG.warn("Failed to delete spooled input: {}", path, e);
+ }
+ }
+ }
+
+ /** Reads a stream the same way {@link #route} does; for callers holding a
plain stream. */
+ public static Routed route(InputStream is, int maxInlineBytes, SpoolTarget
spoolTarget)
+ throws IOException {
+ byte[] head = IOUtils.toByteArray(new BoundedInputStream((long)
maxInlineBytes + 1, is));
+ if (head.length <= maxInlineBytes) {
+ return new Routed(Route.INLINE, new InlineBytes(head), null,
false);
+ }
+ Path target = spoolTarget.create();
+ try (OutputStream out = Files.newOutputStream(target)) {
+ // head was already consumed off the stream; writing it back first
is what keeps
+ // the spooled file the whole document rather than everything past
the threshold.
+ out.write(head);
+ IOUtils.copy(is, out);
+ } catch (IOException e) {
+ // No Routed exists yet, so nobody else can delete the partial
file.
+ try {
+ Files.deleteIfExists(target);
+ } catch (IOException suppressed) {
+ e.addSuppressed(suppressed);
+ }
+ throw e;
+ }
+ return new Routed(Route.SPOOLED, null, target, true);
+ }
+}
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 bc26f26b49..07bc0c6f09 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
@@ -41,6 +41,7 @@ import com.fasterxml.jackson.databind.JsonNode;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.parser.ParseContext;
+import org.apache.tika.pipes.api.ComponentIds;
import org.apache.tika.pipes.api.FetchEmitTuple;
import org.apache.tika.pipes.api.emitter.EmitKey;
import org.apache.tika.pipes.api.fetcher.FetchKey;
@@ -52,15 +53,39 @@ public class FetchEmitTupleDeserializer extends
JsonDeserializer<FetchEmitTuple>
ID, FETCHER, FETCH_KEY, EMITTER, EMIT_KEY, FETCH_RANGE_START,
FETCH_RANGE_END,
METADATA_KEY, PARSE_CONTEXT, ON_PARSE_EXCEPTION);
+ private final boolean restricted;
+
+ /**
+ * Deserializer for untrusted input: refuses system component ids. This is
the default
+ * because forgetting to restrict a new caller must not be the thing that
opens the gate.
+ */
+ public FetchEmitTupleDeserializer() {
+ this(true);
+ }
+
+ private FetchEmitTupleDeserializer(boolean restricted) {
+ this.restricted = restricted;
+ }
+
+ /**
+ * Deserializer for the parent-to-child IPC, whose tuples the host builds
itself and which
+ * must therefore be able to name {@code __} components. The parseContext
stays restricted in
+ * both modes, so a request that slipped past the REST gate still cannot
bind a wire-blocked
+ * component here.
+ */
+ public static FetchEmitTupleDeserializer internal() {
+ return new FetchEmitTupleDeserializer(false);
+ }
+
@Override
public FetchEmitTuple deserialize(JsonParser jsonParser,
DeserializationContext deserializationContext) throws IOException,
JacksonException {
JsonNode root = jsonParser.readValueAsTree();
rejectUnknownKeys(root);
String id = readVal(ID, root, null, true);
- String fetcherId = readVal(FETCHER, root, null, true);
+ String fetcherId = normalizeId(readVal(FETCHER, root, null, true),
"fetcher");
String fetchKey = readVal(FETCH_KEY, root, null, true);
- String emitterName = readVal(EMITTER, root, "", false);
+ String emitterName = normalizeId(readVal(EMITTER, root, "", false),
"emitter");
String emitKey = readVal(EMIT_KEY, root, "", false);
long fetchRangeStart = readLong(FETCH_RANGE_START, root, -1l, false);
long fetchRangeEnd = readLong(FETCH_RANGE_END, root, -1l, false);
@@ -77,6 +102,23 @@ public class FetchEmitTupleDeserializer extends
JsonDeserializer<FetchEmitTuple>
onParseException);
}
+ /**
+ * An absent emitter is the empty string, so blank passes through
untouched; anything else is
+ * canonicalized here so the id that is checked is the id later used for
lookup.
+ */
+ private String normalizeId(String id, String what) throws IOException {
+ if (id == null || id.isBlank()) {
+ return id;
+ }
+ try {
+ return restricted
+ ? ComponentIds.requireUserId(id, what, "a request")
+ : ComponentIds.requireLegalId(id, what, "an internal
tuple");
+ } catch (IllegalArgumentException e) {
+ throw new IOException(e.getMessage(), e);
+ }
+ }
+
private static void rejectUnknownKeys(JsonNode root) throws IOException {
for (Iterator<String> it = root.fieldNames(); it.hasNext(); ) {
String name = it.next();
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 1de3451cc2..d35f45a66d 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
@@ -58,7 +58,8 @@ public class JsonPipesIpc {
// Add pipes-specific serializers
SimpleModule pipesModule = new SimpleModule();
pipesModule.addSerializer(FetchEmitTuple.class, new
FetchEmitTupleSerializer());
- pipesModule.addDeserializer(FetchEmitTuple.class, new
FetchEmitTupleDeserializer());
+ // Parent-to-child IPC: the host builds these tuples itself and they
name __ components.
+ pipesModule.addDeserializer(FetchEmitTuple.class,
FetchEmitTupleDeserializer.internal());
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/server/EmitHandler.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java
index 924c22c8c5..8458e7ceec 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java
@@ -127,9 +127,11 @@ class EmitHandler {
// For CONTENT_ONLY mode: force direct emission only when a real
emitter is configured,
// so the content is emitted as a raw byte stream via
emitContentOnly() instead of
// being returned as passback JSON to the parent AsyncEmitter.
+ // getAllIds, not getSupported: this is an existence check, and a
host emitter
+ // (__unpack) is configured but deliberately absent from the
caller-facing list.
boolean forceEmit = parseMode == ParseMode.CONTENT_ONLY
&& emitterId != null
- && emitterManager.getSupported().contains(emitterId);
+ && emitterManager.getAllIds().contains(emitterId);
boolean willEmit = forceEmit || shouldEmit(parseMode, parseData,
emitDataTuple, parseContext);
if (willEmit) {
return emit(t.getId(), emitKey, parseMode == ParseMode.UNPACK,
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java
index 1abbe78031..e5116b63bb 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java
@@ -29,6 +29,7 @@ import org.apache.tika.pipes.api.FetchEmitTuple;
import org.apache.tika.pipes.api.PipesResult;
import org.apache.tika.pipes.api.fetcher.Fetcher;
import org.apache.tika.pipes.api.fetcher.FetcherNotFoundException;
+import org.apache.tika.pipes.core.fetcher.BytesFetcher;
import org.apache.tika.pipes.core.fetcher.FetcherManager;
import org.apache.tika.utils.ExceptionUtils;
@@ -36,6 +37,8 @@ class FetchHandler {
private static final Logger LOG =
LoggerFactory.getLogger(FetchHandler.class);
+ private static final BytesFetcher BYTES_FETCHER = new BytesFetcher();
+
private final FetcherManager fetcherManager;
public FetchHandler(FetcherManager fetcherManager) {
@@ -57,8 +60,14 @@ class FetchHandler {
}
private FetcherOrResult getFetcher(FetchEmitTuple t) {
+ String fetcherId = t.getFetchKey().getFetcherId();
+ // Built in, not configured: the bytes come with the request, so there
is nothing for an
+ // operator to point at and no reason for it to occupy an id in the
ConfigStore.
+ if (BytesFetcher.FETCHER_ID.equals(fetcherId)) {
+ return new FetcherOrResult(BYTES_FETCHER, null);
+ }
try {
- return new
FetcherOrResult(fetcherManager.getFetcher(t.getFetchKey().getFetcherId()),
null);
+ return new FetcherOrResult(fetcherManager.getFetcher(fetcherId),
null);
} catch (FetcherNotFoundException e) {
String noFetcherMsg =
getNoFetcherMsg(t.getFetchKey().getFetcherId());
LOG.warn(noFetcherMsg);
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 c1a92265a8..937dbb174b 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
@@ -80,6 +80,62 @@ public class TikaPipesConfigTest extends TikaTest {
assertEquals(atMin, config.getMaxIpcPayloadBytes());
}
+ /**
+ * The inline/ipc pair is validated after binding, so an inline threshold
that is only
+ * legal because ipc was raised must load no matter which key comes first.
+ */
+ @Test
+ void testPayloadLimitPairIsKeyOrderIndependent() throws Exception {
+ String inlineFirst = """
+ {"pipes": {"maxInlineBytes": 99614720, "maxIpcPayloadBytes":
209715200}}
+ """;
+ String ipcFirst = """
+ {"pipes": {"maxIpcPayloadBytes": 209715200, "maxInlineBytes":
99614720}}
+ """;
+ for (String json : new String[]{inlineFirst, ipcFirst}) {
+ PipesConfig config = PipesConfig.load(TikaJsonConfig.load(
+ new
ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8))));
+ assertEquals(99614720, config.getMaxInlineBytes());
+ assertEquals(209715200, config.getMaxIpcPayloadBytes());
+ }
+ }
+
+ /** Lowering only ipc must re-check the untouched inline default (10MB >
90% of 5MB). */
+ @Test
+ void testLoweringIpcAloneRechecksInlineDefault() throws Exception {
+ String json = """
+ {"pipes": {"maxIpcPayloadBytes": 5242880}}
+ """;
+ TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(
+ new
ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8)));
+ assertThrows(Exception.class, () -> PipesConfig.load(tikaJsonConfig));
+ }
+
+ @Test
+ void testInconsistentPayloadPairRejectedInEitherOrder() throws Exception {
+ // inline 95MB against the default 100MB ipc limit fails the 10%
headroom rule
+ String inlineFirst = """
+ {"pipes": {"maxInlineBytes": 99614720}}
+ """;
+ String withExplicitIpc = """
+ {"pipes": {"maxIpcPayloadBytes": 104857600, "maxInlineBytes":
99614720}}
+ """;
+ for (String json : new String[]{inlineFirst, withExplicitIpc}) {
+ TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(
+ new
ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8)));
+ assertThrows(Exception.class, () ->
PipesConfig.load(tikaJsonConfig));
+ }
+ }
+
+ @Test
+ void testCheckPayloadLimitsForSetterBuiltConfigs() {
+ PipesConfig config = new PipesConfig();
+ config.setMaxInlineBytes(99614720);
+ assertThrows(IllegalArgumentException.class,
config::checkPayloadLimits);
+ config.setMaxIpcPayloadBytes(209715200);
+ config.checkPayloadLimits();
+ }
+
@Test
void testMaxIpcPayloadBytesFromJsonRejectsZero() throws Exception {
String json = """
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/config/ConfigMergerTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/config/ConfigMergerTest.java
index fea3fcbfae..c1471ce3fe 100644
---
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/config/ConfigMergerTest.java
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/config/ConfigMergerTest.java
@@ -20,6 +20,7 @@ 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;
@@ -150,7 +151,8 @@ public class ConfigMergerTest {
ConfigMerger.MergeResult result = ConfigMerger.mergeOrCreate(null,
overrides);
assertNotNull(result.fetcherId());
- assertTrue(result.fetcherId().startsWith("tika-internal-fetcher-"));
+
assertTrue(result.fetcherId().startsWith(ConfigMerger.GENERATED_FETCHER_PREFIX),
+ result.fetcherId());
// Verify fetcher exists with generated ID
ObjectMapper mapper = new ObjectMapper();
@@ -317,4 +319,38 @@ public class ConfigMergerTest {
Files.deleteIfExists(result.configPath());
}
+
+ /**
+ * A user config naming a host component would otherwise be merged into
the host's own entry
+ * by getOrCreateObject -- silently repointing, say, tika-server's upload
fetcher rather than
+ * failing. This is the only point where user config and host overrides
are still separable.
+ */
+ @Test
+ public void testUserConfigCannotDefineSystemId() throws IOException {
+ Path userConfig = tempDir.resolve("user-config.json");
+ Files.writeString(userConfig,
+
"{\"fetchers\":{\"__tika-server\":{\"file-system-fetcher\":{\"basePath\":\"/evil\"}}}}");
+
+ ConfigOverrides overrides = ConfigOverrides.builder()
+ .addFetcher("__tika-server", "file-system-fetcher",
Map.of("basePath", "/spool"))
+ .build();
+
+ IOException e = assertThrows(IOException.class,
+ () -> ConfigMerger.mergeOrCreate(userConfig, overrides));
+ assertTrue(e.getMessage().contains("is reserved"), e.getMessage());
+ }
+
+ /** The host's own overrides must still be able to use the namespace. */
+ @Test
+ public void testHostOverrideMayUseSystemId() throws IOException {
+ ConfigOverrides overrides = ConfigOverrides.builder()
+ .addFetcher("__tika-server", "file-system-fetcher",
Map.of("basePath", "/spool"))
+ .build();
+
+ ConfigMerger.MergeResult result = ConfigMerger.mergeOrCreate(null,
overrides);
+ JsonNode root = new
ObjectMapper().readTree(result.configPath().toFile());
+ assertTrue(root.get("fetchers").has("__tika-server"));
+
+ Files.deleteIfExists(result.configPath());
+ }
}
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/fetcher/PayloadRouterTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/fetcher/PayloadRouterTest.java
new file mode 100644
index 0000000000..2a64a92a84
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/fetcher/PayloadRouterTest.java
@@ -0,0 +1,228 @@
+/*
+ * 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.fetcher;
+
+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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.io.TikaInputStream;
+
+/**
+ * The inline-vs-spool decision has to be made from bytes actually read -- a
declared length is
+ * absent under chunked transfer encoding and client-supplied besides -- so
the boundary
+ * arithmetic and the draining of the already-read prefix are where this can
silently truncate a
+ * document.
+ */
+public class PayloadRouterTest {
+
+ @TempDir
+ Path tmp;
+
+ private static byte[] body(int n) {
+ byte[] b = new byte[n];
+ for (int i = 0; i < n; i++) {
+ b[i] = (byte) (i % 251);
+ }
+ return b;
+ }
+
+ private PayloadRouter.SpoolTarget target(AtomicInteger calls) {
+ return () -> {
+ calls.incrementAndGet();
+ return Files.createTempFile(tmp, "spool-", ".bin");
+ };
+ }
+
+ private PayloadRouter.Routed route(byte[] content, int threshold,
AtomicInteger calls)
+ throws IOException {
+ try (TikaInputStream tis = TikaInputStream.get(new
ByteArrayInputStream(content))) {
+ return PayloadRouter.route(tis, threshold, target(calls));
+ }
+ }
+
+ @Test
+ public void underThresholdIsInlined() throws IOException {
+ AtomicInteger spools = new AtomicInteger();
+ byte[] expected = body(99);
+ try (PayloadRouter.Routed r = route(expected, 100, spools)) {
+ assertTrue(r.isInline());
+ assertArrayEquals(expected, r.inlineBytes().getBytes());
+ assertNull(r.path());
+ assertEquals(0, spools.get(), "should not have created a spool
file");
+ }
+ }
+
+ /** Exactly at the threshold still inlines; only strictly larger spills. */
+ @Test
+ public void atThresholdIsInlined() throws IOException {
+ AtomicInteger spools = new AtomicInteger();
+ byte[] expected = body(100);
+ try (PayloadRouter.Routed r = route(expected, 100, spools)) {
+ assertTrue(r.isInline());
+ assertArrayEquals(expected, r.inlineBytes().getBytes());
+ assertEquals(0, spools.get());
+ }
+ }
+
+ /**
+ * One byte over spills -- and the spilled file must hold the whole
document, not just what
+ * was left after the threshold probe already consumed the head.
+ */
+ @Test
+ public void oneOverThresholdSpillsWholeBody() throws IOException {
+ AtomicInteger spools = new AtomicInteger();
+ byte[] expected = body(101);
+ try (PayloadRouter.Routed r = route(expected, 100, spools)) {
+ assertFalse(r.isInline());
+ assertEquals(1, spools.get());
+ assertArrayEquals(expected, Files.readAllBytes(r.path()));
+ }
+ }
+
+ @Test
+ public void largeBodySpillsWholeBody() throws IOException {
+ AtomicInteger spools = new AtomicInteger();
+ byte[] expected = body(64 * 1024 + 7);
+ try (PayloadRouter.Routed r = route(expected, 4096, spools)) {
+ assertFalse(r.isInline());
+ assertEquals(expected.length, Files.size(r.path()));
+ assertArrayEquals(expected, Files.readAllBytes(r.path()));
+ }
+ }
+
+ /**
+ * TikaInputStream spills its own cache to disk at 1MB, which would mean a
disk write we
+ * thought we had avoided plus a heap copy on top. That cache is opt-in
(enableRewind), so
+ * routing a body larger than 1MB but under the inline threshold must
still inline and must
+ * leave the stream with no file behind it. If someone later rewinds
before routing, this
+ * fails rather than silently reintroducing the write.
+ */
+ @Test
+ public void aboveTikaInputStreamCacheThresholdStillInlines() throws
IOException {
+ AtomicInteger spools = new AtomicInteger();
+ byte[] expected = body(3 * 1024 * 1024);
+ try (TikaInputStream tis = TikaInputStream.get(new
ByteArrayInputStream(expected));
+ PayloadRouter.Routed r =
+ PayloadRouter.route(tis, 10 * 1024 * 1024,
target(spools))) {
+ assertTrue(r.isInline(), "3MB should inline under a 10MB
threshold");
+ assertEquals(0, spools.get(), "no spool file should have been
created");
+ assertFalse(tis.hasFile(), "TikaInputStream must not have spilled
its cache to disk");
+ assertArrayEquals(expected, r.inlineBytes().getBytes());
+ }
+ }
+
+ /** A file-backed stream keeps its file: reading it into heap would be
strictly worse. */
+ @Test
+ public void fileBackedStreamKeepsItsFile() throws IOException {
+ AtomicInteger spools = new AtomicInteger();
+ Path existing = Files.createTempFile(tmp, "existing-", ".bin");
+ byte[] expected = body(50);
+ Files.write(existing, expected);
+
+ try (TikaInputStream tis = TikaInputStream.get(existing);
+ PayloadRouter.Routed r = PayloadRouter.route(tis, 10_000,
target(spools))) {
+ assertEquals(PayloadRouter.Route.EXISTING_FILE, r.route());
+ assertEquals(existing, r.path());
+ assertEquals(0, spools.get());
+ }
+ assertTrue(Files.exists(existing), "must not delete a file it did not
create");
+ }
+
+ @Test
+ public void spooledFileIsDeletedOnClose() throws IOException {
+ AtomicInteger spools = new AtomicInteger();
+ Path spooled;
+ try (PayloadRouter.Routed r = route(body(500), 100, spools)) {
+ spooled = r.path();
+ assertTrue(Files.exists(spooled));
+ }
+ assertFalse(Files.exists(spooled), "spool file should be deleted on
close");
+ }
+
+ /** A zero threshold turns inlining off. */
+ @Test
+ public void zeroThresholdAlwaysSpills() throws IOException {
+ AtomicInteger spools = new AtomicInteger();
+ byte[] expected = body(1);
+ try (PayloadRouter.Routed r = route(expected, 0, spools)) {
+ assertFalse(r.isInline());
+ assertArrayEquals(expected, Files.readAllBytes(r.path()));
+ }
+ }
+
+ @Test
+ public void emptyBodyIsInlined() throws IOException {
+ AtomicInteger spools = new AtomicInteger();
+ try (PayloadRouter.Routed r = route(new byte[0], 100, spools)) {
+ assertTrue(r.isInline());
+ assertEquals(0, r.inlineBytes().length());
+ }
+ }
+
+ /**
+ * If the source dies after the spool file is created, no Routed exists
yet, so route()
+ * itself must delete the partial file.
+ */
+ @Test
+ public void sourceFailureDuringSpoolDeletesPartialFile() throws
IOException {
+ AtomicInteger spools = new AtomicInteger();
+ InputStream failing = new InputStream() {
+ private int count = 0;
+
+ @Override
+ public int read() throws IOException {
+ if (count < 200) {
+ count++;
+ return 'x';
+ }
+ throw new IOException("source died mid-stream");
+ }
+ };
+ try (TikaInputStream tis = TikaInputStream.get(failing)) {
+ assertThrows(IOException.class, () -> PayloadRouter.route(tis,
100, target(spools)));
+ }
+ assertEquals(1, spools.get(), "spool file should have been created
before the failure");
+ try (var files = Files.list(tmp)) {
+ assertEquals(0, files.count(), "partial spool file must be
deleted");
+ }
+ }
+
+ @Test
+ public void plainStreamOverloadRoutesTheSameWay() throws IOException {
+ AtomicInteger spools = new AtomicInteger();
+ byte[] expected = body(101);
+ try (InputStream is = new ByteArrayInputStream(expected);
+ PayloadRouter.Routed r = PayloadRouter.route(is, 100,
target(spools))) {
+ assertFalse(r.isInline());
+ assertArrayEquals(expected, Files.readAllBytes(r.path()));
+ }
+ }
+}
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
new file mode 100644
index 0000000000..574bd351b8
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/SystemComponentIdWireTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.StringReader;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.smile.SmileFactory;
+import org.junit.jupiter.api.Test;
+
+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.emitter.EmitKey;
+import org.apache.tika.pipes.api.fetcher.FetchKey;
+
+/**
+ * A request must not be able to name a {@code __} component -- that is what
keeps /pipes and
+ * /async away from the fetchers and emitters tika-server wires up for /tika
and /unpack. The
+ * parent-to-child IPC builds those same tuples itself and must still be able
to name them.
+ */
+public class SystemComponentIdWireTest {
+
+ private static String tuple(String fetcherId, String emitterId) {
+ return "{\"id\":\"t\",\"fetcher\":\"" + fetcherId +
"\",\"fetchKey\":\"k\"," +
+ "\"emitter\":\"" + emitterId + "\",\"emitKey\":\"ek\"}";
+ }
+
+ /**
+ * The surrounding-whitespace cases are the point of normalizing before
validating: checking
+ * the raw id and resolving the trimmed one would let these bind the
component they name.
+ */
+ @Test
+ public void requestCannotNameSystemFetcher() {
+ for (String fetcherId : new String[]{"__tika-server", "
__tika-server", "__tika-server ",
+ "\\t__unpack"}) {
+ Exception e = assertThrows(Exception.class,
+ () -> JsonFetchEmitTuple.fromJson(new
StringReader(tuple(fetcherId, "e"))),
+ "should have rejected fetcher id '" + fetcherId + "'");
+ assertTrue(root(e).contains("is reserved"),
+ "expected reserved rejection for '" + fetcherId + "', got:
" + root(e));
+ }
+ }
+
+ @Test
+ public void requestCannotNameSystemEmitter() {
+ for (String emitterId : new String[]{"__unpack", " __unpack"}) {
+ Exception e = assertThrows(Exception.class,
+ () -> JsonFetchEmitTuple.fromJson(new
StringReader(tuple("f", emitterId))),
+ "should have rejected emitter id '" + emitterId + "'");
+ assertTrue(root(e).contains("is reserved"),
+ "expected reserved rejection for '" + emitterId + "', got:
" + root(e));
+ }
+ }
+
+ @Test
+ public void asyncEndpointCannotNameSystemFetcher() {
+ Exception e = assertThrows(Exception.class, () ->
JsonFetchEmitTupleList.fromJson(
+ new StringReader("{\"tuples\":[" + tuple("__tika-server", "e")
+ "]}")));
+ assertTrue(root(e).contains("is reserved"), "expected reserved
rejection, got: " + root(e));
+ }
+
+ /**
+ * The NBSP case is why the rule is an allowlist rather than a trim:
trim() and strip() both
+ * leave U+00A0, so such an id would be neither reserved nor resolvable.
+ */
+ @Test
+ public void requestCannotUseIllegalIdCharacters() {
+ for (String fetcherId : new String[]{"my fetcher", "../etc", "a/b",
"f\u00A0oo", "f:oo"}) {
+ Exception e = assertThrows(Exception.class,
+ () -> JsonFetchEmitTuple.fromJson(new
StringReader(tuple(fetcherId, "e"))),
+ "should have rejected fetcher id '" + fetcherId + "'");
+ assertTrue(root(e).contains("Illegal"),
+ "expected charset rejection for '" + fetcherId + "', got:
" + root(e));
+ }
+ }
+
+ @Test
+ public void legalIdsAreTrimmed() throws Exception {
+ FetchEmitTuple t = JsonFetchEmitTuple.fromJson(new
StringReader(tuple(" my-fetcher ", " e ")));
+ assertEquals("my-fetcher", t.getFetchKey().getFetcherId());
+ assertEquals("e", t.getEmitKey().getEmitterId());
+ }
+
+ @Test
+ public void internalIpcMayNameSystemComponents() throws Exception {
+ FetchEmitTuple t = new FetchEmitTuple("t", new
FetchKey("__tika-server", "k"),
+ new EmitKey("__unpack", "ek"), new Metadata(), new
ParseContext(),
+ FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP);
+ FetchEmitTuple back = JsonPipesIpc.fromBytes(JsonPipesIpc.toBytes(t),
FetchEmitTuple.class);
+ assertEquals("__tika-server", back.getFetchKey().getFetcherId());
+ assertEquals("__unpack", back.getEmitKey().getEmitterId());
+ }
+
+ @Test
+ public void internalIpcStillRejectsIllegalIdCharacters() throws Exception {
+ byte[] smile = new ObjectMapper(new SmileFactory())
+ .writeValueAsBytes(new ObjectMapper().readTree(tuple("a/b",
"e")));
+ Exception e = assertThrows(Exception.class,
+ () -> 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-fork-parser/src/main/java/org/apache/tika/pipes/fork/PipesForkParser.java
b/tika-pipes/tika-pipes-fork-parser/src/main/java/org/apache/tika/pipes/fork/PipesForkParser.java
index 8c57272827..a3b62115ae 100644
---
a/tika-pipes/tika-pipes-fork-parser/src/main/java/org/apache/tika/pipes/fork/PipesForkParser.java
+++
b/tika-pipes/tika-pipes-fork-parser/src/main/java/org/apache/tika/pipes/fork/PipesForkParser.java
@@ -21,12 +21,14 @@ import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Map;
+import java.util.UUID;
import org.apache.tika.config.EmbeddedLimits;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
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;
@@ -39,6 +41,9 @@ import org.apache.tika.pipes.core.PipesException;
import org.apache.tika.pipes.core.PipesParser;
import org.apache.tika.pipes.core.config.ConfigMerger;
import org.apache.tika.pipes.core.config.ConfigOverrides;
+import org.apache.tika.pipes.core.fetcher.BytesFetcher;
+import org.apache.tika.pipes.core.fetcher.InlineBytes;
+import org.apache.tika.pipes.core.fetcher.PayloadRouter;
import org.apache.tika.sax.ContentHandlerFactory;
/**
@@ -136,6 +141,8 @@ public class PipesForkParser implements Closeable {
*/
public PipesForkParser(PipesForkParserConfig config) throws IOException,
TikaConfigException {
this.config = config;
+ // Jackson-deserialized configs are checked on binding; setter-built
ones are checked here.
+ config.getPipesConfig().checkPayloadLimits();
ConfigMerger.MergeResult mergeResult = createTikaConfigFile();
this.tikaConfigPath = mergeResult.configPath();
this.internalFetcherId = mergeResult.fetcherId();
@@ -247,10 +254,25 @@ public class PipesForkParser implements Closeable {
*/
public PipesForkResult parse(TikaInputStream tis, Metadata metadata,
ParseContext parseContext)
throws IOException, InterruptedException, PipesException,
TikaException {
- // Get the path - this will spool to a temp file if the stream doesn't
have
- // an underlying file. The temp file is managed by TikaInputStream and
will
- // be cleaned up when the TikaInputStream is closed.
- return parseInternal(tis.getPath(), metadata, parseContext);
+ // A stream already backed by a file keeps it; otherwise small content
rides inline and
+ // only large content is written out. Shared with tika-server so both
use one threshold.
+ try (PayloadRouter.Routed routed = PayloadRouter.route(tis,
+ config.getPipesConfig().getMaxInlineBytes(),
+ () -> Files.createTempFile("tika-fork-", ".tmp"))) {
+ if (routed.isInline()) {
+ parseContext.set(InlineBytes.class, routed.inlineBytes());
+ try {
+ String name =
metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY);
+ return parseWithKey(new FetchKey(BytesFetcher.FETCHER_ID,
name == null ? "" : name),
+ UUID.randomUUID().toString(), metadata,
parseContext);
+ } finally {
+ // The payload is request-owned; a caller reusing this
context must not
+ // retain (and re-serialize) this document's bytes on
later parses.
+ parseContext.set(InlineBytes.class, null);
+ }
+ }
+ return parseInternal(routed.path(), metadata, parseContext);
+ }
}
/**
@@ -259,10 +281,14 @@ public class PipesForkParser implements Closeable {
private PipesForkResult parseInternal(Path path, Metadata metadata,
ParseContext parseContext)
throws IOException, InterruptedException, PipesException,
TikaException {
String absolutePath = path.toAbsolutePath().toString();
- String id = absolutePath;
-
// Use the internal fetcher ID generated by ConfigMerger (UUID-based)
- FetchKey fetchKey = new FetchKey(internalFetcherId, absolutePath);
+ return parseWithKey(new FetchKey(internalFetcherId, absolutePath),
absolutePath,
+ metadata, parseContext);
+ }
+
+ private PipesForkResult parseWithKey(FetchKey fetchKey, String id,
Metadata metadata,
+ ParseContext parseContext)
+ throws IOException, InterruptedException, PipesException,
TikaException {
EmitKey emitKey = new EmitKey("", id); // Empty emitter name since
we're using PASSBACK_ALL
// Add content handler factory and parse mode to parse context
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 c8287629b8..725c446062 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
@@ -19,9 +19,11 @@ package org.apache.tika.pipes.fork;
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.ByteArrayInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
@@ -39,8 +41,10 @@ import org.junit.jupiter.api.io.TempDir;
import org.apache.tika.io.TikaInputStream;
import org.apache.tika.metadata.HttpHeaders;
import org.apache.tika.metadata.Metadata;
+import org.apache.tika.parser.ParseContext;
import org.apache.tika.pipes.api.ParseMode;
import org.apache.tika.pipes.api.PipesResult;
+import org.apache.tika.pipes.core.fetcher.InlineBytes;
import org.apache.tika.sax.BasicContentHandlerFactory;
public class PipesForkParserTest {
@@ -103,6 +107,30 @@ public class PipesForkParserTest {
}
}
+ /**
+ * The inline payload is request-owned: a caller reusing the ParseContext
must not have this
+ * document's bytes retained and re-serialized into later requests.
+ */
+ @Test
+ public void testInlinePayloadNotRetainedInCallerContext() throws Exception
{
+ PipesForkParserConfig config = new PipesForkParserConfig()
+ .setPluginsDir(PLUGINS_DIR)
+ .setHandlerType(BasicContentHandlerFactory.HANDLER_TYPE.TEXT)
+ .setParseMode(ParseMode.RMETA)
+ .addJvmArg("-Xmx256m");
+
+ ParseContext parseContext = new ParseContext();
+ byte[] content = "inline body".getBytes(StandardCharsets.UTF_8);
+ try (PipesForkParser parser = new PipesForkParser(config);
+ TikaInputStream tis = TikaInputStream.get(new
ByteArrayInputStream(content))) {
+ PipesForkResult result = parser.parse(tis, new Metadata(),
parseContext);
+ assertTrue(result.isSuccess(), "Parse should succeed. Status: " +
result.getStatus()
+ + ", message: " + result.getMessage());
+ assertNull(parseContext.get(InlineBytes.class),
+ "inline payload must not outlive its request in the
caller's context");
+ }
+ }
+
@Test
public void testParseWithMetadata() throws Exception {
// Create a simple HTML file
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 c4fb5b6b27..54fa3da1c2 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
@@ -21,7 +21,6 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Collections;
import java.util.List;
-import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@@ -39,6 +38,7 @@ 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.ComponentIds;
import org.apache.tika.pipes.api.FetchEmitTuple;
import org.apache.tika.pipes.api.ParseMode;
import org.apache.tika.pipes.api.PipesResult;
@@ -51,6 +51,9 @@ import org.apache.tika.pipes.core.PipesConfig;
import org.apache.tika.pipes.core.PipesException;
import org.apache.tika.pipes.core.PipesParser;
import org.apache.tika.pipes.core.extractor.UnpackConfig;
+import org.apache.tika.pipes.core.fetcher.BytesFetcher;
+import org.apache.tika.pipes.core.fetcher.InlineBytes;
+import org.apache.tika.pipes.core.fetcher.PayloadRouter;
import org.apache.tika.server.core.TikaServerParseException;
/**
@@ -69,11 +72,12 @@ public class PipesParsingHelper {
* The fetcher ID used for reading temp files.
* This fetcher is configured with basePath = inputTempDirectory.
*/
- public static final String DEFAULT_FETCHER_ID = "tika-server-fetcher";
+ public static final String DEFAULT_FETCHER_ID = "__tika-server";
private final PipesParser pipesParser;
private final PipesConfig pipesConfig;
private final Path inputTempDirectory;
+ private final int maxInlineBytes;
private final Path unpackEmitterBasePath;
/**
@@ -92,6 +96,7 @@ public class PipesParsingHelper {
this.pipesParser = pipesParser;
this.pipesConfig = pipesConfig;
this.inputTempDirectory = inputTempDirectory;
+ this.maxInlineBytes = pipesConfig.getMaxInlineBytes();
this.unpackEmitterBasePath = unpackEmitterBasePath;
if (inputTempDirectory == null ||
!Files.isDirectory(inputTempDirectory)) {
@@ -172,19 +177,27 @@ public class PipesParsingHelper {
public List<Metadata> parse(TikaInputStream tis, Metadata metadata,
ParseContext parseContext, ParseMode
parseMode) throws IOException {
String requestId = UUID.randomUUID().toString();
- Path tempFile = null;
+ PayloadRouter.Routed routed = null;
String callerSuppliedName =
metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY);
try {
- // Spool input to our dedicated temp directory with proper suffix
- String suffix = getSuffix(metadata);
- tempFile = Files.createTempFile(inputTempDirectory, "tika-",
suffix);
- // getPath() spools once via Tika's TemporaryResources; copying
file->file
- // avoids re-reading the stream (which would decode the spool a
second time).
- Files.copy(tis.getPath(), tempFile,
java.nio.file.StandardCopyOption.REPLACE_EXISTING);
-
- String relativeName = tempFile.getFileName().toString();
- LOG.debug("parse: spooled to {} ({} bytes)", relativeName,
Files.size(tempFile));
+ routed = PayloadRouter.route(tis, maxInlineBytes,
+ () -> Files.createTempFile(inputTempDirectory, "tika-",
getSuffix(metadata)));
+
+ String relativeName = null;
+ FetchKey fetchKey;
+ if (routed.isInline()) {
+ parseContext.set(InlineBytes.class, routed.inlineBytes());
+ // Fetch key doubles as the caller's filename, so no spool
name to scrub later.
+ fetchKey = new FetchKey(BytesFetcher.FETCHER_ID,
+ callerSuppliedName == null ? "" : callerSuppliedName);
+ LOG.debug("parse: {} bytes inline",
routed.inlineBytes().length());
+ } else {
+ relativeName = routed.path().getFileName().toString();
+ fetchKey = new FetchKey(DEFAULT_FETCHER_ID, relativeName);
+ LOG.debug("parse: spooled to {} ({} bytes)", relativeName,
+ Files.size(routed.path()));
+ }
// Set parse mode in context
parseContext.set(ParseMode.class, parseMode);
@@ -195,9 +208,6 @@ public class PipesParsingHelper {
// explicitly per-request rather than relying on the parser-level
default.
parseContext.set(EmitStrategyConfig.class, new
EmitStrategyConfig(EmitStrategy.PASSBACK_ALL));
- // Create FetchEmitTuple with relative filename (basePath is
configured in fetcher)
- FetchKey fetchKey = new FetchKey(DEFAULT_FETCHER_ID, relativeName);
-
FetchEmitTuple tuple = new FetchEmitTuple(
requestId,
fetchKey,
@@ -211,7 +221,9 @@ public class PipesParsingHelper {
// Process result
List<Metadata> metadataList = processResult(result);
- stripSpoolIdentity(metadataList, relativeName, callerSuppliedName);
+ if (relativeName != null) {
+ stripSpoolIdentity(metadataList, relativeName,
callerSuppliedName);
+ }
return metadataList;
} catch (InterruptedException e) {
@@ -220,13 +232,11 @@ public class PipesParsingHelper {
} catch (PipesException e) {
throw new TikaServerParseException(e);
} finally {
- // Clean up temp file
- if (tempFile != null) {
- try {
- Files.deleteIfExists(tempFile);
- } catch (IOException e) {
- LOG.warn("Failed to delete temp file: {}", tempFile, e);
- }
+ // The payload is request-owned: contexts are per-request today,
but do not let
+ // safety depend on that call-site discipline.
+ parseContext.set(InlineBytes.class, null);
+ if (routed != null) {
+ routed.close();
}
}
}
@@ -454,7 +464,7 @@ public class PipesParsingHelper {
* This emitter must be configured in tika-config.json with a basePath
* pointing to a writable temp directory.
*/
- public static final String UNPACK_EMITTER_ID = "unpack-emitter";
+ public static final String UNPACK_EMITTER_ID = "__unpack";
/**
* Fetcher/emitter ids the server wires up for its own request plumbing.
Both are rooted at
@@ -467,10 +477,11 @@ public class PipesParsingHelper {
* it builds the tuples for /tika, /rmeta, and /unpack, which is exactly
the use being
* reserved.
*/
- private static final Set<String> RESERVED_COMPONENT_IDS =
- Set.of(DEFAULT_FETCHER_ID, UNPACK_EMITTER_ID);
-
/**
+ * Backstop for ids the tuple deserializer cannot reach. {@code fetcher}
and {@code emitter}
+ * are already refused there for any tuple parsed from a request; the
UnpackConfig emitter is
+ * buried in a parse-context component, so it is checked only here.
+ *
* @throws BadRequestException if a caller-supplied tuple names a
server-internal component.
*/
public static void rejectReservedComponentIds(FetchEmitTuple t) {
@@ -484,9 +495,9 @@ public class PipesParsingHelper {
}
private static void checkNotReserved(String id, String kind) {
- if (id != null && RESERVED_COMPONENT_IDS.contains(id)) {
+ if (ComponentIds.isSystem(id)) {
throw new BadRequestException(
- "'" + id + "' is reserved for tika-server's internal use
and may not be named as a "
+ "'" + id.trim() + "' is reserved for tika-server's
internal use and may not be named as a "
+ kind + " by a request");
}
}
@@ -514,7 +525,7 @@ public class PipesParsingHelper {
public UnpackResult parseUnpack(TikaInputStream tis, Metadata metadata,
ParseContext parseContext, boolean
saveAll) throws IOException {
String requestId = UUID.randomUUID().toString();
- Path tempFile = null;
+ PayloadRouter.Routed routed = null;
String callerSuppliedName =
metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY);
// The child emits the zip during parse, before we know whether the
request
// succeeds. Unless we hand it off to the caller (who streams then
deletes it),
@@ -523,16 +534,23 @@ public class PipesParsingHelper {
boolean handedOff = false;
try {
- // Spool input to our dedicated temp directory with proper suffix
- String suffix = getSuffix(metadata);
- tempFile = Files.createTempFile(inputTempDirectory,
"tika-unpack-", suffix);
- // getPath() spools once via Tika's TemporaryResources; copying
file->file
- // avoids re-reading the stream (which would decode the spool a
second time).
- Files.copy(tis.getPath(), tempFile,
java.nio.file.StandardCopyOption.REPLACE_EXISTING);
-
- String relativeName = tempFile.getFileName().toString();
- LOG.debug("parseUnpack: spooled to {} ({} bytes), requestId={}",
- relativeName, Files.size(tempFile), requestId);
+ routed = PayloadRouter.route(tis, maxInlineBytes, () ->
+ Files.createTempFile(inputTempDirectory, "tika-unpack-",
getSuffix(metadata)));
+
+ String relativeName = null;
+ FetchKey fetchKey;
+ if (routed.isInline()) {
+ parseContext.set(InlineBytes.class, routed.inlineBytes());
+ fetchKey = new FetchKey(BytesFetcher.FETCHER_ID,
+ callerSuppliedName == null ? "" : callerSuppliedName);
+ LOG.debug("parseUnpack: {} bytes inline, requestId={}",
+ routed.inlineBytes().length(), requestId);
+ } else {
+ relativeName = routed.path().getFileName().toString();
+ fetchKey = new FetchKey(DEFAULT_FETCHER_ID, relativeName);
+ LOG.debug("parseUnpack: spooled to {} ({} bytes),
requestId={}",
+ relativeName, Files.size(routed.path()), requestId);
+ }
// Set parse mode to UNPACK
parseContext.set(ParseMode.class, ParseMode.UNPACK);
@@ -566,8 +584,6 @@ public class PipesParsingHelper {
parseContext.set(UnpackConfig.class, unpackConfig);
- // Create FetchEmitTuple with relative filename (basePath is
configured in fetcher)
- FetchKey fetchKey = new FetchKey(DEFAULT_FETCHER_ID, relativeName);
EmitKey emitKey = new EmitKey(UNPACK_EMITTER_ID, requestId);
FetchEmitTuple tuple = new FetchEmitTuple(
@@ -627,17 +643,16 @@ public class PipesParsingHelper {
boolean isFrictionless = unpackConfig.getOutputFormat() ==
UnpackConfig.OUTPUT_FORMAT.FRICTIONLESS;
Path zipFile = getEmittedZipPath(requestId, isFrictionless);
- stripSpoolIdentity(metadataList, relativeName, callerSuppliedName);
+ if (relativeName != null) {
+ stripSpoolIdentity(metadataList, relativeName,
callerSuppliedName);
+ }
handedOff = true;
return new UnpackResult(zipFile, metadataList);
} finally {
- // Clean up temp file
- if (tempFile != null) {
- try {
- Files.deleteIfExists(tempFile);
- } catch (IOException e) {
- LOG.warn("Failed to delete temp file: {}", tempFile, e);
- }
+ // See parse(): the inline payload must not outlive its request.
+ parseContext.set(InlineBytes.class, null);
+ if (routed != null) {
+ routed.close();
}
if (!handedOff) {
deleteEmittedZips(requestId);
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/UnpackerResource.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/UnpackerResource.java
index 5a5ef25250..b94be06c4f 100644
---
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/UnpackerResource.java
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/UnpackerResource.java
@@ -55,26 +55,11 @@ import org.apache.tika.parser.ParseContext;
* <li>POST /unpack/all - Extract all with config (multipart)</li>
* </ul>
* <p>
- * <b>Configuration Requirements:</b>
+ * <b>Configuration:</b>
* <p>
- * Your tika-config.json must include:
- * <pre>
- * {
- * "fetchers": {
- * "file-system-fetcher": {
- * "class": "org.apache.tika.pipes.fetcher.fs.FileSystemFetcher",
- * "allowAbsolutePaths": true
- * }
- * },
- * "emitters": {
- * "unpack-emitter": {
- * "class": "org.apache.tika.pipes.emitter.fs.FileSystemEmitter",
- * "basePath": "/tmp/tika-unpack",
- * "onExists": "replace"
- * }
- * }
- * }
- * </pre>
+ * None required. The server wires up its own {@code __}-prefixed fetcher and
emitter against
+ * temp directories it owns, confined by {@code basePath}. Those ids are
reserved and a request
+ * that names one is rejected.
* <p>
* <b>Multipart Configuration (POST endpoints):</b>
* <p>
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/ForcedSpoolPathTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/ForcedSpoolPathTest.java
new file mode 100644
index 0000000000..855f209d40
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/ForcedSpoolPathTest.java
@@ -0,0 +1,115 @@
+/*
+ * 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.server.core;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import jakarta.ws.rs.core.Response;
+import org.apache.cxf.jaxrs.JAXRSServerFactoryBean;
+import org.apache.cxf.jaxrs.client.WebClient;
+import org.apache.cxf.jaxrs.lifecycle.SingletonResourceProvider;
+import org.junit.jupiter.api.Test;
+
+import org.apache.tika.server.core.resource.RecursiveMetadataResource;
+import org.apache.tika.server.core.resource.TikaResource;
+import org.apache.tika.server.core.writer.JSONMessageBodyWriter;
+import org.apache.tika.server.core.writer.MetadataListMessageBodyWriter;
+
+/**
+ * With {@code maxInlineBytes} at 0 every request takes the spool branch,
which the rest of the
+ * suite never reaches -- the test corpus is far below the 10MB default, so
those runs all go
+ * inline. This is the only coverage of the branch that writes to the
fetcher's basePath and then
+ * scrubs the spool filename back out of the returned metadata.
+ */
+public class ForcedSpoolPathTest extends CXFTestBase {
+
+ private static final String TIKA_PATH = "/tika";
+ private static final String RMETA_PATH = "/rmeta";
+ private static final String TEST_DOC =
"test-documents/mock/hello_world.xml";
+
+ @Override
+ protected InputStream getPipesConfigInputStream() throws IOException {
+ InputStream base = super.getPipesConfigInputStream();
+ if (base == null) {
+ return null;
+ }
+ JsonNode config = new
com.fasterxml.jackson.databind.ObjectMapper().readTree(base);
+ ((ObjectNode) config.get("pipes")).put("maxInlineBytes", 0);
+ return new ByteArrayInputStream(
+ config.toString().getBytes(UTF_8));
+ }
+
+ @Override
+ protected String getPipesInputPath() {
+ return "target/pipes-input-forced-spool";
+ }
+
+ @Override
+ protected void setUpResources(JAXRSServerFactoryBean sf) {
+ sf.setResourceClasses(TikaResource.class,
RecursiveMetadataResource.class);
+ sf.setResourceProvider(TikaResource.class, new
SingletonResourceProvider(tikaResource));
+ sf.setResourceProvider(RecursiveMetadataResource.class,
+ new SingletonResourceProvider(new
RecursiveMetadataResource(tikaResource)));
+ }
+
+ @Override
+ protected void setUpProviders(JAXRSServerFactoryBean sf) {
+ List<Object> providers = new ArrayList<>();
+ providers.add(new TikaServerParseExceptionMapper());
+ providers.add(new BadRequestExceptionMapper());
+ providers.add(new JSONMessageBodyWriter());
+ providers.add(new MetadataListMessageBodyWriter());
+ sf.setProviders(providers);
+ }
+
+ @Test
+ public void spooledTikaRequestParses() throws Exception {
+ Response response = WebClient
+ .create(endPoint + TIKA_PATH)
+ .accept("text/plain")
+ .put(ClassLoader.getSystemResourceAsStream(TEST_DOC));
+ String content = getStringFromInputStream((InputStream)
response.getEntity());
+ assertEquals(200, response.getStatus());
+ assertContains("hello world", content);
+ }
+
+ /**
+ * The spool filename is server-internal and names a file already deleted;
it must not leak
+ * out as the document's identity.
+ */
+ @Test
+ public void spooledRequestDoesNotLeakSpoolName() throws Exception {
+ Response response = WebClient
+ .create(endPoint + RMETA_PATH)
+ .accept("application/json")
+ .put(ClassLoader.getSystemResourceAsStream(TEST_DOC));
+ String json = getStringFromInputStream((InputStream)
response.getEntity());
+ assertEquals(200, response.getStatus());
+ assertFalse(json.contains("tika-"),
+ "spool filename leaked into the response: " + json);
+ }
+}
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/resource/ReservedComponentIdTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/resource/ReservedComponentIdTest.java
index 76972bb193..8e0c73b5cc 100644
---
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/resource/ReservedComponentIdTest.java
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/resource/ReservedComponentIdTest.java
@@ -17,6 +17,7 @@
package org.apache.tika.server.core.resource;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import jakarta.ws.rs.BadRequestException;
@@ -30,7 +31,7 @@ import org.apache.tika.pipes.api.fetcher.FetchKey;
import org.apache.tika.pipes.core.extractor.UnpackConfig;
/**
- * The server configures {@code tika-server-fetcher} and {@code
unpack-emitter} against its own
+ * The server configures {@code __tika-server} and {@code __unpack} against
its own
* spool directories. They exist on every running server, so unlike an unknown
id these would
* resolve if a /pipes or /async caller named them -- reading another
request's pending upload,
* or planting a file where the unpack download path serves from.
@@ -41,6 +42,17 @@ import org.apache.tika.pipes.core.extractor.UnpackConfig;
*/
public class ReservedComponentIdTest {
+ /**
+ * cxf-unpack-test-template.json has to spell the emitter id out --
JsonConfigHelper
+ * substitutes textual values, never field names -- so renaming the
constant without the
+ * template surfaces only as "Archive is not a ZIP archive" from the
unpack tests.
+ */
+ @Test
+ public void testUnpackTemplateIdMatchesConstant() {
+ assertEquals("__unpack", PipesParsingHelper.UNPACK_EMITTER_ID,
+ "cxf-unpack-test-template.json hard-codes this id; update it
too");
+ }
+
@Test
public void testReservedFetcherRejected() {
assertThrows(BadRequestException.class, () ->
PipesParsingHelper.rejectReservedComponentIds(
diff --git
a/tika-server/tika-server-core/src/test/resources/configs/cxf-unpack-test-template.json
b/tika-server/tika-server-core/src/test/resources/configs/cxf-unpack-test-template.json
index e1b628a84f..500480ebff 100644
---
a/tika-server/tika-server-core/src/test/resources/configs/cxf-unpack-test-template.json
+++
b/tika-server/tika-server-core/src/test/resources/configs/cxf-unpack-test-template.json
@@ -7,7 +7,7 @@
}
},
"emitters": {
- "unpack-emitter": {
+ "__unpack": {
"file-system-emitter": {
"basePath": "UNPACK_EMITTER_BASE_PATH",
"onExists": "REPLACE"
