feat(streaming): declare every Kafka record shape in FLUID, checked before deploy by fas89 · Pull Request #817 · Agenticstiger/forge-cli

fas89 · 2026-11-04T04:38:48Z

Stacked on #784 (feat/managed-flink-iceberg-streaming), after #809 (run 2) and #815 (re-sync).

The managed Flink → Iceberg lane now takes every shape of Kafka record, declared in FLUID:

flat records;
nested records kept nested;
nested records flattened: struct fields lifted, arrays exploded, map entries looked up;
nested records rebuilt into another nested shape;
the record key and Kafka metadata as columns.

forge-cli writes the job's query from the contract and checks every output column against it before anything deploys. Shared stream fragments work across contracts.

What a contract says (FLUID 0.7.7 preview only)

properties:
  language: flink_sql
  stream:                       # $defs/streamSource: what the records are
    kind: kafka
    topic: orders
    format: json                # or avro-confluent
    schema: {$ref: ../fragments/streams/orders.value.yaml}   # FLUID columns
    key: {format: raw, schema: [{name: order_key, type: string}]}
    metadata: [{name: kafka_offset, from: offset}, {name: kafka_ts, from: timestamp}]
  mapping:                      # $defs/recordMapping: how they become rows
    explode: {path: items, as: item}
    columns:
      - {name: country, from: customer.address.country}
      - {name: channel, from: "attributes['channel']"}
      - {name: sku, from: item.sku}
      - {name: order.key, from: order_key}      # a dotted name builds a nested output
  parameters:
    flink:
      source: {bootstrapServers: kafka:9092, groupId: g1}   # connection only now

