This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4828 in repository https://gitbox.apache.org/repos/asf/tika.git
commit 22c9bc0a61eb8401f38bc31f4ab74c79b2a88b46 Author: tallison <[email protected]> AuthorDate: Thu Aug 20 14:24:06 2026 -0400 TIKA-4828 -- improve caching while digesting embedded files --- .../apache/tika/digest/InputStreamDigester.java | 8 +- .../java/org/apache/tika/io/CacheMemoryBudget.java | 94 +++++++++++ .../java/org/apache/tika/io/CachingSource.java | 7 +- .../java/org/apache/tika/io/ReopenableSource.java | 173 +++++++++++++++++++++ .../main/java/org/apache/tika/io/StreamCache.java | 72 +++++++-- .../java/org/apache/tika/io/TikaInputSource.java | 11 ++ .../java/org/apache/tika/io/TikaInputStream.java | 50 ++++++ .../java/org/apache/tika/parser/pkg/ZipParser.java | 9 +- .../apache/tika/pipes/core/server/PipesServer.java | 15 ++ .../pipes/core/server/SharedServerResources.java | 6 + 10 files changed, 428 insertions(+), 17 deletions(-) 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..27fac5aa83 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; @@ -116,7 +117,12 @@ public class InputStreamDigester implements Digester { @Override public void digest(TikaInputStream tis, Metadata metadata, ParseContext parseContext) throws IOException { - tis.enableRewind(); + // Bridge the shared memory budget (if any) from the ParseContext into the IO layer, so + // that caching this object for rewind stays in memory up to the budget instead of spilling + // per-object at 1MB. TikaInputStream itself 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/CacheMemoryBudget.java b/tika-core/src/main/java/org/apache/tika/io/CacheMemoryBudget.java new file mode 100644 index 0000000000..850c284746 --- /dev/null +++ b/tika-core/src/main/java/org/apache/tika/io/CacheMemoryBudget.java @@ -0,0 +1,94 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.tika.io; + +import java.util.concurrent.atomic.AtomicLong; + +/** + * A shared, bounded memory budget for in-memory stream caching. + * <p> + * When a {@code CacheMemoryBudget} is available, {@link StreamCache} keeps embedded-object + * content in memory as long as it can reserve against this budget, spilling to a temp file only + * once the budget is exhausted -- rather than at a fixed per-object threshold. This lets small + * embedded objects stay in RAM (avoiding per-object temp-file spills) while still bounding total + * cache heap across concurrent parses. + * <p> + * Intended usage: place a single, process-wide instance in the {@link + * org.apache.tika.parser.ParseContext} (e.g. seeded by the pipes server / batch runner into every + * document's context). It is bridged to the IO layer explicitly at the point of caching (see + * {@link org.apache.tika.digest.InputStreamDigester#digest} -> {@link + * TikaInputStream#enableRewind(CacheMemoryBudget)}); {@code TikaInputStream} itself never depends + * on {@code ParseContext}. When no budget is present, callers fall back to the historic per-object + * default ({@link StreamCache} 1MB threshold). + * <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 + */ + 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/CachingSource.java b/tika-core/src/main/java/org/apache/tika/io/CachingSource.java index 6ed283ee8f..dbfc11114e 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 @@ -187,6 +187,11 @@ class CachingSource extends InputStream implements TikaInputSource { @Override public void enableRewind() throws IOException { + enableRewind(null); + } + + @Override + public void enableRewind(CacheMemoryBudget budget) throws IOException { // Already in caching or file mode - no-op if (cachingStream != null || fileStream != null) { return; @@ -199,7 +204,7 @@ 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; } 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..c6f5eb902a --- /dev/null +++ b/tika-core/src/main/java/org/apache/tika/io/ReopenableSource.java @@ -0,0 +1,173 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.tika.io; + +import java.io.BufferedInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.file.Files; +import java.nio.file.Path; + +import org.apache.commons.io.IOUtils; +import org.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, {@link #enableRewind()} + * is a no-op and {@link #seekTo(long)}/rewind simply re-open the source and skip. + * This avoids the memory-then-disk caching that {@link CachingSource} performs for + * one-shot streams -- notably the per-embedded-object spill during digesting. + * A temp file is only created if {@link #getPath} is called (i.e. a parser or + * detector genuinely needs a File on disk). + */ +class ReopenableSource extends InputStream implements TikaInputSource { + + private final IOSupplier<InputStream> opener; + private final TemporaryResources tmp; + private long length; + + private InputStream currentStream; // lazily opened + private long position; + private Path spilledPath; + private long markPosition = -1; + + ReopenableSource(IOSupplier<InputStream> opener, TemporaryResources tmp, long length) { + this.opener = opener; + this.tmp = tmp; + this.length = length; + 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(); + } + // Re-open from the beginning (from the spilled file if we've already spilled, + // otherwise from the supplier) and skip forward. + 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) { + // A caller needs a real File -- materialize once from a fresh stream. + Path p = tmp.createTempFile(suffix); + try (InputStream in = opener.get(); OutputStream out = Files.newOutputStream(p)) { + IOUtils.copy(in, out); + } + spilledPath = p; + if (length < 0) { + length = Files.size(p); + } + } + return spilledPath; + } + + @Override + public long getLength() { + return length; + } + + @Override + public void enableRewind() throws IOException { + // No-op: the source can be re-opened, so no caching is needed to rewind. + } + + @Override + public void close() throws IOException { + 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..8699317ac7 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 @@ -33,9 +33,18 @@ 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; + // Memory storage (null after spill) private byte[] memoryBuffer; private int memorySize; @@ -51,18 +60,28 @@ class StreamCache implements Closeable { private boolean closed; StreamCache(TemporaryResources tmp) { - this(tmp, null, DEFAULT_MEMORY_THRESHOLD); + this(tmp, null, DEFAULT_MEMORY_THRESHOLD, null); } /** Suffix up front: a threshold spill precedes any getPath(suffix) call (TIKA-3903). */ StreamCache(TemporaryResources tmp, String suffix) { - this(tmp, suffix, DEFAULT_MEMORY_THRESHOLD); + this(tmp, suffix, DEFAULT_MEMORY_THRESHOLD, null); } StreamCache(TemporaryResources tmp, String suffix, int memoryThreshold) { + this(tmp, suffix, memoryThreshold, null); + } + + /** When {@code budget} is non-null it governs memory-vs-spill instead of memoryThreshold. */ + StreamCache(TemporaryResources tmp, String suffix, CacheMemoryBudget budget) { + this(tmp, suffix, DEFAULT_MEMORY_THRESHOLD, budget); + } + + 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 +97,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 +120,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,11 +134,38 @@ class StreamCache implements Closeable { totalSize += len; } + /** + * Decide whether {@code additional} more bytes can stay in memory. With no budget this is the + * historic fixed per-object threshold; with a budget it reserves against the shared pool + * (all-or-nothing) and spills once the pool is exhausted. + */ + private boolean canKeepInMemory(int additional) { + if (budget == null) { + return memorySize + (long) additional <= memoryThreshold; + } + if ((long) memorySize + additional > MAX_ARRAY_SIZE) { + return false; + } + if (budget.tryReserve(additional) == additional) { + reserved += additional; + return true; + } + return false; + } + + private void releaseReserved() { + if (budget != null && reserved > 0) { + 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)); + int cap = (budget == null) ? memoryThreshold : MAX_ARRAY_SIZE; + int newSize = Math.min(cap, Math.max(memoryBuffer.length * 2, needed)); byte[] newBuffer = new byte[newSize]; System.arraycopy(memoryBuffer, 0, newBuffer, 0, memorySize); memoryBuffer = newBuffer; @@ -141,9 +187,10 @@ class StreamCache implements Closeable { spillOutputStream.write(memoryBuffer, 0, memorySize); } - // Release memory buffer + // Release memory buffer (and give any reserved budget back -- we're on disk now) memoryBuffer = null; memorySize = 0; + releaseReserved(); } /** @@ -258,6 +305,7 @@ class StreamCache implements Closeable { } closed = true; memoryBuffer = null; + releaseReserved(); 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..c7958c9e28 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 @@ -65,4 +65,15 @@ interface TikaInputSource extends Closeable { * @throws IOException if position is not 0 */ void enableRewind() throws IOException; + + /** + * Like {@link #enableRewind()}, but supplies a shared {@link CacheMemoryBudget} that governs + * how much may be held in memory before spilling to disk. Only sources that cache + * (CachingSource) use it; sources that are inherently rewindable ignore it. + * + * @param budget shared memory budget, or {@code null} for the historic per-object default + */ + default void enableRewind(CacheMemoryBudget budget) throws IOException { + enableRewind(); + } } 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..6c0b4ffa04 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 @@ -33,6 +33,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 +102,38 @@ 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 only created if {@link #getPath()} is later called (a parser/detector needing a File). + * + * @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); + return new TikaInputStream(inputSource, tmp, ext); + } + public static TikaInputStream get(InputStream stream) { return get(stream, new TemporaryResources(), null); } @@ -484,6 +517,23 @@ public class TikaInputStream extends TaggedInputStream { } } + /** + * Like {@link #enableRewind()}, but supplies a shared {@link CacheMemoryBudget} that governs + * how much is held in memory before spilling to disk (only used by stream-backed sources that + * cache). {@code null} falls back to the historic per-object default. This lets a caller that + * has the budget (e.g. the digester, which is handed a ParseContext) bridge it to the IO layer + * without TikaInputStream itself depending on ParseContext. + * + * @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(budget); + } + } + @Override public String toString() { String str = "TikaInputStream of "; 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..8882ef9828 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; @@ -498,8 +497,12 @@ 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: the entry can be re-read from the random-access ZipFile, + // so enableRewind/rewind (e.g. during per-object digesting) re-opens the entry + // instead of buffering/spilling it to disk. A temp file is created only if a + // parser/detector actually needs a File via getPath(). See ReopenableSource. + 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-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..28ee93ec54 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,16 @@ 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. + // Tunable via -Dtika.pipes.cacheMemoryBudgetBytes; <=0 disables (falls back to the 1MB default). + static final CacheMemoryBudget CACHE_MEMORY_BUDGET = initCacheMemoryBudget(); + + private static CacheMemoryBudget initCacheMemoryBudget() { + long bytes = Long.getLong("tika.pipes.cacheMemoryBudgetBytes", 256L * 1024 * 1024); + return bytes > 0 ? new CacheMemoryBudget(bytes) : null; + } + 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 +688,10 @@ public class PipesServer implements AutoCloseable { // EmbeddedDocumentExtractor + UnpackedByteCount in PipesWorker's UNPACK-mode setup. // Request-level values override config defaults mergedContext.copyFrom(requestContext); + // Seed the process-wide cache memory budget (unless the request already set one). + if (CACHE_MEMORY_BUDGET != null && mergedContext.get(CacheMemoryBudget.class) == null) { + mergedContext.set(CacheMemoryBudget.class, CACHE_MEMORY_BUDGET); + } 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..f4a8fcab8a 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 @@ -25,6 +25,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.filter.MetadataFilter; import org.apache.tika.metadata.writelimiter.MetadataWriteLimiterFactory; import org.apache.tika.parser.AutoDetectParser; @@ -158,6 +159,11 @@ public class SharedServerResources { // content extraction for every non-UNPACK parse mode. // Request-level values override config defaults mergedContext.copyFrom(requestContext); + // Seed the process-wide cache memory budget (unless the request already set one). + if (PipesServer.CACHE_MEMORY_BUDGET != null + && mergedContext.get(CacheMemoryBudget.class) == null) { + mergedContext.set(CacheMemoryBudget.class, PipesServer.CACHE_MEMORY_BUDGET); + } return mergedContext; }
