ClassLoader aware DSBulk

[aymkhalil]

Description
DSBulk TypeSafe configs (namely dsbulk-reference.conf & driver-reference.conf and ServiceLoader dependent loaders use the Thread.currentThread().getContextClassLoader() by default (ex1, ex2). However, sometimes it is desirable to use a different ClassLoader (i.e. the ClassLoader from which the DSBulk dependency code is loading, which may or may not be the same as the context class loader).

Scenario
I have I use case where I wanted to extend another CLI runtime with DSBulk capabilities by using NAR archives which are used for class path isolation (this is a real uses case for pulsar admin extensions, here is the enry point for loading the NAR archive for reference).

DSbulk is used via it BulkLoader API:

// assume getClass().getClassLoader != Thread.currentThread().getContextClassLoader()
BulkLoader loader =  new BulkLoader(args);
loader.run();

At runtime the Thread.currentThread().getContextClassLoader() will resolve to an AppClassLoader that is not aware of the DSBulk dependency and hence confs/service will not be loaded. Here is a sample error from the config use case:

com.typesafe.config.ConfigException$Missing: dsbulk-reference: No configuration setting found for key 'dsbulk'   

For workflows, I've run into un issue where the UnloadWorkflow was not loaded although it is available on the NAR class path.

Proposal
Update BulkLoader to accept a class loader programatailly, and propagate that to the TypeSafe & ServiceLoader APIs. For example:

  public ClassLoaderAwareBulkLoader(ClassLoader configClassLoader, String... args) {
        super(args);
        this.args = args;
        this.configClassLoader = configClassLoader;
    }
    
   @NonNull
    public ExitStatus run() {
 
            // pass class loader down to configs
            ClassLoaderAwareCommandLineParser parser = new ClassLoaderAwareCommandLineParser(configClassLoader, this.args);
            ParsedCommandLine result = parser.parse();
            Config config = result.getConfig();
            // pass class loader down to service loaders
            Workflow workflow = result.getWorkflowProvider().newWorkflow(configClassLoader, config);
            ....
    }

And then TypeSafe configs can be called like:

public static Config createReferenceConfig(ClassLoader classLoader) {
        return ConfigFactory.parseResourcesAnySyntax("dsbulk-reference",
                ConfigParseOptions.defaults().setClassLoader(classLoader))
                .withFallback(ConfigFactory.parseResourcesAnySyntax("driver-reference",
                        ConfigParseOptions.defaults().setClassLoader(classLoader)))
                .withFallback(ConfigFactory.defaultReferenceUnresolved());
    }

and service loaders:

ServiceLoader<WorkflowProvider> loader = ServiceLoader.load(WorkflowProvider.class, configClassLoader);

┆Issue is synchronized with this Jira Task by Unito