[
https://issues.apache.org/jira/browse/FLINK-15992?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Victor Wong updated FLINK-15992:
--------------------------------
Description:
*Background*
As a streaming service maintainer in our company, to ensure our users depend on
the correct version of Kafka and flink-kafka, we add "flink-connector-kafka"
into "fink-dist/lib" directory.
*Problem*
When submitting flink-sql jobs, we encountered below exceptions:
{code:java}
Caused by: org.apache.flink.table.api.NoMatchingTableFactoryException: Could
not find a suitable table factory for
'org.apache.flink.table.factories.DeserializationSchemaFactory' in
the classpath.
{code}
But we have add "org.apache.flink.formats.json.JsonRowFormatFactory" in
"META-INF/services/org.apache.flink.table.factories.TableFactory", which
implements DeserializationSchemaFactory.
*Debug*
We find that it was caused by this:
{code:java}
//
org.apache.flink.streaming.connectors.kafka.KafkaTableSourceSinkFactoryBase#getSerializationSchema
final SerializationSchemaFactory<Row> formatFactory =
TableFactoryService.find(
SerializationSchemaFactory.class,
properties,
this.getClass().getClassLoader());
{code}
It uses `this.getClass().getClassLoader()`, which will be BootStrapClassLoader
of flink.
I think we could replace it with
`Thread.currentThread().getContextClassLoader()` to solve this.
There is a related issue: https://issues.apache.org/jira/browse/FLINK-15552
was:
*Background*
As a streaming service maintainer in our company, to ensure our users depend on
the correct version of Kafka and flink-kafka, we add "flink-connector-kafka"
into "fink-dist/lib" directory.
*Problem*
When submitting flink-sql jobs, we encountered below exceptions:
{code:java}
Caused by: org.apache.flink.table.api.NoMatchingTableFactoryException: Could
not find a suitable table factory for
'org.apache.flink.table.factories.DeserializationSchemaFactory' in
the classpath.
{code}
*Debug*
We find that it was caused by this:
{code:java}
//
org.apache.flink.streaming.connectors.kafka.KafkaTableSourceSinkFactoryBase#getSerializationSchema
final SerializationSchemaFactory<Row> formatFactory =
TableFactoryService.find(
SerializationSchemaFactory.class,
properties,
this.getClass().getClassLoader());
{code}
It uses `this.getClass().getClassLoader()`, which will be BootStrapClassLoader
of fink.
We could replace it with `Thread.currentThread().getContextClassLoader()` to
solve this.
There is a related issue: https://issues.apache.org/jira/browse/FLINK-15552
> Incorrect classloader when finding TableFactory
> -----------------------------------------------
>
> Key: FLINK-15992
> URL: https://issues.apache.org/jira/browse/FLINK-15992
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Reporter: Victor Wong
> Priority: Major
>
> *Background*
> As a streaming service maintainer in our company, to ensure our users depend
> on the correct version of Kafka and flink-kafka, we add
> "flink-connector-kafka" into "fink-dist/lib" directory.
> *Problem*
> When submitting flink-sql jobs, we encountered below exceptions:
> {code:java}
> Caused by: org.apache.flink.table.api.NoMatchingTableFactoryException: Could
> not find a suitable table factory for
> 'org.apache.flink.table.factories.DeserializationSchemaFactory' in
> the classpath.
> {code}
> But we have add "org.apache.flink.formats.json.JsonRowFormatFactory" in
> "META-INF/services/org.apache.flink.table.factories.TableFactory", which
> implements DeserializationSchemaFactory.
> *Debug*
> We find that it was caused by this:
> {code:java}
> //
> org.apache.flink.streaming.connectors.kafka.KafkaTableSourceSinkFactoryBase#getSerializationSchema
> final SerializationSchemaFactory<Row> formatFactory =
> TableFactoryService.find(
> SerializationSchemaFactory.class,
> properties,
> this.getClass().getClassLoader());
> {code}
> It uses `this.getClass().getClassLoader()`, which will be
> BootStrapClassLoader of flink.
> I think we could replace it with
> `Thread.currentThread().getContextClassLoader()` to solve this.
> There is a related issue: https://issues.apache.org/jira/browse/FLINK-15552
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
