Confluent Cloud Flink SQL

SkillDatabases & data

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.

Available today. Use it from your connected AI after setup.

Add ahel to your AI once: Claude, ChatGPT, Cursor, Claude Code or Codex. Then ask it to use this.

Then ask your AI: use the Confluent Cloud Flink SQL skill

What this skill tells your AI

The instructions your AI receives, as published by confluentinc/agent-skills in skills/confluent-cloud-flink-sql/SKILL.md and read by ahel’s review.

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

  1. CC Flink ≠ Apache Flink. Verify every API, SQL construct, and runtime behavior against:
  2. No mocks in verification. Integration claims require real confluent CLI runs. Unit tests may mock; anything calling itself "end-to-end verification" may not.
  3. 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.
  4. Secrets never in repo. CREATE CONNECTION parameters are Terraform-injected, never hardcoded. Gitignore .tfvars, .tfstate*, *.secret* from day one.
  5. EXPLAIN before CREATE. Always EXPLAIN a query before statement create — catches parse/type errors without consuming CFUs.
  6. 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:

FileWhen to load
references/dialect-traps.mdBefore writing ANY Flink SQL — 22 CC-vs-OSS traps, single source of truth
references/cli-reference.mdBefore running confluent CLI — flag schemas, carry-over recipe, timing, token expiry
references/sql-patterns-cc.mdWhen writing SQL — CC-validated patterns: windows, joins, dedup, MATCH_RECOGNIZE, JSON, External Tables
references/formats-and-serialization.mdWhen configuring table formats — 7 supported formats, id-encoding, consume flags
references/troubleshooting-cc.mdWhen debugging errors — CC-specific error to cause to fix
references/reserved-words.mdWhen 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:

  1. EXPLAIN the query in flink shell — catches syntax and type errors for free.
  2. Write a minimal reproducer.
  3. 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.
  4. Run: confluent flink statement create <name> --sql "$(cat repro.sql)" --compute-pool <id> --database <cluster> --environment <env> --wait
  5. 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)
  6. 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

  1. references/dialect-traps.md (this skill) — canonical, consolidated
  2. Per-project CLAUDE.md — references this skill, adds project-specific context
  3. 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

Signals

GitHub stars
58
Forks
11
Last commit
Sep 2026
Advanced
Item type
skill
Key
confluent-cloud-flink-sql
Source
github.com/confluentinc/agent-skills