By name by default. An output column the mapping does not list is read from the same path. Flat and nested-kept-nested streams need no mapping at all.
Static contract check (dbt's model-contract preflight, done statically). Every output column's type and nullability is checked against its path, before deploy:
- struct fields bind by name, never by position;
- only lossless widenings and binary to string are allowed;
- a required output fed by a path that can be null fails.
- Before this change, a SELECT producing 1 of 6 columns compiled.
Null propagation. A struct rebuilt from one nullable source struct is null where that struct is. Its required fields stay required where the struct is present.
Kafka key and metadata.
- The key comes through key.format, key.fields, value.fields-include=EXCEPT_KEY and an optional key.fields-prefix.
- Metadata columns are METADATA FROM '<key>' VIRTUAL, with types fixed by forge-cli from Flink 1.20's connector table.
- A metadata timestamp can be the event-time column.
Contract SQL is an operator's call. properties.sql still works, bound to the output by name.
- It runs with the job's catalog identity, so it can read any lake.<ns>.<table>. fluid apply therefore refuses it unless FLUID_FLINK_CONTRACT_SQL_ALLOWLIST lists it.
- The recommended entry is pinned to the reviewed SQL: <product>/<build>@sha256:<digest>.
Interop.
- ODCS export carries each column's source as transformSourceObjects/transformLogic, and an explode as dataGranularityDescription. It still round-trips.
- fluid diff treats the stream as a reader schema. A mapping, explode, topic, format or contract-SQL change is classified.

Design and decisions

The founder chose these via AskUserQuestion on 2026-11-04:

a declarative mapping, with operator-gated SQL;
FLUID-level $defs;
Kafka key and metadata as the only v1 extras;
evolution handled by refusal plus fluid diff rules.

Prior art read for the design: Flink 1.20's Kafka connector and formats; Iceberg's Flink type mapping and DDL; Confluent Schema Registry compatibility; RisingWave, Materialize, Arroyo, Paimon, Flink CDC; Estuary, Doris and the Kafka Connect Flatten SMT; Redpanda Iceberg Topics; Tableflow; dbt model contracts; ODCS 3.2. Not in scope: protobuf (Flink 1.20 needs compiled classes) and the Debezium formats (they need upsert into Iceberg).

Fixes to the lane found on the way

Security (HIGH): two builds sharing an id were both checked as the last one. That let one build skip the contract-SQL gate and every run-2 allowlist. Builds are now paired with their artifacts by position, and repeated ids are refused.
Live fixture: Docker Hub and Quay no longer serve MinIO's images, so the live lane could not start. It now runs on Adobe S3Mock 5.2.3.
Event time: an event-time column of precision above 3 compiled to a WATERMARK that Flink refuses. It is now refused at compile.
Schema: the 0.7.7 schema had dropped sql as a required field for any build with a stream. It is optional now only for a managed build.

Evidence

Unit lane: ruff, black, mypy --strict and lint-imports pass; 570 tests pass; fluid_build.streaming coverage is 94%.
Live Docker lane (flink-iceberg.yml, run_live): see the run linked below.
- nested fidelity with task recovery and drift refusal;
- both shapes from keyed records with a header, read back with PyIceberg and compared value by value;
- a planner test that EXPLAINs every generated query in the runtime image's Flink.
Reviews: an Opus security review (re-verified after the fixes), a defect scan (logic, runtime and cross-file hunters plus CLI journeys), and a scoped world-class inspection.

Not in this PR

docs/flink-iceberg.md still describes the old parameters.flink.source.schema + SELECT format. Docs live in forge_docs, so updating it is a follow-up there.
Iceberg partition transforms (day, bucket, …) in icebergConfig.partitionSpec compile into DDL that Flink cannot parse, because Iceberg's Flink DDL has no hidden partitioning. This predates this change, and the fix is pending a decision.

…ts before deploy The managed Flink lane read its records from parameters.flink.source and ran a hand-written SELECT that nothing checked against the exposed schema: a query producing one of six columns, or a string into an int column, compiled. 0.7.7 (preview) adds two $defs for embedded-logic builds: - properties.stream (streamSource): kind kafka, the topic, the value format and schema (FLUID columns, the shape of exposes[].contract.schema), an optional key (raw, json or avro-confluent; key.prefix when a key field repeats a value field) and Kafka metadata read as typed VIRTUAL columns. - properties.mapping (recordMapping): explode one array into rows, and columns read from another path (customer.address.country, attributes['channel'], item.sku), nested output fields included by dotted name, with a safe cast. forge-cli writes the query from them: - An output column the mapping does not list is read from the same path, so flat and nested streams land with no mapping. - Every output column's type and nullability is checked against its path before anything deploys: dbt's model-contract preflight check, done statically because the contract declares every source type. - Struct fields bind by name, never by position. A struct built from one nullable source struct stays null where that struct is. - No contract text reaches the job's SQL. parameters.flink.source keeps the connection only. properties.sql stays possible, and is now bound to the output columns by name.

…lists it A contract's SELECT runs with the job's catalog identity. It can read any table that identity can (lake.<namespace>.<table>, through a join or UNION ALL) and copy it into the product's own table. Disabling SQL hints does not close this. fluid apply now refuses properties.sql unless the operator lists the product (or product/build) in FLUID_FLINK_CONTRACT_SQL_ALLOWLIST. An unset list allows nothing, the same stance as the other managed-Flink allowlists. fluid validate warns about it. A build that declares properties.mapping needs no listing: forge-cli writes its query, and the query reads only the stream.

Properties of an expose that a build fills from properties.stream get ODCS v3 property lineage: - transformSourceObjects: the topic; - transformLogic: the source path, including for nested fields mapped on their own. An explode becomes the schema object's dataGranularityDescription. Imported transform* fields are kept as written, so FLUID-emitted ODCS still round-trips. A build that runs its own SQL reports no lineage.

properties.stream is a reader schema, so fluid diff applies the column rules with one exception: - A field the job stops reading is non-breaking: an output that still reads it fails fluid validate. - A field the job newly requires is breaking: a record without it stops the job. This is FORWARD compatibility seen from the consumer. A changed topic, value or key format, metadata source or explode is breaking. An output column read from another path keeps its type, and is reported as non-breaking for review.

examples/flink-iceberg/shapes reads the orders topic, kept in one stream fragment with its key and Kafka metadata, into two products: - order-items flattens each event to one row per item. It lifts nested fields, explodes items, looks up a map entry and a header, and keeps the key and the Kafka position. - orders-nested writes each event as a nested row in a shape of its own: structs built from fields elsewhere in the record, a customer subset that stays null where the customer is, and the Kafka position as one struct. A fluid.workspace.yaml lets both reach the shared fragment. The fixture's task manager gets two slots, so both jobs run on one stack.

…h headers The live lane now also runs the shapes example: two jobs from the one stream fragment, fed records with a per-record key and a header, read back with PyIceberg and compared with what was produced. The comparison covers: - one row per item, a null element included; - lifted fields, the map lookup and the header cast; - the key column and the Kafka partition and offset; - structs rebuilt, null-propagated and moved; - the Kafka struct. The workflow also runs when the changelog modules change.

…s are gone The live lane could not start: Docker Hub no longer serves minio/minio or minio/mc, and Quay refuses them too. So the existing nested-fidelity proof failed at compose up, before any job ran. The fixture's object store is now Adobe's S3Mock 5.2.3 (Apache-2.0): - It creates the warehouse bucket at startup, so the mc init container goes. - It accepts any credentials. - It serves path-style requests, which is how the catalog, the Flink job and the PyIceberg readback already address it.

…und by id apply, diff and verify looked a build up by its id. Two builds sharing an id were therefore both checked as the last one. A contract could put properties.sql reading lake.secret_ns.payroll, an unlisted image, a cluster-admin service account, a production Secret or an OAuth2 credential sent to its own host in the first build. A clean build reusing the id came last. Then the contract-SQL gate and every operator allowlist checked in reconcile passed it. compile_builds pairs each build with its artifacts by position, and refuses build ids that repeat. reconcile, inspect_drift and _verify_once use those pairs. The id lookup predates the gate (#809), but the gate depended on it. Found by the security review of the Kafka record shapes change.

@sha256

…was reviewed A FLUID_FLINK_CONTRACT_SQL_ALLOWLIST entry naming a product or build approved any SQL that build later ran. A reviewed SELECT could be edited into one that reads lake.hr.salaries with no new review. An entry can now be <product>/<build>@sha256:<digest of the stripped SQL>. That entry approves the reviewed SQL only. The refusal prints it, and recommends it in place of an earlier pin for the same build. It says when the SQL changed after it was listed. Product and build entries still work, as the operator's broader choice.

…ning its own SQL A build moving from its mapping to properties.sql that reads other tables got no fluid diff finding at all. The stream rules now report: - SQL added: breaking, because it can read any table the job's catalog identity can; - SQL changed: breaking, for review; - SQL removed in favour of the mapping: non-breaking. Whitespace around the SQL is not a change. Found by the security review of the Kafka record shapes change.

…apping error is about Every output path scanned all mapping entries for fields mapped under it, so compile time grew with the square of the mapping's size: 8000 columns took 5.2 s. An index built once takes 0.11 s. A path into the record that went wrong said "the struct has no field adress". It now names where the path got to: "customer has no field adress", and "event_id is string, not a map".

… proof The proof that the runner refuses a table that no longer matches the contract changed the output's type alone. The compiler now refuses that contract before anything runs, because its output cannot be read from its stream. The proof now changes both, so the contract agrees with itself and only the existing table disagrees.

…essage, not a crash Found by the defect scan of the record shapes change: - The topics . and .. passed fluid validate. Kafka forbids both, so the job failed on deploy. - An event-time column of precision above 3 compiled into a WATERMARK that Flink refuses: its rowtime takes precision 0 to 3, and a stream timestamp defaults to 6. The base branch had this too. Probing Flink 1.20's planner confirmed it. - A stream key that is not an object raised AttributeError, and a metadata "from" that is not a string raised TypeError. fluid apply and fluid verify then printed "Unexpected error" with no path. - An empty key.prefix was accepted, although the schema refuses it. - validation_warnings raised TypeError on a contract whose builds is not a list. Each now fails with a ValueError that names the field.

…eping required fields orders-nested failed Flink's validator in the live lane. Probing Flink 1.20's planner showed the cause: a CAST(ROW(...)) as a CASE or IF branch is refused, while a bare ROW branch and a CAST of the whole CASE are accepted. A struct read from one nullable source struct is now: CAST(CASE WHEN <struct> IS NULL THEN NULL ELSE ROW(<fields>) END AS ROW<...>) The fields of that row are now checked assuming the source struct is present, because the row is null wherever it is not. A required field of an optional struct therefore stays required when the struct is rebuilt (a subset, a reorder, a widening), as it already did when the struct was copied whole. The struct itself still cannot be declared required. tests/integration/test_flink_iceberg_live.py gains a planner test. It EXPLAINs every example's query, and the hardest mappings, in the runtime image's own Flink. This needs no broker or catalog: the output is a blackhole table with the expose's columns. The diff tests in test_standards.py change the stream with the output, so their contracts stay valid once fluid diff validates both sides.

The 0.7.7 embedded-logic pattern dropped the sql requirement for any build that declared a stream. A python build or an external Flink build with a stream and no query therefore validated cleanly, and nothing ran it. sql is now optional only where the managed lane writes the query: a build with a stream and parameters.flink.management. A mapping on a build the lane does not manage is reported by fluid validate, because nothing compiles it. Found by the inspection of the record shapes change.

…pped, and a cast changed Gaps found by the security review and the defect scan: - A build renamed or added with properties.sql got no finding, because diff_streams compared only ids present on both sides. It is now stream_contract_sql_added (breaking). - A build that stops or starts declaring its stream is reported, as info. - A changed cast on a mapped column is reported as a changed source. - Field labels read as a mapping names them ("customer.id", "items[].price"), not as the schema nests them ("customer.fields.id").

…s, and the gate's order The inspection found guarantees that held but that no test held them to: - a mismatch inside an array element or a map value; - a nullable element; - nullability carried through a widening, a conversion and an exploded element; - the path grammar as the thing that refuses an unsafe path; - the contract-SQL gate refusing before any kubectl call, and an allowed build going on to apply.

…and #820 (#821) Real merges of main (through 85174c0) and of #817 and #819 into the managed Flink feature branch, plus the schema_validator fix #811's guard required. Landed as a merge commit so main stays an ancestor.

docs/flink-iceberg.md came in with #784 but still described the old parameters.flink.source.schema, which #817 replaced with properties.stream and properties.mapping. Docs live in forge_docs, so the guide moves there, rewritten for #817, #809 and #819, as Agenticstiger/forge_docs#196. The README and the example now link to the published page (founder decision, 2026-11-04). That link resolves once forge_docs#196 merges after #784.

Managed Flink to Iceberg streaming preview, in FLUID 0.7.7 (author: Dustlab). - Recursive record types. - Kafka record shapes declared in FLUID (properties.stream and properties.mapping). - Generated FlinkDeployment resources. - Plan, apply, verify and diff integration. - Checkpoint and catalog verification. - fluid generate flink. The operator security gate covers several things: - Allowlists for the kube context, namespace, image, service account and Secrets. - Listed hostnames for every destination the job contacts: the catalog, the token endpoint, S3, the Kafka brokers and the Schema Registry. - NAME@host pairing for catalogAuth. - A pinned-IP request guard. - Refusals: no env: refs in catalog text, an https-only OAuth2 endpoint, no SQL hints, and a contract-SQL allowlist. Hardened through stacked PRs #795, #809, #817, #819, #825 and #826, and syncs #815, #821 and #824. The guide is in forge_docs (advanced/managed-flink-iceberg). The remaining live proof is tracked on Trello.

fas89 added 17 commits November 4, 2026 05:35

github-actions Bot added provider Changes or requests related to providers ci Continuous integration and automation changes tests Test coverage or test infrastructure changes labels Nov 4, 2026

fas89 merged commit a89966a into feat/managed-flink-iceberg-streaming Nov 4, 2026
25 of 30 checks passed

This was referenced Nov 4, 2026

fix(streaming): refuse Iceberg partition transforms that Flink's DDL cannot create #819

Merged

chore(streaming): re-sync #784 with main after #802, #811, #813 and #814 #821

Merged

fas89 mentioned this pull request Nov 4, 2026

docs(streaming): move the managed Flink guide to forge_docs #825

Merged

Navigation Menu

Repository navigation

feat(streaming): declare every Kafka record shape in FLUID, checked before deploy - #817

feat(streaming): declare every Kafka record shape in FLUID, checked before deploy#817
fas89 merged 17 commits into
feat/managed-flink-iceberg-streamingfrom
feat/flink-kafka-record-shapes

fas89 commented Nov 4, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

fas89 commented Nov 4, 2026

What a contract says (FLUID 0.7.7 preview only)

Design and decisions

Fixes to the lane found on the way

Evidence

Not in this PR

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant