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 81f2eb01d6 TIKA-4828 -- improve caching while digesting embedded files
(#3042)
81f2eb01d6 is described below
commit 81f2eb01d696005f3cff5b61e7778f43da375dab
Author: Tim Allison <[email protected]>
AuthorDate: Fri Aug 21 19:48:24 2026 -0400
TIKA-4828 -- improve caching while digesting embedded files (#3042)
---
CHANGES.txt | 13 +
docs/modules/ROOT/pages/pipes/configuration.adoc | 22 +-
.../java/org/apache/tika/digest/DigestHelper.java | 3 +-
.../apache/tika/digest/InputStreamDigester.java | 8 +-
.../java/org/apache/tika/io/ByteArraySource.java | 9 +-
.../java/org/apache/tika/io/CacheMemoryBudget.java | 86 ++++++
.../org/apache/tika/io/CachingInputStream.java | 34 +++
.../java/org/apache/tika/io/CachingSource.java | 40 ++-
.../main/java/org/apache/tika/io/FileSource.java | 10 +-
.../apache/tika/io/MemorySeekableByteChannel.java | 115 ++++++++
.../java/org/apache/tika/io/ReopenableSource.java | 319 +++++++++++++++++++++
.../main/java/org/apache/tika/io/StreamCache.java | 114 ++++++--
.../java/org/apache/tika/io/TikaInputSource.java | 18 +-
.../java/org/apache/tika/io/TikaInputStream.java | 70 ++++-
.../parser/multiple/AbstractMultipleParser.java | 3 +-
.../org/apache/tika/io/CacheMemoryBudgetTest.java | 108 +++++++
.../tika/io/MemorySeekableByteChannelTest.java | 166 +++++++++++
.../org/apache/tika/io/ReopenableSourceTest.java | 303 +++++++++++++++++++
.../org/apache/tika/io/StreamCacheBudgetTest.java | 158 ++++++++++
.../apache/tika/io/TikaInputStreamChannelTest.java | 159 ++++++++++
.../org/apache/tika/io/TikaInputStreamTest.java | 10 +-
.../org/apache/tika/parser/crypto/Pkcs7Parser.java | 6 +-
.../parser/microsoft/onenote/OneNoteParser.java | 8 +-
.../org/apache/tika/parser/epub/EpubParser.java | 3 +-
.../apache/tika/parser/odf/OpenDocumentParser.java | 6 +-
.../apache/tika/parser/pkg/CompressorParser.java | 17 +-
.../org/apache/tika/parser/pkg/SevenZParser.java | 11 +-
.../java/org/apache/tika/parser/pkg/ZipParser.java | 14 +-
.../tika/parser/pkg/ZipEmbeddedDigestTest.java | 162 +++++++++++
.../detect/zip/DefaultZipContainerDetector.java | 29 +-
.../org/apache/tika/zip/utils/ZipFileHelper.java | 51 ++++
.../org/apache/tika/zip/utils/ZipSalvager.java | 9 +-
.../tika/pipes/core/server/ParseHandler.java | 4 +-
.../apache/tika/pipes/core/server/PipesServer.java | 47 +++
.../pipes/core/server/SharedServerResources.java | 1 +
.../core/server/CacheMemoryBudgetSeedingTest.java | 45 +++
36 files changed, 2100 insertions(+), 81 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 9e3a699fff..5dde33b1b4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,18 @@
Release 4.1.0 - ???
+ * Digesting embedded documents no longer buffers each embedded object to a
+ temp file. Zip entries are re-read from the parent archive on rewind, and
+ a new process-wide CacheMemoryBudget (seeded by the pipes forked server;
+ default 256MB, clamped to a quarter of the fork's heap; tunable via
+ -Dtika.pipes.cacheMemoryBudgetBytes in the config's forkedJvmArgs, <=0
+ disables) lets embedded objects stay in memory past the per-object 1MB
+ threshold. New public API on TikaInputStream: get(IOSupplier,...),
+ enableRewind(CacheMemoryBudget), getSeekableByteChannel(). Zip/7z/epub/odf
+ parsing and zip container detection now read through seekable channels, so
+ after detection/parsing a TikaInputStream may no longer be file-backed
+ (hasFile() false); getPath()/getFile() still work and spool on demand
+ (TIKA-4828).
+
* Pipes now carries the caller-supplied Content-Type across the worker's
fresh-metadata boundary as a soft detection hint, so every forked-parse
endpoint (/tika, /meta, /rmeta, /unpack, /async, /pipes, plus tika-grpc
diff --git a/docs/modules/ROOT/pages/pipes/configuration.adoc
b/docs/modules/ROOT/pages/pipes/configuration.adoc
index 763f2a5612..02a309661e 100644
--- a/docs/modules/ROOT/pages/pipes/configuration.adoc
+++ b/docs/modules/ROOT/pages/pipes/configuration.adoc
@@ -45,7 +45,7 @@ how many forked JVMs to run, timeouts, memory management, and
parse behavior.
|`forkedJvmArgs`
|`[]`
-|JVM arguments for forked processes (e.g., `["-Xmx512m", "-Xms256m"]`). Unless
you supply your own, Tika auto-injects `-XX:ActiveProcessorCount` to right-size
each fork's GC and JIT thread pools, and `-XX:MaxRAMPercentage` to keep the
forks' combined heap ceiling under the host's memory; see
xref:pipes/cpu-sizing.adoc[Forked-JVM CPU and Heap Sizing].
+|JVM arguments for forked processes (e.g., `["-Xmx512m", "-Xms256m"]`). Unless
you supply your own, Tika auto-injects `-XX:ActiveProcessorCount` to right-size
each fork's GC and JIT thread pools, and `-XX:MaxRAMPercentage` to keep the
forks' combined heap ceiling under the host's memory; see
xref:pipes/cpu-sizing.adoc[Forked-JVM CPU and Heap Sizing]. This is also where
the fork's system properties go, e.g. `-Dtika.pipes.cacheMemoryBudgetBytes`
below -- setting them on the parent JVM has [...]
|`javaPath`
|`java`
@@ -66,6 +66,26 @@ cannot, and its subdirectories survive. On a RAM-backed
filesystem those leaks c
rather than disk, and `/dev/shm` is commonly sized at half of RAM -- so if you
point
`tempDirectory` at one, sweep it on service start.
+=== Embedded-object cache memory budget
+
+Each fork holds a process-wide in-memory budget for stream caching (chiefly
the rewind
+buffers used when digesting embedded documents), so small embedded objects
stay in RAM
+instead of spilling to a temp file at the per-object 1MB threshold. The
default is 256MB per
+fork, clamped to a quarter of the fork's max heap; the effective value is
logged at fork
+startup. Tune it with a system property in `forkedJvmArgs` (plain bytes, no
unit suffix;
+`<=0` disables the budget and restores the per-object threshold):
+
+[source,json]
+----
+"forkedJvmArgs": ["-Xmx1g", "-Dtika.pipes.cacheMemoryBudgetBytes=134217728"]
+----
+
+Size `-Xmx` with the budget in mind: the budget is additional heap the fork
may use on top
+of its parsing working set, and in per-client mode every fork holds its own
budget
+(`numClients` x budget in aggregate). In shared-server mode all concurrent
parses in the
+single forked server share one budget, so each in-flight document gets a
smaller slice of
+the same value.
+
== Timeouts
See also xref:pipes/timeouts.adoc[Timeouts] for the full timeout model.
diff --git a/tika-core/src/main/java/org/apache/tika/digest/DigestHelper.java
b/tika-core/src/main/java/org/apache/tika/digest/DigestHelper.java
index a5f751b1ae..359ebf9f27 100644
--- a/tika-core/src/main/java/org/apache/tika/digest/DigestHelper.java
+++ b/tika-core/src/main/java/org/apache/tika/digest/DigestHelper.java
@@ -23,6 +23,7 @@ import java.nio.file.Path;
import org.apache.tika.extractor.DefaultEmbeddedStreamTranslator;
import org.apache.tika.extractor.EmbeddedStreamTranslator;
+import org.apache.tika.io.CacheMemoryBudget;
import org.apache.tika.io.TemporaryResources;
import org.apache.tika.io.TikaInputStream;
import org.apache.tika.metadata.Metadata;
@@ -83,7 +84,7 @@ public class DigestHelper {
// The translator consumes `tis` (e.g. OLE2), so enableRewind() before
and rewind()
// after -- otherwise the caller would see an exhausted stream.
if (EMBEDDED_STREAM_TRANSLATOR.shouldTranslate(tis, metadata)) {
- tis.enableRewind();
+ tis.enableRewind(context.get(CacheMemoryBudget.class));
try (TemporaryResources tmp = new TemporaryResources()) {
Path tmpBytes = tmp.createTempFile();
try (OutputStream os = Files.newOutputStream(tmpBytes)) {
diff --git
a/tika-core/src/main/java/org/apache/tika/digest/InputStreamDigester.java
b/tika-core/src/main/java/org/apache/tika/digest/InputStreamDigester.java
index 56ee7ea21e..00941602d3 100644
--- a/tika-core/src/main/java/org/apache/tika/digest/InputStreamDigester.java
+++ b/tika-core/src/main/java/org/apache/tika/digest/InputStreamDigester.java
@@ -23,6 +23,7 @@ import java.security.Provider;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import org.apache.tika.io.CacheMemoryBudget;
import org.apache.tika.io.TikaInputStream;
import org.apache.tika.metadata.HttpHeaders;
import org.apache.tika.metadata.Metadata;
@@ -110,13 +111,16 @@ public class InputStreamDigester implements Digester {
*
* @param tis TikaInputStream to digest
* @param metadata metadata in which to store the digest information
- * @param parseContext ParseContext -- not actually used yet, but there
for future expansion
+ * @param parseContext ParseContext, which may carry a {@link
CacheMemoryBudget}
* @throws IOException on IO problem or IllegalArgumentException if
algorithm couldn't be found
*/
@Override
public void digest(TikaInputStream tis, Metadata metadata, ParseContext
parseContext)
throws IOException {
- tis.enableRewind();
+ // Bridge the shared budget from ParseContext; TikaInputStream never
sees ParseContext
+ CacheMemoryBudget budget =
+ (parseContext == null) ? null :
parseContext.get(CacheMemoryBudget.class);
+ tis.enableRewind(budget);
MessageDigest messageDigest = newMessageDigest();
byte[] buffer = new byte[8192];
diff --git a/tika-core/src/main/java/org/apache/tika/io/ByteArraySource.java
b/tika-core/src/main/java/org/apache/tika/io/ByteArraySource.java
index d2d3bbd61a..72c94f728e 100644
--- a/tika-core/src/main/java/org/apache/tika/io/ByteArraySource.java
+++ b/tika-core/src/main/java/org/apache/tika/io/ByteArraySource.java
@@ -19,6 +19,7 @@ package org.apache.tika.io;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.nio.channels.SeekableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -120,10 +121,16 @@ class ByteArraySource extends InputStream implements
TikaInputSource {
}
@Override
- public void enableRewind() throws IOException {
+ public void enableRewind(CacheMemoryBudget budget) throws IOException {
// No-op: byte array is always rewindable
}
+ @Override
+ public SeekableByteChannel getSeekableByteChannel() {
+ // The array stays valid even after a getPath() spill copy
+ return new MemorySeekableByteChannel(data, length);
+ }
+
// Mark/reset support using position tracking
private int markPosition = -1;
diff --git a/tika-core/src/main/java/org/apache/tika/io/CacheMemoryBudget.java
b/tika-core/src/main/java/org/apache/tika/io/CacheMemoryBudget.java
new file mode 100644
index 0000000000..58af58c33c
--- /dev/null
+++ b/tika-core/src/main/java/org/apache/tika/io/CacheMemoryBudget.java
@@ -0,0 +1,86 @@
+/*
+ * 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.io;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * A shared, bounded memory budget for in-memory stream caching.
+ * <p>
+ * When present, stream caches hold content in memory past the per-object
threshold by
+ * reserving buffer <em>capacity</em> against this budget, spilling to a temp
file when a
+ * reservation fails. Content up to the per-object threshold is not accounted
here. Place a
+ * single, process-wide instance in the {@link
org.apache.tika.parser.ParseContext}; it is
+ * bridged to the IO layer via {@link
TikaInputStream#enableRewind(CacheMemoryBudget)}.
+ * <p>
+ * Thread-safe; a single instance may be shared across concurrent parses.
+ */
+public final class CacheMemoryBudget {
+
+ private final long maxBytes;
+ private final AtomicLong reserved = new AtomicLong();
+
+ /**
+ * @param maxBytes maximum total bytes that may be held in memory across
all caches sharing
+ * this budget; must be positive (to disable budgeting,
pass no budget at all)
+ */
+ public CacheMemoryBudget(long maxBytes) {
+ if (maxBytes <= 0) {
+ throw new IllegalArgumentException("maxBytes must be > 0: " +
maxBytes);
+ }
+ this.maxBytes = maxBytes;
+ }
+
+ /**
+ * Attempts to reserve {@code n} bytes. All-or-nothing: either the full
amount is reserved
+ * (return {@code n}) or nothing is (return {@code 0}, signalling the
caller to spill).
+ *
+ * @param n bytes requested
+ * @return {@code n} if reserved, else {@code 0}
+ */
+ public long tryReserve(long n) {
+ if (n <= 0) {
+ return 0;
+ }
+ while (true) {
+ long cur = reserved.get();
+ if (cur + n > maxBytes) {
+ return 0;
+ }
+ if (reserved.compareAndSet(cur, cur + n)) {
+ return n;
+ }
+ }
+ }
+
+ /**
+ * Releases {@code n} previously-reserved bytes back to the budget.
+ */
+ public void release(long n) {
+ if (n > 0) {
+ reserved.addAndGet(-n);
+ }
+ }
+
+ public long getMaxBytes() {
+ return maxBytes;
+ }
+
+ public long getReservedBytes() {
+ return reserved.get();
+ }
+}
diff --git a/tika-core/src/main/java/org/apache/tika/io/CachingInputStream.java
b/tika-core/src/main/java/org/apache/tika/io/CachingInputStream.java
index c17537ba22..ff97c31485 100644
--- a/tika-core/src/main/java/org/apache/tika/io/CachingInputStream.java
+++ b/tika-core/src/main/java/org/apache/tika/io/CachingInputStream.java
@@ -18,7 +18,10 @@ package org.apache.tika.io;
import java.io.IOException;
import java.io.InputStream;
+import java.nio.channels.FileChannel;
+import java.nio.channels.SeekableByteChannel;
import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
/**
* Package-private InputStream wrapper that caches all bytes read,
@@ -42,6 +45,37 @@ class CachingInputStream extends InputStream {
this.sourceExhausted = false;
}
+ /**
+ * Returns a read-only random-access channel over the full content,
draining any unread
+ * source bytes into the cache first; in-memory when the content fit, a
file channel when
+ * it spilled. Does not change the logical read position.
+ */
+ SeekableByteChannel getSeekableByteChannel() throws IOException {
+ drainSourceToCache();
+ SeekableByteChannel channel = cache.getInMemorySeekableByteChannel();
+ if (channel != null) {
+ return channel;
+ }
+ // The drain overflowed the memory budget/threshold -> serve from the
spill file.
+ return FileChannel.open(cache.toFile(), StandardOpenOption.READ);
+ }
+
+ /**
+ * Drains any remaining source bytes into the cache without changing the
logical read
+ * position.
+ */
+ private void drainSourceToCache() throws IOException {
+ if (sourceExhausted) {
+ return;
+ }
+ byte[] buffer = new byte[8192];
+ int n;
+ while ((n = source.read(buffer)) != -1) {
+ cache.append(buffer, 0, n);
+ }
+ sourceExhausted = true;
+ }
+
@Override
public int read() throws IOException {
if (position < cache.size()) {
diff --git a/tika-core/src/main/java/org/apache/tika/io/CachingSource.java
b/tika-core/src/main/java/org/apache/tika/io/CachingSource.java
index 6ed283ee8f..70d4fc6a63 100644
--- a/tika-core/src/main/java/org/apache/tika/io/CachingSource.java
+++ b/tika-core/src/main/java/org/apache/tika/io/CachingSource.java
@@ -20,8 +20,11 @@ import java.io.BufferedInputStream;
import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
+import java.nio.channels.FileChannel;
+import java.nio.channels.SeekableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
import org.apache.commons.io.IOUtils;
@@ -33,9 +36,9 @@ import org.apache.tika.utils.StringUtils;
* Input source that wraps a raw InputStream with optional caching.
* <p>
* Starts in passthrough mode using {@link BufferedInputStream} for basic
- * mark/reset support. When {@link #enableRewind()} is called (at position 0),
- * switches to caching mode using {@link CachingInputStream} which enables
- * full rewind/seek capability.
+ * mark/reset support. When {@link #enableRewind(CacheMemoryBudget)} is called
+ * (at position 0), switches to caching mode using {@link CachingInputStream}
+ * which enables full rewind/seek capability.
* <p>
* If caching is not enabled, {@link #seekTo(long)} will fail for any position
* other than the current position.
@@ -186,7 +189,7 @@ class CachingSource extends InputStream implements
TikaInputSource {
}
@Override
- public void enableRewind() throws IOException {
+ public void enableRewind(CacheMemoryBudget budget) throws IOException {
// Already in caching or file mode - no-op
if (cachingStream != null || fileStream != null) {
return;
@@ -199,11 +202,36 @@ class CachingSource extends InputStream implements
TikaInputSource {
}
// Switch to caching mode
- StreamCache cache = new StreamCache(tmp, suffix);
+ StreamCache cache = new StreamCache(tmp, suffix, budget);
cachingStream = new CachingInputStream(passthroughStream, cache);
passthroughStream = null;
}
+ @Override
+ public SeekableByteChannel getSeekableByteChannel() throws IOException {
+ if (spilledPath != null) {
+ return FileChannel.open(spilledPath, StandardOpenOption.READ);
+ }
+ // If still in passthrough mode, switch to caching first (same rule as
getPath)
+ if (cachingStream == null) {
+ if (passthroughPosition != 0) {
+ throw new IOException(
+ "Cannot create seekable view: position is " +
passthroughPosition +
+ ", must be 0. Call enableRewind() before
reading.");
+ }
+ enableRewind(null);
+ }
+ SeekableByteChannel channel = cachingStream.getSeekableByteChannel();
+ // Record the drained length like getPath() does (feeds
SecureContentHandler's
+ // zip-bomb ratio)
+ length = channel.size();
+ if (metadata != null &&
+ StringUtils.isBlank(metadata.get(HttpHeaders.CONTENT_LENGTH)))
{
+ metadata.set(HttpHeaders.CONTENT_LENGTH, Long.toString(length));
+ }
+ return channel;
+ }
+
@Override
public void seekTo(long position) throws IOException {
if (fileStream != null) {
@@ -245,7 +273,7 @@ class CachingSource extends InputStream implements
TikaInputSource {
"Cannot spill to file: position is " +
passthroughPosition +
", must be 0. Call enableRewind() before
reading if you need file access.");
}
- enableRewind();
+ enableRewind(null);
}
// Spill to file and switch to file-backed mode
diff --git a/tika-core/src/main/java/org/apache/tika/io/FileSource.java
b/tika-core/src/main/java/org/apache/tika/io/FileSource.java
index 143bb2d562..fbd06d7a0d 100644
--- a/tika-core/src/main/java/org/apache/tika/io/FileSource.java
+++ b/tika-core/src/main/java/org/apache/tika/io/FileSource.java
@@ -19,8 +19,11 @@ package org.apache.tika.io;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.nio.channels.FileChannel;
+import java.nio.channels.SeekableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
import org.apache.commons.io.IOUtils;
@@ -120,10 +123,15 @@ class FileSource extends InputStream implements
TikaInputSource {
}
@Override
- public void enableRewind() throws IOException {
+ public void enableRewind(CacheMemoryBudget budget) throws IOException {
// No-op: file is always rewindable
}
+ @Override
+ public SeekableByteChannel getSeekableByteChannel() throws IOException {
+ return FileChannel.open(path, StandardOpenOption.READ);
+ }
+
// Mark/reset support using seekTo
private long markPosition = -1;
diff --git
a/tika-core/src/main/java/org/apache/tika/io/MemorySeekableByteChannel.java
b/tika-core/src/main/java/org/apache/tika/io/MemorySeekableByteChannel.java
new file mode 100644
index 0000000000..71d71d93ed
--- /dev/null
+++ b/tika-core/src/main/java/org/apache/tika/io/MemorySeekableByteChannel.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.io;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.NonWritableChannelException;
+import java.nio.channels.SeekableByteChannel;
+
+/**
+ * Read-only {@link SeekableByteChannel} over an in-memory byte array. Wraps
the array directly
+ * (no copy); callers must not mutate the backing array while the channel is
in use.
+ */
+class MemorySeekableByteChannel implements SeekableByteChannel {
+
+ private final byte[] data;
+ private final int length;
+ private final Runnable onClose;
+ private long position;
+ private boolean open = true;
+
+ MemorySeekableByteChannel(byte[] data, int length) {
+ this(data, length, null);
+ }
+
+ /** {@code onClose} runs exactly once, on the first {@link #close()} --
used by sources
+ * to unpin a budget reservation held while this channel references their
buffer. */
+ MemorySeekableByteChannel(byte[] data, int length, Runnable onClose) {
+ this.data = data;
+ this.length = length;
+ this.onClose = onClose;
+ }
+
+ @Override
+ public int read(ByteBuffer dst) throws IOException {
+ ensureOpen();
+ if (position >= length) {
+ return -1;
+ }
+ int pos = (int) position;
+ int n = Math.min(dst.remaining(), length - pos);
+ dst.put(data, pos, n);
+ position += n;
+ return n;
+ }
+
+ @Override
+ public int write(ByteBuffer src) {
+ throw new NonWritableChannelException();
+ }
+
+ @Override
+ public long position() throws IOException {
+ ensureOpen();
+ return position;
+ }
+
+ @Override
+ public SeekableByteChannel position(long newPosition) throws IOException {
+ ensureOpen();
+ if (newPosition < 0) {
+ throw new IllegalArgumentException("negative position: " +
newPosition);
+ }
+ // Setting a position beyond size is legal per SeekableByteChannel;
reads return EOF
+ position = newPosition;
+ return this;
+ }
+
+ @Override
+ public long size() throws IOException {
+ ensureOpen();
+ return length;
+ }
+
+ @Override
+ public SeekableByteChannel truncate(long size) {
+ throw new NonWritableChannelException();
+ }
+
+ @Override
+ public boolean isOpen() {
+ return open;
+ }
+
+ @Override
+ public void close() {
+ if (open) {
+ open = false;
+ if (onClose != null) {
+ onClose.run();
+ }
+ }
+ }
+
+ private void ensureOpen() throws IOException {
+ if (!open) {
+ throw new ClosedChannelException();
+ }
+ }
+}
diff --git a/tika-core/src/main/java/org/apache/tika/io/ReopenableSource.java
b/tika-core/src/main/java/org/apache/tika/io/ReopenableSource.java
new file mode 100644
index 0000000000..f758f1a295
--- /dev/null
+++ b/tika-core/src/main/java/org/apache/tika/io/ReopenableSource.java
@@ -0,0 +1,319 @@
+/*
+ * 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.io;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.channels.FileChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.Arrays;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.io.function.IOSupplier;
+
+/**
+ * Input source backed by a re-openable stream supplier (e.g. an entry in a
+ * random-access {@code ZipFile}, which can be re-opened via
+ * {@code zipFile.getInputStream(entry)}).
+ * <p>
+ * Because the underlying content can be re-read on demand, rewinding simply
+ * re-opens the source, avoiding the memory-then-disk caching that
+ * {@link CachingSource} performs for one-shot streams. A temp file is created
+ * only if {@link #getPath} is called or {@link #getSeekableByteChannel()} is
+ * asked for content that does not fit in memory.
+ */
+class ReopenableSource extends InputStream implements TikaInputSource {
+
+ // Per-object bytes that may be buffered in memory without a budget
reservation;
+ // matches StreamCache's default per-object threshold.
+ private static final int IN_MEMORY_FLOOR = 1024 * 1024;
+
+ private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
+
+ private final IOSupplier<InputStream> opener;
+ private final TemporaryResources tmp;
+ private final String suffix;
+ private long length;
+
+ private InputStream currentStream; // lazily opened
+ private long position;
+ private Path spilledPath;
+ private long markPosition = -1;
+
+ private CacheMemoryBudget budget;
+ // Full content retained after an in-memory drain, with its budget
reservation,
+ // so repeated channel requests don't re-read (re-decompress) the entry.
+ private byte[] retainedBuffer;
+ private int retainedLength;
+ private long retainedReservation;
+ private int channelPins;
+ private boolean closed;
+
+ ReopenableSource(IOSupplier<InputStream> opener, TemporaryResources tmp,
long length,
+ String suffix) {
+ this.opener = opener;
+ this.tmp = tmp;
+ this.length = length;
+ this.suffix = suffix;
+ this.position = 0;
+ }
+
+ private void ensureOpen() throws IOException {
+ if (currentStream == null) {
+ currentStream = new BufferedInputStream(
+ spilledPath != null ? Files.newInputStream(spilledPath) :
opener.get());
+ }
+ }
+
+ @Override
+ public int read() throws IOException {
+ ensureOpen();
+ int b = currentStream.read();
+ if (b != -1) {
+ position++;
+ }
+ return b;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ ensureOpen();
+ int n = currentStream.read(b, off, len);
+ if (n > 0) {
+ position += n;
+ }
+ return n;
+ }
+
+ @Override
+ public long skip(long n) throws IOException {
+ ensureOpen();
+ long skipped = IOUtils.skip(currentStream, n);
+ position += skipped;
+ return skipped;
+ }
+
+ @Override
+ public int available() throws IOException {
+ ensureOpen();
+ return currentStream.available();
+ }
+
+ @Override
+ public void seekTo(long newPosition) throws IOException {
+ if (newPosition < 0) {
+ throw new IOException("Cannot seek to negative position: " +
newPosition);
+ }
+ if (currentStream != null) {
+ currentStream.close();
+ }
+ currentStream = new BufferedInputStream(
+ spilledPath != null ? Files.newInputStream(spilledPath) :
opener.get());
+ if (newPosition > 0) {
+ IOUtils.skipFully(currentStream, newPosition);
+ }
+ this.position = newPosition;
+ }
+
+ @Override
+ public boolean hasPath() {
+ return spilledPath != null;
+ }
+
+ @Override
+ public Path getPath(String suffix) throws IOException {
+ if (spilledPath == null) {
+ Path p = tmp.createTempFile(suffix == null ? this.suffix : suffix);
+ try (OutputStream out = Files.newOutputStream(p)) {
+ if (retainedBuffer != null) {
+ out.write(retainedBuffer, 0, retainedLength);
+ } else {
+ try (InputStream in = opener.get()) {
+ IOUtils.copy(in, out);
+ }
+ }
+ }
+ spilledPath = p;
+ // The spooled size is ground truth, even over a lying declared
length
+ length = Files.size(p);
+ }
+ return spilledPath;
+ }
+
+ @Override
+ public long getLength() {
+ return length;
+ }
+
+ @Override
+ public void enableRewind(CacheMemoryBudget budget) throws IOException {
+ if (position != 0) {
+ throw new IOException("Cannot enable rewind: position is " +
position +
+ ", must be 0. Call enableRewind() before reading.");
+ }
+ // No caching needed to rewind (the source re-opens); the budget is
kept for
+ // on-demand in-memory buffering in getSeekableByteChannel().
+ if (this.budget == null) {
+ this.budget = budget;
+ }
+ }
+
+ @Override
+ public SeekableByteChannel getSeekableByteChannel() throws IOException {
+ if (retainedBuffer != null) {
+ return retainedChannel();
+ }
+ if (spilledPath == null && tryBufferInMemory()) {
+ return retainedChannel();
+ }
+ return FileChannel.open(getPath(null), StandardOpenOption.READ);
+ }
+
+ /** Channels pin the retained buffer: the reservation is released only
once this source
+ * is closed AND no handed-out channel still references the array. */
+ private SeekableByteChannel retainedChannel() {
+ channelPins++;
+ return new MemorySeekableByteChannel(retainedBuffer, retainedLength,
() -> {
+ channelPins--;
+ maybeReleaseRetained();
+ });
+ }
+
+ private void maybeReleaseRetained() {
+ if (closed && channelPins == 0 && retainedReservation > 0) {
+ budget.release(retainedReservation);
+ retainedReservation = 0;
+ }
+ }
+
+ /**
+ * Drains a fresh stream into memory if it fits within the per-object
floor plus what
+ * can be reserved from the shared budget, retaining the buffer (and its
reservation)
+ * until {@link #close()}. The declared length is a sizing hint only -- it
can lie, so
+ * the cap is enforced during the read. Does not disturb this source's
read position.
+ */
+ private boolean tryBufferInMemory() throws IOException {
+ if (length > MAX_ARRAY_SIZE || (length > IN_MEMORY_FLOOR && budget ==
null)) {
+ return false;
+ }
+ long reservedHere = 0;
+ // Reservation invariant: reservedHere == max(0, data.length -
IN_MEMORY_FLOOR)
+ if (length > IN_MEMORY_FLOOR) {
+ long extra = length - IN_MEMORY_FLOOR;
+ if (budget.tryReserve(extra) != extra) {
+ return false;
+ }
+ reservedHere = extra;
+ }
+ byte[] data = new byte[length > 0 ? (int) length : 8192];
+ int total = 0;
+ boolean fits = false;
+ try (InputStream in = opener.get()) {
+ while (true) {
+ if (total == data.length) {
+ int peek = in.read();
+ if (peek == -1) {
+ fits = true;
+ break;
+ }
+ long newCapacity =
+ Math.min(MAX_ARRAY_SIZE, Math.max((long)
data.length * 2, 8192));
+ if (newCapacity > IN_MEMORY_FLOOR) {
+ if (budget == null) {
+ newCapacity = IN_MEMORY_FLOOR;
+ } else {
+ long delta = newCapacity - Math.max(data.length,
IN_MEMORY_FLOOR);
+ if (delta > 0) {
+ if (budget.tryReserve(delta) != delta) {
+ break;
+ }
+ reservedHere += delta;
+ }
+ }
+ }
+ if (newCapacity <= data.length) {
+ break;
+ }
+ data = Arrays.copyOf(data, (int) newCapacity);
+ data[total++] = (byte) peek;
+ continue;
+ }
+ int r = in.read(data, total, data.length - total);
+ if (r == -1) {
+ fits = true;
+ break;
+ }
+ total += r;
+ }
+ } finally {
+ if (!fits && reservedHere > 0) {
+ budget.release(reservedHere);
+ }
+ }
+ if (!fits) {
+ return false;
+ }
+ // Trim over-allocation from a lying declared length, releasing the
excess reservation
+ if (data.length - total > 8192) {
+ data = Arrays.copyOf(data, Math.max(total, 1));
+ long target = Math.max(0, (long) data.length - IN_MEMORY_FLOOR);
+ if (reservedHere > target) {
+ budget.release(reservedHere - target);
+ reservedHere = target;
+ }
+ }
+ retainedBuffer = data;
+ retainedLength = total;
+ retainedReservation = reservedHere;
+ // The full read is ground truth, even over a lying declared length
+ length = total;
+ return true;
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed = true;
+ retainedBuffer = null;
+ maybeReleaseRetained();
+ if (currentStream != null) {
+ currentStream.close();
+ }
+ }
+
+ @Override
+ public synchronized void mark(int readlimit) {
+ markPosition = position;
+ }
+
+ @Override
+ public synchronized void reset() throws IOException {
+ if (markPosition < 0) {
+ throw new IOException("Mark not set");
+ }
+ seekTo(markPosition);
+ }
+
+ @Override
+ public boolean markSupported() {
+ return true;
+ }
+}
diff --git a/tika-core/src/main/java/org/apache/tika/io/StreamCache.java
b/tika-core/src/main/java/org/apache/tika/io/StreamCache.java
index 9da3d69991..d3e4aaae7d 100644
--- a/tika-core/src/main/java/org/apache/tika/io/StreamCache.java
+++ b/tika-core/src/main/java/org/apache/tika/io/StreamCache.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.RandomAccessFile;
+import java.nio.channels.SeekableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -33,9 +34,20 @@ class StreamCache implements Closeable {
private static final int DEFAULT_MEMORY_THRESHOLD = 1024 * 1024; // 1MB
+ // Max size of the in-memory byte[] (a single JVM array); with a budget we
grow past the
+ // per-object threshold but never past this.
+ private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
+
private final int memoryThreshold;
private final TemporaryResources tmp;
+ // Optional shared memory budget; when non-null it governs the
memory-vs-spill decision
+ // instead of the fixed per-object memoryThreshold.
+ private final CacheMemoryBudget budget;
+ private long reserved;
+ // In-memory channels handed out that still reference memoryBuffer (see
maybeReleaseReserved)
+ private int channelPins;
+
// Memory storage (null after spill)
private byte[] memoryBuffer;
private int memorySize;
@@ -50,19 +62,19 @@ class StreamCache implements Closeable {
private boolean closed;
- StreamCache(TemporaryResources tmp) {
- this(tmp, null, DEFAULT_MEMORY_THRESHOLD);
- }
-
- /** Suffix up front: a threshold spill precedes any getPath(suffix) call
(TIKA-3903). */
- StreamCache(TemporaryResources tmp, String suffix) {
- this(tmp, suffix, DEFAULT_MEMORY_THRESHOLD);
+ /**
+ * Suffix up front: a threshold spill precedes any getPath(suffix) call
(TIKA-3903).
+ * A non-null {@code budget} allows in-memory growth past the per-object
threshold.
+ */
+ StreamCache(TemporaryResources tmp, String suffix, CacheMemoryBudget
budget) {
+ this(tmp, suffix, DEFAULT_MEMORY_THRESHOLD, budget);
}
- StreamCache(TemporaryResources tmp, String suffix, int memoryThreshold) {
+ StreamCache(TemporaryResources tmp, String suffix, int memoryThreshold,
CacheMemoryBudget budget) {
this.tmp = tmp;
this.suffix = suffix;
this.memoryThreshold = memoryThreshold;
+ this.budget = budget;
this.memoryBuffer = new byte[Math.min(memoryThreshold, 8192)];
this.memorySize = 0;
this.totalSize = 0;
@@ -78,12 +90,12 @@ class StreamCache implements Closeable {
if (memoryBuffer != null) {
// Still in memory mode
- if (memorySize >= memoryThreshold) {
- spillToFile();
- spillOutputStream.write(b);
- } else {
+ if (canKeepInMemory(1)) {
ensureMemoryCapacity(memorySize + 1);
memoryBuffer[memorySize++] = (byte) b;
+ } else {
+ spillToFile();
+ spillOutputStream.write(b);
}
} else {
// Already spilled to file
@@ -101,13 +113,13 @@ class StreamCache implements Closeable {
}
if (memoryBuffer != null) {
- if (memorySize + len > memoryThreshold) {
- spillToFile();
- spillOutputStream.write(b, off, len);
- } else {
+ if (canKeepInMemory(len)) {
ensureMemoryCapacity(memorySize + len);
System.arraycopy(b, off, memoryBuffer, memorySize, len);
memorySize += len;
+ } else {
+ spillToFile();
+ spillOutputStream.write(b, off, len);
}
} else {
spillOutputStream.write(b, off, len);
@@ -115,17 +127,62 @@ class StreamCache implements Closeable {
totalSize += len;
}
+ /**
+ * Decide whether {@code additional} more bytes can stay in memory.
Content up to the
+ * per-object threshold always may (budget or not, so budget exhaustion is
never worse
+ * than the no-budget default); beyond it, the shared budget must cover
the *capacity*
+ * the backing array will actually grow to (arrays grow in doubling steps,
so logical
+ * bytes would under-count real heap). Invariant:
+ * {@code reserved == max(0, memoryBuffer.length - memoryThreshold)}.
+ */
+ private boolean canKeepInMemory(int additional) {
+ long needed = memorySize + (long) additional;
+ if (needed <= memoryThreshold) {
+ return true;
+ }
+ if (budget == null || needed > MAX_ARRAY_SIZE) {
+ return false;
+ }
+ if (needed <= memoryBuffer.length) {
+ return true;
+ }
+ long targetCapacity =
+ Math.min(MAX_ARRAY_SIZE, Math.max((long) memoryBuffer.length *
2, needed));
+ long delta = targetCapacity - Math.max(memoryThreshold,
memoryBuffer.length);
+ if (budget.tryReserve(delta) != delta) {
+ return false;
+ }
+ reserved += delta;
+ return true;
+ }
+
+ /** Releases the reservation once the buffer is no longer held by this
cache (spilled or
+ * closed) AND no handed-out in-memory channel still pins the array. */
+ private void maybeReleaseReserved() {
+ if (budget != null && reserved > 0 && channelPins == 0 &&
+ (closed || memoryBuffer == null)) {
+ budget.release(reserved);
+ reserved = 0;
+ }
+ }
+
private void ensureMemoryCapacity(int needed) {
if (needed <= memoryBuffer.length) {
return;
}
- int newSize = Math.min(memoryThreshold, Math.max(memoryBuffer.length *
2, needed));
+ // With a budget, growth is capped at the reserved capacity (see
canKeepInMemory)
+ long cap = (budget == null) ? memoryThreshold
+ : Math.min((long) memoryThreshold + reserved, MAX_ARRAY_SIZE);
+ int newSize = (int) Math.min(cap, Math.max((long) memoryBuffer.length
* 2, needed));
byte[] newBuffer = new byte[newSize];
System.arraycopy(memoryBuffer, 0, newBuffer, 0, memorySize);
memoryBuffer = newBuffer;
}
private void spillToFile() throws IOException {
+ if (closed) {
+ throw new IOException("StreamCache is closed");
+ }
if (spillFile != null) {
return; // Already spilled
}
@@ -144,6 +201,7 @@ class StreamCache implements Closeable {
// Release memory buffer
memoryBuffer = null;
memorySize = 0;
+ maybeReleaseReserved();
}
/**
@@ -216,6 +274,9 @@ class StreamCache implements Closeable {
* After this call, the cache is in file-backed mode.
*/
Path toFile() throws IOException {
+ if (closed) {
+ throw new IOException("StreamCache is closed");
+ }
if (spillFile == null) {
spillToFile();
}
@@ -251,6 +312,24 @@ class StreamCache implements Closeable {
return spillFile != null;
}
+ /**
+ * If the full content is currently held in memory (not spilled, not
closed), returns a
+ * read-only random-access channel over it (no copy, no disk). Otherwise
{@code null}.
+ * The caller is responsible only for content that has actually been
cached so far.
+ */
+ SeekableByteChannel getInMemorySeekableByteChannel() {
+ if (closed || memoryBuffer == null) {
+ return null;
+ }
+ // The channel pins the array: the budget reservation is held until
this cache no
+ // longer needs the buffer (spill/close) AND all handed-out channels
are closed.
+ channelPins++;
+ return new MemorySeekableByteChannel(memoryBuffer, memorySize, () -> {
+ channelPins--;
+ maybeReleaseReserved();
+ });
+ }
+
@Override
public void close() throws IOException {
if (closed) {
@@ -258,6 +337,7 @@ class StreamCache implements Closeable {
}
closed = true;
memoryBuffer = null;
+ maybeReleaseReserved();
if (spillOutputStream != null) {
spillOutputStream.close();
diff --git a/tika-core/src/main/java/org/apache/tika/io/TikaInputSource.java
b/tika-core/src/main/java/org/apache/tika/io/TikaInputSource.java
index 81bc69eaba..53d8942399 100644
--- a/tika-core/src/main/java/org/apache/tika/io/TikaInputSource.java
+++ b/tika-core/src/main/java/org/apache/tika/io/TikaInputSource.java
@@ -18,6 +18,7 @@ package org.apache.tika.io;
import java.io.Closeable;
import java.io.IOException;
+import java.nio.channels.SeekableByteChannel;
import java.nio.file.Path;
/**
@@ -62,7 +63,22 @@ interface TikaInputSource extends Closeable {
* <p>
* Must be called when position is 0, otherwise throws IOException.
*
+ * @param budget shared memory budget governing how much a caching source
may hold in
+ * memory before spilling, or {@code null} for the
per-object default;
+ * inherently rewindable sources ignore it
* @throws IOException if position is not 0
*/
- void enableRewind() throws IOException;
+ void enableRewind(CacheMemoryBudget budget) throws IOException;
+
+ /**
+ * Returns a read-only random-access channel over this source's full
content: content
+ * already in memory (byte[], unspilled cache) is served from memory;
file-backed or
+ * spilled content from a file channel; unread stream content is drained
through the
+ * cache, which decides memory-vs-disk during the drain. Fails for a
stream-backed
+ * source that has been partially read without rewind enabled. Callers own
closing the
+ * returned channel. Does not change this source's read position.
+ *
+ * @throws IOException if the source is partially read and cannot be
rewound
+ */
+ SeekableByteChannel getSeekableByteChannel() throws IOException;
}
diff --git a/tika-core/src/main/java/org/apache/tika/io/TikaInputStream.java
b/tika-core/src/main/java/org/apache/tika/io/TikaInputStream.java
index 320b64f68c..6de71cfa8e 100644
--- a/tika-core/src/main/java/org/apache/tika/io/TikaInputStream.java
+++ b/tika-core/src/main/java/org/apache/tika/io/TikaInputStream.java
@@ -26,6 +26,7 @@ import java.net.URISyntaxException;
import java.net.URL;
import java.net.URLConnection;
import java.nio.channels.FileChannel;
+import java.nio.channels.SeekableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -33,6 +34,7 @@ import java.sql.Blob;
import java.sql.SQLException;
import org.apache.commons.io.IOUtils;
+import org.apache.commons.io.function.IOSupplier;
import org.apache.commons.io.input.TaggedInputStream;
import org.apache.tika.metadata.HttpHeaders;
@@ -101,6 +103,40 @@ public class TikaInputStream extends TaggedInputStream {
return new TikaInputStream(inputSource, tmp, ext);
}
+ /**
+ * Creates a TikaInputStream from a re-openable stream supplier. Unlike
+ * {@link #get(InputStream, TemporaryResources, Metadata)} -- which caches
a one-shot
+ * stream to memory/disk so it can be rewound -- the supplier is
re-invoked to re-read
+ * the content, so rewinding (e.g. during digesting) never spills to disk.
A temp file
+ * is created only if {@link #getPath()} is later called (a
parser/detector needing a
+ * File) or {@link #getSeekableByteChannel()} is asked for content that
does not fit
+ * in memory.
+ *
+ * @param opener supplies a fresh InputStream over the same content on
each call
+ * @param tmp temporary resources for any on-demand {@link
#getPath()} spill
+ * @param metadata metadata used for extension/length hints; may be null
+ */
+ public static TikaInputStream get(IOSupplier<InputStream> opener,
TemporaryResources tmp,
+ Metadata metadata) {
+ if (opener == null) {
+ throw new NullPointerException("The opener must not be null");
+ }
+ String ext = getExtension(metadata);
+ long length = -1;
+ if (metadata != null) {
+ String cl = metadata.get(HttpHeaders.CONTENT_LENGTH);
+ if (cl != null) {
+ try {
+ length = Long.parseLong(cl);
+ } catch (NumberFormatException e) {
+ length = -1;
+ }
+ }
+ }
+ TikaInputSource inputSource = new ReopenableSource(opener, tmp,
length, ext);
+ return new TikaInputStream(inputSource, tmp, ext);
+ }
+
public static TikaInputStream get(InputStream stream) {
return get(stream, new TemporaryResources(), null);
}
@@ -478,10 +514,42 @@ public class TikaInputStream extends TaggedInputStream {
* (position is not 0); rewind support cannot be enabled
retroactively
*/
public void enableRewind() throws IOException {
+ enableRewind(null);
+ }
+
+ /**
+ * Like {@link #enableRewind()}, but supplies a shared {@link
CacheMemoryBudget} governing
+ * how much may be held in memory before spilling to disk (used only by
stream-backed
+ * sources); {@code null} falls back to the per-object default.
+ *
+ * @param budget shared memory budget, or {@code null}
+ * @throws IOException if bytes have already been read (position is not 0)
+ */
+ public void enableRewind(CacheMemoryBudget budget) throws IOException {
TikaInputSource source = inputSource();
if (source != null) {
- source.enableRewind();
+ source.enableRewind(budget);
+ }
+ }
+
+ /**
+ * Returns a read-only random-access {@link SeekableByteChannel} over this
stream's full
+ * content. Unlike {@link #getPath()}/{@link #getFile()}, this never
forces content that is
+ * already in memory onto disk: in-memory content is served from memory,
file-backed or
+ * spilled content from a file channel, and unread stream content is
drained through the
+ * cache which decides memory-vs-disk as it goes. Use this when random
access is needed
+ * (e.g. reading a zip central directory); reserve {@code getFile()} for
callers that truly
+ * need a {@link java.io.File}. The caller owns closing the returned
channel. Does not
+ * disturb this stream's read position.
+ *
+ * @throws IOException if this stream has been partially read without
rewind enabled
+ */
+ public SeekableByteChannel getSeekableByteChannel() throws IOException {
+ TikaInputSource source = inputSource();
+ if (source == null) {
+ throw new IOException("No TikaInputSource available");
}
+ return source.getSeekableByteChannel();
}
@Override
diff --git
a/tika-core/src/main/java/org/apache/tika/parser/multiple/AbstractMultipleParser.java
b/tika-core/src/main/java/org/apache/tika/parser/multiple/AbstractMultipleParser.java
index 3e928fd751..ee7c55210b 100644
---
a/tika-core/src/main/java/org/apache/tika/parser/multiple/AbstractMultipleParser.java
+++
b/tika-core/src/main/java/org/apache/tika/parser/multiple/AbstractMultipleParser.java
@@ -33,6 +33,7 @@ import org.xml.sax.ContentHandler;
import org.xml.sax.SAXException;
import org.apache.tika.exception.TikaException;
+import org.apache.tika.io.CacheMemoryBudget;
import org.apache.tika.io.TemporaryResources;
import org.apache.tika.io.TikaInputStream;
import org.apache.tika.metadata.Metadata;
@@ -244,7 +245,7 @@ public abstract class AbstractMultipleParser implements
Parser {
ContentHandlerFactory handlerFactory, Metadata
originalMetadata,
ParseContext context) throws IOException, SAXException,
TikaException {
// Enable rewind capability since we rewind between multiple parsers
- tis.enableRewind();
+ tis.enableRewind(context == null ? null :
context.get(CacheMemoryBudget.class));
// Track the metadata between parsers, so we can apply our policy
Metadata lastMetadata = cloneMetadata(originalMetadata);
diff --git
a/tika-core/src/test/java/org/apache/tika/io/CacheMemoryBudgetTest.java
b/tika-core/src/test/java/org/apache/tika/io/CacheMemoryBudgetTest.java
new file mode 100644
index 0000000000..c24c53facb
--- /dev/null
+++ b/tika-core/src/test/java/org/apache/tika/io/CacheMemoryBudgetTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.io;
+
+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.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.jupiter.api.Test;
+
+public class CacheMemoryBudgetTest {
+
+ @Test
+ public void testCtorRejectsNonPositive() {
+ assertThrows(IllegalArgumentException.class, () -> new
CacheMemoryBudget(0));
+ assertThrows(IllegalArgumentException.class, () -> new
CacheMemoryBudget(-1));
+ }
+
+ @Test
+ public void testAllOrNothing() {
+ CacheMemoryBudget budget = new CacheMemoryBudget(100);
+ assertEquals(60, budget.tryReserve(60));
+ // only 40 left; a 60-byte request must reserve nothing, not 40
+ assertEquals(0, budget.tryReserve(60));
+ assertEquals(60, budget.getReservedBytes());
+ assertEquals(40, budget.tryReserve(40));
+ assertEquals(0, budget.tryReserve(1));
+ }
+
+ @Test
+ public void testZeroAndNegativeRequests() {
+ CacheMemoryBudget budget = new CacheMemoryBudget(100);
+ assertEquals(0, budget.tryReserve(0));
+ assertEquals(0, budget.tryReserve(-5));
+ assertEquals(0, budget.getReservedBytes());
+ }
+
+ @Test
+ public void testReleaseRestoresCapacity() {
+ CacheMemoryBudget budget = new CacheMemoryBudget(100);
+ assertEquals(100, budget.tryReserve(100));
+ assertEquals(0, budget.tryReserve(1));
+ budget.release(30);
+ assertEquals(30, budget.tryReserve(30));
+ budget.release(100);
+ assertEquals(0, budget.getReservedBytes());
+ assertEquals(100, budget.tryReserve(100));
+ }
+
+ @Test
+ public void testConcurrency() throws Exception {
+ final CacheMemoryBudget budget = new CacheMemoryBudget(1000);
+ final AtomicLong netReserved = new AtomicLong();
+ int threads = 8;
+ ExecutorService executor = Executors.newFixedThreadPool(threads);
+ try {
+ List<Callable<Void>> tasks = new ArrayList<>();
+ for (int t = 0; t < threads; t++) {
+ final long seed = 42L + t;
+ tasks.add(() -> {
+ Random random = new Random(seed);
+ for (int i = 0; i < 5000; i++) {
+ long n = 1 + random.nextInt(100);
+ long got = budget.tryReserve(n);
+ if (got > 0) {
+ netReserved.addAndGet(got);
+ long reserved = budget.getReservedBytes();
+ assertTrue(reserved <= 1000, "over-reserved: " +
reserved);
+ budget.release(got);
+ netReserved.addAndGet(-got);
+ }
+ }
+ return null;
+ });
+ }
+ for (Future<Void> f : executor.invokeAll(tasks)) {
+ f.get();
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ assertEquals(0, netReserved.get());
+ assertEquals(0, budget.getReservedBytes());
+ }
+}
diff --git
a/tika-core/src/test/java/org/apache/tika/io/MemorySeekableByteChannelTest.java
b/tika-core/src/test/java/org/apache/tika/io/MemorySeekableByteChannelTest.java
new file mode 100644
index 0000000000..4c982e278a
--- /dev/null
+++
b/tika-core/src/test/java/org/apache/tika/io/MemorySeekableByteChannelTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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.io;
+
+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.assertThrows;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.FileChannel;
+import java.nio.channels.NonWritableChannelException;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Conformance test: MemorySeekableByteChannel must behave like a read-only
FileChannel over
+ * the same bytes -- external code (commons-compress ZipFile/SevenZFile) is
handed either and
+ * is entitled to spec behavior for both. Each test runs against both kinds.
+ */
+public class MemorySeekableByteChannelTest {
+
+ private static final byte[] DATA =
+
"0123456789abcdefghijklmnopqrstuvwxyz".getBytes(StandardCharsets.UTF_8);
+
+ private static final List<String> KINDS = Arrays.asList("memory", "file");
+
+ @TempDir
+ Path tmp;
+
+ private SeekableByteChannel open(String kind) throws IOException {
+ if ("memory".equals(kind)) {
+ return new MemorySeekableByteChannel(DATA, DATA.length);
+ }
+ Path p = Files.createTempFile(tmp, "msbc", ".bin");
+ Files.write(p, DATA);
+ return FileChannel.open(p, StandardOpenOption.READ);
+ }
+
+ @Test
+ public void testSequentialRead() throws Exception {
+ for (String kind : KINDS) {
+ try (SeekableByteChannel channel = open(kind)) {
+ assertEquals(DATA.length, channel.size(), kind);
+ ByteBuffer buf = ByteBuffer.allocate(10);
+ assertEquals(10, channel.read(buf), kind);
+ assertArrayEquals(Arrays.copyOfRange(DATA, 0, 10),
buf.array(), kind);
+ assertEquals(10, channel.position(), kind);
+ buf.clear();
+ int total = 10;
+ int n;
+ while ((n = channel.read(buf)) != -1) {
+ total += n;
+ buf.clear();
+ }
+ assertEquals(DATA.length, total, kind);
+ }
+ }
+ }
+
+ @Test
+ public void testPositionRoundTripIncludingBeyondSize() throws Exception {
+ for (String kind : KINDS) {
+ try (SeekableByteChannel channel = open(kind)) {
+ channel.position(5);
+ assertEquals(5, channel.position(), kind);
+ ByteBuffer one = ByteBuffer.allocate(1);
+ channel.read(one);
+ assertEquals(DATA[5], one.get(0), kind);
+
+ // Setting a position beyond size is legal; position() must
echo it and
+ // reads must return EOF
+ long beyond = DATA.length + 100L;
+ channel.position(beyond);
+ assertEquals(beyond, channel.position(), kind);
+ one.clear();
+ assertEquals(-1, channel.read(one), kind);
+ assertEquals(beyond, channel.position(), kind);
+ assertEquals(DATA.length, channel.size(), kind);
+
+ channel.position(DATA.length);
+ one.clear();
+ assertEquals(-1, channel.read(one), kind);
+ }
+ }
+ }
+
+ @Test
+ public void testNegativePositionRejected() throws Exception {
+ for (String kind : KINDS) {
+ try (SeekableByteChannel channel = open(kind)) {
+ assertThrows(IllegalArgumentException.class, () ->
channel.position(-1), kind);
+ }
+ }
+ }
+
+ @Test
+ public void testZeroRemainingBufferReadsZero() throws Exception {
+ for (String kind : KINDS) {
+ try (SeekableByteChannel channel = open(kind)) {
+ assertEquals(0, channel.read(ByteBuffer.allocate(0)), kind);
+ }
+ }
+ }
+
+ @Test
+ public void testReadOnly() throws Exception {
+ for (String kind : KINDS) {
+ try (SeekableByteChannel channel = open(kind)) {
+ assertThrows(NonWritableChannelException.class,
+ () -> channel.write(ByteBuffer.allocate(1)), kind);
+ assertThrows(NonWritableChannelException.class, () ->
channel.truncate(1), kind);
+ }
+ }
+ }
+
+ @Test
+ public void testClosedChannel() throws Exception {
+ for (String kind : KINDS) {
+ SeekableByteChannel channel = open(kind);
+ channel.close();
+ assertFalse(channel.isOpen(), kind);
+ assertThrows(ClosedChannelException.class,
+ () -> channel.read(ByteBuffer.allocate(1)), kind);
+ assertThrows(ClosedChannelException.class, channel::position,
kind);
+ assertThrows(ClosedChannelException.class, () ->
channel.position(0), kind);
+ assertThrows(ClosedChannelException.class, channel::size, kind);
+ channel.close(); // idempotent
+ }
+ }
+
+ @Test
+ public void testPartialLengthView() throws Exception {
+ // MemorySeekableByteChannel may expose a prefix of a larger array
+ try (SeekableByteChannel channel = new MemorySeekableByteChannel(DATA,
10)) {
+ assertEquals(10, channel.size());
+ ByteBuffer buf = ByteBuffer.allocate(100);
+ assertEquals(10, channel.read(buf));
+ assertEquals(-1, channel.read(buf));
+ }
+ }
+}
diff --git
a/tika-core/src/test/java/org/apache/tika/io/ReopenableSourceTest.java
b/tika-core/src/test/java/org/apache/tika/io/ReopenableSourceTest.java
new file mode 100644
index 0000000000..d3142dd461
--- /dev/null
+++ b/tika-core/src/test/java/org/apache/tika/io/ReopenableSourceTest.java
@@ -0,0 +1,303 @@
+/*
+ * 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.io;
+
+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.assertInstanceOf;
+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.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Path;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.commons.io.function.IOSupplier;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+public class ReopenableSourceTest {
+
+ private static final int FLOOR = 1024 * 1024; //
ReopenableSource.IN_MEMORY_FLOOR
+
+ private final TemporaryResources tmp = new TemporaryResources();
+
+ @AfterEach
+ public void tearDown() throws IOException {
+ tmp.close();
+ }
+
+ private static byte[] data(int size) {
+ byte[] data = new byte[size];
+ for (int i = 0; i < size; i++) {
+ data[i] = (byte) (i * 31 + 7);
+ }
+ return data;
+ }
+
+ private static IOSupplier<InputStream> countingOpener(byte[] data,
AtomicInteger opens) {
+ return () -> {
+ opens.incrementAndGet();
+ return new ByteArrayInputStream(data);
+ };
+ }
+
+ private static byte[] readFully(SeekableByteChannel channel) throws
IOException {
+ ByteBuffer buf = ByteBuffer.allocate((int) channel.size());
+ while (buf.hasRemaining() && channel.read(buf) != -1) {
+ // keep reading
+ }
+ return buf.array();
+ }
+
+ @Test
+ public void testReadAndRewindReopens() throws Exception {
+ byte[] data = data(1000);
+ AtomicInteger opens = new AtomicInteger();
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, null)) {
+ source.enableRewind(null);
+ byte[] first = source.readAllBytes();
+ assertArrayEquals(data, first);
+ assertEquals(1, opens.get());
+ source.seekTo(0);
+ byte[] second = source.readAllBytes();
+ assertArrayEquals(data, second);
+ assertEquals(2, opens.get());
+ }
+ }
+
+ @Test
+ public void testMarkReset() throws Exception {
+ byte[] data = data(1000);
+ AtomicInteger opens = new AtomicInteger();
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, null)) {
+ byte[] buf = new byte[100];
+ source.readNBytes(buf, 0, buf.length);
+ source.mark(0);
+ source.readNBytes(buf, 0, buf.length);
+ source.reset();
+ int b = source.read();
+ assertEquals(data[100] & 0xFF, b);
+ }
+ }
+
+ @Test
+ public void testResetWithoutMarkThrows() throws Exception {
+ try (ReopenableSource source = new ReopenableSource(
+ countingOpener(data(10), new AtomicInteger()), tmp, 10, null))
{
+ assertThrows(IOException.class, source::reset);
+ }
+ }
+
+ @Test
+ public void testEnableRewindAfterReadThrows() throws Exception {
+ try (ReopenableSource source = new ReopenableSource(
+ countingOpener(data(10), new AtomicInteger()), tmp, 10, null))
{
+ source.enableRewind(null); // at 0: fine
+ source.read();
+ assertThrows(IOException.class, () -> source.enableRewind(null));
+ }
+ }
+
+ @Test
+ public void testGetPathSpillsOnceWithSuffix() throws Exception {
+ byte[] data = data(1000);
+ AtomicInteger opens = new AtomicInteger();
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, ".zip")) {
+ assertFalse(source.hasPath());
+ Path p = source.getPath(null);
+ assertTrue(source.hasPath());
+ assertTrue(p.getFileName().toString().endsWith(".zip"));
+ assertEquals(1, opens.get());
+ assertEquals(p, source.getPath(null));
+ assertEquals(1, opens.get());
+ // post-spill reads come from the file, not the opener
+ source.seekTo(0);
+ assertArrayEquals(data, source.readAllBytes());
+ assertEquals(1, opens.get());
+ }
+ }
+
+ @Test
+ public void testChannelSmallContentInMemoryAndRetained() throws Exception {
+ byte[] data = data(50_000);
+ AtomicInteger opens = new AtomicInteger();
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, null)) {
+ try (SeekableByteChannel channel =
source.getSeekableByteChannel()) {
+ assertInstanceOf(MemorySeekableByteChannel.class, channel);
+ assertArrayEquals(data, readFully(channel));
+ }
+ assertEquals(1, opens.get());
+ assertFalse(source.hasPath());
+ // second channel served from the retained buffer -- no re-read
+ try (SeekableByteChannel channel =
source.getSeekableByteChannel()) {
+ assertArrayEquals(data, readFully(channel));
+ }
+ assertEquals(1, opens.get());
+ }
+ }
+
+ @Test
+ public void testChannelDoesNotDisturbReadPosition() throws Exception {
+ byte[] data = data(1000);
+ AtomicInteger opens = new AtomicInteger();
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, null)) {
+ byte[] buf = new byte[10];
+ source.readNBytes(buf, 0, buf.length);
+ try (SeekableByteChannel channel =
source.getSeekableByteChannel()) {
+ readFully(channel);
+ }
+ int b = source.read();
+ assertEquals(data[10] & 0xFF, b);
+ }
+ }
+
+ @Test
+ public void testChannelOverFloorWithoutBudgetSpills() throws Exception {
+ byte[] data = data(FLOOR + 1);
+ AtomicInteger opens = new AtomicInteger();
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, null)) {
+ try (SeekableByteChannel channel =
source.getSeekableByteChannel()) {
+ assertInstanceOf(FileChannel.class, channel);
+ assertEquals(data.length, channel.size());
+ }
+ assertTrue(source.hasPath());
+ // declared length routed straight to spill: exactly one full read
+ assertEquals(1, opens.get());
+ }
+ }
+
+ @Test
+ public void testChannelOverFloorWithBudgetStaysInMemory() throws Exception
{
+ byte[] data = data(FLOOR + 100);
+ AtomicInteger opens = new AtomicInteger();
+ CacheMemoryBudget budget = new CacheMemoryBudget(16L * 1024 * 1024);
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, null)) {
+ source.enableRewind(budget);
+ try (SeekableByteChannel channel =
source.getSeekableByteChannel()) {
+ assertInstanceOf(MemorySeekableByteChannel.class, channel);
+ assertArrayEquals(data, readFully(channel));
+ }
+ assertFalse(source.hasPath());
+ assertTrue(budget.getReservedBytes() > 0, "capacity beyond the
floor is reserved");
+ }
+ assertEquals(0, budget.getReservedBytes(), "close() releases the
retained reservation");
+ }
+
+ @Test
+ public void testChannelOutlivingSourceHoldsReservation() throws Exception {
+ byte[] data = data(FLOOR + 100);
+ CacheMemoryBudget budget = new CacheMemoryBudget(16L * 1024 * 1024);
+ ReopenableSource source = new ReopenableSource(
+ countingOpener(data, new AtomicInteger()), tmp, data.length,
null);
+ source.enableRewind(budget);
+ SeekableByteChannel channel = source.getSeekableByteChannel();
+ source.close();
+ assertTrue(budget.getReservedBytes() > 0,
+ "reservation must be held while a channel still pins the
retained buffer");
+ assertArrayEquals(data, readFully(channel));
+ channel.close();
+ assertEquals(0, budget.getReservedBytes(), "last channel close
releases");
+ channel.close(); // idempotent: no double-release
+ assertEquals(0, budget.getReservedBytes());
+ }
+
+ @Test
+ public void testChannelBudgetExhaustedSpills() throws Exception {
+ byte[] data = data(FLOOR + 2048);
+ AtomicInteger opens = new AtomicInteger();
+ CacheMemoryBudget budget = new CacheMemoryBudget(1024); // too small
for beyond-floor
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, null)) {
+ source.enableRewind(budget);
+ try (SeekableByteChannel channel =
source.getSeekableByteChannel()) {
+ assertInstanceOf(FileChannel.class, channel);
+ assertEquals(data.length, channel.size());
+ }
+ assertEquals(0, budget.getReservedBytes(), "failed reservation
fully released");
+ }
+ }
+
+ @Test
+ public void testLyingDeclaredLengthCorrected() throws Exception {
+ byte[] data = data(500);
+ AtomicInteger opens = new AtomicInteger();
+ // declared length lies high
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
400_000, null)) {
+ assertEquals(400_000, source.getLength());
+ try (SeekableByteChannel channel =
source.getSeekableByteChannel()) {
+ assertEquals(500, channel.size());
+ }
+ assertEquals(500, source.getLength(), "full read overrides the
lying declared length");
+ }
+ // declared length lies high, spill path
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
400_000, null)) {
+ source.getPath(null);
+ assertEquals(500, source.getLength());
+ }
+ // declared length lies low: growth still captures everything
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
100, null)) {
+ try (SeekableByteChannel channel =
source.getSeekableByteChannel()) {
+ assertArrayEquals(data, readFully(channel));
+ }
+ assertEquals(500, source.getLength());
+ }
+ }
+
+ @Test
+ public void testUnknownLengthChannel() throws Exception {
+ byte[] data = data(100_000);
+ AtomicInteger opens = new AtomicInteger();
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
-1, null)) {
+ assertEquals(-1, source.getLength());
+ try (SeekableByteChannel channel =
source.getSeekableByteChannel()) {
+ assertInstanceOf(MemorySeekableByteChannel.class, channel);
+ assertArrayEquals(data, readFully(channel));
+ }
+ assertEquals(data.length, source.getLength());
+ }
+ }
+
+ @Test
+ public void testEmptyContent() throws Exception {
+ AtomicInteger opens = new AtomicInteger();
+ try (ReopenableSource source =
+ new ReopenableSource(countingOpener(new byte[0], opens),
tmp, 0, null)) {
+ try (SeekableByteChannel channel =
source.getSeekableByteChannel()) {
+ assertEquals(0, channel.size());
+ assertEquals(-1, channel.read(ByteBuffer.allocate(1)));
+ }
+ }
+ }
+}
diff --git
a/tika-core/src/test/java/org/apache/tika/io/StreamCacheBudgetTest.java
b/tika-core/src/test/java/org/apache/tika/io/StreamCacheBudgetTest.java
new file mode 100644
index 0000000000..0b2b7a941b
--- /dev/null
+++ b/tika-core/src/test/java/org/apache/tika/io/StreamCacheBudgetTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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.io;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+public class StreamCacheBudgetTest {
+
+ private static final int THRESHOLD = 1024;
+
+ private final TemporaryResources tmp = new TemporaryResources();
+
+ @AfterEach
+ public void tearDown() throws IOException {
+ tmp.close();
+ }
+
+ private static void append(StreamCache cache, int n) throws IOException {
+ byte[] data = new byte[n];
+ cache.append(data, 0, n);
+ }
+
+ @Test
+ public void testNoBudgetSpillsAtThreshold() throws Exception {
+ try (StreamCache cache = new StreamCache(tmp, null, THRESHOLD, null)) {
+ append(cache, THRESHOLD);
+ assertFalse(cache.isFileBacked());
+ append(cache, 1);
+ assertTrue(cache.isFileBacked());
+ }
+ }
+
+ @Test
+ public void testBudgetAllowsGrowthPastThreshold() throws Exception {
+ CacheMemoryBudget budget = new CacheMemoryBudget(1024 * 1024);
+ try (StreamCache cache = new StreamCache(tmp, null, THRESHOLD,
budget)) {
+ append(cache, 10 * THRESHOLD);
+ assertFalse(cache.isFileBacked());
+ // channel must be closed: an open channel pins the reservation
past cache close
+ try (java.nio.channels.SeekableByteChannel channel =
+ cache.getInMemorySeekableByteChannel()) {
+ assertNotNull(channel);
+ }
+ assertTrue(budget.getReservedBytes() >= 9 * THRESHOLD,
+ "capacity beyond the threshold is reserved");
+ }
+ assertEquals(0, budget.getReservedBytes(), "close() releases the
reservation");
+ }
+
+ @Test
+ public void testUpToThresholdNeverTouchesBudget() throws Exception {
+ CacheMemoryBudget budget = new CacheMemoryBudget(1);
+ try (StreamCache cache = new StreamCache(tmp, null, THRESHOLD,
budget)) {
+ // an exhausted/tiny budget must not force sub-threshold content
to disk
+ append(cache, THRESHOLD);
+ assertFalse(cache.isFileBacked());
+ assertEquals(0, budget.getReservedBytes());
+ }
+ }
+
+ @Test
+ public void testBudgetExhaustedSpillsAndReleases() throws Exception {
+ CacheMemoryBudget budget = new CacheMemoryBudget(2048);
+ try (StreamCache cache = new StreamCache(tmp, null, THRESHOLD,
budget)) {
+ append(cache, 100 * THRESHOLD);
+ assertTrue(cache.isFileBacked());
+ assertEquals(0, budget.getReservedBytes(), "spill returns the
reservation");
+ assertNull(cache.getInMemorySeekableByteChannel());
+ assertEquals(100 * THRESHOLD, cache.size());
+ }
+ }
+
+ @Test
+ public void testTwoCachesShareBudget() throws Exception {
+ CacheMemoryBudget budget = new CacheMemoryBudget(8 * THRESHOLD);
+ try (StreamCache first = new StreamCache(tmp, null, THRESHOLD, budget);
+ StreamCache second = new StreamCache(tmp, null, THRESHOLD,
budget)) {
+ append(first, 8 * THRESHOLD);
+ assertFalse(first.isFileBacked());
+ // pool is (mostly) consumed; the second cache must spill past its
threshold
+ append(second, 100 * THRESHOLD);
+ assertTrue(second.isFileBacked());
+ first.close();
+ // after the first cache releases, a third can grow again
+ try (StreamCache third = new StreamCache(tmp, null, THRESHOLD,
budget)) {
+ append(third, 4 * THRESHOLD);
+ assertFalse(third.isFileBacked());
+ }
+ }
+ }
+
+ @Test
+ public void testReadBackIntact() throws Exception {
+ CacheMemoryBudget budget = new CacheMemoryBudget(1024 * 1024);
+ try (StreamCache cache = new StreamCache(tmp, null, THRESHOLD,
budget)) {
+ byte[] data = new byte[5000];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = (byte) i;
+ }
+ cache.append(data, 0, data.length);
+ byte[] out = new byte[data.length];
+ assertEquals(data.length, cache.readAt(0, out, 0, out.length));
+ for (int i = 0; i < data.length; i++) {
+ assertEquals(data[i], out[i]);
+ }
+ }
+ }
+
+ @Test
+ public void testChannelOutlivingCacheHoldsReservation() throws Exception {
+ CacheMemoryBudget budget = new CacheMemoryBudget(1024 * 1024);
+ StreamCache cache = new StreamCache(tmp, null, THRESHOLD, budget);
+ append(cache, 10 * THRESHOLD);
+ java.nio.channels.SeekableByteChannel channel =
cache.getInMemorySeekableByteChannel();
+ assertNotNull(channel);
+ cache.close();
+ assertTrue(budget.getReservedBytes() > 0,
+ "reservation must be held while a channel still pins the
buffer");
+ channel.close();
+ assertEquals(0, budget.getReservedBytes(), "last channel close
releases");
+ channel.close(); // idempotent: no double-release
+ assertEquals(0, budget.getReservedBytes());
+ }
+
+ @Test
+ public void testUseAfterCloseThrows() throws Exception {
+ StreamCache cache = new StreamCache(tmp, null, THRESHOLD, null);
+ append(cache, 10);
+ cache.close();
+ assertThrows(IOException.class, () -> cache.append(new byte[1], 0, 1));
+ assertThrows(IOException.class, cache::toFile);
+ assertNull(cache.getInMemorySeekableByteChannel());
+ }
+}
diff --git
a/tika-core/src/test/java/org/apache/tika/io/TikaInputStreamChannelTest.java
b/tika-core/src/test/java/org/apache/tika/io/TikaInputStreamChannelTest.java
new file mode 100644
index 0000000000..c0135f7618
--- /dev/null
+++ b/tika-core/src/test/java/org/apache/tika/io/TikaInputStreamChannelTest.java
@@ -0,0 +1,159 @@
+/*
+ * 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.io;
+
+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.assertInstanceOf;
+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.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.metadata.HttpHeaders;
+import org.apache.tika.metadata.Metadata;
+
+/**
+ * TikaInputStream.getSeekableByteChannel() across the source types.
+ */
+public class TikaInputStreamChannelTest {
+
+ @TempDir
+ Path tmp;
+
+ private static byte[] data(int size) {
+ byte[] data = new byte[size];
+ for (int i = 0; i < size; i++) {
+ data[i] = (byte) (i * 17 + 3);
+ }
+ return data;
+ }
+
+ private static byte[] readFully(SeekableByteChannel channel) throws
IOException {
+ ByteBuffer buf = ByteBuffer.allocate((int) channel.size());
+ while (buf.hasRemaining() && channel.read(buf) != -1) {
+ // keep reading
+ }
+ return buf.array();
+ }
+
+ @Test
+ public void testByteArrayStaysInMemory() throws Exception {
+ byte[] data = data(10_000);
+ try (TikaInputStream tis = TikaInputStream.get(data)) {
+ try (SeekableByteChannel channel = tis.getSeekableByteChannel()) {
+ assertArrayEquals(data, readFully(channel));
+ }
+ assertFalse(tis.hasFile(), "in-memory content must not be forced
to disk");
+ }
+ }
+
+ @Test
+ public void testFileBacked() throws Exception {
+ byte[] data = data(10_000);
+ Path p = Files.createTempFile(tmp, "chan", ".bin");
+ Files.write(p, data);
+ try (TikaInputStream tis = TikaInputStream.get(p)) {
+ try (SeekableByteChannel channel = tis.getSeekableByteChannel()) {
+ assertInstanceOf(FileChannel.class, channel);
+ assertArrayEquals(data, readFully(channel));
+ }
+ }
+ }
+
+ @Test
+ public void testUnreadStreamDrainsAndSetsContentLength() throws Exception {
+ byte[] data = data(10_000);
+ Metadata metadata = new Metadata();
+ try (TikaInputStream tis =
+ TikaInputStream.get(new ByteArrayInputStream(data),
metadata)) {
+ try (SeekableByteChannel channel = tis.getSeekableByteChannel()) {
+ assertArrayEquals(data, readFully(channel));
+ }
+ assertEquals(Long.toString(data.length),
metadata.get(HttpHeaders.CONTENT_LENGTH));
+ assertEquals(data.length, tis.getLength());
+ // read position was not disturbed by the drain
+ assertEquals(data[0] & 0xFF, tis.read());
+ }
+ }
+
+ @Test
+ public void testPartiallyReadStreamThrows() throws Exception {
+ byte[] data = data(100);
+ try (TikaInputStream tis = TikaInputStream.get(new
ByteArrayInputStream(data))) {
+ tis.read();
+ assertThrows(IOException.class, tis::getSeekableByteChannel);
+ }
+ }
+
+ @Test
+ public void testPartiallyReadWithRewindEnabledWorks() throws Exception {
+ byte[] data = data(100);
+ try (TikaInputStream tis = TikaInputStream.get(new
ByteArrayInputStream(data))) {
+ tis.enableRewind();
+ tis.read();
+ try (SeekableByteChannel channel = tis.getSeekableByteChannel()) {
+ assertArrayEquals(data, readFully(channel));
+ }
+ assertEquals(data[1] & 0xFF, tis.read());
+ }
+ }
+
+ @Test
+ public void testAfterGetPathServesFileChannel() throws Exception {
+ byte[] data = data(100);
+ try (TikaInputStream tis = TikaInputStream.get(new
ByteArrayInputStream(data))) {
+ tis.getPath();
+ try (SeekableByteChannel channel = tis.getSeekableByteChannel()) {
+ assertInstanceOf(FileChannel.class, channel);
+ assertArrayEquals(data, readFully(channel));
+ }
+ }
+ }
+
+ @Test
+ public void testStreamWithBudgetPastPerObjectThreshold() throws Exception {
+ // over the 1MB per-object threshold: without a budget this drains to
a spill file,
+ // with a budget it stays in memory
+ byte[] data = data(2 * 1024 * 1024);
+ try (TikaInputStream tis = TikaInputStream.get(new
ByteArrayInputStream(data))) {
+ try (SeekableByteChannel channel = tis.getSeekableByteChannel()) {
+ assertInstanceOf(FileChannel.class, channel);
+ }
+ }
+ CacheMemoryBudget budget = new CacheMemoryBudget(16L * 1024 * 1024);
+ try (TikaInputStream tis = TikaInputStream.get(new
ByteArrayInputStream(data))) {
+ tis.enableRewind(budget);
+ try (SeekableByteChannel channel = tis.getSeekableByteChannel()) {
+ assertFalse(channel instanceof FileChannel);
+ assertArrayEquals(data, readFully(channel));
+ }
+ assertTrue(budget.getReservedBytes() > 0);
+ }
+ assertEquals(0, budget.getReservedBytes(), "closing the stream
releases the reservation");
+ }
+}
diff --git
a/tika-core/src/test/java/org/apache/tika/io/TikaInputStreamTest.java
b/tika-core/src/test/java/org/apache/tika/io/TikaInputStreamTest.java
index 1ac4f4a37a..b5498f9945 100644
--- a/tika-core/src/test/java/org/apache/tika/io/TikaInputStreamTest.java
+++ b/tika-core/src/test/java/org/apache/tika/io/TikaInputStreamTest.java
@@ -685,7 +685,7 @@ public class TikaInputStreamTest {
try (TemporaryResources tmp = new TemporaryResources()) {
CachingSource source = new CachingSource(
new ByteArrayInputStream(data), tmp, -1, metadata, null);
- source.enableRewind(); // Enable caching for spill support
+ source.enableRewind(null); // Enable caching for spill support
// Read all data
byte[] buffer = new byte[data.length];
@@ -721,7 +721,7 @@ public class TikaInputStreamTest {
try (TemporaryResources tmp = new TemporaryResources()) {
CachingSource source = new CachingSource(
new ByteArrayInputStream(data), tmp, -1, metadata, null);
- source.enableRewind(); // Enable caching for seek/spill support
+ source.enableRewind(null); // Enable caching for seek/spill support
// Read and spill
IOUtils.toByteArray(source);
@@ -742,7 +742,7 @@ public class TikaInputStreamTest {
try (TemporaryResources tmp = new TemporaryResources()) {
CachingSource source = new CachingSource(
new ByteArrayInputStream(data), tmp, -1, null, null);
- source.enableRewind(); // Enable caching for seek support
+ source.enableRewind(null); // Enable caching for seek support
// Read first 5 bytes
byte[] buf = new byte[5];
@@ -768,7 +768,7 @@ public class TikaInputStreamTest {
try (TemporaryResources tmp = new TemporaryResources()) {
CachingSource source = new CachingSource(
new ByteArrayInputStream(data), tmp, -1, null, null);
- source.enableRewind(); // Enable caching for spill/seek support
+ source.enableRewind(null); // Enable caching for spill/seek support
// Read first 5 bytes
byte[] buf = new byte[5];
@@ -1052,7 +1052,7 @@ public class TikaInputStreamTest {
TemporaryResources tmp = new TemporaryResources();
CachingSource source = new CachingSource(
new ByteArrayInputStream(new byte[1024 * 1024 + 1]), tmp, -1,
null, null);
- source.enableRewind();
+ source.enableRewind(null);
IOUtils.toByteArray(source);
source.seekTo(0);
assertEquals(0, source.read(), "precondition: reading from the spilled
cache");
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-crypto-module/src/main/java/org/apache/tika/parser/crypto/Pkcs7Parser.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-crypto-module/src/main/java/org/apache/tika/parser/crypto/Pkcs7Parser.java
index 78f2e68180..a901bc347a 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-crypto-module/src/main/java/org/apache/tika/parser/crypto/Pkcs7Parser.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-crypto-module/src/main/java/org/apache/tika/parser/crypto/Pkcs7Parser.java
@@ -18,7 +18,7 @@ package org.apache.tika.parser.crypto;
import java.io.IOException;
import java.io.InputStream;
-import java.nio.file.Files;
+import java.nio.channels.Channels;
import java.util.Set;
import org.bouncycastle.asn1.ASN1ObjectIdentifier;
@@ -101,7 +101,7 @@ public class Pkcs7Parser implements Parser {
private void parseSignedData(TikaInputStream tis, ContentHandler handler,
Metadata metadata,
ParseContext context)
throws IOException, SAXException, TikaException {
- try (InputStream is = Files.newInputStream(tis.getPath())) {
+ try (InputStream is =
Channels.newInputStream(tis.getSeekableByteChannel())) {
DigestCalculatorProvider digestCalculatorProvider =
new
JcaDigestCalculatorProviderBuilder().setProvider("BC").build();
CMSSignedDataParser parser = new
CMSSignedDataParser(digestCalculatorProvider, is);
@@ -131,7 +131,7 @@ public class Pkcs7Parser implements Parser {
private void extractCompressedContent(TikaInputStream tis, ContentHandler
handler,
Metadata metadata, ParseContext
context)
throws IOException, SAXException, TikaException {
- try (InputStream is = Files.newInputStream(tis.getPath())) {
+ try (InputStream is =
Channels.newInputStream(tis.getSeekableByteChannel())) {
CMSCompressedDataParser parser = new CMSCompressedDataParser(is);
CMSTypedStream content = parser.getContent(new
ZlibExpanderProvider());
BoundedInputStream inflated;
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-microsoft-module/src/main/java/org/apache/tika/parser/microsoft/onenote/OneNoteParser.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-microsoft-module/src/main/java/org/apache/tika/parser/microsoft/onenote/OneNoteParser.java
index d3c4ec2515..b7e498cebf 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-microsoft-module/src/main/java/org/apache/tika/parser/microsoft/onenote/OneNoteParser.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-microsoft-module/src/main/java/org/apache/tika/parser/microsoft/onenote/OneNoteParser.java
@@ -17,7 +17,8 @@
package org.apache.tika.parser.microsoft.onenote;
import java.io.IOException;
-import java.nio.file.Files;
+import java.io.InputStream;
+import java.nio.channels.Channels;
import java.time.Instant;
import java.util.Arrays;
import java.util.Collections;
@@ -163,7 +164,10 @@ public class OneNoteParser implements Parser {
} else if (header.isLegacyOrAlternativePackaging()) {
try {
AlternativePackaging alternatePackageOneStoreFile = new
AlternativePackaging();
- byte[] bytes = Files.readAllBytes(tis.getPath());
+ byte[] bytes;
+ try (InputStream is =
Channels.newInputStream(tis.getSeekableByteChannel())) {
+ bytes = is.readAllBytes();
+ }
//enable streaming deserialization
alternatePackageOneStoreFile.doDeserializeFromByteArray(bytes, 0);
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-miscoffice-module/src/main/java/org/apache/tika/parser/epub/EpubParser.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-miscoffice-module/src/main/java/org/apache/tika/parser/epub/EpubParser.java
index 078da8e478..bc525b6aff 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-miscoffice-module/src/main/java/org/apache/tika/parser/epub/EpubParser.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-miscoffice-module/src/main/java/org/apache/tika/parser/epub/EpubParser.java
@@ -64,6 +64,7 @@ import org.apache.tika.sax.EmbeddedContentHandler;
import org.apache.tika.sax.XHTMLBalancingHandler;
import org.apache.tika.sax.XHTMLContentHandler;
import org.apache.tika.utils.XMLReaderUtils;
+import org.apache.tika.zip.utils.ZipFileHelper;
/**
* Epub parser
@@ -152,7 +153,7 @@ public class EpubParser implements Parser {
return bufferedParseZipFile((ZipFile) tis.getOpenContainer(),
bodyHandler,
normalizer, xhtml, metadata, context);
}
- try (ZipFile zipFile =
ZipFile.builder().setFile(tis.getPath().toFile()).get()) {
+ try (ZipFile zipFile = ZipFileHelper.open(tis, null)) {
return bufferedParseZipFile(zipFile, bodyHandler, normalizer,
xhtml, metadata, context);
}
}
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-miscoffice-module/src/main/java/org/apache/tika/parser/odf/OpenDocumentParser.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-miscoffice-module/src/main/java/org/apache/tika/parser/odf/OpenDocumentParser.java
index a624ecf9d1..c500689ab4 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-miscoffice-module/src/main/java/org/apache/tika/parser/odf/OpenDocumentParser.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-miscoffice-module/src/main/java/org/apache/tika/parser/odf/OpenDocumentParser.java
@@ -58,6 +58,7 @@ import org.apache.tika.sax.EmbeddedContentHandler;
import org.apache.tika.sax.EndDocumentShieldingContentHandler;
import org.apache.tika.sax.XHTMLContentHandler;
import org.apache.tika.utils.XMLReaderUtils;
+import org.apache.tika.zip.utils.ZipFileHelper;
/**
* OpenOffice parser
@@ -167,14 +168,13 @@ public class OpenDocumentParser implements Parser {
EmbeddedDocumentExtractor embeddedDocumentExtractor =
EmbeddedDocumentUtil.getEmbeddedDocumentExtractor(context);
- // Open the Zip stream
- // Use a File if we can, and an already open zip is even better
+ // Open the Zip stream; an already open zip from the detector is best.
ZipFile zipFile = null;
Object container = tis.getOpenContainer();
if (container instanceof ZipFile) {
zipFile = (ZipFile) container;
} else {
- zipFile = ZipFile.builder().setFile(tis.getFile()).get();
+ zipFile = ZipFileHelper.open(tis, null);
tis.setOpenContainer(zipFile);
}
// Prepare to handle the content
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/CompressorParser.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/CompressorParser.java
index b37abf9309..68d3198479 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/CompressorParser.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/CompressorParser.java
@@ -35,7 +35,7 @@ import static
org.apache.tika.metadata.HttpHeaders.CONTENT_TYPE;
import java.io.IOException;
import java.io.InputStream;
-import java.nio.file.Files;
+import java.nio.channels.Channels;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -251,13 +251,14 @@ public class CompressorParser implements Parser {
// TIKA-4221 / COMPRESS-721 workaround: commons-compress'
Pack200CompressorInputStream
// reflects into java.io internals (FilterInputStream.in /
FileInputStream.path) to
// bound its input, which throws InaccessibleObjectException
on Java 17+. A
- // TikaInputStream is a FilterInputStream, so it triggers
this. Spool to a file and
- // reopen via Files.newInputStream (a ChannelInputStream) --
the one input type
- // commons-compress does not reflect into.
Pack200CompressorInputStream reads its
- // input fully in the constructor (IN_MEMORY) and then serves
bytes from an in-memory
- // buffer, so the channel stream can be closed immediately
afterward. Remove this once
- // Tika depends on a commons-compress release that contains
the COMPRESS-721 fix.
- try (InputStream packStream =
Files.newInputStream(tis.getPath())) {
+ // TikaInputStream is a FilterInputStream, so it triggers
this. Re-read via
+ // Channels.newInputStream (a ChannelInputStream) -- the one
input type
+ // commons-compress does not reflect into -- without forcing
in-memory content to
+ // disk. Pack200CompressorInputStream reads its input fully in
the constructor
+ // (IN_MEMORY) and then serves bytes from an in-memory buffer,
so the channel
+ // stream can be closed immediately afterward. Remove this
once Tika depends on a
+ // commons-compress release that contains the COMPRESS-721 fix.
+ try (InputStream packStream =
Channels.newInputStream(tis.getSeekableByteChannel())) {
cis =
factory.createCompressorInputStream(CompressorStreamFactory.PACK200,
packStream);
}
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/SevenZParser.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/SevenZParser.java
index 14cab9f9cd..1bf93fb580 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/SevenZParser.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/SevenZParser.java
@@ -20,6 +20,7 @@ import static
org.apache.tika.detect.zip.PackageConstants.SEVENZ;
import java.io.IOException;
import java.io.InputStream;
+import java.nio.channels.SeekableByteChannel;
import java.util.Collections;
import java.util.Set;
@@ -113,10 +114,12 @@ public class SevenZParser extends AbstractArchiveParser {
}
SevenZFile sevenZFile;
+ // SevenZFile.close() closes the channel it was built on
+ SeekableByteChannel channel = tis.getSeekableByteChannel();
try {
// Use setMaxMemoryLimitKiB (direct KiB); setMaxMemoryLimitKb
divides the arg by 1024.
SevenZFile.Builder builder = new SevenZFile.Builder()
- .setFile(tis.getFile())
+ .setSeekableByteChannel(channel)
.setMaxMemoryLimitKiB(defaultConfig.getMemoryLimitInKb());
if (password == null) {
sevenZFile = builder.get();
@@ -124,11 +127,17 @@ public class SevenZParser extends AbstractArchiveParser {
sevenZFile = builder.setPassword(password.toCharArray()).get();
}
} catch (PasswordRequiredException e) {
+ channel.close();
throw new EncryptedDocumentException(e);
} catch (MemoryLimitException e) {
// The limit can be exceeded at open time (assertValidity) as well
as lazily on the
// first getNextEntry() when the LZMA/LZMA2 dictionary is
allocated.
+ channel.close();
throw new TikaMemoryLimitException(e.getMessage());
+ } catch (IOException | RuntimeException e) {
+ // commons-compress throws unchecked exceptions on corrupt headers
+ channel.close();
+ throw e;
}
metadata.set(HttpHeaders.CONTENT_TYPE, SEVENZ.toString());
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/ZipParser.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/ZipParser.java
index de5b8453c2..c99264fdc0 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/ZipParser.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/ZipParser.java
@@ -20,7 +20,6 @@ import static org.apache.tika.detect.zip.PackageConstants.JAR;
import static org.apache.tika.detect.zip.PackageConstants.ZIP;
import java.io.IOException;
-import java.io.InputStream;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.nio.file.attribute.FileTime;
@@ -62,6 +61,7 @@ import org.apache.tika.mime.MediaType;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.sax.XHTMLContentHandler;
import org.apache.tika.utils.ParserUtils;
+import org.apache.tika.zip.utils.ZipFileHelper;
/**
* Parser for ZIP and JAR archives using file-based access for complete
metadata extraction.
@@ -229,11 +229,7 @@ public class ZipParser extends AbstractArchiveParser {
// This handles cases where parser is called without detector
ZipFile zipFile = null;
try {
- ZipFile.Builder builder = ZipFile.builder().setFile(tis.getFile());
- if (config.getEntryEncoding() != null) {
- builder.setCharset(config.getEntryEncoding());
- }
- zipFile = builder.get();
+ zipFile = ZipFileHelper.open(tis, config.getEntryEncoding());
tis.setOpenContainer(zipFile);
} catch (IOException e) {
// ZipFile failed - fall back to streaming
@@ -498,8 +494,10 @@ public class ZipParser extends AbstractArchiveParser {
if (extractor.shouldParseEmbedded(entryMetadata, context)) {
TemporaryResources tmp = new TemporaryResources();
- try (InputStream entryStream = zipFile.getInputStream(entry)) {
- TikaInputStream tis = TikaInputStream.get(entryStream, tmp,
entryMetadata);
+ // Re-openable source: rewind (e.g. for digesting) re-opens the
entry instead of
+ // buffering/spilling it
+ try (TikaInputStream tis = TikaInputStream.get(
+ () -> zipFile.getInputStream(entry), tmp, entryMetadata)) {
extractor.parseEmbedded(tis, xhtml, entryMetadata, context,
true);
} catch (UnsupportedZipFeatureException e) {
EmbeddedDocumentUtil.recordEmbeddedStreamException(e,
parentMetadata);
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/test/java/org/apache/tika/parser/pkg/ZipEmbeddedDigestTest.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/test/java/org/apache/tika/parser/pkg/ZipEmbeddedDigestTest.java
new file mode 100644
index 0000000000..0010254efe
--- /dev/null
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/test/java/org/apache/tika/parser/pkg/ZipEmbeddedDigestTest.java
@@ -0,0 +1,162 @@
+/*
+ * 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.parser.pkg;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import org.apache.commons.codec.binary.Hex;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.digest.DigesterFactory;
+import org.apache.tika.digest.InputStreamDigester;
+import org.apache.tika.io.CacheMemoryBudget;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.TikaCoreProperties;
+import org.apache.tika.parser.ParseContext;
+
+/**
+ * Digest-then-parse round trip over zip entries: with a DigesterFactory in
the context,
+ * every embedded entry must get a correct digest AND its content must still
be extracted
+ * afterward (the re-openable entry source rewinds by re-opening the entry).
+ */
+public class ZipEmbeddedDigestTest extends AbstractPkgTest {
+
+ private static final String DIGEST_KEY = "tk:digest:MD5";
+
+ @TempDir
+ Path tmp;
+
+ private static byte[] entryBytes(String name, int size) {
+ byte[] data = new byte[size];
+ byte[] seed = name.getBytes(StandardCharsets.UTF_8);
+ for (int i = 0; i < size; i++) {
+ data[i] = (byte) (seed[i % seed.length] + i);
+ }
+ return data;
+ }
+
+ /** name -> content; includes a nested zip to exercise embedded random
access. */
+ private Map<String, byte[]> buildEntries() throws Exception {
+ Map<String, byte[]> entries = new HashMap<>();
+ entries.put("small.txt", "hello embedded
world".getBytes(StandardCharsets.UTF_8));
+ entries.put("bigger.dat", entryBytes("bigger.dat", 300_000));
+ ByteArrayOutputStream inner = new ByteArrayOutputStream();
+ try (ZipOutputStream zos = new ZipOutputStream(inner)) {
+ zos.putNextEntry(new ZipEntry("inner.txt"));
+ zos.write("nested entry".getBytes(StandardCharsets.UTF_8));
+ zos.closeEntry();
+ }
+ entries.put("nested.zip", inner.toByteArray());
+ return entries;
+ }
+
+ private Path buildZip(Map<String, byte[]> entries) throws Exception {
+ Path zip = Files.createTempFile(tmp, "digest-test", ".zip");
+ try (ZipOutputStream zos = new
ZipOutputStream(Files.newOutputStream(zip))) {
+ for (Map.Entry<String, byte[]> e : entries.entrySet()) {
+ zos.putNextEntry(new ZipEntry(e.getKey()));
+ zos.write(e.getValue());
+ zos.closeEntry();
+ }
+ }
+ return zip;
+ }
+
+ private static String md5(byte[] data) throws Exception {
+ return
Hex.encodeHexString(MessageDigest.getInstance("MD5").digest(data));
+ }
+
+ private ParseContext digestContext(CacheMemoryBudget budget) {
+ ParseContext context = new ParseContext();
+ context.set(DigesterFactory.class, () -> new InputStreamDigester(
+ "MD5", DIGEST_KEY, Hex::encodeHexString));
+ if (budget != null) {
+ context.set(CacheMemoryBudget.class, budget);
+ }
+ return context;
+ }
+
+ private void assertDigestsAndContent(List<Metadata> metadataList,
+ Map<String, byte[]> entries) throws
Exception {
+ Map<String, Metadata> byName = new HashMap<>();
+ for (Metadata m : metadataList) {
+ String name = m.get(TikaCoreProperties.RESOURCE_NAME_KEY);
+ if (name != null) {
+ byName.put(name, m);
+ }
+ }
+ for (Map.Entry<String, byte[]> e : entries.entrySet()) {
+ Metadata m = byName.get(e.getKey());
+ assertTrue(m != null, "missing embedded metadata for " +
e.getKey());
+ assertEquals(md5(e.getValue()), m.get(DIGEST_KEY),
+ "wrong digest for " + e.getKey());
+ }
+ // Parse-after-digest still works: the container's content lists its
entries...
+ String containerContent =
metadataList.get(0).get(TikaCoreProperties.TIKA_CONTENT);
+ assertTrue(containerContent != null &&
containerContent.contains("small.txt"),
+ "container content missing after digest rewind");
+ // ...and the nested zip was digested AND recursed into (its entry
parsed at depth 2),
+ // proving the rewind-after-digest of a zip entry yields parseable
content
+ Metadata inner = byName.get("inner.txt");
+ assertTrue(inner != null, "nested zip entry not recursed after digest
rewind");
+ assertEquals(md5("nested entry".getBytes(StandardCharsets.UTF_8)),
+ inner.get(DIGEST_KEY), "wrong digest for inner.txt");
+ }
+
+ @Test
+ public void testEmbeddedDigestsWithoutBudget() throws Exception {
+ Map<String, byte[]> entries = buildEntries();
+ Path zip = buildZip(entries);
+ List<Metadata> metadataList = getRecursiveMetadata(zip,
digestContext(null), false);
+ assertDigestsAndContent(metadataList, entries);
+ }
+
+ @Test
+ public void testEmbeddedDigestsWithBudget() throws Exception {
+ Map<String, byte[]> entries = buildEntries();
+ Path zip = buildZip(entries);
+ CacheMemoryBudget budget = new CacheMemoryBudget(64L * 1024 * 1024);
+ List<Metadata> metadataList = getRecursiveMetadata(zip,
digestContext(budget), false);
+ assertDigestsAndContent(metadataList, entries);
+ assertEquals(0, budget.getReservedBytes(),
+ "budget reservations must all be released after the parse");
+ }
+
+ @Test
+ public void testEmbeddedDigestsWithTinyBudget() throws Exception {
+ // an effectively exhausted budget must not break digests or extraction
+ Map<String, byte[]> entries = buildEntries();
+ Path zip = buildZip(entries);
+ CacheMemoryBudget budget = new CacheMemoryBudget(1);
+ List<Metadata> metadataList = getRecursiveMetadata(zip,
digestContext(budget), false);
+ assertDigestsAndContent(metadataList, entries);
+ assertEquals(0, budget.getReservedBytes());
+ }
+}
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-zip-commons/src/main/java/org/apache/tika/detect/zip/DefaultZipContainerDetector.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-zip-commons/src/main/java/org/apache/tika/detect/zip/DefaultZipContainerDetector.java
index 32eaff2fdc..20d26e06bd 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-zip-commons/src/main/java/org/apache/tika/detect/zip/DefaultZipContainerDetector.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-zip-commons/src/main/java/org/apache/tika/detect/zip/DefaultZipContainerDetector.java
@@ -19,8 +19,8 @@ package org.apache.tika.detect.zip;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
+import java.nio.channels.Channels;
+import java.nio.channels.SeekableByteChannel;
import java.util.ArrayList;
import java.util.List;
@@ -46,6 +46,7 @@ import org.apache.tika.metadata.Zip;
import org.apache.tika.mime.MediaType;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.parser.ParsingIntent;
+import org.apache.tika.zip.utils.ZipFileHelper;
import org.apache.tika.zip.utils.ZipSalvager;
/**
@@ -213,7 +214,8 @@ public class DefaultZipContainerDetector implements
Detector {
}
/**
- * This will call TikaInputStream's getFile(). If there are no exceptions,
+ * Opens the zip via {@link ZipFileHelper#open} (random access from memory
when the
+ * content is already cached there, a file otherwise). If there are no
exceptions,
* it will place the ZipFile in TikaInputStream's openContainer and leave
it
* open.
* <p>
@@ -232,7 +234,7 @@ public class DefaultZipContainerDetector implements
Detector {
// Try to open ZipFile directly
ZipFile zip = null;
try {
- zip = ZipFile.builder().setFile(tis.getFile()).get();
+ zip = ZipFileHelper.open(tis, null);
metadata.set(Zip.DETECTOR_ZIPFILE_OPENED, true);
} catch (IOException e) {
// ZipFile failed to open (truncated/corrupt)
@@ -287,8 +289,8 @@ public class DefaultZipContainerDetector implements
Detector {
if (LOG.isDebugEnabled()) {
LOG.debug("Falling back to streaming detection");
}
- try {
- return detectStreamingFromPath(tis.getPath(), metadata, false);
+ try (SeekableByteChannel channel = tis.getSeekableByteChannel()) {
+ return detectStreamingFromChannel(channel, metadata, false);
} catch (IOException e) {
//swallow
}
@@ -328,11 +330,18 @@ public class DefaultZipContainerDetector implements
Detector {
return finalDetect(detectContext);
}
- private MediaType detectStreamingFromPath(Path p, Metadata metadata,
boolean allowStoredEntries)
- throws IOException {
+ /**
+ * Streaming detection over a seekable channel (rewound to 0 for each
attempt), so
+ * in-memory content need not be spilled to a path. The channel is
close-shielded from
+ * the ZipArchiveInputStream; the caller owns closing it.
+ */
+ private MediaType detectStreamingFromChannel(SeekableByteChannel channel,
Metadata metadata,
+ boolean allowStoredEntries)
throws IOException {
+ channel.position(0);
StreamingDetectContext detectContext = new StreamingDetectContext();
try (ZipArchiveInputStream zis = new ZipArchiveInputStream(
- Files.newInputStream(p), "UTF8", false, allowStoredEntries)) {
+ CloseShieldInputStream.wrap(Channels.newInputStream(channel)),
+ "UTF8", false, allowStoredEntries)) {
ZipArchiveEntry zae = zis.getNextEntry();
while (zae != null) {
MediaType mt = detect(zae, zis, detectContext);
@@ -346,7 +355,7 @@ public class DefaultZipContainerDetector implements
Detector {
zfe.getFeature() ==
UnsupportedZipFeatureException.Feature.DATA_DESCRIPTOR) {
// Set hint for parser that DATA_DESCRIPTOR support is required
metadata.set(Zip.DETECTOR_DATA_DESCRIPTOR_REQUIRED, true);
- return detectStreamingFromPath(p, metadata, true);
+ return detectStreamingFromChannel(channel, metadata, true);
}
} catch (SecurityException e) {
throw e;
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-zip-commons/src/main/java/org/apache/tika/zip/utils/ZipFileHelper.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-zip-commons/src/main/java/org/apache/tika/zip/utils/ZipFileHelper.java
new file mode 100644
index 0000000000..2caae570f5
--- /dev/null
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-zip-commons/src/main/java/org/apache/tika/zip/utils/ZipFileHelper.java
@@ -0,0 +1,51 @@
+/*
+ * 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.zip.utils;
+
+import java.io.IOException;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.charset.Charset;
+
+import org.apache.commons.compress.archivers.zip.ZipFile;
+
+import org.apache.tika.io.TikaInputStream;
+
+public class ZipFileHelper {
+
+ /**
+ * Opens a {@link ZipFile} over {@link
TikaInputStream#getSeekableByteChannel()}, so
+ * in-memory content is read without spilling to disk. On success the
returned
+ * {@code ZipFile.close()} closes the channel it was built on; on failure
the channel
+ * is closed before the exception propagates.
+ */
+ public static ZipFile open(TikaInputStream tis, Charset charset) throws
IOException {
+ SeekableByteChannel channel = tis.getSeekableByteChannel();
+ try {
+ ZipFile.Builder builder =
ZipFile.builder().setSeekableByteChannel(channel);
+ if (charset != null) {
+ builder.setCharset(charset);
+ }
+ return builder.get();
+ } catch (IOException | RuntimeException e) {
+ channel.close();
+ throw e;
+ }
+ }
+
+ private ZipFileHelper() {
+ }
+}
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-zip-commons/src/main/java/org/apache/tika/zip/utils/ZipSalvager.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-zip-commons/src/main/java/org/apache/tika/zip/utils/ZipSalvager.java
index ec30cb3dd6..4538117e75 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-zip-commons/src/main/java/org/apache/tika/zip/utils/ZipSalvager.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-zip-commons/src/main/java/org/apache/tika/zip/utils/ZipSalvager.java
@@ -57,7 +57,8 @@ public class ZipSalvager {
* <li>Returns null</li>
* </ul>
*
- * @param tis the TikaInputStream (must be file-backed)
+ * @param tis the TikaInputStream (must be rewindable, i.e. unread or
with rewind
+ * enabled; it need not be file-backed)
* @param metadata the metadata to update with hints
* @param charset optional charset for entry names (may be null)
* @return the opened ZipFile, or null if opening and salvaging both failed
@@ -65,11 +66,7 @@ public class ZipSalvager {
public static ZipFile tryToOpenZipFile(TikaInputStream tis, Metadata
metadata, Charset charset) {
// First, try direct open
try {
- ZipFile.Builder builder = new
ZipFile.Builder().setFile(tis.getFile());
- if (charset != null) {
- builder.setCharset(charset);
- }
- ZipFile zipFile = builder.get();
+ ZipFile zipFile = ZipFileHelper.open(tis, charset);
// Direct open succeeded
metadata.set(Zip.DETECTOR_ZIPFILE_OPENED, true);
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ParseHandler.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ParseHandler.java
index fab0a3f53d..e8aa9375fb 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ParseHandler.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ParseHandler.java
@@ -18,7 +18,7 @@ package org.apache.tika.pipes.core.server;
import java.io.IOException;
import java.io.InputStream;
-import java.nio.file.Files;
+import java.nio.channels.Channels;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
@@ -163,7 +163,7 @@ class ParseHandler {
if (unpackConfig != null &&
unpackConfig.isIncludeOriginal()) {
UnpackHandler unpackHandler =
parseContext.get(UnpackHandler.class);
- try (InputStream is = Files.newInputStream(tis.getPath())) {
+ try (InputStream is =
Channels.newInputStream(tis.getSeekableByteChannel())) {
unpackHandler.add(0, metadata, is);
} catch (IOException e) {
LOG.warn("problem reading source file into embedded document
byte store", e);
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
index fc40d3beb1..c359f89227 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
@@ -52,6 +52,7 @@ import org.apache.tika.config.loader.TikaLoader;
import org.apache.tika.detect.Detector;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
+import org.apache.tika.io.CacheMemoryBudget;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.metadata.filter.MetadataFilter;
import org.apache.tika.metadata.writelimiter.MetadataWriteLimiterFactory;
@@ -90,6 +91,51 @@ public class PipesServer implements AutoCloseable {
private static final Logger LOG =
LoggerFactory.getLogger(PipesServer.class);
+ // Process-wide budget bounding total in-memory stream caching
(embedded-object digest rewind
+ // buffers, etc.) so small embedded objects stay in RAM instead of
spilling per-object at 1MB.
+ // NOTE: read in THIS (forked server) JVM -- set it via the config's
forkedJvmArgs, not on the
+ // parent JVM. Tunable via -Dtika.pipes.cacheMemoryBudgetBytes; <=0
disables (falls back to
+ // the 1MB per-object default). Clamped to a quarter of the fork's max
heap.
+ static final String CACHE_MEMORY_BUDGET_BYTES_PROP =
"tika.pipes.cacheMemoryBudgetBytes";
+
+ static final CacheMemoryBudget CACHE_MEMORY_BUDGET =
initCacheMemoryBudget();
+
+ private static CacheMemoryBudget initCacheMemoryBudget() {
+ long bytes = 256L * 1024 * 1024;
+ String val = System.getProperty(CACHE_MEMORY_BUDGET_BYTES_PROP);
+ if (val != null) {
+ try {
+ bytes = Long.parseLong(val.trim());
+ } catch (NumberFormatException e) {
+ LOG.warn("Could not parse -D{}={} (plain bytes required, no
unit suffix); " +
+ "using the default {}",
CACHE_MEMORY_BUDGET_BYTES_PROP, val, bytes);
+ }
+ }
+ if (bytes <= 0) {
+ LOG.info("Cache memory budget disabled; per-object 1MB spill
threshold applies");
+ return null;
+ }
+ long clamp = Runtime.getRuntime().maxMemory() / 4;
+ if (bytes > clamp) {
+ LOG.warn("Cache memory budget {} exceeds a quarter of max heap;
clamping to {}",
+ bytes, clamp);
+ bytes = clamp;
+ }
+ LOG.info("Cache memory budget: {} bytes", bytes);
+ return new CacheMemoryBudget(bytes);
+ }
+
+ /**
+ * Seeds the process-wide cache memory budget into a merged per-request
ParseContext.
+ * The budget is not a registered component, so neither the config nor a
wire request
+ * can supply one -- this is the only way it enters a context in a pipes
fork.
+ */
+ static void seedCacheMemoryBudget(ParseContext mergedContext) {
+ if (CACHE_MEMORY_BUDGET != null) {
+ mergedContext.set(CacheMemoryBudget.class, CACHE_MEMORY_BUDGET);
+ }
+ }
+
public static final int AUTH_TOKEN_LENGTH_BYTES = 32;
/** Env var the parent manager sets so the child can watch the parent's
@@ -677,6 +723,7 @@ public class PipesServer implements AutoCloseable {
// EmbeddedDocumentExtractor + UnpackedByteCount in PipesWorker's
UNPACK-mode setup.
// Request-level values override config defaults
mergedContext.copyFrom(requestContext);
+ seedCacheMemoryBudget(mergedContext);
return mergedContext;
}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
index d1b5ed3a2c..a488a4922e 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
@@ -158,6 +158,7 @@ public class SharedServerResources {
// content extraction for every non-UNPACK parse mode.
// Request-level values override config defaults
mergedContext.copyFrom(requestContext);
+ PipesServer.seedCacheMemoryBudget(mergedContext);
return mergedContext;
}
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/CacheMemoryBudgetSeedingTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/CacheMemoryBudgetSeedingTest.java
new file mode 100644
index 0000000000..f44cae7f5c
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/CacheMemoryBudgetSeedingTest.java
@@ -0,0 +1,45 @@
+/*
+ * 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.server;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+import org.junit.jupiter.api.Test;
+
+import org.apache.tika.io.CacheMemoryBudget;
+import org.apache.tika.parser.ParseContext;
+
+public class CacheMemoryBudgetSeedingTest {
+
+ @Test
+ public void testDefaultBudgetClampedToHeap() {
+ // No -Dtika.pipes.cacheMemoryBudgetBytes in the surefire JVM -> the
256MB default,
+ // clamped to a quarter of max heap
+ assertNotNull(PipesServer.CACHE_MEMORY_BUDGET);
+ long expected = Math.min(256L * 1024 * 1024,
Runtime.getRuntime().maxMemory() / 4);
+ assertEquals(expected, PipesServer.CACHE_MEMORY_BUDGET.getMaxBytes());
+ }
+
+ @Test
+ public void testSeeding() {
+ ParseContext context = new ParseContext();
+ PipesServer.seedCacheMemoryBudget(context);
+ assertSame(PipesServer.CACHE_MEMORY_BUDGET,
context.get(CacheMemoryBudget.class));
+ }
+}
