xiangfu0 commented on code in PR #19238:
URL: https://github.com/apache/pinot/pull/19238#discussion_r3787087097
##########
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:
Thanks—agreed that the failure type should reach the server log. Passing the
original exception to the existing four-argument constructor would log its
message for 4xx or the full throwable for 5xx, and those values can include the
submitted path or URI query parameters. I added a TYPE_ONLY mode and use it in
both branches, so the response remains generic while the log records only the
exception class. Tests cover both status families with sentinel URI data.
##########
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:
Good catch on the resolution scope. I moved URI logging around the full
resolve/use lifecycle, so resolution, initialization, copy/build/upload, and
linkage failures are covered. URI logging remains type-only because filesystem
exception messages and causes can contain source paths or signed URI
parameters. The old inner URI log was removed; multipart logging is unchanged.
##########
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:
Removed the unused three-argument overload. The remaining overload requires
every caller to state the local-filesystem policy explicitly.
##########
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:
Closing a replaced mapping is unsafe because callers can retain the
non-closing wrapper returned by create() and may still be using the old Hadoop
or S3 client. Instead, I added atomic, configuration-aware registerIfNeeded()
for the repeated Spark and Hadoop executor setup paths. Equivalent
registrations reuse one initialized client; a differing explicit class or
configuration still replaces the mapping without closing retained wrappers.
Tests cover equivalent reuse, differing mappings, nullable compatibility, and
concurrent initialization.
##########
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:
Replaced the reflection assertion with a behavioral close-aware filesystem
test. It initializes two adapters with the same Hadoop configuration, closes
the first, and verifies the second remains usable. The old shared
FileSystem.get() behavior fails this test, while independently owned clients
pass.
--
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]
