Greetings Flinkers, I'm trying to build a container for Flink that supports SQL queries on Kafka topics, and Kafka, the jobmanager, taskmanager, sql-client, and sql-gateway all come up and run. but when I create a table thus:

CREATE TABLE FOO (id STRING, name STRING)

WITH (

'connector.type'='kafka',

'connector.version'='universal',

'connector.topic'='foo',

'format.type'='json',

'connector.properties.bootstrap.servers'='kafka:29091',

'connector.properties.group.id'='flink'

);

I get this error when I try to select from the table:

[ERROR] Could not execute SQL statement. Reason:

org.apache.flink.table.factories.NoMatchingTableFactoryException: Could not find a suitable table factory for 'org.apache.flink.table.factories.TableSourceFactory' in

the classpath.

Reason: Required context properties mismatch.

The following properties are requested:

connector.properties.bootstrap.servers=kafka:29091

connector.properties.group.id=flink

connector.topic=foo

connector.type=kafka

connector.version=universal

format.type=json

schema.0.data-type=VARCHAR(<tel:2147483647|2147483647>)

schema.0.name=id

schema.1.data-type=VARCHAR(<tel:2147483647|2147483647>)

schema.1.name=name

The following factories have been considered:

org.apache.flink.table.sources.CsvBatchTableSourceFactory

org.apache.flink.table.sources.CsvAppendTableSourceFactory

I interpreted the error as a symptom of not having the appropriate JAR files in /opt/flink/lib, so I added flink-sql-connector-kafka.jar, flink-connector-kafka.jar, and flink-table-api-java-1.17.0.jar to the image, but the error persists. The container is based off the Flink 1.17.0 container image on Docker Hub.

I guess they changed recently, as this code had worked with 1.12.

Well Flink 1.12 is 3 years old 😅 These properties were removed with Flink 1.14, as was highlighted in the release notes https://nightlies.apache.org/flink/flink-docs-release-1.14/release-notes/flink-1.14/#remove-tableenvironmentconnect Can you post your DDL? It looks like you still might be using incorrect property options

Geez, 2 years old. I am really late to this party!

CREATE TABLE FOO (id STRING, name STRING)
WITH (
  'connector'='kafka', 
  'topic'='foo',
  'format'='json',        'properties.bootstrap.servers'='kafka:29091',
    'properties.group.id'='flink',
    'scan.startup.mode'='earliest-offset'
);

Just wondering what does

'properties.group.id'='flink'

is used for when flink maintains the checkpoint

That's documented at https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/kafka/#consumer-offset-committing

Committing offset is only for exposing the progress of consumer and consuming group for monitoring.

Thanks for the response Martijn..! For example if we have table T1, T2 and T3 and all have consumer group name same as table name Now i have flink-sql job that performs

insert into table3 select .. from T1 inner join T2

semantic In this case, does t3 becomes the consumer group and kafka would display each table how far is consumer group T3 source operator reading from T1 and T2 kafka topics ??

No, because T3 is a sink and not reading from Kafka. T1 and T2 are sources and reading from Kafka, so they would be reported as the consumer groups

hmm..this is interesting so if Job1 performing

insert into table3 select .. from T1 inner join T2

and Job2 performing

insert into T4 select .. from T1 left outer join T2

then at a time T1 will be representing progress of T1 from Job1 or Job2 .. how would it represent both??

You would mess up your own metrics most likely. For Flink, it doesn’t need the consumergroups.