> ## Content Index > Fetch the complete content index at: https://www.ggorantala.dev/llms.txt > Use this file to discover other available public pages before exploring further. # Flink: Could Not Find Any Factory for Identifier 'kafka' - URL: https://www.ggorantala.dev/flink-could-not-find-any-factory-for-identifier-kafka/ - Published: 2026-09-01T19:33:39.000Z - Updated: 2026-09-01T19:33:39.000Z - Description: Flink throws ValidationException: Could not find any factory for identifier 'kafka' in the classpath. Here is why SPI discovery fails and how to fix it. - Author: Gopi Gorantala - Tags: apache-flink, flink-errors, ValidationException, flink-sql, flink-kafka-connector, flink-classloading, stream-processing, Java ## 1\. The Error You run a Flink SQL job that reads from Kafka. The `CREATE TABLE` succeeds. The `SELECT` blows up: ```log org.apache.flink.table.api.ValidationException: Unable to create a source for reading table 'default_catalog.default_database.payments'. Table options are: 'connector'='kafka' 'format'='json' 'properties.bootstrap.servers'='kafka:9092' 'topic'='payments' at org.apache.flink.table.factories.FactoryUtil.createDynamicTableSource(FactoryUtil.java:166) at org.apache.flink.table.factories.FactoryUtil.createDynamicTableSource(FactoryUtil.java:186) ... Caused by: org.apache.flink.table.api.ValidationException: Cannot discover a connector using option: 'connector'='kafka' at org.apache.flink.table.factories.FactoryUtil.enrichNoMatchingConnectorError(FactoryUtil.java:798) at org.apache.flink.table.factories.FactoryUtil.discoverTableFactory(FactoryUtil.java:772) ... 3 more Caused by: org.apache.flink.table.api.ValidationException: Could not find any factory for identifier 'kafka' that implements 'org.apache.flink.table.factories.DynamicTableFactory' in the classpath. Available factory identifiers are: blackhole datagen filesystem print at org.apache.flink.table.factories.FactoryUtil.discoverFactory(FactoryUtil.java:545) at org.apache.flink.table.factories.FactoryUtil.discoverTableFactory(FactoryUtil.java:768) ... 4 more ``` **That last block is the whole diagnosis.** `blackhole`, `datagen`, `filesystem`, `print` are the four factories bundled in `flink-dist`. If that is the entire list, the Kafka connector is not on the classpath the planner is looking at — full stop. Older Flink versions (1.11–1.13) say `DynamicTableSourceFactory` instead of `DynamicTableFactory`. Same cause. Flink CDC has its own variant: `RuntimeException: Cannot find factory with identifier "mysql" in the classpath` from `FactoryDiscoveryUtils` — same SPI mechanism, different exception class. **Applies to:** Flink 1.14 through 2.x, all deployment modes (standalone, YARN, Kubernetes application mode, Flink Kubernetes Operator, SQL Client, embedded Table API in an IDE). This is a classpath problem, not a runtime or cluster problem. ## 2\. How to Reproduce It Docker Compose, Flink 1.20, Kafka in KRaft mode: ```yaml services: kafka: image: apache/kafka:3.8.0 hostname: kafka environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_LISTENERS: PLAINTEXT://kafka:9092,CONTROLLER://kafka:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092 KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 jobmanager: image: flink:1.20-scala_2.12-java17 ports: ["8081:8081"] command: jobmanager environment: FLINK_PROPERTIES: "jobmanager.rpc.address: jobmanager" taskmanager: image: flink:1.20-scala_2.12-java17 depends_on: [jobmanager] command: taskmanager environment: FLINK_PROPERTIES: | jobmanager.rpc.address: jobmanager taskmanager.numberOfTaskSlots: 4 ``` Bring it up and open the SQL Client: ```bash docker compose up -d docker compose exec jobmanager ./bin/sql-client.sh ``` ```sql CREATE TABLE payments ( payment_id STRING, amount DECIMAL(18, 2), event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'payments', 'properties.bootstrap.servers' = 'kafka:9092', 'properties.group.id' = 'flink-poc', 'scan.startup.mode' = 'earliest-offset', 'format' = 'json' ); SELECT * FROM payments; ``` The `CREATE TABLE` returns `[INFO] Execute statement succeeded.` — Flink only stores the options in the catalog and does not validate the connector. The `SELECT` fails with the exception above. Before touching anything, confirm what the planner can actually see: ```sql SHOW CREATE TABLE payments; -- options are stored fine ``` ```bash # ground truth: is the connector jar in lib/ at all? docker compose exec jobmanager ls -la /opt/flink/lib/ # for a fat jar, does the service registration survive? unzip -p target/my-job.jar META-INF/services/org.apache.flink.table.factories.Factory ``` If that last command prints nothing or prints only your own factories, you have found the bug — see §4. ## 3\. Why It Happens — Surface Level Flink does not hardcode `'connector' = 'kafka'` anywhere. Connectors are plugins, discovered at runtime through Java's `ServiceLoader` SPI. Something has to publish a `Factory` implementation whose `factoryIdentifier()` returns `"kafka"`, and that publication lives in a text file inside the jar: `META-INF/services/org.apache.flink.table.factories.Factory`. Four things break it, in descending order of frequency: 1. **The connector jar is not on the classpath at all.** The Kafka connector has never shipped in `flink-dist`. Since Flink 1.17 it is externalised into its own release train entirely. 2. **You added the DataStream connector instead of the SQL connector.** `flink-connector-kafka` gives you `KafkaSource`/`KafkaSink` for the DataStream API. It does *not* by itself give the SQL planner a usable, shaded set of Kafka clients. For SQL Client and `lib/`, you want `flink-sql-connector-kafka`. 3. **Maven Shade merged the fat jar and destroyed the service file.** Silent, and the most expensive to debug. 4. **The dependency is `provided` scope** (copied from a Flink tutorial) so it never makes it into the artifact. ## 4\. Why It Happens — Under the Hood `FactoryUtil.discoverFactory(ClassLoader, Class, String)` does roughly this: ```java final List factories = discoverFactories(classLoader); // ServiceLoader.load(Factory.class, cl) final List foundFactories = factories.stream() .filter(f -> factoryClass.isAssignableFrom(f.getClass())) .collect(Collectors.toList()); final List matchingFactories = foundFactories.stream() .filter(f -> f.factoryIdentifier().equals(factoryIdentifier)) .collect(Collectors.toList()); if (matchingFactories.isEmpty()) { throw new ValidationException(/* ... "Available factory identifiers are:" ... */); } ``` Three consequences fall out of that. **The "Available factory identifiers" list is generated from what SPI actually loaded.** It is not a static list of "supported connectors". It is a live dump of the classpath. Treat it as diagnostic output, not documentation. **The classloader matters, and it is the user classloader.** Flink's `FlinkUserCodeClassLoader` defaults to child-first resolution (`classloader.resolve-order: child-first`), *except* for the prefixes in `classloader.parent-first-patterns.default`, which includes `org.apache.flink.`. Kafka connector classes are `org.apache.flink.connector.kafka.*`, so they match that pattern. Parent-first means *try the parent, fall back to the child if the parent doesn't have it* — so bundling in a fat jar still works when `lib/` is empty. But if `lib/` contains an **older** connector version, the parent wins and your bundled newer one is shadowed. You get `NoSuchMethodError` or `NoClassDefFoundError` rather than this exception, which is why "I fixed the classpath and now I have a different error" is the normal second act of this bug. **Shading is where the silent failure lives.** `META-INF/services/org.apache.flink.table.factories.Factory` is one file path. Multiple dependencies define it. Maven Shade's default behaviour is last-one-wins overwrite. If `flink-connector-kafka`, `flink-json`, and your own module each carry that file, you end up with exactly one survivor and Kafka silently vanishes from SPI. The jar is present, the classes are present, `javap` finds `KafkaDynamicTableFactory` — and Flink still cannot see it, because SPI reads the file, not the classes. This is why the fix in §5 includes a shade transformer, not just a dependency. The same SPI mechanism is what backs formats (`'format' = 'avro-confluent'` needs `flink-avro-confluent-registry`) and catalogs. If you fix Kafka and immediately hit `Could not find any factory for identifier 'avro-confluent'`, you are looking at the same bug wearing a different hat. ## 5\. The Fix ### Path A — SQL Client / cluster-wide (`lib/`) Drop the **shaded SQL uber jar** into `lib/` on JobManager *and* every TaskManager, then restart the cluster. `lib/` is read at process start; hot-copying a jar into a running container does nothing. ```yaml # docker-compose.yml — before jobmanager: image: flink:1.20-scala_2.12-java17 command: jobmanager # docker-compose.yml — after jobmanager: image: flink:1.20-scala_2.12-java17 command: jobmanager volumes: - ./lib/flink-sql-connector-kafka-3.3.0-1.20.jar:/opt/flink/lib/flink-sql-connector-kafka-3.3.0-1.20.jar ``` Apply the identical volume mount to the `taskmanager` service. Then: ```bash docker compose down && docker compose up -d docker compose exec jobmanager ls /opt/flink/lib/ | grep kafka ``` For a one-off session without restarting, the SQL Client accepts a jar directly: ```bash ./bin/sql-client.sh --jar /opt/flink/opt/flink-sql-connector-kafka-3.3.0-1.20.jar ``` or, from inside the session (Flink 1.15+): ```sql ADD JAR '/opt/flink/opt/flink-sql-connector-kafka-3.3.0-1.20.jar'; ``` **Version coordinates matter.** Since Flink 1.17 the Kafka connector is released separately, and the artifact version encodes both connector and Flink versions: `3.3.0-1.20` means connector 3.3.0 built for Flink 1.20\. For Flink 2.0 the coordinate is `4.0.1-2.0`. There is no `flink-sql-connector-kafka:1.20.0` — that assumption is the source of a lot of 404s from Maven Central. Scala suffixes were dropped from this connector in 1.15; if you are copying a `_2.12` artifact id from a blog post, that post predates 1.15. ### Path B — packaged job jar (Table API in Java) ```xml org.apache.flink flink-connector-kafka 3.3.0-1.20 provided ``` ```xml org.apache.flink flink-connector-kafka 3.3.0-1.20 org.apache.flink flink-json 1.20.0 ``` And the part everyone forgets — the shade transformer that merges service files instead of overwriting them: ```xml org.apache.maven.plugins maven-shade-plugin 3.5.1 package shade dev.ggorantala.PaymentsJob *:* META-INF/*.SF META-INF/*.DSA META-INF/*.RSA ``` Gradle equivalent: the Shadow plugin's `mergeServiceFiles()` call in the `shadowJar` block. Verify before you deploy: ```bash mvn clean package unzip -p target/payments-job.jar META-INF/services/org.apache.flink.table.factories.Factory # must contain: org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicTableFactory ``` ### Which path when | Situation | Use | | ------------------------------------------------------ | ------------------------------------------------------------------------------ | | SQL Client, ad-hoc exploration, shared session cluster | flink-sql-connector-kafka in lib/ | | Packaged Java/Table API job, application mode | flink-connector-kafka in the fat jar, compile scope | | Kubernetes Operator FlinkDeployment | bake into a custom image, not an initContainer copy | | Running from an IDE | compile scope + flink-table-planner-loader, flink-table-runtime, flink-clients | Do not put both `flink-connector-kafka` and `flink-sql-connector-kafka` on the same classpath. The SQL jar shades and relocates `org.apache.kafka.*`; the plain one does not. Having both produces duplicate-class conflicts that surface later, at checkpoint or commit time, far from the actual cause. ## 6\. Best Practices & The Better Design Stop treating the connector classpath as a deployment-time detail. Bake it into the image and make the image the unit of promotion: ```dockerfile FROM flink:1.20-scala_2.12-java17 ARG KAFKA_CONNECTOR_VERSION=3.3.0-1.20 ARG FLINK_VERSION=1.20.0 RUN curl -fsSL -o /opt/flink/lib/flink-sql-connector-kafka-${KAFKA_CONNECTOR_VERSION}.jar \ https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/${KAFKA_CONNECTOR_VERSION}/flink-sql-connector-kafka-${KAFKA_CONNECTOR_VERSION}.jar \ && curl -fsSL -o /opt/flink/lib/flink-avro-confluent-registry-${FLINK_VERSION}.jar \ https://repo1.maven.org/maven2/org/apache/flink/flink-avro-confluent-registry/${FLINK_VERSION}/flink-avro-confluent-registry-${FLINK_VERSION}.jar # fail the build, not the job, if SPI registration is broken RUN unzip -p /opt/flink/lib/flink-sql-connector-kafka-${KAFKA_CONNECTOR_VERSION}.jar \ META-INF/services/org.apache.flink.table.factories.Factory | grep -q KafkaDynamicTableFactory ``` Three properties this buys you: `curl -f` fails the build on a wrong version coordinate rather than producing a working image with a missing jar; the `grep -q` guard turns a silent runtime `ValidationException` into a red CI pipeline; and the connector version is pinned in source control next to the Flink version, so upgrading Flink forces a conscious decision about the connector. For the Flink Kubernetes Operator, reference that image directly rather than mounting jars at runtime: ```yaml apiVersion: flink.apache.org/v1beta1 kind: FlinkDeployment metadata: name: payments-pipeline spec: image: registry.internal/flink-payments:1.20-kafka3.3.0 flinkVersion: v1_20 flinkConfiguration: taskmanager.numberOfTaskSlots: "4" state.backend.type: rocksdb execution.checkpointing.interval: "60s" serviceAccount: flink jobManager: resource: { memory: "2048m", cpu: 1 } taskManager: resource: { memory: "4096m", cpu: 2 } job: jarURI: local:///opt/flink/usrlib/payments-job.jar parallelism: 4 upgradeMode: savepoint ``` Note FLINK-34991: when the operator hits this error, the CRD status often surfaces an HA/leader-election message instead of the underlying `ValidationException`. Read the JobManager pod logs, not the CR status, when a `FlinkDeployment` refuses to start. ## 7\. How to Prevent It Long-Term - **CI gate on SPI registration.** After `mvn package`, assert that `META-INF/services/org.apache.flink.table.factories.Factory` in the shaded jar contains every factory identifier the job's SQL references. A three-line shell step catches every shade regression. - **Smoke-test the classpath, not the cluster.** A CI job that runs `sql-client.sh` against the built image with a `datagen` → `kafka` → `blackhole` pipeline validates connector discovery in seconds without needing real infrastructure. - **Pin connector versions in a dependency-management BOM** shared across teams, so a Flink minor upgrade cannot silently leave a stale connector behind. - **Alert on the error class, not the message.** `ValidationException` at job submission is always a packaging defect and never a transient failure. Route it differently from `NoResourceAvailableException` or checkpoint alerts — retrying it is pointless, and a restart strategy that keeps retrying it will burn your failure budget for nothing. - **Ban runtime jar mounting in production manifests.** `initContainer` jar copies drift between JobManager and TaskManager and produce the maddening variant where the job submits successfully and then fails on the TaskManager. **Related failure modes worth linking:** once the connector is discoverable, the next two errors POC teams hit are `InvalidTxnTimeoutException` when `transaction.timeout.ms` exceeds the broker's `transaction.max.timeout.ms`, and windows that never fire because an idle Kafka partition blocks watermark progress. Both live one step downstream of this one. ## 8\. Key Takeaways - The "Available factory identifiers are:" list is a live dump of the classpath. If it shows only `blackhole`, `datagen`, `filesystem`, `print`, nothing beyond `flink-dist` was loaded. - `flink-sql-connector-kafka` (shaded uber jar, for `lib/` and SQL Client) is not `flink-connector-kafka` (DataStream API, for your fat jar). Never both on one classpath. - Connector versions encode two numbers — `3.3.0-1.20` is connector 3.3.0 for Flink 1.20, `4.0.1-2.0` for Flink 2.0\. There is no `flink-sql-connector-kafka:1.20.0`. - Discovery is `ServiceLoader` reading `META-INF/services/...`, not classpath scanning. Maven Shade without `ServicesResourceTransformer` overwrites that file and the connector disappears while its classes remain. - Jars in `lib/` are read at process start. Copying a jar into a running container fixes nothing until you restart JobManager and every TaskManager.