Github user arina-ielchiieva commented on a diff in the pull request:
https://github.com/apache/drill/pull/701#discussion_r102920190
--- Diff:
exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionImplementationRegistry.java
---
@@ -260,76 +293,101 @@ public RemoteFunctionRegistry
getRemoteFunctionRegistry() {
}
/**
- * Attempts to load and register functions from remote function registry.
- * First checks if there is no missing jars.
- * If yes, enters synchronized block to prevent other loading the same
jars.
- * Again re-checks if there are no missing jars in case someone has
already loaded them (double-check lock).
- * If there are still missing jars, first copies jars to local udf area
and prepares {@link JarScan} for each jar.
- * Jar registration timestamp represented in milliseconds is used as
suffix.
- * Then registers all jars at the same time. Returns true when finished.
- * In case if any errors during jars coping or registration, logs errors
and proceeds.
+ * Purpose of this method is to synchronize remote and local function
registries if needed
+ * and to inform if function registry was changed after given version.
*
- * If no missing jars are found, checks current local registry version.
- * Returns false if versions match, true otherwise.
+ * To make synchronization as much light-weigh as possible, first only
versions of both registries are checked
+ * without any locking. If synchronization is needed, enters
synchronized block to prevent others loading the same jars.
+ * The need of synchronization is checked again (double-check lock)
before comparing jars.
+ * If any missing jars are found, they are downloaded to local udf area,
each is wrapped into {@link JarScan}.
+ * Once jar download is finished, all missing jars are registered in one
batch.
+ * In case if any errors during jars download / registration, these
errors are logged.
*
- * @param version local function registry version
- * @return true if new jars were registered or local function registry
version is different, false otherwise
+ * During registration local function registry is updated with remote
function registry version it is synced with.
+ * When at least one jar of the missing jars failed to download /
register,
+ * local function registry version are not updated but jars that where
successfully downloaded / registered
+ * are added to local function registry.
+ *
+ * If synchronization between remote and local function registry was not
needed,
+ * checks if given registry version matches latest sync version
+ * to inform if function registry was changed after given version.
+ *
+ * @param version remote function registry local function registry was
based on
+ * @return true if remote and local function registries were
synchronized after given version
*/
- public boolean loadRemoteFunctions(long version) {
- List<String> missingJars = getMissingJars(remoteFunctionRegistry,
localFunctionRegistry);
- if (!missingJars.isEmpty()) {
+ public boolean syncWithRemoteRegistry(long version) {
+ if
(doSyncFunctionRegistries(remoteFunctionRegistry.getRegistryVersion(),
localFunctionRegistry.getVersion())) {
synchronized (this) {
- missingJars = getMissingJars(remoteFunctionRegistry,
localFunctionRegistry);
- if (!missingJars.isEmpty()) {
- logger.info("Starting dynamic UDFs lazy-init process.\n" +
- "The following jars are going to be downloaded and
registered locally: " + missingJars);
+ long localRegistryVersion = localFunctionRegistry.getVersion();
+ if
(doSyncFunctionRegistries(remoteFunctionRegistry.getRegistryVersion(),
localRegistryVersion)) {
+ DataChangeVersion remoteVersion = new DataChangeVersion();
+ List<String> missingJars =
getMissingJars(this.remoteFunctionRegistry, localFunctionRegistry,
remoteVersion);
List<JarScan> jars = Lists.newArrayList();
- for (String jarName : missingJars) {
- Path binary = null;
- Path source = null;
- URLClassLoader classLoader = null;
- try {
- binary = copyJarToLocal(jarName, remoteFunctionRegistry);
- source = copyJarToLocal(JarUtil.getSourceName(jarName),
remoteFunctionRegistry);
- URL[] urls = {binary.toUri().toURL(),
source.toUri().toURL()};
- classLoader = new URLClassLoader(urls);
- ScanResult scanResult = scan(classLoader, binary, urls);
- localFunctionRegistry.validate(jarName, scanResult);
- jars.add(new JarScan(jarName, scanResult, classLoader));
- } catch (Exception e) {
- deleteQuietlyLocalJar(binary);
- deleteQuietlyLocalJar(source);
- if (classLoader != null) {
- try {
- classLoader.close();
- } catch (Exception ex) {
- logger.warn("Problem during closing class loader for
{}", jarName, e);
+ if (!missingJars.isEmpty()) {
+ logger.info("Starting dynamic UDFs lazy-init process.\n" +
+ "The following jars are going to be downloaded and
registered locally: " + missingJars);
+ for (String jarName : missingJars) {
+ Path binary = null;
+ Path source = null;
+ URLClassLoader classLoader = null;
+ try {
+ binary = copyJarToLocal(jarName,
this.remoteFunctionRegistry);
+ source = copyJarToLocal(JarUtil.getSourceName(jarName),
this.remoteFunctionRegistry);
+ URL[] urls = {binary.toUri().toURL(),
source.toUri().toURL()};
+ classLoader = new URLClassLoader(urls);
+ ScanResult scanResult = scan(classLoader, binary, urls);
+ localFunctionRegistry.validate(jarName, scanResult);
+ jars.add(new JarScan(jarName, scanResult, classLoader));
+ } catch (Exception e) {
+ deleteQuietlyLocalJar(binary);
+ deleteQuietlyLocalJar(source);
+ if (classLoader != null) {
+ try {
+ classLoader.close();
+ } catch (Exception ex) {
+ logger.warn("Problem during closing class loader for
{}", jarName, e);
+ }
}
+ logger.error("Problem during remote functions load from
{}", jarName, e);
}
- logger.error("Problem during remote functions load from {}",
jarName, e);
}
}
- if (!jars.isEmpty()) {
- localFunctionRegistry.register(jars);
- return true;
- }
+ long latestRegistryVersion = jars.size() != missingJars.size() ?
+ localRegistryVersion : remoteVersion.getVersion();
+ localFunctionRegistry.register(jars, latestRegistryVersion);
+ return true;
}
}
}
+
return version != localFunctionRegistry.getVersion();
}
/**
- * First finds path to marker file url, otherwise throws {@link
JarValidationException}.
- * Then scans jar classes according to list indicated in marker files.
- * Additional logic is added to close {@link URL} after {@link
ConfigFactory#parseURL(URL)}.
- * This is extremely important for Windows users where system doesn't
allow to delete file if it's being used.
+ * Checks if local function registry should be synchronized with remote
function registry.
+ * If remote function registry version is -1, it means that remote
function registry is unreachable
+ * or is not configured thus we skip synchronization and return false.
+ * In all other cases synchronization is needed if remote and local
function registries versions do not match.
*
- * @param classLoader unique class loader for jar
- * @param path local path to jar
- * @param urls urls associated with the jar (ex: binary and source)
- * @return scan result of packages, classes, annotations found in jar
+ * @param remoteVersion remote function registry version
+ * @param localVersion local function registry version
+ * @return true is local registry should be refreshed, false otherwise
*/
+ private boolean doSyncFunctionRegistries(long remoteVersion, long
localVersion) {
--- End diff --
Agree. Done.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
