Confluent Cloud for Apache Flink compared with Apache Flink

Confluent Cloud for Apache Flink extends Apache Flink with fully managed infrastructure and automatic metadata mapping from Apache Kafka® topics to Apache Flink® tables. It also adds Schema Registry integration for Avro, JSON Schema, and Protobuf formats and autoscaling. This page describes the key differences, behavioral changes, and current limitations of Confluent Cloud for Apache Flink compared with self-managed Apache Flink.

Quick comparison: Confluent Cloud for Apache Flink and Apache Flink

This table summarizes the Confluent Cloud for Apache Flink behaviors that most often surprise users — and AI assistants — familiar with Apache Flink. Each row links to a fuller explanation later on this page.

Topic	Apache Flink	Confluent Cloud for Apache Flink
Table creation	You must define every table with `CREATE TABLE`.	Tables are automatically inferred from Kafka topics. `CREATE TABLE` is usually unnecessary.
Delivery guarantee	The `kafka` connector writes with at-least-once by default and reads uncommitted messages, so output is visible to downstream statements immediately.	Exactly-once by default. Statements read only committed messages, so output is visible to downstream statements only after the transaction commits, approximately every minute, and each chained statement adds another commit wait.
Table options	The `kafka` and `upsert-kafka` connectors accept many options, including `properties.`, `topic`, and `sink.`.	Supports a smaller set of options. Unsupported options fail with an error that lists the supported options.
DROP TABLE behavior	Metadata-only. The underlying topic and data remain.	Deletes the underlying Kafka topic and all of its data.
Kafka connector option	`'connector' = 'kafka'` or `'connector' = 'upsert-kafka'`.	`'connector' = 'confluent'` — a single connector covers both append and upsert behavior.
SHOW CREATE TABLE round-trip	Re-running the output against an existing table returns “table already exists.”	Returns a Schema Registry subject mismatch error, which can be misread as a schema-incompatibility problem.
Topic deletion and schemas	Deleting a topic outside Flink doesn’t clean up Schema Registry subjects.	Same — but `DROP TABLE` (with the default `TopicNameStrategy`) cleans up subjects automatically.
Watermark strategy	You must define a watermark strategy explicitly for every table.	Confluent Cloud for Apache Flink applies a default watermark strategy to every table. Override it only when needed.

Additional features

The following list shows features provided by Confluent Cloud for Apache Flink that go beyond what Apache Flink offers.

Auto-inference of environments, clusters, topics, and schemas

Most users never need to write a CREATE TABLE statement in Confluent Cloud for Apache Flink. Every Kafka topic in the current cluster is already visible as a table, with its key and value schemas mapped from Schema Registry. Confluent Cloud for Apache Flink maps environments, clusters, topics, and schemas automatically from Confluent Cloud to the corresponding Apache Flink concepts of catalogs, databases, tables, and table schemas. In Apache Flink, by contrast, you must define and configure every table and its schemas, including authentication and authorization to Apache Kafka®.

Use CREATE TABLE only when you need to customize a schema, set a non-default serialization format, configure a changelog mode, or create a CREATE TABLE AS SELECT derived table.

Autoscaling

Autopilot scales up and scales down the compute resources that SQL statements use in Confluent Cloud. The autoscaling process is based on parallelism, which is the number of parallel operations that occur when the SQL statement is running. A SQL statement performs at its best when it has the optimal resources for its required parallelism.

Default system column implementation

Confluent Cloud for Apache Flink has a default implementation for a system column named $rowtime. This column maps to the Kafka record timestamp, which can be either LogAppendTime or CreateTime.

Default watermark strategy

Flink requires a watermark strategy for a variety of features, such as windowing and temporal joins. Confluent Cloud for Apache Flink has a default watermark strategy that applies to all tables/topics and depends on the $rowtime system column. Apache Flink requires you to define a watermark strategy manually. For more information, see Event time and watermarks.

Because the default strategy targets general usage, some cases require a custom strategy, for example, when your data has out-of-orderness that significantly exceeds a fixed out-of-orderness tolerance of 180 milliseconds. You can override the default strategy with a custom strategy by using the ALTER TABLE statement.

