This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new bf0793af430 Fix protobuf descriptor file temp directory leak (#19378)
bf0793af430 is described below
commit bf0793af430a9216267d43bb675bec9a50f162a7
Author: Mayank Shrivastava <[email protected]>
AuthorDate: Wed Sep 16 18:07:25 2026 -0700
Fix protobuf descriptor file temp directory leak (#19378)
ProtoBufUtils.getDescriptorFileInputStream() copied the descriptor file
to a temporary directory (prefix "pinot-protobuf") via
getFileCopiedToLocal() and wrapped it in a FileInputStream. Neither was
ever released, so every ProtoBufMessageDecoder.init() and
ProtoBufRecordReader.init() leaked one directory under java.io.tmpdir
plus one open file descriptor.
Replaced it with openDescriptorFile(), which resolves the scheme and
returns PinotFS.open(uri) directly - no local copy is made, and local
file:// URIs go through LocalPinotFS.open() just the same. Both callers
now read the descriptor inside try-with-resources so the stream is
closed once parsing completes.
Not in scope: ProtoBufCodeGenMessageDecoder passes its JAR through
getFileCopiedToLocal(), which still creates a per-decoder temp directory
that nothing deletes. Fixing that requires an owned classloader/segment
lifecycle or a JAR cache keyed by URI and version, and is left to a
follow-up so this change stays a small, complete fix for the descriptor
path.
Testing:
- Added ProtoBufTempFileLeakTest, which registers a ClasspathPinotFS
under a custom scheme so both consumers are configured with a remote
(non-file:) descriptor URI and therefore exercise the PinotFS path.
The fake filesystem counts open()/copyToLocalFile() calls and tracks
stream closure, so the assertions read the actual filesystem
interactions rather than scanning the shared java.io.tmpdir, which
races with any other test creating directories with the same prefix.
- testMessageDecoderInitStreamsDescriptorWithoutLocalCopy
- testRecordReaderInitStreamsDescriptorWithoutLocalCopy
Each asserts copyToLocalFile was never called, the descriptor was
opened exactly once, and the stream was closed.
- Both tests fail against the pre-fix code (copyToLocalFile count 1).
The record reader test was additionally verified by mutation: removing
its try-with-resources fails the close assertion. The decoder's stream
is also closed by DynamicSchema.parseFrom, so that assertion confirms
the stream is closed after init rather than attributing the close.
- Full module suite: 169/169 tests pass.
---
.../protobuf/ProtoBufMessageDecoder.java | 7 +-
.../inputformat/protobuf/ProtoBufRecordReader.java | 7 +-
.../plugin/inputformat/protobuf/ProtoBufUtils.java | 25 ++-
.../protobuf/ProtoBufTempFileLeakTest.java | 185 +++++++++++++++++++++
4 files changed, 211 insertions(+), 13 deletions(-)
diff --git
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoder.java
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoder.java
index cee7fac5c7e..5c79564f2ac 100644
---
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoder.java
+++
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoder.java
@@ -48,9 +48,10 @@ public class ProtoBufMessageDecoder implements
StreamMessageDecoder<byte[]> {
"Protocol Buffer schema descriptor file must be provided");
_protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- InputStream descriptorFileInputStream =
ProtoBufUtils.getDescriptorFileInputStream(
- props.get(DESCRIPTOR_FILE_PATH));
- Descriptors.Descriptor descriptor =
buildProtoBufDescriptor(descriptorFileInputStream);
+ Descriptors.Descriptor descriptor;
+ try (InputStream fin =
ProtoBufUtils.openDescriptorFile(props.get(DESCRIPTOR_FILE_PATH))) {
+ descriptor = buildProtoBufDescriptor(fin);
+ }
_recordExtractor = new ProtoBufRecordExtractor();
_recordExtractor.init(fieldsToRead, null);
DynamicMessage dynamicMessage =
DynamicMessage.getDefaultInstance(descriptor);
diff --git
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordReader.java
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordReader.java
index edd76987b7b..201b08f2a9c 100644
---
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordReader.java
+++
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufRecordReader.java
@@ -79,10 +79,9 @@ public class ProtoBufRecordReader implements RecordReader {
private Descriptors.Descriptor
buildProtoBufDescriptor(ProtoBufRecordReaderConfig protoBufRecordReaderConfig)
throws IOException {
- try {
- InputStream fin = ProtoBufUtils.getDescriptorFileInputStream(
- protoBufRecordReaderConfig.getDescriptorFile().toString());
- DescriptorProtos.FileDescriptorSet set =
DescriptorProtos.FileDescriptorSet.parseFrom(fin);
+ try (InputStream in = ProtoBufUtils.openDescriptorFile(
+ protoBufRecordReaderConfig.getDescriptorFile().toString())) {
+ DescriptorProtos.FileDescriptorSet set =
DescriptorProtos.FileDescriptorSet.parseFrom(in);
Descriptors.FileDescriptor fileDescriptor =
Descriptors.FileDescriptor.buildFrom(set.getFile(0), new
Descriptors.FileDescriptor[]{});
return fileDescriptor.getMessageTypes().get(0);
diff --git
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java
index 6fd0e2866a9..9ec6110054c 100644
---
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java
+++
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java
@@ -21,7 +21,6 @@ package org.apache.pinot.plugin.inputformat.protobuf;
import com.google.protobuf.Descriptors;
import com.google.protobuf.ProtobufInternalUtils;
import java.io.File;
-import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URI;
import java.nio.file.Files;
@@ -39,6 +38,9 @@ public class ProtoBufUtils {
private ProtoBufUtils() {
}
+ /// Copies a remote file to a local temp directory and returns the local
copy. Used by the codegen
+ /// decoder path for JAR files that must remain on disk for the lifetime of
the classloader.
+ /// The caller is responsible for managing the lifetime of the returned file
and its parent directory.
public static File getFileCopiedToLocal(String filePath)
throws Exception {
URI fileURI = URI.create(filePath);
@@ -60,11 +62,6 @@ public class ProtoBufUtils {
}
}
- public static InputStream getDescriptorFileInputStream(String
descriptorFilePath)
- throws Exception {
- return new FileInputStream(getFileCopiedToLocal(descriptorFilePath));
- }
-
public static File createLocalFile(URI srcURI, File dstDir) {
String sourceURIPath = srcURI.getPath();
File dstFile = new File(dstDir, new File(sourceURIPath).getName());
@@ -73,6 +70,22 @@ public class ProtoBufUtils {
return dstFile;
}
+ /// Opens a descriptor file (local or remote) and returns a stream over its
contents.
+ /// The caller is responsible for closing the returned stream. No temporary
files are created.
+ ///
+ /// @param descriptorFilePath URI string pointing to a `.desc` protobuf
descriptor file
+ /// @return an open [InputStream] over the descriptor file contents
+ public static InputStream openDescriptorFile(String descriptorFilePath)
+ throws Exception {
+ URI fileURI = URI.create(descriptorFilePath);
+ String scheme = fileURI.getScheme();
+ if (scheme == null) {
+ scheme = PinotFSFactory.LOCAL_PINOT_FS_SCHEME;
+ }
+ PinotFS pinotFS = PinotFSFactory.create(scheme);
+ return pinotFS.open(fileURI);
+ }
+
public static String getFullJavaName(Descriptors.Descriptor descriptor) {
String prefix;
if (null != descriptor.getContainingType()) {
diff --git
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufTempFileLeakTest.java
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufTempFileLeakTest.java
new file mode 100644
index 00000000000..477f83f6ca9
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufTempFileLeakTest.java
@@ -0,0 +1,185 @@
+/**
+ * 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.pinot.plugin.inputformat.protobuf;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.filesystem.LocalPinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import static
org.apache.pinot.plugin.inputformat.protobuf.ProtoBufTestDataGenerator.getFieldsInSampleRecord;
+import static
org.apache.pinot.plugin.inputformat.protobuf.ProtoBufTestDataGenerator.getSampleRecordMessage;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+/// Verifies that reading a protobuf descriptor file does not copy it to a
local temporary file.
+///
+/// Both consumers of [ProtoBufUtils#openDescriptorFile] are configured with a
remote (non-`file:`) URI backed by
+/// [ClasspathPinotFS], which counts filesystem interactions. Each test
asserts that the descriptor was streamed via
+/// [org.apache.pinot.spi.filesystem.PinotFS#open] and never staged through
`copyToLocalFile`, and that the consumer
+/// closed the stream it was handed. Assertions are made against these
counters rather than by scanning the shared
+/// `java.io.tmpdir`, which races with any other test creating directories
with the same prefix.
+public class ProtoBufTempFileLeakTest {
+ private static final String REMOTE_SCHEME = "proto-test-remote";
+ private static final String SAMPLE_DESCRIPTOR_URI = REMOTE_SCHEME +
":///sample.desc";
+
+ @BeforeClass
+ public void setUp() {
+ PinotFSFactory.register(REMOTE_SCHEME, ClasspathPinotFS.class.getName(),
null);
+ }
+
+ @BeforeMethod
+ public void resetCounters() {
+ ClasspathPinotFS.reset();
+ }
+
+ /// [ProtoBufMessageDecoder#init] must stream the descriptor and close it,
without staging a local copy.
+ @Test
+ public void testMessageDecoderInitStreamsDescriptorWithoutLocalCopy()
+ throws Exception {
+ Map<String, String> decoderProps = new HashMap<>();
+ decoderProps.put(ProtoBufMessageDecoder.DESCRIPTOR_FILE_PATH,
SAMPLE_DESCRIPTOR_URI);
+ ProtoBufMessageDecoder decoder = new ProtoBufMessageDecoder();
+ decoder.init(decoderProps, getFieldsInSampleRecord(), "");
+
+ // Verify decoding works off the remotely-read descriptor
+ GenericRow destination = new GenericRow();
+ decoder.decode(getSampleRecordMessage().toByteArray(), destination);
+ assertEquals(destination.getValue("email"), "[email protected]");
+
+ assertStreamedAndClosed();
+ }
+
+ /// [ProtoBufRecordReader#init] must stream the descriptor and close it,
without staging a local copy.
+ @Test
+ public void testRecordReaderInitStreamsDescriptorWithoutLocalCopy()
+ throws Exception {
+ File tempDataDir =
Files.createTempDirectory("protobuf-descriptor-test-data").toFile();
+ try {
+ File dataFile = new File(tempDataDir, "test.data");
+ try (FileOutputStream out = new FileOutputStream(dataFile)) {
+ getSampleRecordMessage().writeDelimitedTo(out);
+ }
+
+ ProtoBufRecordReaderConfig config = new ProtoBufRecordReaderConfig();
+ config.setDescriptorFile(URI.create(SAMPLE_DESCRIPTOR_URI));
+
+ try (ProtoBufRecordReader reader = new ProtoBufRecordReader()) {
+ reader.init(dataFile, getFieldsInSampleRecord(), config);
+ assertTrue(reader.hasNext());
+ GenericRow row = reader.next(new GenericRow());
+ assertEquals(row.getValue("email"), "[email protected]");
+ }
+ } finally {
+ FileUtils.deleteDirectory(tempDataDir);
+ }
+
+ assertStreamedAndClosed();
+ }
+
+ /// Asserts the descriptor was read exactly once via `open`, never staged
through `copyToLocalFile`, and that
+ /// every stream handed to the consumer was closed.
+ private static void assertStreamedAndClosed() {
+ assertEquals(ClasspathPinotFS.getCopyToLocalFileCount(), 0,
+ "Descriptor must be streamed, not copied to a local temporary file");
+ assertEquals(ClasspathPinotFS.getOpenCount(), 1, "Descriptor should be
opened exactly once");
+ assertEquals(ClasspathPinotFS.getClosedCount(),
ClasspathPinotFS.getOpenCount(),
+ "Every opened descriptor stream must be closed");
+ }
+
+ /// PinotFS that serves classpath resources and records how it was used.
Counters are static because
+ /// [PinotFSFactory] instantiates and wraps the filesystem itself, leaving
no handle on the instance.
+ public static class ClasspathPinotFS extends LocalPinotFS {
+ private static final AtomicInteger OPEN_COUNT = new AtomicInteger();
+ private static final AtomicInteger CLOSED_COUNT = new AtomicInteger();
+ private static final AtomicInteger COPY_TO_LOCAL_FILE_COUNT = new
AtomicInteger();
+
+ static void reset() {
+ OPEN_COUNT.set(0);
+ CLOSED_COUNT.set(0);
+ COPY_TO_LOCAL_FILE_COUNT.set(0);
+ }
+
+ static int getOpenCount() {
+ return OPEN_COUNT.get();
+ }
+
+ static int getClosedCount() {
+ return CLOSED_COUNT.get();
+ }
+
+ static int getCopyToLocalFileCount() {
+ return COPY_TO_LOCAL_FILE_COUNT.get();
+ }
+
+ @Override
+ public InputStream open(URI uri)
+ throws IOException {
+ OPEN_COUNT.incrementAndGet();
+ return new FilterInputStream(openResource(uri)) {
+ private boolean _closed;
+
+ /// Counts at most once per stream: protobuf parsing closes the stream
it consumes, and the caller's
+ /// try-with-resources closes it again. Both are correct; the
assertion cares that it was closed at all.
+ @Override
+ public void close()
+ throws IOException {
+ if (!_closed) {
+ _closed = true;
+ CLOSED_COUNT.incrementAndGet();
+ }
+ super.close();
+ }
+ };
+ }
+
+ @Override
+ public void copyToLocalFile(URI srcUri, File dstFile)
+ throws Exception {
+ COPY_TO_LOCAL_FILE_COUNT.incrementAndGet();
+ try (InputStream in = openResource(srcUri)) {
+ FileUtils.copyInputStreamToFile(in, dstFile);
+ }
+ }
+
+ private static InputStream openResource(URI uri)
+ throws IOException {
+ String name = new File(uri.getPath()).getName();
+ InputStream in =
ClasspathPinotFS.class.getClassLoader().getResourceAsStream(name);
+ if (in == null) {
+ throw new IOException("Classpath resource not found: " + name);
+ }
+ return in;
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
