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 cc9fd84d49 TIKA-4908: close archive entry streams so their cache
budget is released (#3199)
cc9fd84d49 is described below
commit cc9fd84d49a27afc38b24953518ab0ed5b73bc11
Author: Tim Allison <[email protected]>
AuthorDate: Fri Sep 18 16:00:58 2026 -0400
TIKA-4908: close archive entry streams so their cache budget is released
(#3199)
---
CHANGES.txt | 3 +
.../java/org/apache/tika/io/CachingSource.java | 5 +
.../java/org/apache/tika/io/ReopenableSource.java | 12 +-
.../java/org/apache/tika/io/TikaInputStream.java | 6 +
.../tika/io/TikaInputStreamOwnershipTest.java | 160 +++++++++++++++++++++
.../org/apache/tika/parser/pkg/PackageParser.java | 5 +-
.../org/apache/tika/parser/pkg/SevenZParser.java | 5 +-
.../java/org/apache/tika/parser/pkg/ZipParser.java | 3 +-
.../tika/parser/pkg/PackageEmbeddedBudgetTest.java | 87 +++++++++++
9 files changed, 278 insertions(+), 8 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 3336a59aac..26446a0a4a 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,8 @@
Release 4.1.0 - unreleased
+ * Fix a cache-budget leak: tar, 7z and streamed zip entries never released
+ their in-memory cache's reservation (TIKA-4908).
+
* Mp3Parser no longer writes the literal "null" (a missing album) or the
duration into the body; an untagged MP3 has an empty body, and a TEXT
inference binding no longer embeds it (TIKA-4907).
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 47255ccfa4..d77b86779b 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
@@ -204,6 +204,11 @@ class CachingSource extends InputStream implements
TikaInputSource {
// Switch to caching mode
StreamCache cache = new StreamCache(tmp, suffix, budget);
+ if (tmp != null) {
+ // tmp owns the cache: a scope that disposes tmp without closing
this stream still
+ // frees the buffer and its budget reservation. The wrapped stream
is not tmp's to close.
+ tmp.addResource(cache);
+ }
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
index ea6df9397e..8ce3d278e1 100644
--- a/tika-core/src/main/java/org/apache/tika/io/ReopenableSource.java
+++ b/tika-core/src/main/java/org/apache/tika/io/ReopenableSource.java
@@ -255,8 +255,13 @@ class ReopenableSource extends InputStream implements
TikaInputSource {
});
}
+ private void dropRetained() {
+ retainedBuffer = null;
+ maybeReleaseRetained();
+ }
+
private void maybeReleaseRetained() {
- if (closed && channelPins == 0 && retainedReservation > 0) {
+ if ((closed || retainedBuffer == null) && channelPins == 0 &&
retainedReservation > 0) {
budget.release(retainedReservation);
retainedReservation = 0;
}
@@ -337,6 +342,11 @@ class ReopenableSource extends InputStream implements
TikaInputSource {
retainedBuffer = data;
retainedLength = total;
retainedReservation = reservedHere;
+ if (tmp != null && reservedHere > 0) {
+ // tmp owns the retention: disposing tmp without closing this
source still releases
+ // the reservation (once no handed-out channel pins the array)
+ tmp.addResource(this::dropRetained);
+ }
// The full read is ground truth, even over a lying declared length
length = total;
lengthMeasured = true;
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 65ad794795..09e6148f60 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
@@ -85,6 +85,12 @@ public class TikaInputStream extends TaggedInputStream {
/**
* Returns the backing TikaInputSource, or null if using protected
constructor.
+ * <p>
+ * {@code tmp} owns what this call creates, never {@code stream} itself: a
cache built for
+ * rewind (its buffer, {@link CacheMemoryBudget} reservation and spill
file) is registered
+ * with {@code tmp}, so disposing {@code tmp} frees it even if the
returned stream is never
+ * closed, while the caller's stream stays open. If {@code stream} is
already a
+ * TikaInputStream it is returned as-is.
*/
private TikaInputSource inputSource() {
return in instanceof TikaInputSource ? (TikaInputSource) in : null;
diff --git
a/tika-core/src/test/java/org/apache/tika/io/TikaInputStreamOwnershipTest.java
b/tika-core/src/test/java/org/apache/tika/io/TikaInputStreamOwnershipTest.java
new file mode 100644
index 0000000000..4953a370c3
--- /dev/null
+++
b/tika-core/src/test/java/org/apache/tika/io/TikaInputStreamOwnershipTest.java
@@ -0,0 +1,160 @@
+/*
+ * 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.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.channels.SeekableByteChannel;
+
+import org.apache.commons.io.IOUtils;
+import org.junit.jupiter.api.Test;
+
+import org.apache.tika.metadata.HttpHeaders;
+import org.apache.tika.metadata.Metadata;
+
+/**
+ * {@code TikaInputStream.get(stream, tmp, ...)}: tmp owns the cache the call
creates and never
+ * the caller's stream. A scope that disposes tmp without closing the stream
must still release
+ * the cache's budget reservation; the caller's stream, or a TikaInputStream
passed through
+ * get() as-is, must never be closed on their behalf.
+ */
+public class TikaInputStreamOwnershipTest {
+
+ private static final int BIG = 3 * 1024 * 1024; // past StreamCache's 1 MB
threshold and
+ // ReopenableSource's
in-memory floor
+
+ 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 class CloseCountingInputStream extends ByteArrayInputStream
{
+ int closes = 0;
+
+ CloseCountingInputStream(byte[] buf) {
+ super(buf);
+ }
+
+ @Override
+ public void close() throws IOException {
+ closes++;
+ super.close();
+ }
+ }
+
+ private static CacheMemoryBudget budget() {
+ return new CacheMemoryBudget(64L * 1024 * 1024);
+ }
+
+ @Test
+ public void tmpOwnsTheCacheNotTheCallersStream() throws Exception {
+ CacheMemoryBudget budget = budget();
+ CloseCountingInputStream raw = new CloseCountingInputStream(data(BIG));
+ TemporaryResources tmp = new TemporaryResources();
+ TikaInputStream tis = TikaInputStream.get(raw, tmp, new Metadata());
+ tis.enableRewind(budget);
+ assertEquals(BIG, IOUtils.toByteArray(tis).length);
+ assertTrue(budget.getReservedBytes() > 0, "a cache past the threshold
reserves budget");
+
+ tmp.close(); // the TikaInputStream itself is never closed
+
+ assertEquals(0, budget.getReservedBytes(),
+ "disposing tmp must release the cache it created and its
reservation");
+ assertEquals(0, raw.closes, "the caller's stream is not tmp's to
close");
+ }
+
+ @Test
+ public void callersTikaInputStreamIsNotRegistered() throws Exception {
+ byte[] data = data(64);
+ TikaInputStream outer = TikaInputStream.get(data);
+ TemporaryResources tmp = new TemporaryResources();
+ TikaInputStream same = TikaInputStream.get(outer, tmp, new Metadata());
+ assertSame(outer, same, "an existing TikaInputStream is returned
as-is");
+
+ tmp.close();
+
+ assertEquals(data.length, IOUtils.toByteArray(outer).length,
+ "closing the scope's tmp must not close a caller-supplied
stream");
+ outer.close();
+ }
+
+ @Test
+ public void closingBothWaysReleasesOnce() throws Exception {
+ CacheMemoryBudget budget = budget();
+ TemporaryResources tmp = new TemporaryResources();
+ TikaInputStream tis =
+ TikaInputStream.get(new ByteArrayInputStream(data(BIG)), tmp,
new Metadata());
+ tis.enableRewind(budget);
+ IOUtils.toByteArray(tis);
+ assertTrue(budget.getReservedBytes() > 0);
+
+ tis.close();
+ tmp.close();
+ tis.close();
+
+ assertEquals(0, budget.getReservedBytes(), "released exactly once,
never negative");
+ }
+
+ private static TikaInputStream reopenable(byte[] data, TemporaryResources
tmp) {
+ Metadata metadata = new Metadata();
+ metadata.set(HttpHeaders.CONTENT_LENGTH,
Integer.toString(data.length));
+ return TikaInputStream.get(() -> new ByteArrayInputStream(data), tmp,
metadata);
+ }
+
+ @Test
+ public void tmpOwnsAReopenableSourcesRetention() throws Exception {
+ CacheMemoryBudget budget = budget();
+ TemporaryResources tmp = new TemporaryResources();
+ TikaInputStream tis = reopenable(data(BIG), tmp);
+ tis.enableRewind(budget);
+ try (SeekableByteChannel channel = tis.getSeekableByteChannel()) {
+ assertEquals(BIG, channel.size());
+ }
+ assertTrue(budget.getReservedBytes() > 0, "retention past the floor
reserves budget");
+
+ tmp.close(); // the TikaInputStream itself is never closed
+
+ assertEquals(0, budget.getReservedBytes(),
+ "disposing tmp must release the retained buffer's
reservation");
+ }
+
+ @Test
+ public void aPinnedChannelOutlivesTmpAndReleasesOnClose() throws Exception
{
+ CacheMemoryBudget budget = budget();
+ TemporaryResources tmp = new TemporaryResources();
+ TikaInputStream tis = reopenable(data(BIG), tmp);
+ tis.enableRewind(budget);
+ SeekableByteChannel channel = tis.getSeekableByteChannel();
+ long reserved = budget.getReservedBytes();
+ assertTrue(reserved > 0);
+
+ tmp.close();
+
+ assertEquals(reserved, budget.getReservedBytes(),
+ "a handed-out channel still pins the array; nothing is
released under it");
+ assertEquals(BIG, channel.size());
+ channel.close();
+ assertEquals(0, budget.getReservedBytes(), "the last pin releases the
reservation");
+ }
+}
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/PackageParser.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/PackageParser.java
index cba7fc9d22..1bc5b317b7 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/PackageParser.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/PackageParser.java
@@ -176,8 +176,9 @@ public class PackageParser extends AbstractArchiveParser {
if (extractor.shouldParseEmbedded(entrydata, context)) {
TemporaryResources tmp = new TemporaryResources();
- try {
- TikaInputStream tis = TikaInputStream.get(archive, tmp,
entrydata);
+ // close the stream, not only the temp dir: an in-memory cache
is not owned by
+ // tmp, and its budget reservation is released only on close
+ try (TikaInputStream tis = TikaInputStream.get(archive, tmp,
entrydata)) {
extractor.parseEmbedded(tis, xhtml, entrydata, context,
true);
} finally {
tmp.dispose();
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 1bf93fb580..b0c97c1973 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
@@ -182,9 +182,8 @@ public class SevenZParser extends AbstractArchiveParser {
if (extractor.shouldParseEmbedded(entrydata, context)) {
TemporaryResources tmp = new TemporaryResources();
- try {
- TikaInputStream tis = TikaInputStream.get(
- new SevenZEntryInputStream(sevenZFile), tmp,
entrydata);
+ try (TikaInputStream tis = TikaInputStream.get(
+ new SevenZEntryInputStream(sevenZFile), tmp, entrydata)) {
extractor.parseEmbedded(tis, xhtml, entrydata, context, true);
} finally {
tmp.dispose();
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 405bb314bb..c21ece862b 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
@@ -548,8 +548,7 @@ public class ZipParser extends AbstractArchiveParser {
if (extractor.shouldParseEmbedded(entryMetadata, context)) {
TemporaryResources tmp = new TemporaryResources();
- try {
- TikaInputStream tis = TikaInputStream.get(zis, tmp,
entryMetadata);
+ try (TikaInputStream tis = TikaInputStream.get(zis, tmp,
entryMetadata)) {
extractor.parseEmbedded(tis, xhtml, entryMetadata, context,
true);
} catch (UnsupportedZipFeatureException e) {
EmbeddedDocumentUtil.recordEmbeddedStreamException(e,
parentMetadata, context);
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/test/java/org/apache/tika/parser/pkg/PackageEmbeddedBudgetTest.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/test/java/org/apache/tika/parser/pkg/PackageEmbeddedBudgetTest.java
new file mode 100644
index 0000000000..e73e9b1fd4
--- /dev/null
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/test/java/org/apache/tika/parser/pkg/PackageEmbeddedBudgetTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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.assertNotNull;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Base64;
+import java.util.List;
+
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
+import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
+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.parser.ParseContext;
+
+/**
+ * An archive entry's in-memory cache reserves against the shared {@link
CacheMemoryBudget}
+ * when it is digested (rewind is enabled, the cache grows past the 1 MB
threshold). That
+ * reservation is released only when the entry's TikaInputStream is closed;
disposing the
+ * TemporaryResources alone does not reach it. Leaked reservations pin the
budget at its
+ * maximum after a few thousand documents, after which every cache spills to
disk.
+ */
+public class PackageEmbeddedBudgetTest extends AbstractPkgTest {
+
+ @TempDir
+ Path tmp;
+
+ /** Three entries over the 1 MB StreamCache threshold, so each one
reserves budget. */
+ private Path buildTar() throws Exception {
+ Path tar = tmp.resolve("entries.tar");
+ try (TarArchiveOutputStream out =
+ new TarArchiveOutputStream(Files.newOutputStream(tar))) {
+ for (int i = 0; i < 3; i++) {
+ byte[] data = new byte[2 * 1024 * 1024];
+ for (int j = 0; j < data.length; j++) {
+ data[j] = (byte) ('a' + (j + i) % 26);
+ }
+ TarArchiveEntry entry = new TarArchiveEntry("entry-" + i +
".txt");
+ entry.setSize(data.length);
+ out.putArchiveEntry(entry);
+ out.write(data);
+ out.closeArchiveEntry();
+ }
+ }
+ return tar;
+ }
+
+ @Test
+ public void budgetIsReleasedAfterEachEntry() throws Exception {
+ CacheMemoryBudget budget = new CacheMemoryBudget(64L * 1024 * 1024);
+ ParseContext context = new ParseContext();
+ context.set(DigesterFactory.class, () -> new
InputStreamDigester("MD5", "tk:digest:MD5",
+ bytes -> Base64.getEncoder().encodeToString(bytes)));
+ context.set(CacheMemoryBudget.class, budget);
+
+ List<Metadata> metadataList = getRecursiveMetadata(buildTar(),
context, false);
+
+ assertEquals(4, metadataList.size(), "container plus three entries");
+ for (int i = 1; i < metadataList.size(); i++) {
+ assertNotNull(metadataList.get(i).get("tk:digest:MD5"), "entry " +
i + " digested");
+ }
+ assertEquals(0, budget.getReservedBytes(),
+ "every entry's in-memory cache must return its reservation on
close");
+ }
+}
