Flink: Could Not Find Any Factory for Identifier 'kafka'

1. The Error

You run a Flink SQL job that reads from Kafka. The CREATE TABLE succeeds. The SELECT blows up:

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:

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:

docker compose up -d
docker compose exec jobmanager ./bin/sql-client.sh

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:

SHOW CREATE TABLE payments;   -- options are stored fine

# 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<T>, String) does roughly this:

final List<Factory> factories = discoverFactories(classLoader);   // ServiceLoader.load(Factory.class, cl)
final List<Factory> foundFactories = factories.stream()
    .filter(f -> factoryClass.isAssignableFrom(f.getClass()))
    .collect(Collectors.toList());
final List<Factory> 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.

# 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:

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:

./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+):

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)

<!-- before: DataStream connector, provided scope -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka</artifactId>
  <version>3.3.0-1.20</version>
  <scope>provided</scope>
</dependency>

<!-- after: compile scope so it lands in the fat jar -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka</artifactId>
  <version>3.3.0-1.20</version>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-json</artifactId>
  <version>1.20.0</version>
</dependency>

And the part everyone forgets — the shade transformer that merges service files instead of overwriting them:

<plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-shade-plugin</artifactId>
  <version>3.5.1</version>
  <executions>
    <execution>
      <phase>package</phase>
      <goals><goal>shade</goal></goals>
      <configuration>
        <transformers>
          <!-- non-negotiable: merges META-INF/services/* across dependencies -->
          <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
          <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
            <mainClass>dev.ggorantala.PaymentsJob</mainClass>
          </transformer>
        </transformers>
        <filters>
          <filter>
            <artifact>*:*</artifact>
            <excludes>
              <exclude>META-INF/*.SF</exclude>
              <exclude>META-INF/*.DSA</exclude>
              <exclude>META-INF/*.RSA</exclude>
            </excludes>
          </filter>
        </filters>
      </configuration>
    </execution>
  </executions>
</plugin>

Gradle equivalent: the Shadow plugin's mergeServiceFiles() call in the shadowJar block.

Verify before you deploy:

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

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:

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:

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.