Eventstream exec: a real Kafka broker behind Fabric Spark options

Status: shipped, opt-in sidecar, graded Real. Eventstream item management always works. Execution — a real Kafka topic the Fabric notebook API can subscribe to — is real when an Apache Kafka KRaft broker is attached. No broker → honest 501. --profile eventstream stays opt-in. Sail is the default engine; the JVM overlay keeps a native spark-sql-kafka source. Neither path maps Eventstream onto rate. Destinations and operators sit on the Custom HTTP produce path: Lakehouse (Delta append), Reflex (item job), Eventhouse (Kusto direct ingest), and Filter / GroupBy / tumbling Window on the produce batch. Kafka stays the raw source; dests see operator output.

This is the same move as Eventhouse (25-rti-kusto.md): terminate the Fabric names ourselves, relay bytes to a real engine.

Why Apache Kafka, not Redpanda

The family rule is no BSL surprise. Redpanda was Apache-2.0 and then moved to the Business Source License; pinning it would put a license change under an opt-in profile the way an unpinned latest tag puts a binary change under a witness. apache/kafka is ASF, Apache License 2.0, KRaft (no ZooKeeper), and multi-arch (linux/amd64 + linux/arm64), so it runs natively on Apple silicon — unlike kustainer, which needs AVX2.

The image is digest-pinned in docker-compose.yml. Refresh with:

docker buildx imagetools inspect apache/kafka:3.9.1

The two Fabric surfaces (one product)

Fabric Eventstream is not “the item or the notebook API”. Both exist:

• Item: POST /v1/workspaces/{ws}/eventstreams — sources, streams, operators, destinations. This slice mints a DefaultStream datasourceId and a Kafka topic {itemId}.{datasourceId}.
• Notebook (notebook-with-event-stream):

df_raw = spark.readStream.format("kafka").options(**{
    "eventstream.itemid": item_id,
    "eventstream.datasourceid": datasource_id,
}).load()
# schema: key, value, topic, partition, offset, timestamp
df_raw.writeStream.foreachBatch(showDf).outputMode("append").start()

Microsoft’s adapter uses the notebook Entra token to resolve those IDs to a Kafka endpoint. User code has no kafka.bootstrap.servers. It is not rate (timestamp / value). Mapping Eventstream options onto rate would paint the engine matrix green with the wrong schema; that stays forbidden.

What the emulator does

notebook  --format kafka + eventstream.*-->  spark-agent
spark-agent  --GET /v1/eventstreams/{item}/sources/{ds}-->  fabric-emulator
  JVM:  rewrite to kafka.bootstrap.servers + subscribe → OSS kafka source
  Sail: GET …/sources/{ds}/events → LocalRelation (Kafka columns)
        foreachBatch runs in the agent (Sail cannot pickle UDFs)
notebook  --format kafka + bootstrap/subscribe/pattern/assign-->  spark-agent
  JVM:  spark-sql-kafka jar
  Sail: kafka-python consume → createDataFrame (bytes on Sail)
notebook  --write/writeStream format kafka-->  spark-agent
  JVM:  spark-sql-kafka jar
  Sail: kafka-python produce from collected rows
producer / Custom HTTP  -->  Kafka
POST …/eventstreams  -->  mint datasourceId + CreateTopics

1. Broker sidecar — apache/kafka:3.9.1, --profile eventstream.
2. Item create provisions a stream — DefaultStream id, topic name stored on the item, CreateTopics against the broker when one is attached. Unknown IDs fail at resolve (404), not as an empty stream.
3. Custom source — POST …/eventstreams/{id}/sources/{ds}/events writes JSON key/value bytes into that topic. Not thirty connectors. When a destination is bound, those same bytes (or the operator output) are drained after the produce succeeds.
4. Lakehouse destination — emulator-native POST …/eventstreams/{id}/destinations with {type, itemId, table}. Fabric’s topology JSON has no public REST (same as Reflex triggers). type is Lakehouse, Reflex, or Eventhouse.
5. Reflex destination — same bind surface, {type: Reflex, itemId}. A trigger on that Reflex with eventType: Microsoft.Fabric.Eventstream.EventReceived and source.itemId the Eventstream starts the action job per produced event (invokeType: EventTriggered, TriggerEvent.Key / .Value).
6. Eventhouse destination — {type: Eventhouse, itemId, table} and optional database. After produce, operator output is ingested with .create-merge + .ingest inline into the isolated engine database for that KQL Database — the path kustainer actually supports. This is not Fabric’s streaming-ingest protocol and not Kusto.Ingest. No engine attached → produce 502 naming --kql-url.
7. Operators — emulator-native POST …/eventstreams/{id}/operators. Filter (eq/ne/gt/gte/lt/lte/contains/exists), GroupBy (count/sum/min/max/avg), and tumbling Window (this batch only; stamps _window_start). Join / Union / Expand and hopping/sliding are refused by name. Kafka consume is still unfiltered.
8. Spark adapter, both engines — python/spark_agent/eventstream_kafka.py:
   • JVM: wraps classic readStream.format("kafka").load() when both eventstream options are set: Entra-gated lookup, then kafka.bootstrap.servers + subscribe on the real OSS Kafka source (spark-sql-kafka-0-10_2.12 matching Spark 3.5.5).
   • Sail: the engine has no Kafka source and rejects foreachBatch at start(). The wrap consumes through GET /v1/eventstreams/{item}/sources/{ds}/events, materialises a LocalRelation with the Kafka schema, and runs foreachBatch in the agent. One micro-batch, announced on stderr, no checkpoint — the same class of wrap as CDF / JSON multiLine. Native format("kafka") with kafka.bootstrap.servers plus subscribe / subscribePattern / assign is the same wrap without the Fabric IDs: driver consume → Kafka-schema LocalRelation on Sail. JSON offsets, includeHeaders, SASL PLAIN, GSSAPI, PEM SSL, and JKS/P12 truststores are honoured; a kafka sink produces from the driver. JVM native kafka still uses the jar.

