[jira] [Created] (FLINK-31243) KryoSerializer when loaded from user code classloader cannot load Scala extensions from app classloader

["Amit Gurdasani (Jira)" <ji...@apache.org>]

Amit Gurdasani created FLINK-31243:
--------------------------------------

             Summary: KryoSerializer when loaded from user code classloader cannot load Scala extensions from app classloader
                 Key: FLINK-31243
                 URL: https://issues.apache.org/jira/browse/FLINK-31243
             Project: Flink
          Issue Type: Bug
          Components: API / Core
    Affects Versions: 1.16.1, 1.15.3
         Environment: OS: Amazon Linux 2

JVM: Amazon Corretto 11

 
            Reporter: Amit Gurdasani


The [KryoSerializer|https://github.com/apache/flink/blob/9bf0d9f2c2bcb2bc0c8ab6228bb0a9e76e10ad70/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java] uses Class.forName() to dynamically load Scala extensions by name. This seems to imply that it references only its own classloader to find these extensions. By default, as the application classloader is favored for KryoSerializer, this implies that unless the flink-scala artifact is available to the application classloader, the Scala extensions cannot be loaded. Scala applications that include flink-scala are therefore unable to benefit from the Scala extensions to the Kryo Serializer.

Exception looks like this:
{noformat}
java.lang.ClassNotFoundException: org.apache.flink.runtime.types.FlinkScalaKryoInstantiator
    at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:581)
    at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:522)
    at java.base/java.lang.Class.forName0(Native Method)
    at java.base/java.lang.Class.forName(Class.java:315)
    at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.getKryoInstance(KryoSerializer.java:486)
    at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.checkKryoInitialized(KryoSerializer.java:521)
    at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.getKryo(KryoSerializer.java:720)
    at software.amazon.kinesisanalytics.kryotest.Main.main(Main.java:16)
    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.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
    at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
    at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
    at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:84)
    at org.apache.flink.client.deployment.application.DetachedApplicationRunner.run(DetachedApplicationRunner.java:70)
    at org.apache.flink.runtime.webmonitor.handlers.JarRunOverrideHandler.lambda$handleRequest$3(JarRunOverrideHandler.java:239)
    at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829){noformat}
Example code resulting in this issue:

Main class for Flink application:
{noformat}
package software.amazon.kinesisanalytics.kryotest;

import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.Serializable;

public class Main {
    private static class Something implements Serializable {
        public static long serialVersionUID = 289034745902347830L;
    }

    public static void main(String... args) {
        StreamExecutionEnvironment executionEnvironment = new StreamExecutionEnvironment();
        KryoSerializer<Something> serializer = new KryoSerializer<>(Something.class, executionEnvironment.getConfig());
        serializer.getKryo();
    }
}
{noformat}
build.gradle for Flink application:
{code:java}
plugins {
    id 'application'
    id 'java'
    id 'com.github.johnrengelman.shadow' version '7.1.2'
}

group 'software.amazon.kinesisanalytics'
version '0.1'

repositories {
    mavenCentral()
}

dependencies {
    compileOnly 'org.apache.flink:flink-core:1.15.2'
    compileOnly 'org.apache.flink:flink-streaming-java:1.15.2'
    implementation 'org.apache.flink:flink-scala_2.12:1.15.2'
}

shadowJar {
    dependencies {
        exclude(dependency('com.esotericsoftware.kryo:.*:.*'))
        exclude(dependency('com.esotericsoftware.minlog:.*:.*'))
        exclude(dependency('com.twitter:.*:.*'))
        exclude(dependency('org.apache.flink:flink-core:.*'))
        exclude(dependency('org.apache.flink:flink-streaming-java:.*'))
        exclude(dependency('org.scala-lang:.*:.*'))
    }
}

mainClassName = 'software.amazon.kinesisanalytics.kryotest.Main'
 {code}
Note that the application jar does not include Kryo itself, nor flink-core, but does include flink-scala.

Placing flink-scala in the application classpath eliminates the error, but as I understand it, the [point of eliminating Scala|https://flink.apache.org/2022/02/22/scala-free-in-one-fifteen/] from the Flink application classloader was to allow the only Scala dependencies to be loaded by the user code classloader. This issue prevents that from being achieved for the Scala extensions to the Kryo Serializer.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)