[
https://issues.apache.org/jira/browse/FLINK-22632?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Zhong Zhuang updated FLINK-22632:
---------------------------------
Description:
Hello, community.
I'm using Hive catalog, I'm trying to load UDF through HTTP.
I registered a UDF, then I submited a FlinkSQL job with command line:
{code:java}
flink run ... -C "http://someHost/plusTwoFunc.jar"; ...{code}
When the job started, it throws an Exception:(:
{code:java}
Caused by: java.lang.ClassNotFoundException: com.slankka.flink.udf.PlusTwoFunc
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:264)
at
org.apache.flink.table.catalog.CatalogFunctionImpl.isGeneric(CatalogFunctionImpl.java:72)
{code}
I know that user code is loaded by {{SafetyNetWrapperClassLoader}} . and
{{FlinkUserCodeClassLoader}} , {{SafetyNetWrapperClassLoader}} is a facade
classloader, {{FlinkUserCodeClassLoader}} is the real classloader whose
classpath contains the jar dependency, but it failed to load udf in
{{CatalogFunctionImpl}}
{code:java}
@Override
public boolean isGeneric() {
if (functionLanguage == FunctionLanguage.PYTHON) {
return true;
}
try {
Class c = Class.forName(className); //ClassNotFound
if (UserDefinedFunction.class.isAssignableFrom(c)) {
return true;
}
} catch (ClassNotFoundException e) {
throw new RuntimeException(String.format("Can't resolve udf class %s",
className), e);
}
return false;
}
{code}
When I change to:
{code:java}
Thread.currentThread().getContextClassLoader().loadClass(className);
{code}
Then it works fine.
My proposal is CatalogFunctionImpl.isGeneric should using ContextClassloader,
that is {{FlinkUserCodeClassLoader}} proxied by {{SafetyNetWrapperClassLoader}}
was:
Hello, community.
I'm using Hive catalog, I'm trying to load UDF through HTTP.
I registered a UDF, then I submited a FlinkSQL job with command line:
{code:java}
flink run ... -S "http://someHost/plusTwoFunc.jar"; ...{code}
When the job started, it throws an Exception:(:
{code:java}
Caused by: java.lang.ClassNotFoundException: com.slankka.flink.udf.PlusTwoFunc
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:264)
at
org.apache.flink.table.catalog.CatalogFunctionImpl.isGeneric(CatalogFunctionImpl.java:72)
{code}
I know that user code is loaded by {{SafetyNetWrapperClassLoader}} . and
{{FlinkUserCodeClassLoader}} , {{SafetyNetWrapperClassLoader}} is a facade
classloader, {{FlinkUserCodeClassLoader}} is the real classloader whose
classpath contains the jar dependency, but it failed to load udf in
{{CatalogFunctionImpl}}
{code:java}
@Override
public boolean isGeneric() {
if (functionLanguage == FunctionLanguage.PYTHON) {
return true;
}
try {
Class c = Class.forName(className); //ClassNotFound
if (UserDefinedFunction.class.isAssignableFrom(c)) {
return true;
}
} catch (ClassNotFoundException e) {
throw new RuntimeException(String.format("Can't resolve udf class %s",
className), e);
}
return false;
}
{code}
When I change to:
{code:java}
Thread.currentThread().getContextClassLoader().loadClass(className);
{code}
Then it works fine.
My proposal is CatalogFunctionImpl.isGeneric should using ContextClassloader,
that is {{FlinkUserCodeClassLoader}} proxied by {{SafetyNetWrapperClassLoader}}
> CatalogFunctionImpl.isGeneric should use ContextClassLoader
> -----------------------------------------------------------
>
> Key: FLINK-22632
> URL: https://issues.apache.org/jira/browse/FLINK-22632
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
> Affects Versions: 1.12.2, 1.12.3
> Reporter: Zhong Zhuang
> Priority: Major
> Labels: easyfix
> Fix For: 1.13.0
>
> Original Estimate: 0.5h
> Remaining Estimate: 0.5h
>
> Hello, community.
> I'm using Hive catalog, I'm trying to load UDF through HTTP.
> I registered a UDF, then I submited a FlinkSQL job with command line:
> {code:java}
> flink run ... -C "http://someHost/plusTwoFunc.jar"; ...{code}
> When the job started, it throws an Exception:(:
> {code:java}
> Caused by: java.lang.ClassNotFoundException: com.slankka.flink.udf.PlusTwoFunc
> at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Native Method)
> at java.lang.Class.forName(Class.java:264)
> at
> org.apache.flink.table.catalog.CatalogFunctionImpl.isGeneric(CatalogFunctionImpl.java:72)
> {code}
> I know that user code is loaded by {{SafetyNetWrapperClassLoader}} . and
> {{FlinkUserCodeClassLoader}} , {{SafetyNetWrapperClassLoader}} is a facade
> classloader, {{FlinkUserCodeClassLoader}} is the real classloader whose
> classpath contains the jar dependency, but it failed to load udf in
> {{CatalogFunctionImpl}}
> {code:java}
> @Override
> public boolean isGeneric() {
> if (functionLanguage == FunctionLanguage.PYTHON) {
> return true;
> }
> try {
> Class c = Class.forName(className); //ClassNotFound
> if (UserDefinedFunction.class.isAssignableFrom(c)) {
> return true;
> }
> } catch (ClassNotFoundException e) {
> throw new RuntimeException(String.format("Can't resolve udf class
> %s", className), e);
> }
> return false;
> }
> {code}
> When I change to:
> {code:java}
> Thread.currentThread().getContextClassLoader().loadClass(className);
> {code}
> Then it works fine.
> My proposal is CatalogFunctionImpl.isGeneric should using ContextClassloader,
> that is {{FlinkUserCodeClassLoader}} proxied by
> {{SafetyNetWrapperClassLoader}}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
