Explicitly set the Pulsar function classloader

[merlimat]

Motivation

When a function is trying to create objects through the client API that involve loading an implementation class, it will fail because it will attempt to use the function's own class loader, and that does not include Pulsar client implementation classes.

Instead, we should be recognizing that we are in "Pulsar Function" mode and directly use the framework class loader for all the implementation classes.

[eolivelli]

Overall I like this approach. Please take a look to my comment

[eolivelli]

Sorry I wasn't clear regarding the system property.
I may be 'pulsar.allow.override.system.classloader', it defaults to false.
If you set it to true then you can set a new value.
The value of the system property should be populated in a static block and put into a private final static method.

BTW lgtm in this form

[eolivelli]

LGTM

[Anonymitaet]

@merlimat thanks for your contribution. For this PR, do we need to update docs?

[eolivelli]

Tests are failing

rror:  testIsAuthorizedRole(org.apache.pulsar.functions.worker.rest.api.FunctionsImplTest)  Time elapsed: 0.014 s  <<< FAILURE!
java.lang.ClassCastException: class org.apache.pulsar.common.policies.data.TenantInfoImpl$TenantInfoImplBuilder cannot be cast to class org.apache.pulsar.common.policies.data.TenantInfo$Builder (org.apache.pulsar.common.policies.data.TenantInfoImpl$TenantInfoImplBuilder is in unnamed module of loader 'app'; org.apache.pulsar.common.policies.data.TenantInfo$Builder is in unnamed module of loader org.powermock.core.classloader.javassist.JavassistMockClassLoader @5f803481)
	at org.apache.pulsar.common.policies.data.TenantInfo.builder(TenantInfo.java:36)
	at org.apache.pulsar.functions.worker.rest.api.FunctionsImplTest.testIsAuthorizedRole(FunctionsImplTest.java:250)
	at org.apache.pulsar.functions.worker.rest.api.FunctionsImplTest_$$_jvstb61_0._d12testIsAuthorizedRole(FunctionsImplTest_$$_jvstb61_0.java)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.powermock.modules.testng.internal.PowerMockTestNGMethodHandler.invoke(PowerMockTestNGMethodHandler.java:50)
	at 

[eolivelli]

@merlimat @jerrypeng

I believe that the better way to deal with the classpath is to implement a simpler ClassLoader hierarchy.
In which the Function is loaded in a classloader (NarClassLoader) that is child of the CL that loads the Pulsar Client Implementation and it follows the standard CL delegation model (parent-first).

This way the function is forced to use the same version of the classes that are running within Pulsar core.

The downside is that if you want to use a library that is already inside Pulsar runtime, like Netty, you have to relocate it inside the function NAR.

When we choose to let GenericRecord.getNativeObject return the "native" object we drew this path.
Otherwise it is not possible for a Function to use the returned object.

for instance in a Sink:

void write(Record record) {
Object nativeAvroRecord = record.getValue().getNativeObject();
org.apache.avro.GenericRecord record = (xxx) nativeAvroRecord;
}

the org.apache.avro.GenericRecord class for "nativeAvroRecord" but be the same class that is loaded by the function to represent "record", otherwise you will see a ClassCastException.

If we implement "parent-first" strategy, when the JMV loads the class for "record" (using the same classloader that loaded the Sink class), it will return the same o.a.avro.GenericRecord class that built the "nativeAvroRecord" object.

I believe that there is no other way to do it.

@lhotari gave me a link to how Gradle solved a similar problem (if want to filter out some classes/resources and keep only a subset)

https://github.com/gradle/gradle/blob/master/subprojects/base-services/src/main/java/org/gradle/internal/classloader/FilteringClassLoader.java

[eolivelli]

why do we need this cache ?
the JVM is very good in dealing with Class.forName.

[eolivelli]

I believe that this will not fix the problems we have.

Therefore I discovered this other bad problem about using AVRO (and JsonNode...)
#11274

basically now (2.8.0) you cannot use the org.apache.avro...GenericRecord returned by GenericRecord.getNativeObject()

[eolivelli]

An alternative approach:
#11636