[
https://issues.apache.org/jira/browse/FLINK-15552?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17016775#comment-17016775
]
Leonard Xu commented on FLINK-15552:
------------------------------------
Hi,[~jark] [~Terry1897]
e2e test
[test_sql_client_kafka011.sh|https://github.com/apache/flink/blob/master/flink-end-to-end-tests/test-scripts/test_sql_client_kafka011.sh]
can run normally because it defines Kafka source Table in YAML way as
environment file which will create when SqlClient start , and it will use
FlinkUserCodeClassLoaders#ParentFirstClassLoader, so TableFactoryService#find()
can find correct KafkaTableSource class from user defined jars(like
flink-sql-connector-kafka-0.11_2.11-1.10-SNAPSHOT.jar) :
{code:java}
//ExecutionContext.java
// create class loader and load user defined jars
classLoader = FlinkUserCodeClassLoaders.parentFirst(
dependencies.toArray(new URL[dependencies.size()]),
this.getClass().getClassLoader());
//create table soure
environment.getTables().forEach((name, entry) -> {
if (entry instanceof SourceTableEntry || entry instanceof
SourceSinkTableEntry) {
tableSources.put(name, createTableSource(environment.getExecution(),
entry.asMap(), classLoader));
}
if (entry instanceof SinkTableEntry || entry instanceof
SourceSinkTableEntry) {
tableSinks.put(name, createTableSink(environment.getExecution(),
entry.asMap(), classLoader));
}
});
});
// register table sources
tableSources.forEach(tableEnv::registerTableSource);
....
// use FlinkUserCodeClassLoaders#ParentFirstClassLoader here
private static TableSource<?> createTableSource(ExecutionEntry execution,
Map<String, String> sourceProperties, ClassLoader classLoader) {
if (execution.isStreamingPlanner()) {
final TableSourceFactory<?> factory = (TableSourceFactory<?>)
TableFactoryService.find(TableSourceFactory.class, sourceProperties,
classLoader);
return factory.createTableSource(sourceProperties);
} else if (execution.isBatchPlanner()) {
final BatchTableSourceFactory<?> factory = (BatchTableSourceFactory<?>)
TableFactoryService.find(BatchTableSourceFactory.class,
sourceProperties, classLoader);
return factory.createBatchTableSource(sourceProperties);
}
throw new SqlExecutionException("Unsupported execution type for sources.");
}
{code}
when we define a Kafka source/sink table in cli way, DDL sql will translate to
tEnv.sqlUpdate(ddl), this will not throw exception right now because
tEnv.sqlUpdate() is executing lazily. when we input `select * from
kafkaTable`, tEnv will create kafkaTable and execute the query by:
{code:java}
//LocalExecutor.java
// create table
final Table table = createTable(context, context.getTableEnvironment(), query);
...
/**
* Creates a table using the given query in the given table environment.
*/
private <C> Table createTable(ExecutionContext<C> context, TableEnvironment
tableEnv, String selectQuery) {
// parse and validate query
try {
return context.wrapClassLoader(() -> tableEnv.sqlQuery(selectQuery));
} catch (Throwable t) {
// catch everything such that the query does not crash the executor
throw new SqlExecutionException("Invalid SQL statement.", t);
}
}
...
//DatabaseCalciteSchema.java
private Table convertCatalogTable(ObjectPath tablePath, CatalogTable table,
@Nullable TableFactory tableFactory) {
final TableSource<?> tableSource;
if (tableFactory != null) {
if (tableFactory instanceof TableSourceFactory) {
tableSource = ((TableSourceFactory)
tableFactory).createTableSource(tablePath, table);
} else {
throw new TableException(
"Cannot query a sink-only table. TableFactory provided by catalog
must implement TableSourceFactory");
}
} else {
tableSource = TableFactoryUtil.findAndCreateTableSource(table);
}{code}
there are no `classLoader` parameter in
TableFactoryUtil#findAndCreateTableSource(table) , and TableFactoryUtil will
use its default classLoader which can not load KafkaTableSource class from use
defined jars(like flink-sql-connector-kafka-0.11_2.11-1.10-SNAPSHOT.jar).
Do you have any suggestion to fix this ?
> SQL Client can not correctly create kafka table using --library to indicate a
> kafka connector directory
> -------------------------------------------------------------------------------------------------------
>
> Key: FLINK-15552
> URL: https://issues.apache.org/jira/browse/FLINK-15552
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Client, Table SQL / Runtime
> Reporter: Terry Wang
> Priority: Major
> Fix For: 1.10.0
>
>
> How to Reproduce:
> first, I start a sql client and using `-l` to point to a kafka connector
> directory.
> `
> bin/sql-client.sh embedded -l /xx/connectors/kafka/
> `
> Then, I create a Kafka Table like following
> `
> Flink SQL> CREATE TABLE MyUserTable (
> > content String
> > ) WITH (
> > 'connector.type' = 'kafka',
> > 'connector.version' = 'universal',
> > 'connector.topic' = 'test',
> > 'connector.properties.zookeeper.connect' = 'localhost:2181',
> > 'connector.properties.bootstrap.servers' = 'localhost:9092',
> > 'connector.properties.group.id' = 'testGroup',
> > 'connector.startup-mode' = 'earliest-offset',
> > 'format.type' = 'csv'
> > );
> [INFO] Table has been created.
> `
> Then I select from just created table and an exception been thrown:
> `
> Flink SQL> select * from MyUserTable;
> [ERROR] Could not execute SQL statement. Reason:
> org.apache.flink.table.api.NoMatchingTableFactoryException: Could not find a
> suitable table factory for
> 'org.apache.flink.table.factories.TableSourceFactory' in
> the classpath.
> Reason: Required context properties mismatch.
> The matching candidates:
> org.apache.flink.table.sources.CsvBatchTableSourceFactory
> Mismatched properties:
> 'connector.type' expects 'filesystem', but is 'kafka'
> The following properties are requested:
> connector.properties.bootstrap.servers=localhost:9092
> connector.properties.group.id=testGroup
> connector.properties.zookeeper.connect=localhost:2181
> connector.startup-mode=earliest-offset
> connector.topic=test
> connector.type=kafka
> connector.version=universal
> format.type=csv
> schema.0.data-type=VARCHAR(2147483647)
> schema.0.name=content
> The following factories have been considered:
> org.apache.flink.table.sources.CsvBatchTableSourceFactory
> org.apache.flink.table.sources.CsvAppendTableSourceFactory
> `
> Potential Reasons:
> Now we use `TableFactoryUtil#findAndCreateTableSource` to convert a
> CatalogTable to TableSource, but when call `TableFactoryService.find` we
> don't pass current classLoader to this method, the default loader will be
> BootStrapClassLoader, which can not find our factory.
> I verified in my box, it's truly caused by this behavior.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
