Confluent Cloud Flink SQL
Enforce the CC-Flink-vs-OSS-Flink dialect boundary and the CLI-driven verification loop for any Confluent Cloud Flink SQL work. Apache Flink OSS training data is a trap — CC rejects or silently mishandles a long list of otherwise-valid Flink SQL constructs.
Scope note: this skill's reference material is built around the OSS-vs-CC dialect boundary — traps in constructs that exist in both dialects but behave differently. It does not yet catalog CC-only DDL that has no OSS counterpart (e.g. CREATE MATERIALIZED TABLE, CREATE MODEL/AI_COMPLETE, CREATE AGENT, USE CATALOG). For those, verify directly against the CC Flink SQL reference rather than expecting a trap entry here.
Non-negotiables
- CC Flink ≠ Apache Flink. Verify every API, SQL construct, and runtime behavior against:
- No mocks in verification. Integration claims require real
confluent CLI runs. Unit tests may mock; anything calling itself "end-to-end verification" may not.
- Record decisions somewhere durable. Ask the user where they want dialect traps and verification notes tracked (e.g.
docs/flink-dialect-traps.md in their project) before writing any new file — don't assume a docs/ layout.
- Secrets never in repo.
CREATE CONNECTION parameters are Terraform-injected, never hardcoded. Gitignore .tfvars, .tfstate*, *.secret* from day one.
- EXPLAIN before CREATE. Always
EXPLAIN a query before statement create — catches parse/type errors without consuming CFUs.
- Don't invent identifiers. Use
<placeholder> for any topic, table, statement, or resource name you haven't verified.
Reference files
Load these on demand when the topic matches — do not read them all upfront:
| File |
When to load |
| references/dialect-traps.md |
Before writing ANY Flink SQL — 22 CC-vs-OSS traps, single source of truth |
| references/cli-reference.md |
Before running confluent CLI — flag schemas, carry-over recipe, timing, token expiry |
| references/sql-patterns-cc.md |
When writing SQL — CC-validated patterns: windows, joins, dedup, MATCH_RECOGNIZE, JSON, External Tables |
| references/formats-and-serialization.md |
When configuring table formats — 7 supported formats, id-encoding, consume flags |
| references/troubleshooting-cc.md |
When debugging errors — CC-specific error to cause to fix |
| references/reserved-words.md |
When hitting parse errors — must-backquote identifiers |
Red flags
Stop and consult references/dialect-traps.md if you catch yourself writing any of these:
- DataStream API (Java/Scala) — not supported on CC. Table API + PTF are GA in Java; Python Table API is Open Preview with no PTF yet
CREATE CATALOG ... — catalog = CC environment, not creatable
SET 'execution.checkpointing.*' — CC-managed, not settable
CREATE TABLE ... WITH ('connector' = 'kafka', ...) — tables auto-map from topics
'value.format' = 'json' — must be 'json-registry' (or another SR-backed format)
WITH cte AS (...) INSERT INTO ... — CC requires the CTE AFTER INSERT INTO
GROUP BY TUMBLE(ts, INTERVAL ...) — must use the TVF form: TUMBLE(TABLE t, DESCRIPTOR(ts), ...)
LATERAL TABLE(UNNEST(...)) — parse error; use CROSS JOIN UNNEST(...)
PROCTIME() — not supported; use External Tables/KEY_SEARCH_AGG or an event-time temporal join (avoid a regular join against an upsert-kafka topic — it retains the whole table in state)
CREATE FUNCTION f AS '...' without USING JAR — CC UDFs require an uploaded artifact
- Savepoints /
STOP WITH SAVEPOINT — not exposed on CC
--sql-file flag — doesn't exist; use --sql "$(cat file.sql)"
DROP TABLE — deletes the physical Kafka topic and its data on CC, not just metadata; confirm before running
Verification loop
Canonical validation loop for any CC Flink SQL claim:
- EXPLAIN the query in
flink shell — catches syntax and type errors for free.
- Write a minimal reproducer.
- Present the plan and wait for explicit user confirmation before running anything that creates or modifies a real resource. State: the statement name and SQL, the compute pool/database/environment it targets, and any side effects (DDL creates a Kafka topic and Schema Registry subject; every run consumes CFUs). Do not proceed to step 3 without a go-ahead.
- Run:
confluent flink statement create <name> --sql "$(cat repro.sql)" --compute-pool <id> --database <cluster> --environment <env> --wait
- Observe. Consume downstream:
confluent kafka topic consume <topic> --cluster <id> --from-beginning --value-format <matching-format> 2>/dev/null | grep -v '^%' — match <matching-format> to the sink's value.format (see references/formats-and-serialization.md; jsonschema for json-registry, avro for avro-registry, protobuf for proto-registry, string for raw)
- Record the command + output for later reference.
Escalation-required states (no silent workarounds):
- Statement
PENDING > 60s → confluent flink statement exception list <name> --cloud <provider> --region <region>
- UDF deploy "jar not found" →
confluent flink artifact list --cloud <provider> --region <region>
- Schema mismatch →
DESCRIBE <table>, diff against the producer schema
- Egress denied → check
CREATE CONNECTION + USING CONNECTIONS clause
See references/cli-reference.md for full flag schemas and timing expectations.
Anti-patterns
- Apache Flink docs or Stack Overflow answers tagged
apache-flink cited as CC authority
- LLM memory of "Flink SQL syntax" used without CC verification
- Mocking the
confluent CLI in anything claiming end-to-end verification
terraform apply -auto-approve on the first run of a root module
- Committing
.tfvars, .tfstate*, .terraform/, *.secret*
- Swallowing Flink statement exceptions — fail loud; read
statement exception list
- Hardcoded secrets in
CREATE CONNECTION or UDF source
Tutorials
developer.confluent.io/tutorials/#flink mixes OSS and CC tutorials.
Filter rule: Only use tutorials that list "Confluent Cloud" in prerequisites or use confluent flink shell. Apply the dialect trap table to any SQL copied from a tutorial — many target OSS Flink or Kafka Streams and are not CC-compatible as-is.
Source-of-truth hierarchy
references/dialect-traps.md (this skill) — canonical, consolidated
- Per-project
CLAUDE.md — references this skill, adds project-specific context
- A per-project trap log, if the user wants one kept — ask where before creating it
When a new trap is discovered during a session, tell the user so they can decide whether to record it in their project's own notes. Do not edit this skill's own installed files (references/dialect-traps.md or elsewhere) — propose the change and let the user (or a separate PR to this skill's repo) apply it.
References
1---2name: confluent-cloud-flink-sql3description: Write and debug Flink SQL that runs on Confluent Cloud, enforcing the CC-vs-Apache-Flink (OSS) dialect boundary. Use when the working directory is a Confluent Cloud Flink workspace, when a Flink SQL statement needs checking before it runs on a CC compute pool, or when the user mentions CC Flink, Confluent Cloud Flink SQL, the `confluent flink` CLI, a CFU compute pool, `CREATE CONNECTION`, or asks to check or debug Flink SQL whose runtime is Confluent Cloud. Also trigger when a Flink SQL question is posed and nothing establishes an Apache Flink OSS runtime. Do NOT trigger for: building or deploying Flink UDFs in Java (UDF/UDTF/PTF — use flink-udf); a full CDC pipeline from a database through Flink into Tableflow/Iceberg/Delta Lake (use confluent-cloud-cdc-tableflow); Kafka Streams topology work (use kafka-streams-programming); or Flink SQL confirmed to run on Apache Flink OSS, not Confluent Cloud.4---56# Confluent Cloud Flink SQL78Enforce the CC-Flink-vs-OSS-Flink dialect boundary and the CLI-driven verification loop for any Confluent Cloud Flink SQL work. Apache Flink OSS training data is a trap — CC rejects or silently mishandles a long list of otherwise-valid Flink SQL constructs.910Scope note: this skill's reference material is built around the OSS-vs-CC dialect boundary — traps in constructs that exist in both dialects but behave differently. It does not yet catalog CC-only DDL that has no OSS counterpart (e.g. `CREATE MATERIALIZED TABLE`, `CREATE MODEL`/`AI_COMPLETE`, `CREATE AGENT`, `USE CATALOG`). For those, verify directly against the [CC Flink SQL reference](https://docs.confluent.io/cloud/current/flink/reference/overview.md) rather than expecting a trap entry here.1112## Non-negotiables13141. **CC Flink ≠ Apache Flink.** Verify every API, SQL construct, and runtime behavior against:15 - [Confluent Cloud Flink docs](https://docs.confluent.io/cloud/current/flink/overview.md)16 - [CC Flink SQL reference](https://docs.confluent.io/cloud/current/flink/reference/overview.md)17 - [Confluent Terraform provider](https://registry.terraform.io/providers/confluentinc/confluent/latest/docs)18 - Live `confluent flink shell` against the user's compute pool192. **No mocks in verification.** Integration claims require real `confluent` CLI runs. Unit tests may mock; anything calling itself "end-to-end verification" may not.203. **Record decisions somewhere durable.** Ask the user where they want dialect traps and verification notes tracked (e.g. `docs/flink-dialect-traps.md` in their project) before writing any new file — don't assume a `docs/` layout.214. **Secrets never in repo.** `CREATE CONNECTION` parameters are Terraform-injected, never hardcoded. Gitignore `.tfvars`, `.tfstate*`, `*.secret*` from day one.225. **EXPLAIN before CREATE.** Always `EXPLAIN` a query before `statement create` — catches parse/type errors without consuming CFUs.236. **Don't invent identifiers.** Use `<placeholder>` for any topic, table, statement, or resource name you haven't verified.2425## Reference files2627Load these on demand when the topic matches — do not read them all upfront:2829| File | When to load |30|------|-------------|31| [references/dialect-traps.md](references/dialect-traps.md) | Before writing ANY Flink SQL — 22 CC-vs-OSS traps, single source of truth |32| [references/cli-reference.md](references/cli-reference.md) | Before running `confluent` CLI — flag schemas, carry-over recipe, timing, token expiry |33| [references/sql-patterns-cc.md](references/sql-patterns-cc.md) | When writing SQL — CC-validated patterns: windows, joins, dedup, MATCH_RECOGNIZE, JSON, External Tables |34| [references/formats-and-serialization.md](references/formats-and-serialization.md) | When configuring table formats — 7 supported formats, id-encoding, consume flags |35| [references/troubleshooting-cc.md](references/troubleshooting-cc.md) | When debugging errors — CC-specific error to cause to fix |36| [references/reserved-words.md](references/reserved-words.md) | When hitting parse errors — must-backquote identifiers |3738## Red flags3940Stop and consult `references/dialect-traps.md` if you catch yourself writing any of these:4142- DataStream API (Java/Scala) — not supported on CC. Table API + PTF are GA in Java; Python Table API is Open Preview with no PTF yet43- `CREATE CATALOG ...` — catalog = CC environment, not creatable44- `SET 'execution.checkpointing.*'` — CC-managed, not settable45- `CREATE TABLE ... WITH ('connector' = 'kafka', ...)` — tables auto-map from topics46- `'value.format' = 'json'` — must be `'json-registry'` (or another SR-backed format)47- `WITH cte AS (...) INSERT INTO ...` — CC requires the CTE AFTER `INSERT INTO`48- `GROUP BY TUMBLE(ts, INTERVAL ...)` — must use the TVF form: `TUMBLE(TABLE t, DESCRIPTOR(ts), ...)`49- `LATERAL TABLE(UNNEST(...))` — parse error; use `CROSS JOIN UNNEST(...)`50- `PROCTIME()` — not supported; use External Tables/`KEY_SEARCH_AGG` or an event-time temporal join (avoid a regular join against an upsert-kafka topic — it retains the whole table in state)51- `CREATE FUNCTION f AS '...'` without `USING JAR` — CC UDFs require an uploaded artifact52- Savepoints / `STOP WITH SAVEPOINT` — not exposed on CC53- `--sql-file` flag — doesn't exist; use `--sql "$(cat file.sql)"`54- `DROP TABLE` — deletes the physical Kafka topic and its data on CC, not just metadata; confirm before running5556## Verification loop5758Canonical validation loop for any CC Flink SQL claim:59600. **EXPLAIN** the query in `flink shell` — catches syntax and type errors for free.611. Write a minimal reproducer.622. **Present the plan and wait for explicit user confirmation** before running anything that creates or modifies a real resource. State: the statement name and SQL, the compute pool/database/environment it targets, and any side effects (DDL creates a Kafka topic and Schema Registry subject; every run consumes CFUs). Do not proceed to step 3 without a go-ahead.633. Run: `confluent flink statement create <name> --sql "$(cat repro.sql)" --compute-pool <id> --database <cluster> --environment <env> --wait`644. Observe. Consume downstream: `confluent kafka topic consume <topic> --cluster <id> --from-beginning --value-format <matching-format> 2>/dev/null | grep -v '^%'` — match `<matching-format>` to the sink's `value.format` (see [references/formats-and-serialization.md](references/formats-and-serialization.md); `jsonschema` for `json-registry`, `avro` for `avro-registry`, `protobuf` for `proto-registry`, `string` for `raw`)655. Record the command + output for later reference.6667Escalation-required states (no silent workarounds):6869- Statement `PENDING` > 60s → `confluent flink statement exception list <name> --cloud <provider> --region <region>`70- UDF deploy "jar not found" → `confluent flink artifact list --cloud <provider> --region <region>`71- Schema mismatch → `DESCRIBE <table>`, diff against the producer schema72- Egress denied → check `CREATE CONNECTION` + `USING CONNECTIONS` clause7374See [references/cli-reference.md](references/cli-reference.md) for full flag schemas and timing expectations.7576## Anti-patterns7778- Apache Flink docs or Stack Overflow answers tagged `apache-flink` cited as CC authority79- LLM memory of "Flink SQL syntax" used without CC verification80- Mocking the `confluent` CLI in anything claiming end-to-end verification81- `terraform apply -auto-approve` on the first run of a root module82- Committing `.tfvars`, `.tfstate*`, `.terraform/`, `*.secret*`83- Swallowing Flink statement exceptions — fail loud; read `statement exception list`84- Hardcoded secrets in `CREATE CONNECTION` or UDF source8586## Tutorials8788[developer.confluent.io/tutorials/#flink](https://developer.confluent.io/tutorials/#flink) mixes OSS and CC tutorials.8990**Filter rule:** Only use tutorials that list "Confluent Cloud" in prerequisites or use `confluent flink shell`. Apply the dialect trap table to any SQL copied from a tutorial — many target OSS Flink or Kafka Streams and are not CC-compatible as-is.9192## Source-of-truth hierarchy93941. `references/dialect-traps.md` (this skill) — canonical, consolidated952. Per-project `CLAUDE.md` — references this skill, adds project-specific context963. A per-project trap log, if the user wants one kept — ask where before creating it9798When a new trap is discovered during a session, tell the user so they can decide whether to record it in their project's own notes. Do not edit this skill's own installed files (`references/dialect-traps.md` or elsewhere) — propose the change and let the user (or a separate PR to this skill's repo) apply it.99100## References101102- [Confluent Cloud Flink docs](https://docs.confluent.io/cloud/current/flink/overview.md)103- [Confluent Cloud Flink SQL reference](https://docs.confluent.io/cloud/current/flink/reference/overview.md)104- [Confluent Terraform provider](https://registry.terraform.io/providers/confluentinc/confluent/latest/docs)105- [Confluent CLI reference — Flink](https://docs.confluent.io/confluent-cli/current/command-reference/flink/index.md)106- [Tutorials (filter for CC only)](https://developer.confluent.io/tutorials/#flink)