Schema Registry support for Avro, JSON_SR, and Protobuf

Confluent Cloud for Apache Flink supports Schema Registry formats Avro, JSON_SR (JSON Schema), and Protobuf. Apache Flink currently supports only Avro with Schema Registry. When you create or read from Kafka topics, Confluent Cloud for Apache Flink automatically maps between Flink SQL data types and your schema format. For the complete type mapping reference, see Data Type Mappings.

INFORMATION_SCHEMA support

Confluent Cloud for Apache Flink has an implementation for INFORMATION_SCHEMA, which is a system view that provides insights on catalogs, databases, tables, and schemas. This doesn’t exist in Apache Flink.

Behavioral differences

The following list shows differences in behavior between Confluent Cloud for Apache Flink and Apache Flink.

Configuration options

Apache Flink supports various optimization configuration options on different levels, such as Execution Options, Optimizer Options, Table Options, and SQL Client Options. Confluent Cloud for Apache Flink supports only the necessary subset of these options.

The following table maps configuration option names between Confluent Cloud for Apache Flink and Apache Flink. Use it when porting OSS configuration to Confluent Cloud for Apache Flink.

Confluent Cloud for Apache Flink	Apache Flink
client.results-timeout	table.exec.async-lookup.timeout
client.statement-name	–
sql.current-catalog	table.builtin-catalog-name
sql.current-database	table.builtin-database-name
sql.dry-run	–
sql.inline-result	–
sql.local-time-zone	table.local-time-zone
sql.state-ttl	table.exec.state.ttl
sql.tables.scan.bounded.timestamp-millis	scan.bounded.timestamp-millis
sql.tables.scan.bounded.mode	scan.bounded.mode
sql.tables.scan.idle-timeout	table.exec.source.idle-timeout
sql.tables.scan.startup.timestamp-millis	scan.startup.timestamp-millis
sql.tables.scan.startup.mode	scan.startup.mode
sql.tables.scan.watermark-alignment.max-allowed-drift	scan.watermark.alignment.max-drift

In a SET statement, always use the Confluent Cloud for Apache Flink name. A SET statement that sets an Apache Flink option name typically fails with an Unsupported configuration options found error that lists the supported options. The exception is table.local-time-zone, which Confluent Cloud for Apache Flink accepts but doesn’t apply. To set the local time zone, use sql.local-time-zone.

Delivery guarantees default to exactly-once

In Apache Flink, the kafka connector writes with at-least-once by default and reads uncommitted messages, so a downstream statement sees rows as soon as an upstream statement writes them.

In Confluent Cloud for Apache Flink, statements provide exactly-once semantics by default. Statements write to Kafka in transactions that commit approximately every minute, and they read only committed messages. In a pipeline of two or more statements, each statement waits for the upstream transaction to commit before it sees the rows, so each statement in the pipeline adds to the latency.

The Apache Flink options sink.delivery-guarantee and sink.semantic aren’t supported. If you need lower latency and at-least-once semantics are acceptable, set the kafka.consumer.isolation-level option to read-uncommitted on the tables that downstream statements read. Do this only if those statements, and every statement and client downstream of them, tolerate duplicates. For more information, see Latency in multi-statement pipelines.

Table options

For tables that use the confluent connector, Confluent Cloud for Apache Flink supports only the table options listed in CREATE TABLE … WITH and format-specific options, such as value.<format>.subject-names. Tables that use other connectors, such as faker, take the options of that connector. Many options of the Apache Flink kafka and upsert-kafka connectors, such as properties.*, topic, and sink.*, aren’t supported. A CREATE TABLE or ALTER TABLE statement that sets an unsupported option fails with an Unsupported options found error that names the option and lists the supported options. Options that have the same name in both, such as scan.startup.mode and value.format, are supported, but their defaults and values can differ.

A dynamic table option hint supports a smaller set of options and applies only to tables that a query reads from. A query that sets an unsupported option in a hint fails with an Unable to create a source for reading table error. The Unsupported dynamic option(s) line in the error details names the option. A hint on the target table of an INSERT INTO statement fails with a The requested operation is not supported error.

