# 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](../../sr/schemas-manage.md#sr-prv) integration for Avro, JSON Schema, and Protobuf formats and [autoscaling](autopilot.md#flink-sql-autopilot). 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](#flink-comparison-tables-auto-inferred) | You must define every table with `CREATE TABLE`. | Tables are automatically inferred from Kafka topics.
`CREATE TABLE` is usually unnecessary. | | [DROP TABLE behavior](#flink-comparison-drop-table) | Metadata-only. The underlying topic and data remain. | Deletes the underlying Kafka topic and all of its data. | | [Kafka connector option](#flink-comparison-connector-name) | `'connector' = 'kafka'` or `'connector' = 'upsert-kafka'`. | `'connector' = 'confluent'` — a single connector covers both
append and upsert behavior. | | [SHOW CREATE TABLE round-trip](#flink-comparison-show-create-table) | 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](#flink-comparison-topic-delete) | 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](#flink-comparison-watermarks) | 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](../reference/statements/create-table.md#flink-sql-ctas) derived table. ### Autoscaling [Autopilot](autopilot.md#flink-sql-autopilot) scales up and scales down the compute resources that SQL statements use in Confluent Cloud. The autoscaling process is based on [parallelism](overview.md#flink-sql-stream-processing-concepts-parallel-dataflows), 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](../reference/statements/create-table.md#flink-sql-system-columns) named [$rowtime](../reference/statements/create-table.md#flink-sql-system-columns-rowtime). This column maps to the Kafka record timestamp, which can be either `LogAppendTime` or `CreateTime`. ### Default watermark strategy Flink requires a [watermark](../../_glossary.md#term-watermark) strategy for a variety of features, such as [windowing](timely-stream-processing.md#flink-sql-event-time-lateness) and [temporal joins](../reference/queries/joins.md#flink-sql-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](timely-stream-processing.md#flink-sql-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](../reference/statements/alter-table.md#flink-sql-alter-table) statement. ### Schema Registry support for Avro, JSON_SR, and Protobuf Confluent Cloud for Apache Flink supports [Schema Registry](../../sr/schemas-manage.md#sr-prv) 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](../reference/serialization.md#flink-sql-serialization). ### 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](../reference/statements/set.md#flink-sql-set-statement-config-options) 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 | ### 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](../reference/statements/drop-table.md#flink-sql-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](../reference/statements/create-table.md#flink-sql-create-table-with-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](../../topics/tableflow/overview.md#cloud-tableflow) and look up reference data in an [external table](external-tables.md#flink-external-tables). 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](../../connectors/overview.md#kafka-connect-cloud). 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](../reference/statements/drop-table.md#flink-sql-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. ## Related content - [Stream Processing with Confluent Cloud for Apache Flink](../overview.md#ccloud-flink) - [Schema Registry](../../sr/schemas-manage.md#sr-prv) - [Data Type Mappings](../reference/serialization.md#flink-sql-serialization) - [Schema and Statement Evolution](schema-statement-evolution.md#flink-sql-schema-and-statement-evolution) - [Flink SQL Autopilot](autopilot.md#flink-sql-autopilot) - [Compute Pools](compute-pools.md#flink-sql-compute-pools) - [DDL Statements in Confluent Cloud for Apache Flink](../reference/statements/overview.md#flink-sql-statements-overview) - [ALTER TABLE](../reference/statements/alter-table.md#flink-sql-alter-table) #### NOTE This website includes content developed at the [Apache Software Foundation](https://www.apache.org/) under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).