mayankshriv commented on code in PR #19378:
URL: https://github.com/apache/pinot/pull/19378#discussion_r3899633519
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufCodeGenMessageDecoder.java:
##########
@@ -104,4 +123,29 @@ public static Descriptors.Descriptor
getDescriptorForProtoClass(ClassLoader prot
Class<? extends Message> updateMessage = (Class<Message>)
protoMessageClsLoader.loadClass(protoClassName);
return (Descriptors.Descriptor)
updateMessage.getMethod("getDescriptor").invoke(null);
}
+
+ /// Resolves a file path (URI string) to a local [File]. For local files (no
scheme or `file://` scheme),
+ /// the original file is returned directly. For remote files, the file is
copied to a local temporary
+ /// directory. The caller is responsible for the lifetime of the returned
file - for remote files, the
+ /// backing temp directory is intentionally NOT cleaned up because the JAR
must remain accessible for
+ /// lazy class loading at decode time.
+ private static File resolveToLocalFile(String filePath)
+ throws Exception {
+ URI fileURI = URI.create(filePath);
+ String scheme = fileURI.getScheme();
+ if (scheme == null || PinotFSFactory.LOCAL_PINOT_FS_SCHEME.equals(scheme))
{
+ return new File(fileURI.getPath());
+ }
+ PinotFS pinotFS = PinotFSFactory.create(scheme);
+ Path localTmpDir = Files.createTempDirectory(ProtoBufUtils.TMP_DIR_PREFIX);
+ try {
+ File localFile = new File(localTmpDir.toFile(), new
File(fileURI.getPath()).getName());
+ LOGGER.info("Copying protocol buffer JAR from {} to {}", filePath,
localFile.getAbsolutePath());
+ pinotFS.copyToLocalFile(fileURI, localFile);
+ return localFile;
Review Comment:
Good catch. Fixed in two ways:
1. `_localTmpDir` is now an instance field so the decoder tracks ownership
of the temp directory.
2. `init()` wraps the post-copy work (class loading, codegen, compilation)
in a try-catch that calls `cleanupTempDir()` on failure.
This ensures the temp directory is cleaned up if init fails at any point
after the remote copy succeeds. On successful init, the temp dir persists for
the decoder lifetime (needed for lazy class resolution). The segment-offload
case remains a known limitation since `StreamMessageDecoder` does not extend
`Closeable` - documented in the class Javadoc.
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufTempFileLeakTest.java:
##########
@@ -0,0 +1,211 @@
+/**
+ * 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.IOException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.testng.annotations.Test;
+
+import static
org.apache.pinot.plugin.inputformat.protobuf.ProtoBufCodeGenMessageDecoder.PROTOBUF_JAR_FILE_PATH;
+import static
org.apache.pinot.plugin.inputformat.protobuf.ProtoBufCodeGenMessageDecoder.PROTO_CLASS_NAME;
+import static
org.apache.pinot.plugin.inputformat.protobuf.ProtoBufTestDataGenerator.createComplexTypeRecord;
+import static
org.apache.pinot.plugin.inputformat.protobuf.ProtoBufTestDataGenerator.getComplexTypeObject;
+import static
org.apache.pinot.plugin.inputformat.protobuf.ProtoBufTestDataGenerator.getFieldsInSampleRecord;
+import static
org.apache.pinot.plugin.inputformat.protobuf.ProtoBufTestDataGenerator.getSampleRecordMessage;
+import static
org.apache.pinot.plugin.inputformat.protobuf.ProtoBufTestDataGenerator.getSourceFieldsForComplexType;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Verifies that protobuf decoder and reader operations do not leak temporary
directories.
+///
+/// Each test snapshots the set of `pinot-protobuf*` directories in the system
temp directory before the operation,
+/// then asserts that no new ones remain afterward. The functional assertions
confirm the operation itself still works
+/// correctly.
+public class ProtoBufTempFileLeakTest {
+ private static final Path TEMP_DIR =
Path.of(System.getProperty("java.io.tmpdir"));
+
+ /// [ProtoBufMessageDecoder#init] with a descriptor file should not leak a
temp directory.
+ /// This is the streaming consumer path - called once per consumer init.
+ @Test
+ public void testMessageDecoderInitDoesNotLeakTempDir()
+ throws Exception {
+ Set<Path> before = listProtobufTempDirs();
+
+ Map<String, String> decoderProps = new HashMap<>();
+ URL descriptorFile =
getClass().getClassLoader().getResource("sample.desc");
+ decoderProps.put("descriptorFile", descriptorFile.toURI().toString());
+ ProtoBufMessageDecoder decoder = new ProtoBufMessageDecoder();
+ decoder.init(decoderProps, getFieldsInSampleRecord(), "");
+
+ // Verify functional correctness - decoding still works
+ Sample.SampleRecord sampleRecord = getSampleRecordMessage();
+ GenericRow destination = new GenericRow();
+ decoder.decode(sampleRecord.toByteArray(), destination);
+ assertEquals(destination.getValue("email"), "[email protected]");
+ assertEquals(destination.getValue("name"), "Alice");
+ assertEquals(destination.getValue("id"), 18);
+
+ assertNoNewTempDirs(before);
+ }
+
+ /// [ProtoBufMessageDecoder#init] with a complex descriptor should not leak
a temp directory.
+ @Test
+ public void testMessageDecoderComplexDescriptorDoesNotLeakTempDir()
+ throws Exception {
+ Set<Path> before = listProtobufTempDirs();
+
+ Map<String, String> decoderProps = new HashMap<>();
+ URL descriptorFile =
getClass().getClassLoader().getResource("complex_types.desc");
+ decoderProps.put("descriptorFile", descriptorFile.toURI().toString());
+ ProtoBufMessageDecoder decoder = new ProtoBufMessageDecoder();
+ decoder.init(decoderProps, getSourceFieldsForComplexType(), "");
+
+ // Verify functional correctness
+ Map<String, Object> inputRecord = createComplexTypeRecord();
+ GenericRow destination = new GenericRow();
+ decoder.decode(getComplexTypeObject(inputRecord).toByteArray(),
destination);
+ assertNotNull(destination.getValue("string_field"));
+ assertEquals(destination.getValue("string_field"), "hello");
+
+ assertNoNewTempDirs(before);
+ }
+
+ /// [ProtoBufCodeGenMessageDecoder#init] with a JAR file should not leak a
temp directory.
+ /// This is the streaming consumer codegen path.
+ @Test
+ public void testCodeGenDecoderInitDoesNotLeakTempDir()
+ throws Exception {
+ Set<Path> before = listProtobufTempDirs();
+
+ Map<String, String> decoderProps = new HashMap<>();
+ URL jarFile = getClass().getClassLoader().getResource("sample.jar");
Review Comment:
Done. Added a `DelegatingRemotePinotFS` (extends `LocalPinotFS`, delegates
via a static base dir) and a `FailingCopyPinotFS` (always throws on
`copyToLocalFile`), both registered in `@BeforeClass`. Four new tests:
- `testOpenDescriptorFileWithRemoteFS` - exercises the non-file PinotFS path
in `openDescriptorFile`, asserts no temp dirs.
- `testCodeGenDecoderWithRemoteJarCreatesLocalCopy` - verifies remote JAR
creates exactly one temp dir and decoding works.
- `testCodeGenDecoderCleansUpOnCopyFailure` - verifies no leaked temp dir
when `copyToLocalFile` fails.
- `testCodeGenDecoderCleansUpOnInitFailureAfterCopy` - verifies cleanup when
init fails after a successful copy (invalid class name triggers
`ClassNotFoundException`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
