[FLINK-15992] Incorrect classloader when finding TableFactory

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Abandoned
• Component/s: Connectors / Kafka, Table SQL / API, Tests
• Labels: auto-deprioritized-major auto-unassigned pull-request-available
• Assignee: Unassigned
• Reporter: jiasheng55
• Created: 11/Feb/20 10:15
• Updated: 29/Sep/21 02:05
• Resolved: 29/Sep/21 02:05

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:

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.

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:

// org.apache.flink.streaming.connectors.kafka.KafkaTableSourceSinkFactoryBase#getSerializationSchema

		final SerializationSchemaFactory<Row> formatFactory = TableFactoryService.find(
			SerializationSchemaFactory.class,
			properties,
			this.getClass().getClassLoader());

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

Issue Links

• is related to: FLINK-15552 parameters --library and --jar doesn't work for DDL in sqlClient
• relates to: FLINK-15635 Allow passing a ClassLoader to EnvironmentSettings
• links to: GitHub Pull Request #11064