KafkaSource error

[gsbadhan]

Hi team,

I was doing local PoC for kafka -> flink -> fluss.

I was getting error "java.lang.ClassNotFoundException as : org.apache.flink.connector.kafka.source.KafkaSource" after executing select statement on fresh created table i.e. kafka_customers for kafka events.

You can see the error and Docker files attached.

Error logs:

"Flink SQL>

USE CATALOG default_catalog;
[INFO] Execute statement succeeded.

CREATE TEMPORARY TABLE kafka_customers (
c_custkey INT,
c_name STRING,
c_address STRING,
c_phone STRING,
c_mktsegment STRING,
proc_time AS PROCTIME()
) WITH (
'connector' = 'kafka',
'topic' = 'customers',
'properties.bootstrap.servers' = 'kafka:29092',
'properties.group.id' = 'fluss-consumer-group',
'scan.startup.mode' = 'earliest-offset',
'format' = 'json'
);
[INFO] Execute statement succeeded.

show tables;
+-----------------+
| table name |
+-----------------+
| kafka_customers |
| source_customer |
| source_nation |
| source_order |
+-----------------+
4 rows in set

Flink SQL> select * from kafka_customers;
[ERROR] Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: org.apache.flink.connector.kafka.source.KafkaSource"

PFA docker compose file

docker-compose.yml

Dockerfile text:

FROM apache/fluss-quickstart-flink:1.20-0.9.1-incubating
ADD https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/3.4.0-1.20/flink-sql-connector-kafka-3.4.0-1.20.jar /opt/flink/lib/
ADD https://repo1.maven.org/maven2/org/apache/flink/flink-connector-kafka/3.4.0-1.20/flink-connector-kafka-3.4.0-1.20.jar /opt/flink/lib/
ADD https://repo1.maven.org/maven2/org/apache/flink/flink-connector-base/1.20.0/flink-connector-base-1.20.0.jar /opt/flink/lib/
ADD https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/3.4.0/kafka-clients-3.4.0.jar /opt/flink/lib/

[yudin-s]

This looks like a Flink classpath problem, not a Fluss table problem.

The missing class is org.apache.flink.connector.kafka.source.KafkaSource, which belongs to Flink's Kafka connector. Fluss's Flink docs tell you to put the fluss-flink-<flink-version> connector jar into the Flink lib directory; the Fluss quickstart Docker image also copies its prepared jars into /opt/flink/lib. That makes the Fluss catalog/table connector available, but it does not provide connector = 'kafka'.

The Kafka SQL connector docs are explicit that the Kafka connector is not part of the Flink binary distribution. So for your kafka_customers temporary table, add the matching Flink Kafka SQL connector jar for your Flink version to every Flink container that runs SQL jobs, usually JobManager and TaskManager:

# inside / mounted into each Flink container
ls /opt/flink/lib | grep -i kafka

Then restart the Flink cluster / SQL client so the jar is actually on the classpath, recreate the Kafka table, and run the SELECT again.

So the split should be:

• fluss-flink-1.20-...jar or matching version: required for the Fluss catalog.
• Flink Kafka SQL connector jar: required for connector = 'kafka' and KafkaSource.

If adding the Kafka connector jar fixes the ClassNotFoundException, please mark the answer so the classpath issue is easy to find.

[gsbadhan]

we already put these expected jars in classpath but no luck, same error. as i mentioned above, the jar configs we done in Dockerfile and docker-compose.yml file. below are classpath output:

/$ docker compose exec jobmanager ls -lh /opt/flink/lib/ | grep kafka
-rw------- 1 root root 474K Nov 14 2024 flink-connector-kafka-3.4.0-1.20.jar
-rw------- 1 root root 5.4M Nov 14 2024 flink-sql-connector-kafka-3.4.0-1.20.jar
-rw------- 1 root root 4.9M Jan 31 2023 kafka-clients-3.4.0.jar
$
$
$
$
$ docker compose exec taskmanager ls -lh /opt/flink/lib/ | grep kafka
-rw------- 1 root root 474K Nov 14 2024 flink-connector-kafka-3.4.0-1.20.jar
-rw------- 1 root root 5.4M Nov 14 2024 flink-sql-connector-kafka-3.4.0-1.20.jar
-rw------- 1 root root 4.9M Jan 31 2023 kafka-clients-3.4.0.jar
$
$
$ docker compose exec sql-client ls -lh /opt/flink/lib/ | grep kafka
-rw------- 1 root root 474K Nov 14 2024 flink-connector-kafka-3.4.0-1.20.jar
-rw------- 1 root root 5.4M Nov 14 2024 flink-sql-connector-kafka-3.4.0-1.20.jar
-rw------- 1 root root 4.9M Jan 31 2023 kafka-clients-3.4.0.jar

[yudin-s]

The jar list is useful, and the suspicious part is the mode, not the names: all three Kafka jars are -rw------- root root.

So the next thing I would check is whether the actual Flink JVM user can read them. docker compose exec often runs as root unless told otherwise, while the JobManager/TaskManager process may be running as another user. A jar can appear in /opt/flink/lib and still be invisible to the classloader if the process cannot read it.

Try this in each container, especially jobmanager and taskmanager:

id
ps -ef | grep '[f]link'
test -r /opt/flink/lib/flink-connector-kafka-3.4.0-1.20.jar && echo readable || echo not-readable
jar tf /opt/flink/lib/flink-connector-kafka-3.4.0-1.20.jar | grep 'org/apache/flink/connector/kafka/source/KafkaSource.class'

If the process user is not root, rebuild with readable ownership/permissions, for example after the ADD lines:

RUN chmod 0644 /opt/flink/lib/*kafka*.jar /opt/flink/lib/flink-connector-base-1.20.0.jar
# or COPY/ADD them with ownership matching the Flink runtime user

Then recreate the containers, not just the SQL table, so the JobManager/TaskManager JVMs restart with the readable jars.

If jar tf shows KafkaSource.class and the runtime user can read it in every Flink container, the next suspect is that one service is still running an older image/layer, so compare the image IDs for jobmanager, taskmanager, and sql-client after rebuild.

[gsbadhan]

I resolved this error by doing these steps:
`
docker compose exec sql-client /opt/sql-client/sql-client

#execute on flink sql console
ADD JAR '/opt/flink/lib/flink-connector-kafka-3.4.0-1.20.jar';

ADD JAR '/opt/flink/lib/flink-sql-connector-kafka-3.4.0-1.20.jar';

ADD JAR '/opt/flink/lib/kafka-clients-3.4.0.jar';
`

[yudin-s]

Great, that confirms the missing-class failure was still at the Flink connector classpath/registration layer.

For the immediate SQL-client session, ADD JAR ... is a fine workaround because it registers the connector jars for that session. For a repeatable Docker deployment, I would still make the image/container path durable: the same Kafka connector jars need to be readable by the Flink processes and available wherever the job is submitted/executed, otherwise a new SQL client/session or different submit path can hit the same KafkaSource class error again.

If that classpath diagnosis solved the issue, please mark the answer as accepted so the thread is easier to find.

[gsbadhan]

As commented above, the workaround is working for me, so I am closing this issue.

#4039 (reply in thread)