yashmayya commented on code in PR #19238:
URL: https://github.com/apache/pinot/pull/19238#discussion_r3786174688
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/FileIngestionHelper.java:
##########
@@ -179,40 +197,104 @@ public static void copyURIToLocal(Map<String, String>
batchConfigMap, URI source
public static void copyURIToLocal(Map<String, String> batchConfigMap, URI
sourceFileURI, File destFile,
boolean allowLocalFileSystem)
throws Exception {
+ try (ResolvedFileSystem sourceFileSystem =
Review 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.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotIngestionRestletResource.java:
##########
@@ -197,14 +200,12 @@ public void ingestFromURI(
tableNameWithType = DatabaseUtils.translateTableName(tableNameWithType,
headers);
try {
asyncResponse.resume(ingestData(tableNameWithType, batchConfigMapStr,
new DataPayload(new URI(sourceURIStr))));
- } catch (IllegalArgumentException e) {
- asyncResponse.resume(new ControllerApplicationException(LOGGER, String
- .format("Got illegal argument when ingesting file into table: %s.
%s", tableNameWithType, e.getMessage()),
- Response.Status.BAD_REQUEST, e));
- } catch (Exception e) {
- asyncResponse.resume(new ControllerApplicationException(LOGGER,
- String.format("Caught exception when ingesting file into table: %s.
%s", tableNameWithType, e.getMessage()),
- Response.Status.INTERNAL_SERVER_ERROR, e));
+ } 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));
Review 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.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/FileIngestionHelper.java:
##########
@@ -158,12 +171,17 @@ public SuccessResponse buildSegmentAndPush(DataPayload
payload)
SegmentUploader segmentUploader =
PluginManager.get().createInstance(SEGMENT_UPLOADER_CLASS);
segmentUploader.init(tableConfigOverride);
segmentUploader.uploadSegment(segmentTarFile.toURI(), _authProvider);
- LOGGER.info("Uploaded tar: {} to table: {}",
segmentTarFile.getAbsolutePath(), tableNameWithType);
+ LOGGER.info("Uploaded generated segment to table: {}",
tableNameWithType);
return new SuccessResponse(
"Successfully ingested file into table: " + tableNameWithType + " as
segment: " + segmentName);
} catch (Exception e) {
- LOGGER.error("Caught exception when ingesting file to table: {}",
tableNameWithType, e);
+ 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);
+ }
Review 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.
##########
pinot-plugins/pinot-file-system/pinot-hdfs/src/test/java/org/apache/pinot/plugin/filesystem/HadoopPinotFSTest.java:
##########
@@ -479,4 +480,17 @@ public void testInitWithUnauthenticatedUser()
Assert.assertEquals(currentUser, testUser, "The Hadoop login user was
not set correctly!");
}
}
+
+ @Test
+ public void testInstancesOwnDistinctHadoopFileSystems()
+ throws Exception {
+ 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));
+ }
Review 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.
##########
pinot-plugins/pinot-file-system/pinot-hdfs/src/main/java/org/apache/pinot/plugin/filesystem/HadoopPinotFS.java:
##########
@@ -77,7 +77,9 @@ public void init(PinotConfiguration config) {
UserGroupInformation.setLoginUser(ugi);
LOGGER.info("Setting HDFS login user to: {}", globalHadoopUser);
}
- _hadoopFS = org.apache.hadoop.fs.FileSystem.get(_hadoopConf);
+ // 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);
Review 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:
- `SparkSegmentGenerationJobRunner` registers all `pinotFSSpecs` once per
input path inside `pathRDD.foreach`, on the executor.
- `HadoopSegmentCreationMapper` does 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.
--
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]