The following table maps common Apache Flink connector options to their Confluent Cloud for Apache Flink equivalents. Use it when you port DDL from Apache Flink.

Apache Flink option	Confluent Cloud for Apache Flink equivalent
`'connector' = 'kafka'`, `'connector' = 'upsert-kafka'`	`'connector' = 'confluent'` with changelog.mode. For more information, see Kafka connector option is confluent.
`topic`, `properties.bootstrap.servers`	None. The table name is the topic name, and the database is the Kafka cluster.
`properties.group.id`	None. You can’t set the consumer group, and Confluent Cloud for Apache Flink can’t read the offsets of an existing consumer group directly. The `group-offsets` scan.startup.mode uses a consumer group that Confluent Cloud for Apache Flink manages for each statement. To resume a stateless statement from a previous statement, use Carry-over Offsets. To start where an existing consumer group stopped, look up its committed offsets and set them with scan.startup.specific-offsets.
`properties.isolation.level`	kafka.consumer.isolation-level. The values use hyphens: `read-committed` or `read-uncommitted`.
`sink.delivery-guarantee`, `sink.semantic`, `sink.transactional-id-prefix`	None. Streaming writes are always transactional. To lower latency, set `kafka.consumer.isolation-level` to `read-uncommitted` on the reading side, if the reading statements, and every statement and client downstream of them, tolerate duplicates.
`format`, `value.format` = `json`, `avro-confluent`	value.format = `avro-registry`, `json-registry`, `proto-registry`, or `raw`. The `avro-registry` format uses the same Schema Registry wire format as `avro-confluent`. The `json-registry` format isn’t a drop-in replacement for `json`, because it reads and writes the Schema Registry wire format instead of plain JSON. The `raw` format requires a single value column named `val`.
`key.format`	key.format = `avro-registry`, `json-registry`, `proto-registry`, or `raw`. The `raw` format requires a single key column named `key`. The key format applies only if the table has a primary key or a `DISTRIBUTED BY` clause.
`key.fields`, `sink.partitioner`	A DISTRIBUTED BY clause or a PRIMARY KEY constraint. A primary key also changes the default changelog mode to `upsert`. To port an append-only `kafka` table, use `DISTRIBUTED BY`, or a primary key with `'changelog.mode' = 'append'`. By default, Confluent Cloud for Apache Flink excludes key columns from the value, because value.fields-include defaults to `except-key`. To keep the Apache Flink default, which writes key columns to both the key and the value, set it to `all`.
`scan.parallelism`, `sink.parallelism`	None. Autopilot manages parallelism.
`scan.watermark.*`	A WATERMARK clause, or the `sql.tables.scan.idle-timeout` and `sql.tables.scan.watermark-alignment.max-allowed-drift` SET options. The SET options apply to every table that a statement reads, not to a single table.
`scan.topic-partition-discovery.interval`, `sink.buffer-flush.*`	None. Confluent Cloud for Apache Flink manages these settings.

CREATE statements provision underlying resources

When you run a CREATE TABLE statement in Confluent Cloud for Apache Flink, it creates the underlying Kafka topic and a Schema Registry schema in Confluent Cloud. In Apache Flink, a CREATE TABLE statement only registers the object in the Apache Flink catalog and doesn’t create an underlying resource.

Confluent Cloud for Apache Flink doesn’t support temporary tables, because every table maps to a real underlying Kafka topic. Apache Flink supports temporary tables.

DROP TABLE statements delete the underlying Kafka topic and data

In Confluent Cloud for Apache Flink, the DROP TABLE statement permanently deletes the underlying Kafka topic and every record in it. With the default TopicNameStrategy setting, it also deletes the key and value schema subjects from Schema Registry. In Apache Flink, the DROP TABLE statement only removes the table from the Flink catalog and leaves the underlying storage untouched.

Warning

Don’t use DROP TABLE to clear a table you intend to recreate — DROP TABLE deletes the Kafka topic and every record in it along with the table. Stop dependent statements first and confirm the data is reproducible or backed up before you drop a table. For full details, see DROP TABLE.

