You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/02/06 01:37:30 UTC
[GitHub] [pulsar] jerrypeng opened a new pull request #9500: Optimize built-in source/sink startup Part 2
jerrypeng opened a new pull request #9500:
URL: https://github.com/apache/pulsar/pull/9500
### Motivation
As a follow up to https://github.com/apache/pulsar/pull/9413. Allow the ThreadRuntime to leverage the already loaded and cached built-in connector registry so that it does not need to unpack NARs and calculate MD5 checksums as well.
### Modifications
- Moved the jar loading logic from JavaInstanceRunnable to ThreadRuntime.
- Add logic in ThreadRuntime to extract a classloader from a function package, e.g. JAR, or if it is a built-in connector just get it from the ConnectorManager.
- Refactored JavaInstanceRunnable to take in classloader for the user function instead of jar paths.
- Change the RuntimeFactory interface to allow ConnectorsManager to be passed in so that the ThreadRuntimeFactory can 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nlu90 commented on a change in pull request #9500: Optimize built-in source/sink startup Part 2
Posted by GitBox <gi...@apache.org>.
nlu90 commented on a change in pull request #9500:
URL: https://github.com/apache/pulsar/pull/9500#discussion_r572336779
##########
File path: pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java
##########
@@ -84,34 +95,82 @@
this.secretsProvider = secretsProvider;
this.collectorRegistry = collectorRegistry;
this.narExtractionDirectory = narExtractionDirectory;
- this.javaInstanceRunnable = new JavaInstanceRunnable(
- instanceConfig,
- fnCache,
- jarFile,
- pulsarClient,
- pulsarAdmin,
- stateStorageServiceUrl,
- secretsProvider,
- collectorRegistry,
- narExtractionDirectory);
+ this.connectorsManager = connectorsManager;
+ }
+
+ private static ClassLoader getFunctionClassLoader(InstanceConfig instanceConfig,
+ String jarFile,
+ String narExtractionDirectory,
+ FunctionCacheManager fnCache,
+ Optional<ConnectorsManager> connectorsManager) throws Exception {
+
+ if (FunctionCommon.isFunctionCodeBuiltin(instanceConfig.getFunctionDetails())
+ && connectorsManager.isPresent()) {
+ switch (InstanceUtils.calculateSubjectType(instanceConfig.getFunctionDetails())) {
+ case SOURCE:
+ return connectorsManager.get().getConnector(
+ instanceConfig.getFunctionDetails().getSource().getBuiltin()).getClassLoader();
+ case SINK:
+ return connectorsManager.get().getConnector(
+ instanceConfig.getFunctionDetails().getSink().getBuiltin()).getClassLoader();
+ default:
+ return loadJars(jarFile, instanceConfig, narExtractionDirectory, fnCache);
+ }
+ } else {
+ return loadJars(jarFile, instanceConfig, narExtractionDirectory, fnCache);
+ }
+ }
+
+ private static ClassLoader loadJars(String jarFile,
+ InstanceConfig instanceConfig,
+ String narExtractionDirectory,
+ FunctionCacheManager fnCache) throws Exception {
+ ClassLoader fnClassLoader;
+ try {
+ log.info("Load JAR: {}", jarFile);
+ // Let's first try to treat it as a nar archive
+ fnCache.registerFunctionInstanceWithArchive(
+ instanceConfig.getFunctionId(),
+ instanceConfig.getInstanceName(),
+ jarFile, narExtractionDirectory);
+ } catch (FileNotFoundException e) {
+ // create the function class loader
+ fnCache.registerFunctionInstance(
+ instanceConfig.getFunctionId(),
+ instanceConfig.getInstanceName(),
+ Arrays.asList(jarFile),
+ Collections.emptyList());
+ }
+
+ log.info("Initialize function class loader for function {} at function cache manager, functionClassLoader: {}",
+ instanceConfig.getFunctionDetails().getName(), fnCache.getClassLoader(instanceConfig.getFunctionId()));
+
+ fnClassLoader = fnCache.getClassLoader(instanceConfig.getFunctionId());
+ if (null == fnClassLoader) {
+ throw new Exception("No function class loader available.");
+ }
Review comment:
should this null check goes before the `log.info` on line 145 to avoid confusion?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #9500: Optimize built-in source/sink startup Part 2
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #9500:
URL: https://github.com/apache/pulsar/pull/9500#discussion_r572351249
##########
File path: pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java
##########
@@ -84,34 +95,82 @@
this.secretsProvider = secretsProvider;
this.collectorRegistry = collectorRegistry;
this.narExtractionDirectory = narExtractionDirectory;
- this.javaInstanceRunnable = new JavaInstanceRunnable(
- instanceConfig,
- fnCache,
- jarFile,
- pulsarClient,
- pulsarAdmin,
- stateStorageServiceUrl,
- secretsProvider,
- collectorRegistry,
- narExtractionDirectory);
+ this.connectorsManager = connectorsManager;
+ }
+
+ private static ClassLoader getFunctionClassLoader(InstanceConfig instanceConfig,
+ String jarFile,
+ String narExtractionDirectory,
+ FunctionCacheManager fnCache,
+ Optional<ConnectorsManager> connectorsManager) throws Exception {
+
+ if (FunctionCommon.isFunctionCodeBuiltin(instanceConfig.getFunctionDetails())
+ && connectorsManager.isPresent()) {
+ switch (InstanceUtils.calculateSubjectType(instanceConfig.getFunctionDetails())) {
+ case SOURCE:
+ return connectorsManager.get().getConnector(
+ instanceConfig.getFunctionDetails().getSource().getBuiltin()).getClassLoader();
+ case SINK:
+ return connectorsManager.get().getConnector(
+ instanceConfig.getFunctionDetails().getSink().getBuiltin()).getClassLoader();
+ default:
+ return loadJars(jarFile, instanceConfig, narExtractionDirectory, fnCache);
+ }
+ } else {
+ return loadJars(jarFile, instanceConfig, narExtractionDirectory, fnCache);
+ }
+ }
+
+ private static ClassLoader loadJars(String jarFile,
+ InstanceConfig instanceConfig,
+ String narExtractionDirectory,
+ FunctionCacheManager fnCache) throws Exception {
+ ClassLoader fnClassLoader;
+ try {
+ log.info("Load JAR: {}", jarFile);
+ // Let's first try to treat it as a nar archive
+ fnCache.registerFunctionInstanceWithArchive(
+ instanceConfig.getFunctionId(),
+ instanceConfig.getInstanceName(),
+ jarFile, narExtractionDirectory);
+ } catch (FileNotFoundException e) {
+ // create the function class loader
+ fnCache.registerFunctionInstance(
+ instanceConfig.getFunctionId(),
+ instanceConfig.getInstanceName(),
+ Arrays.asList(jarFile),
+ Collections.emptyList());
+ }
+
+ log.info("Initialize function class loader for function {} at function cache manager, functionClassLoader: {}",
+ instanceConfig.getFunctionDetails().getName(), fnCache.getClassLoader(instanceConfig.getFunctionId()));
+
+ fnClassLoader = fnCache.getClassLoader(instanceConfig.getFunctionId());
+ if (null == fnClassLoader) {
+ throw new Exception("No function class loader available.");
+ }
Review comment:
That is the existing logic. I rather not change it in this PR. It is kind of out of scope.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #9500: Optimize built-in source/sink startup Part 2
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #9500:
URL: https://github.com/apache/pulsar/pull/9500#issuecomment-774863729
@nlu90 @freeznet Can you review this?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a change in pull request #9500: Optimize built-in source/sink startup Part 2
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #9500:
URL: https://github.com/apache/pulsar/pull/9500#discussion_r571378547
##########
File path: pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java
##########
@@ -145,6 +204,12 @@ public void stop() {
}
// make sure JavaInstanceRunnable is closed
this.javaInstanceRunnable.close();
+
+ // once the thread quits, clean up the instance
+ fnCache.unregisterFunctionInstance(
+ instanceConfig.getFunctionId(),
+ instanceConfig.getInstanceName());
+ log.info("Unloading JAR files for function {}", instanceConfig);
Review comment:
Do we have to move this line one line above?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] david-streamlio commented on pull request #9500: Optimize built-in source/sink startup Part 2
Posted by GitBox <gi...@apache.org>.
david-streamlio commented on pull request #9500:
URL: https://github.com/apache/pulsar/pull/9500#issuecomment-775235503
LGTM +1
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #9500: Optimize built-in source/sink startup Part 2
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #9500:
URL: https://github.com/apache/pulsar/pull/9500#discussion_r571531891
##########
File path: pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/thread/ThreadRuntime.java
##########
@@ -145,6 +204,12 @@ public void stop() {
}
// make sure JavaInstanceRunnable is closed
this.javaInstanceRunnable.close();
+
+ // once the thread quits, clean up the instance
+ fnCache.unregisterFunctionInstance(
+ instanceConfig.getFunctionId(),
+ instanceConfig.getInstanceName());
+ log.info("Unloading JAR files for function {}", instanceConfig);
Review comment:
sure
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng merged pull request #9500: Optimize built-in source/sink startup Part 2
Posted by GitBox <gi...@apache.org>.
jerrypeng merged pull request #9500:
URL: https://github.com/apache/pulsar/pull/9500
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #9500: Optimize built-in source/sink startup Part 2
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #9500:
URL: https://github.com/apache/pulsar/pull/9500#issuecomment-775456712
Great!
I believe that this work will speed up a lot the integration tests
Thanks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org