This is an automated email from the ASF dual-hosted git repository.
yashmayya 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 faad15f3dc7 Fix temp file leak in batch metadata segment upload
(#19377)
faad15f3dc7 is described below
commit faad15f3dc782dcf7ca7c4e37ab2d04d2c9f3792
Author: Shounak kulkarni <[email protected]>
AuthorDate: Thu Aug 27 21:08:38 2026 +0530
Fix temp file leak in batch metadata segment upload (#19377)
---
.../PinotSegmentUploadDownloadRestletResource.java | 31 +++++++----
...otSegmentUploadDownloadRestletResourceTest.java | 63 ++++++++++++++++++++++
2 files changed, 83 insertions(+), 11 deletions(-)
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
index 5446967d9e3..d9deb410e9d 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
@@ -604,10 +604,10 @@ public class PinotSegmentUploadDownloadRestletResource {
List<SegmentUploadMetadata> segmentUploadMetadataList = new ArrayList<>();
List<File> tempFiles = new ArrayList<>();
List<String> segmentNames = new ArrayList<>();
- Map<String, SegmentMetadataInfo> segmentsMetadataInfoMap =
createSegmentsMetadataInfoMap(multiPart);
- LOGGER.info("Uploading segments in batch mode of size: {}",
segmentsMetadataInfoMap.size());
try {
+ Map<String, SegmentMetadataInfo> segmentsMetadataInfoMap =
createSegmentsMetadataInfoMap(multiPart, tempFiles);
+ LOGGER.info("Uploading segments in batch mode of size: {}",
segmentsMetadataInfoMap.size());
int entryCount = 0;
for (Map.Entry<String, SegmentMetadataInfo> entry :
segmentsMetadataInfoMap.entrySet()) {
String segmentName = entry.getKey();
@@ -1294,26 +1294,31 @@ public class PinotSegmentUploadDownloadRestletResource {
String uuid = UUID.randomUUID().toString();
File segmentMetadataDir =
new File(FileUtils.getTempDirectory(),
SegmentUploadConstants.SEGMENT_METADATA_DIR_PREFIX + uuid);
- FileUtils.copyFile(creationMetaFile, new File(segmentMetadataDir,
V1Constants.SEGMENT_CREATION_META));
- FileUtils.copyFile(metadataPropertiesFile,
- new File(segmentMetadataDir,
V1Constants.MetadataKeys.METADATA_FILE_NAME));
File segmentMetadataTarFile = new File(FileUtils.getTempDirectory(),
SegmentUploadConstants.SEGMENT_METADATA_TAR_FILE_PREFIX + uuid +
TarCompressionUtils.TAR_GZ_FILE_EXTENSION);
- if (segmentMetadataTarFile.exists()) {
- FileUtils.forceDelete(segmentMetadataTarFile);
- }
- TarCompressionUtils.createCompressedTarFile(segmentMetadataDir,
segmentMetadataTarFile);
try {
+ FileUtils.copyFile(creationMetaFile, new File(segmentMetadataDir,
V1Constants.SEGMENT_CREATION_META));
+ FileUtils.copyFile(metadataPropertiesFile,
+ new File(segmentMetadataDir,
V1Constants.MetadataKeys.METADATA_FILE_NAME));
+ if (segmentMetadataTarFile.exists()) {
+ FileUtils.forceDelete(segmentMetadataTarFile);
+ }
+ TarCompressionUtils.createCompressedTarFile(segmentMetadataDir,
segmentMetadataTarFile);
FileUtils.copyFile(segmentMetadataTarFile, destFile);
} finally {
- FileUtils.forceDelete(segmentMetadataTarFile);
+ // Both the staging directory and the intermediate tar file are scoped
to this call. Delete quietly so that a
+ // cleanup failure does not mask the exception that caused it.
+ FileUtils.deleteQuietly(segmentMetadataDir);
+ FileUtils.deleteQuietly(segmentMetadataTarFile);
}
}
// The multipart input would contain a single multipart and this part would
contain the segment metadata
// files (creation.meta, metadata.properties), and an additional mapping file
names 'all_segments_metadata' which
// would contain the mappings from segment names to segment download URI's.
- private static Map<String, SegmentMetadataInfo>
createSegmentsMetadataInfoMap(FormDataMultiPart multiPart) {
+ @VisibleForTesting
+ static Map<String, SegmentMetadataInfo>
createSegmentsMetadataInfoMap(FormDataMultiPart multiPart,
+ List<File> tempFiles) {
List<BodyPart> bodyParts = multiPart.getBodyParts();
validateMultiPartForBatchSegmentUpload(bodyParts);
FormDataBodyPart bodyPartFromReq = (FormDataBodyPart) bodyParts.get(0);
@@ -1322,6 +1327,7 @@ public class PinotSegmentUploadDownloadRestletResource {
File allSegmentsMetadataTarFile = new File(FileUtils.getTempDirectory(),
SegmentUploadConstants.ALL_SEGMENTS_METADATA_TAR_FILE_PREFIX + uuid
+ TarCompressionUtils.TAR_GZ_FILE_EXTENSION);
+ tempFiles.add(allSegmentsMetadataTarFile);
try {
createSegmentFileFromBodyPart(bodyPartFromReq,
allSegmentsMetadataTarFile);
} catch (IOException e) {
@@ -1332,6 +1338,9 @@ public class PinotSegmentUploadDownloadRestletResource {
List<File> segmentsMetadataFiles = new ArrayList<>();
File allSegmentsMetadataDir = new File(FileUtils.getTempDirectory(),
SegmentUploadConstants.ALL_SEGMENTS_METADATA_DIR_PREFIX + uuid);
+ // This directory backs the File handles held by the returned
SegmentMetadataInfo values, so it can only be
+ // removed once the caller is done with the map
+ tempFiles.add(allSegmentsMetadataDir);
try {
FileUtils.forceMkdir(allSegmentsMetadataDir);
List<File> metadataFiles =
TarCompressionUtils.untar(allSegmentsMetadataTarFile, allSegmentsMetadataDir);
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResourceTest.java
index cd6107e581c..e3f6fb0f4b6 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResourceTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResourceTest.java
@@ -26,10 +26,13 @@ import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Method;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.UUID;
+import java.util.stream.Collectors;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.Response;
import org.apache.commons.io.FileUtils;
@@ -38,6 +41,7 @@ import org.apache.pinot.common.utils.TarCompressionUtils;
import org.apache.pinot.controller.ControllerConf;
import
org.apache.pinot.controller.api.exception.ControllerApplicationException;
import org.apache.pinot.controller.api.upload.SegmentMetadataInfo;
+import org.apache.pinot.segment.local.constants.SegmentUploadConstants;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.crypt.NoOpPinotCrypter;
import org.apache.pinot.spi.crypt.PinotCrypterFactory;
@@ -220,11 +224,70 @@ public class
PinotSegmentUploadDownloadRestletResourceTest {
File destFile = new File(_tempDir, "outputSegment");
+ Set<String> tempEntriesBefore =
listTempEntries(SegmentUploadConstants.SEGMENT_METADATA_DIR_PREFIX,
+ SegmentUploadConstants.SEGMENT_METADATA_TAR_FILE_PREFIX);
+
// test
PinotSegmentUploadDownloadRestletResource.createSegmentFileFromSegmentMetadataInfo(metadataInfo,
destFile);
// verify
Assert.assertTrue(FileUtils.getFile(destFile).exists());
+ // The staging directory and the intermediate tar file are scoped to the
call and must not be left behind
+
assertEquals(listTempEntries(SegmentUploadConstants.SEGMENT_METADATA_DIR_PREFIX,
+ SegmentUploadConstants.SEGMENT_METADATA_TAR_FILE_PREFIX),
tempEntriesBefore);
+ }
+
+ @Test
+ public void testCreateSegmentsMetadataInfoMapRegistersTempFilesForCleanup()
+ throws IOException {
+ // setup: an uber tar holding the metadata files of a single segment plus
the download URI mapping file
+ String segmentName = "mySegmentName";
+ String downloadURI = "/path/to/segment/download/uri";
+ File allSegmentsMetadataDir = new File(_tempDir, "allSegmentsMetadata");
+ FileUtils.forceMkdir(allSegmentsMetadataDir);
+ FileUtils.touch(new File(allSegmentsMetadataDir, segmentName +
".creation.meta"));
+ FileUtils.touch(new File(allSegmentsMetadataDir, segmentName +
".metadata.properties"));
+ FileUtils.writeLines(new File(allSegmentsMetadataDir,
SegmentUploadConstants.ALL_SEGMENTS_METADATA_FILENAME),
+ List.of(segmentName, downloadURI));
+ File uberTarFile = new File(_tempDir, "allSegments.tar.gz");
+ TarCompressionUtils.createCompressedTarFile(allSegmentsMetadataDir,
uberTarFile);
+
+ FormDataBodyPart mockBodyPart = mock(FormDataBodyPart.class);
+
when(mockBodyPart.getValueAs(InputStream.class)).thenReturn(FileUtils.openInputStream(uberTarFile));
+ FormDataMultiPart mockMultiPart = mock(FormDataMultiPart.class);
+ when(mockMultiPart.getBodyParts()).thenReturn(List.of(mockBodyPart));
+
+ Set<String> tempEntriesBefore =
listTempEntries(SegmentUploadConstants.ALL_SEGMENTS_METADATA_TAR_FILE_PREFIX,
+ SegmentUploadConstants.ALL_SEGMENTS_METADATA_DIR_PREFIX);
+ List<File> tempFiles = new ArrayList<>();
+
+ // test
+ Map<String, SegmentMetadataInfo> segmentsMetadataInfoMap =
+
PinotSegmentUploadDownloadRestletResource.createSegmentsMetadataInfoMap(mockMultiPart,
tempFiles);
+
+ // verify the map is built and its file handles are still readable, i.e.
cleanup was deferred to the caller
+ assertEquals(segmentsMetadataInfoMap.size(), 1);
+ SegmentMetadataInfo metadataInfo =
segmentsMetadataInfoMap.get(segmentName);
+ assertEquals(metadataInfo.getSegmentDownloadURI(), downloadURI);
+ Assert.assertTrue(metadataInfo.getSegmentCreationMetaFile().exists());
+
Assert.assertTrue(metadataInfo.getSegmentMetadataPropertiesFile().exists());
+
+ // verify both request-scoped temp files were handed to the caller, and
that cleaning them leaves nothing behind
+ assertEquals(tempFiles.size(), 2);
+ tempFiles.forEach(FileUtils::deleteQuietly);
+
assertEquals(listTempEntries(SegmentUploadConstants.ALL_SEGMENTS_METADATA_TAR_FILE_PREFIX,
+ SegmentUploadConstants.ALL_SEGMENTS_METADATA_DIR_PREFIX),
tempEntriesBefore);
+ }
+
+ /// Names of the entries directly under the JVM temp directory that start
with any of the given prefixes. Used to
+ /// assert that a call leaves no residue there, without being confused by
unrelated entries.
+ private static Set<String> listTempEntries(String... prefixes) {
+ String[] names = FileUtils.getTempDirectory().list();
+ if (names == null) {
+ return Set.of();
+ }
+ return Arrays.stream(names).filter(name ->
Arrays.stream(prefixes).anyMatch(name::startsWith))
+ .collect(Collectors.toSet());
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