Kafka connector option is confluent

In Confluent Cloud for Apache Flink, always use the setting 'connector' = 'confluent' for Kafka-backed tables. This single connector covers both append and upsert behavior, and it determines the source or sink type by examining the changelog.mode option. Apache Flink splits this across two connectors — set with 'connector' = 'kafka' and 'connector' = 'upsert-kafka' — where the connector choice combined with the format defines append versus update behavior. When you port DDL from Apache Flink, replace both 'connector' = 'kafka' and 'connector' = 'upsert-kafka' with 'connector' = 'confluent' and set the changelog mode explicitly.

For streaming reads and writes, Confluent Cloud for Apache Flink supports Kafka topics that are on Confluent Cloud. Confluent Cloud for Apache Flink can also read Apache Iceberg™ tables through Confluent Tableflow and look up reference data in an external table. Apache Flink supports other streaming connectors, such as Kinesis, Pulsar, JDBC, and others, along with other Kafka environments, such as on-premises and different cloud service providers.

To bring data from another system into Confluent Cloud for Apache Flink, use Kafka Connect. Confluent Cloud offers 100+ fully managed connectors that stream data from databases, SaaS applications, and other external systems into an Kafka topic. Confluent Cloud for Apache Flink can then read this topic as a table.

SHOW CREATE TABLE output isn’t a re-runnable script

In Confluent Cloud for Apache Flink, the verbatim output of SHOW CREATE TABLE fails with a Schema Registry subject mismatch error when re-run against an existing table. The generated DDL implies registering a new schema subject for a topic that already has one. The table already exists, but the error message describes a schema-incompatibility symptom rather than a condition where the table already exists. Treat SHOW CREATE TABLE output as a human-readable summary, not as an idempotent re-creation script.

In Apache Flink, by contrast, the output of SHOW CREATE TABLE can be re-executed against the same catalog to reproduce a table definition, and re-running it against an existing table fails with an error that states that the table already exists.

Topic deletion and Schema Registry subject cleanup

Deleting a Kafka topic outside Flink never cleans up its Schema Registry subjects, but the DROP TABLE statement in Confluent Cloud for Apache Flink does (with the default TopicNameStrategy setting). In both Confluent Cloud for Apache Flink and Apache Flink, deleting a Kafka topic directly through Kafka APIs, the Confluent Cloud Console, or the Confluent CLI leaves the corresponding key and value schema subjects in Schema Registry — you must delete the subjects explicitly to fully remove the topic.

Confluent Cloud for Apache Flink adds the DROP TABLE shortcut, which Apache Flink doesn’t have: with the default TopicNameStrategy setting, it deletes the topic and the associated key and value subjects in a single statement. The RecordNameStrategy and TopicRecordNameStrategy settings preserve subjects because multiple topics can share them. See DROP TABLE for the matrix of behaviors per naming strategy.

Limitations

The following list shows limitations of Confluent Cloud for Apache Flink compared with Apache Flink.

Windowing functions syntax

Confluent Cloud for Apache Flink supports the TUMBLE, HOP, SESSION, and CUMULATE windowing functions only by using Table-Valued Functions syntax. Apache Flink also supports these windowing functions by using the outdated Group Window Aggregations functions.

Unsupported statements and features

Confluent Cloud for Apache Flink does not support the following statements and features.

ANALYZE statements
CALL statements
CATALOG commands other than SHOW (No CREATE/DROP/ALTER)
DATABASE command other than SHOW (No CREATE/DROP/ALTER)
DELETE statements
DROP CATALOG and DROP DATABASE
JAR statements
LOAD / UNLOAD statements
TRUNCATE statements
UPDATE statements
Processing time operations, such as PROCTIME(), TUMBLE_PROCTIME, HOP_PROCTIME, SESSION_PROCTIME, and CUMULATE_PROCTIME

Limited support for ALTER

Confluent Cloud for Apache Flink has limited support for ALTER TABLE compared with Apache Flink. In Confluent Cloud for Apache Flink, you can use ALTER TABLE only to change the watermark strategy, add a metadata column, or change a parameter value.