GET the item to read properties.streams[0].id — that is the eventstream.datasourceid the notebook snippet needs.

Running it

Default engine (Sail behind Livy):

docker compose --profile eventstream \
  -f docker-compose.yml -f docker-compose.override.yml \
  -f docker-compose.eventstream.yml up

Or make up-eventstream. Two opt-ins, same grammar as RTI: the profile starts the broker; the overlay sets FABRIC_KAFKA_BOOTSTRAP. Without the overlay, create still returns an item and Spark resolution 501s.

Native Kafka source (checkpointed streaming, no LocalRelation) is the JVM overlay: add -f docker-compose.spark-jvm.yml to that command, or make up-jvm plus the eventstream profile and overlay.

The witness

e2e/eventstream runs the same notebook snippet on both engines:

• JVM — CI job eventstream in .github/workflows/spark-jvm.yml (weekly/manual). Real OSS kafka source + real foreachBatch.
• Sail — CI job eventstream-sail in .github/workflows/ci.yml. Emulator consume + LocalRelation + local foreachBatch.

Both create an Eventstream item, bind a Lakehouse destination, produce JSON records through the Custom HTTP source, assert the destination Tables/<name>/_delta_log commit (row count and field names), then run format("kafka") + eventstream options + foreachBatch and assert the Kafka schema (key, value, topic, partition, offset) and row count. A wrong item id must fail. rate must not appear.

Eventhouse destination and operators are Go-unit witnessed against the in-process Kusto stand-in and Lakehouse dest tables. They are not in e2e/eventstream — that job does not start --profile rti.

The one thing a stand-in cannot settle is whether the emitted KQL parses, and the names in it are where that bites: a source field called kind, or a destination bound with {"table": "kind"}, is a name real Kusto refuses bare with SYN0002. The emitter quotes both (['kind']) rather than deciding which names need it, and e2e/rti witnesses the emitted form against kustainer — 36 candidate keywords probed, the nine the engine actually refuses pinned as a set, all 36 legal quoted, and a quoted name shown to resolve to the same entity as the bare one, which is what makes quoting every drain’s table safe (25-rti-kusto.md).

Boundaries (deliberate, not backlog)

• Lakehouse destination is this slice: Custom HTTP produce → Delta append into Tables/<name>. Bind is emulator-native REST. Spark-native format("kafka") writes are not drained; there is no background consumer.
• Reflex destination is this slice: Custom HTTP produce → real item job via the existing Activator fire path. A FileCreated trigger on the same Reflex does not fire. No dest bound → no job, even if a stream trigger exists.
• Eventhouse destination is this slice: Custom HTTP produce → Kusto direct ingest (.create-merge + .ingest inline). Same produce trigger as Lakehouse; no background consumer. Fabric streaming ingest and queued Kusto.Ingest stay refused — the engine does not host them (25-rti-kusto.md).
• Operators are this slice: Filter, GroupBy, tumbling Window on the produce batch. Destinations see the operator output; Kafka / DefaultStream stays the raw source. Join / Union / Expand and hopping/sliding windows stay refused — they need more than one stream or cross-batch state.
• Sail format("kafka") is a driver consume/produce into a Kafka-schema LocalRelation (bytes on Sail). subscribe / subscribePattern / assign, JSON startingOffsets/endingOffsets, includeHeaders, SASL PLAIN, GSSAPI (JAAS keyTab → KRB5_CLIENT_KTNAME), PEM SSL, and JKS/P12 truststores (converted to PEM for kafka-python) are honoured. A kafka sink (write/writeStream.format("kafka")) produces from the driver. Checkpointed streaming (isStreaming, resume from checkpoint) stays on the JVM overlay.
• On Sail the result is materialised (.explain() is a LocalRelation; one micro-batch; isStreaming is false). Checkpointed streaming is the JVM overlay.
• Real-Time Hub explorer UX is not this slice.