Improve Controller ingestFromURI filesystem validation - #19238
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19238 +/- ##
=========================================
Coverage 66.97% 66.98%
Complexity 1423 1423
=========================================
Files 3453 3453
Lines 218936 218975 +39
Branches 34802 34802
=========================================
+ Hits 146638 146672 +34
- Misses 60588 60592 +4
- Partials 11710 11711 +1
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
f02926f to
036d56b
Compare
Validate filesystem policy before creating ingestion state, and avoid registering request-provided filesystems in shared factory state. Request-scoped filesystem instances own their lifecycle. Hadoop-backed instances use independent clients so closing one request does not affect another.
036d56b to
dab1605
Compare
yashmayya
left a comment
yashmayya
left a comment
There was a problem hiding this comment.
Read through the whole change. The ordering and the filesystem lifecycle look right to me:
- Resolving the filesystem before anything else means a rejected request creates no working directory, and holding the resolved instance removes the window where a factory change could swap it out mid-request.
- Dropping
PinotFSFactory.register()from the request path is a good side effect: request threads no longer mutate the static, unsynchronizedPINOT_FS_MAP. - Checking the class before construction works as intended. I confirmed
PluginManager.loadClassonly links the class, it does not initialize it, so nothing in a rejected class runs. isFileSystemInstanceOfwalking the fullNoClosePinotFSchain covers the nested delegate case.LocalPinotFSis the only bundled implementation that can read controller-local paths.HadoopPinotFScannot be aimed at one, because Hadoop'scheckPathrequires the URI scheme to match the filesystem, and thefilescheme is rejected earlier.
Test coverage is thorough. A few comments below. The HadoopPinotFS one is the only one I would like handled before or soon after merge; the rest are small.
| } catch (IllegalArgumentException | URISyntaxException e) { | ||
| asyncResponse.resume( | ||
| new ControllerApplicationException(LOGGER, "Invalid ingestFromURI request", Response.Status.BAD_REQUEST)); | ||
| } catch (Exception | LinkageError e) { | ||
| asyncResponse.resume(new ControllerApplicationException(LOGGER, "Failed to ingest from URI", | ||
| Response.Status.INTERNAL_SERVER_ERROR)); |
There was a problem hiding this comment.
Both branches drop e. The 3-arg ControllerApplicationException(logger, message, status) logs only the message (logger.info(message) for 4xx, logger.error(message) for 5xx), so the cause never reaches the log and a failure here is not debuggable.
Please pass e to the 4-arg constructor. The response body is built from message by WebApplicationExceptionMapper, so it stays generic either way.
| if (payload._payloadType == PayloadType.URI) { | ||
| LOGGER.error("Failed URI ingestion for table: {}, exception type: {}", tableNameWithType, | ||
| e.getClass().getName()); | ||
| } else { | ||
| LOGGER.error("Caught exception when ingesting file to table: {}", tableNameWithType, e); | ||
| } |
There was a problem hiding this comment.
For URI payloads this keeps only the exception class name. Note also that failures from resolveSourceFileSystem are thrown before this try block, so they are not logged here at all.
This log is server side and is not returned to the caller, so I would log the full exception in both branches.
| public static void copyURIToLocal(Map<String, String> batchConfigMap, URI sourceFileURI, File destFile, | ||
| boolean allowLocalFileSystem) | ||
| throws Exception { | ||
| try (ResolvedFileSystem sourceFileSystem = |
There was a problem hiding this comment.
The 3-arg copyURIToLocal just above (line 192) passes allowLocalFileSystem = true and now has no callers anywhere, in main or test code. Please delete it so callers always state the flag.
| // Hadoop's FileSystem.get() returns a process-cached instance. HadoopPinotFS closes its filesystem, so it must | ||
| // own a distinct instance to avoid one PinotFS closing a client that is still in use elsewhere. | ||
| _hadoopFS = org.apache.hadoop.fs.FileSystem.newInstance(_hadoopConf); |
There was a problem hiding this comment.
This is the right ownership fix, but newInstance() also leaves the shared cache, so every instance must now be closed or it stays alive.
PinotFSFactory.register() overwrites the map entry without closing the value it replaces, and two callers re-register in a loop inside a long-lived JVM:
SparkSegmentGenerationJobRunnerregisters allpinotFSSpecsonce per input path insidepathRDD.foreach, on the executor.HadoopSegmentCreationMapperdoes the same per mapper.
With get() these all shared one cached client, so overwriting was harmless. Now each call builds a client that nothing closes. Hadoop's newInstance() goes through Cache.getUnique(), which keeps the instance in the static cache until it is closed, so this holds heap, RPC connections and lease renewer threads.
Can PinotFSFactory.register() close the instance it replaces? That covers every implementation, not only this one. S3PinotFS has the same shape today.
| try (HadoopPinotFS first = new HadoopPinotFS(); HadoopPinotFS second = new HadoopPinotFS()) { | ||
| first.init(new PinotConfiguration()); | ||
| second.init(new PinotConfiguration()); | ||
|
|
||
| Field hadoopFileSystemField = HadoopPinotFS.class.getDeclaredField("_hadoopFS"); | ||
| hadoopFileSystemField.setAccessible(true); | ||
| Assert.assertNotSame(hadoopFileSystemField.get(first), hadoopFileSystemField.get(second)); | ||
| } |
There was a problem hiding this comment.
This reads a private field by reflection. The behaviour that matters is that closing one instance leaves the other usable. Close first, then call something on second and assert it still works. That tests the contract directly and survives a field rename.
3 participants
Summary
ingestFromURIlocal-filesystem policy by resolving and validating the exact filesystem before ingestion work startsExisting behavior on master
Master already contains
dd6520c7267/ #18660, which introduced the default-off setting and the initial direct URI and class checks. This change builds on that implementation to cover remaining delegate, lifecycle, ordering, and error-handling cases.Behavior before this change
Filesystem selection could happen during copying, and endpoint-provided implementations were registered in shared factory state. Rejected requests could also create Controller staging directories before source validation.
Review focus
This PR has an additive
pinot-spisurface and filesystem lifecycle contract change:PinotFSFactory.isFileSystemInstanceOf(...)recursively checks Pinot non-closing delegates without exposing the delegatePinotFS.close()documents that implementations close only resources owned by that instanceHadoopPinotFSuses an independently owned Hadoop client because the adapter closes itFilesystem-plugin and Controller maintainers should review these ownership and compatibility details.
Testing
PinotFSFactoryTest,ControllerConfTest,FileIngestionHelperTest,PinotIngestionRestletResourceStatelessTest, andHadoopPinotFSTestpinot-spi,pinot-controller, andpinot-hdfs