lsyldliu commented on code in PR #20001:
URL: https://github.com/apache/flink/pull/20001#discussion_r916142999
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/resource/ResourceManager.java:
##########
@@ -69,43 +70,67 @@ public URLClassLoader getUserClassLoader() {
return userClassLoader;
}
- public void registerResource(ResourceUri resourceUri) throws IOException {
- // check whether the resource has been registered
- if (resourceInfos.containsKey(resourceUri)) {
- LOG.info(
- "Resource [{}] has been registered, overwriting of
registered resource is not supported "
- + "in the current version, skipping.",
- resourceUri.getUri());
- return;
- }
-
- // here can check whether the resource path is valid
- Path path = new Path(resourceUri.getUri());
- // check resource firstly
- checkResource(path);
-
- URL localUrl;
- // check resource scheme
- String scheme = StringUtils.lowerCase(path.toUri().getScheme());
- // download resource to local path firstly if in remote
- if (scheme != null && !"file".equals(scheme)) {
- localUrl = downloadResource(path);
- } else {
- localUrl = getURLFromPath(path);
+ public void registerResource(List<ResourceUri> resourceUris) throws
IOException {
+ // Due to anyone of the resource in list maybe fail during register,
so we should stage it
+ // before actual register to guarantee transaction process. If all the
resource check
+ // successfully, register them in batch.
+ Map<ResourceUri, URL> stagingResourceLocalURLs = new HashMap<>();
+
+ for (ResourceUri resourceUri : resourceUris) {
+ // check whether the resource has been registered
+ if (resourceInfos.containsKey(resourceUri)) {
+ LOG.info(
+ "Resource [{}] has been registered, overwriting of
registered resource is not supported "
+ + "in the current version, skipping.",
+ resourceUri.getUri());
+ continue;
+ }
+
+ // here can check whether the resource path is valid
+ Path path = new Path(resourceUri.getUri());
+ // check resource firstly
+ checkResource(path);
+
+ URL localUrl;
+ // check resource scheme
+ String scheme = StringUtils.lowerCase(path.toUri().getScheme());
+ // download resource to local path firstly if in remote
+ if (scheme != null && !"file".equals(scheme)) {
+ localUrl = downloadResource(path);
+ } else {
+ localUrl = getURLFromPath(path);
+ }
+
+ // check the jar resource extra
+ if (ResourceType.JAR.equals(resourceUri.getResourceType())) {
+ JarUtils.checkJarFile(localUrl);
+ }
+
+ // add it to staging map
+ stagingResourceLocalURLs.put(resourceUri, localUrl);
}
- // only need add jar resource to classloader
- if (ResourceType.JAR.equals(resourceUri.getResourceType())) {
- // check the Jar file firstly
- JarUtils.checkJarFile(localUrl);
-
- // add it to classloader
- userClassLoader.addURL(localUrl);
- LOG.info("Added jar resource [{}] to class path.", localUrl);
- }
+ // register resource in batch
+ stagingResourceLocalURLs.forEach(
+ (resourceUri, url) -> {
+ // jar resource need add to classloader
+ if
(ResourceType.JAR.equals(resourceUri.getResourceType())) {
+ userClassLoader.addURL(url);
+ LOG.info("Added jar resource [{}] to class path.",
url);
+ }
Review Comment:
File doesn't need add to classloader, you can see hive and spark. BTW, I
also don't know how it work. I think we will know when we need use it.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
