Bdtjtk

I'm using flink table api, using kafka as input source and json as table schema. I got this error when submit my program:
```
The program finished with the following exception:

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error.

    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:546)

    at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421)

    at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:426)

    at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:804)

    at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:280)

    at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:215)

    at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1044)

    at org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1120)

    at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30)

    at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1120)

Caused by: org.apache.flink.table.api.NoMatchingTableFactoryException:     



Could not find a suitable table factory for 'org.apache.flink.table.factories.StreamTableSourceFactory' in

the classpath.    



Reason: No context matches.    



The following properties are requested:

connector.properties.0.key=zookeeper.connect

connector.properties.0.value=localhost:2181

connector.properties.1.key=bootstrap.servers

connector.properties.1.value=localhost:9092

connector.property-version=1

connector.startup-mode=latest-offset

connector.topic=flink_table_test

connector.type=kafka

connector.version=0.10

format.fail-on-missing-field=true

format.json-schema={t'type': 'object',t'properties': {tt'uid': {ttt'type': 'number'tt},tt'name': {ttt'type': 'string'tt},tt'age': {ttt'type': 'number'tt},tt'timestamp': {ttt'type': 'long'tt}t}}

format.property-version=1

format.type=json

schema.0.name=uid

schema.0.type=BIGINT

schema.1.name=name

schema.1.type=VARCHAR

schema.2.name=age

schema.2.type=INT

schema.3.name=ts

schema.3.type=TIMESTAMP

update-mode=append    



The following factories have been considered:

org.apache.flink.formats.json.JsonRowFormatFactory

org.apache.flink.table.sources.CsvBatchTableSourceFactory

org.apache.flink.table.sources.CsvAppendTableSourceFactory

org.apache.flink.table.sinks.CsvBatchTableSinkFactory

org.apache.flink.table.sinks.CsvAppendTableSinkFactory    



    at org.apache.flink.table.factories.TableFactoryService$.filterByContext(TableFactoryService.scala:214)

    at org.apache.flink.table.factories.TableFactoryService$.findInternal(TableFactoryService.scala:130)

    at org.apache.flink.table.factories.TableFactoryService$.find(TableFactoryService.scala:81)

    at org.apache.flink.table.factories.TableFactoryUtil$.findAndCreateTableSource(TableFactoryUtil.scala:49)

    at org.apache.flink.table.descriptors.ConnectTableDescriptor.registerTableSource(ConnectTableDescriptor.scala:44)

    at com.akulaku.data.flink.QueryTable$.main(QueryTable.scala:53)

    at com.akulaku.data.flink.QueryTable.main(QueryTable.scala)

    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)

    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

    at java.lang.reflect.Method.invoke(Method.java:498)

    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529)

    ... 9 more

```

I already add the flink-json-1.6.1.jar into flink lib path.

Could anyone helps?

Try adding the kafka connector sql jar (e.g. flink-connector-kafka-0.11_2.11-1.6.1-sql-jar.jar). You can find the whole list of connector dependencies here.