60-Day Kafka 4 Learning Plan · Week 6 — Day 40 of 60

---

60-Day Kafka 4 Learning Plan · Week 6 — Kafka Connect & ksqlDB Sources: Kafka: The Definitive Guide Ch.11 · docs.ksqldb.io

Goal

Write SQL that runs continuously on Kafka topics — no Java, no Kafka Streams DSL. Learn the three core ksqlDB primitives (STREAM, TABLE, QUERY), deploy ksqlDB locally with Docker Compose, build a real pipeline that filters confirmed orders into a dedicated Kafka topic, and understand that ksqlDB is Kafka Streams underneath — inheriting all of Week 5’s operational concerns.

1. What is ksqlDB?

ksqlDB is a SQL engine that runs continuously on Kafka topics. You write SQL statements; ksqlDB compiles them into Kafka Streams topologies and runs them on the server.

Why ksqlDB?

• No Java required — just SQL
• Runs on top of Kafka Streams internally — all state stored in Kafka
• REST API + CLI for interactive development
• Integrates with Kafka Connect for source/sink pipelines

Set expectations up front: everything from Week 5 — RocksDB state stores, changelog topics, repartition topics, rebalancing, standby replicas — is happening under the hood of every CREATE TABLE/CSAS/CTAS statement here. ksqlDB removes the need to write Kafka Streams Java code, but not the need to understand what a stateful, partitioned, rebalancing Streams application actually does operationally. §8 makes this concrete.

2. Docker Compose — run ksqlDB locally

# docker-compose.yml — add ksqlDB server and CLI
services:
  ksqldb-server:
    image: confluentinc/ksqldb-server:0.29.0
    depends_on: [broker]
    ports:
      - "8088:8088"
    environment:
      KSQL_BOOTSTRAP_SERVERS: broker:9092
      KSQL_LISTENERS: http://0.0.0.0:8088
      KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: "true"
      KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: "true"

  ksqldb-cli:
    image: confluentinc/ksqldb-cli:0.29.0
    depends_on: [ksqldb-server]
    entrypoint: /bin/sh
    tty: true

Connect to the CLI

# Open ksql interactive shell
docker exec -it ksqldb-cli ksql http://ksqldb-server:8088

# Or run a single statement
docker exec ksqldb-cli ksql http://ksqldb-server:8088 --execute "SHOW STREAMS;"

ksqlDB REST API

# List all streams
curl http://localhost:8088/ksql \
  -H "Content-Type: application/vnd.ksql.v1+json" \
  -d '{"ksql": "SHOW STREAMS;"}'

# Execute a statement
curl http://localhost:8088/ksql \
  -H "Content-Type: application/vnd.ksql.v1+json" \
  -d '{"ksql": "CREATE STREAM orders_stream (order_id VARCHAR, amount DOUBLE) WITH (KAFKA_TOPIC='\''labs.events'\'', VALUE_FORMAT='\''JSON'\'');"}'

3. CREATE STREAM — map a topic to SQL

A STREAM is a named view over a Kafka topic. Define the schema and ksqlDB reads records as typed rows.

-- Connect labs.events Kafka topic to a typed SQL stream
CREATE STREAM orders_stream (
  order_id   VARCHAR,
  user_id    VARCHAR,
  amount     DOUBLE,
  status     VARCHAR,
  created_at VARCHAR
)
WITH (
  KAFKA_TOPIC   = 'labs.events',
  VALUE_FORMAT  = 'JSON',
  PARTITIONS    = 3
);

The PARTITIONS = 3 clause here is misleading as written. labs.events already exists as a topic from earlier days’ material (Day 12 onward) — when CREATE STREAM points at an existing topic, the PARTITIONS (and REPLICAS) settings in the WITH clause are silently ignored; they only take effect when ksqlDB is auto-creating a new backing topic that doesn’t exist yet. If you actually need labs.events to have 3 partitions, that has to be set when the topic itself is created (Day 12, kafka-topics.sh --create or a NewTopic bean), not via this CREATE STREAM statement. Leaving PARTITIONS in the WITH clause for an existing topic isn’t harmful, but it can mislead someone reading this SQL into thinking it’s controlling something it isn’t.

Inspect the stream

-- Show all defined streams
SHOW STREAMS;

-- Describe the schema
DESCRIBE orders_stream;

-- Read from beginning (dev only)
SET 'auto.offset.reset' = 'earliest';
SELECT * FROM orders_stream EMIT CHANGES LIMIT 10;

4. Push query — continuous SELECT EMIT CHANGES

A push query runs forever and emits a row for every matching Kafka record. Like subscribing to a filtered stream.

-- Stream all CONFIRMED orders over $100 in real time
SELECT order_id,
       user_id,
       amount,
       status
FROM   orders_stream
WHERE  status = 'CONFIRMED'
  AND  amount > 100
EMIT CHANGES;

Output streams continuously until you cancel (Ctrl+C):

+------------------+-----------+--------+-----------+
| ORDER_ID         | USER_ID   | AMOUNT | STATUS    |
+------------------+-----------+--------+-----------+
| ord-8821         | u-042     | 149.99 | CONFIRMED |
| ord-8834         | u-107     | 299.00 | CONFIRMED |
...

Use cases: real-time dashboards, alerting, driving downstream consumers.

5. CREATE TABLE + pull query

A TABLE aggregates the stream into a materialized view — one row per key, updated as events arrive.

-- Materialized table: order count + total spend per user
CREATE TABLE user_order_stats AS
  SELECT user_id,
         COUNT(*)    AS total_orders,
         SUM(amount) AS total_spend
  FROM   orders_stream
  WHERE  status = 'CONFIRMED'
  GROUP BY user_id
EMIT CHANGES;

A pull query reads the current state once — like a SQL SELECT on a database:

-- Point-in-time lookup for a specific user (no EMIT CHANGES)
SELECT total_orders, total_spend
FROM   user_order_stats
WHERE  user_id = 'u-123';

+---------------+-------------+
| TOTAL_ORDERS  | TOTAL_SPEND |
+---------------+-------------+
| 14            | 2847.50     |
+---------------+-------------+

Pull query requirements: the source must be a TABLE (not a STREAM), and you must filter by the primary key.

6. CSAS — CREATE STREAM AS SELECT

CSAS (Create Stream As Select) writes query results to a new Kafka topic — persistent, continuous stream processing.

-- Filter and transform confirmed orders into a dedicated topic
CREATE STREAM confirmed_orders_stream
  WITH (
    KAFKA_TOPIC  = 'labs.confirmed-orders',
    VALUE_FORMAT = 'JSON',
    PARTITIONS   = 3
  )
AS SELECT
  order_id,
  user_id,
  amount,
  UCASE(status)              AS status,
  ROWTIME                    AS event_ts
FROM   orders_stream
WHERE  status = 'CONFIRMED'
EMIT CHANGES;

This creates:

1. A persistent ksqlDB STREAM named confirmed_orders_stream
2. A new Kafka topic labs.confirmed-orders receiving all matching records
3. A running Kafka Streams topology that keeps filtering events forever

Unlike §3, PARTITIONS = 3 here genuinely takes effect — labs.confirmed-orders doesn’t exist yet, so ksqlDB creates it fresh with the specified partition count. This is the distinction from §3’s warning: the WITH clause’s topic-creation settings only matter for topics ksqlDB is creating, silently no-op for ones that already exist.

Verify the output topic

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic labs.confirmed-orders \
  --from-beginning

7. Push vs Pull queries

8. ksqlDB is Kafka Streams underneath — every Week 5 concern applies

Every CREATE TABLE, CSAS, or CTAS statement spins up a genuine Kafka Streams application internally. This isn’t an abstraction leak to worry about occasionally — it’s the actual execution model, and it means:

Practical implication: disk sizing (Day 29 §8, Day 34 §8’s RocksDB tuning), monitoring consumer group lag for KSQL_KSQL_SERVICE_ID-prefixed groups (Day 6 §7), and standby replica configuration (§10 below) are all real operational concerns for a ksqlDB deployment — SQL syntax doesn’t make the underlying Kafka Streams machinery go away, it just hides the Java code that would otherwise express it.

9. Pull query high availability — standby replicas

Exactly as covered for raw Kafka Streams interactive queries in Day 34 §7, a pull query against a TABLE can hit an availability gap during a rebalance if there’s no standby replica holding a warm copy of the relevant state.

-- Server-level config (ksqldb-server environment)
KSQL_KSQL_STREAMS_NUM_STANDBY_REPLICAS: 1

Same trade-off as Day 34 §7, just configured via ksqlDB’s server properties instead of spring.kafka.streams.properties: standby replicas cost extra disk/network in exchange for pull queries staying available (from a slightly stale replica) during a rebalance, rather than returning an error or NOT_AVAILABLE-style response while the active copy is mid-transfer. For a user_order_stats-style TABLE backing a real-time REST lookup, this is worth setting explicitly rather than accepting the zero-standby default.

10. Persistent query resource cost

Every CSAS/CTAS statement (§6) is a separate, independently-running Kafka Streams application for as long as it’s not TERMINATEd — accumulating several of these on one ksqlDB cluster has real resource cost, not just SQL-statement bookkeeping.

-- Each of these is its own persistent Kafka Streams topology, running forever:
CREATE STREAM confirmed_orders_stream AS SELECT ... FROM orders_stream WHERE status = 'CONFIRMED' EMIT CHANGES;
CREATE STREAM high_value_orders AS SELECT ... FROM orders_stream WHERE amount > 500 EMIT CHANGES;
CREATE TABLE user_order_stats AS SELECT ... FROM orders_stream GROUP BY user_id EMIT CHANGES;

-- List what's actually running and consuming resources
SHOW QUERIES;

-- Stop one that's no longer needed — otherwise it runs (and consumes CPU/memory/disk) forever
TERMINATE query_id;-- List what's actually running and consuming resources
SHOW QUERIES;

Treat abandoned persistent queries as a real cleanup task, not a “just leave it running” default. An exploratory CSAS created during development and forgotten about keeps consuming broker bandwidth, ksqlDB server CPU/memory, and creating its own internal changelog/repartition topics (§8) indefinitely — SHOW QUERIES periodically to audit what’s actually still needed is a reasonable operational habit, the SQL equivalent of checking for orphaned Kafka Streams app instances.

11. Securing the ksqlDB REST API

Port 8088 (§2) has no authentication in this setup — the same pattern flagged for Kafka Connect’s REST API (Day 36 §11) and Schema Registry (Day 23 §9) applies here too, and arguably with higher stakes: the ksqlDB REST API can execute arbitrary SQL, including DROP STREAM ... DELETE TOPIC, against production topics.

KSQL_AUTHENTICATION_METHOD: BASIC
KSQL_AUTHENTICATION_ROLES: ksqldb-user
KSQL_AUTHENTICATION_REALM: KsqlServerProps

Higher stakes than a typical unauthenticated REST API: unlike Connect’s REST API (which manages connector configuration) or Schema Registry (which manages schema metadata), an unauthenticated ksqlDB endpoint lets anyone who reaches it run DROP STREAM ... DELETE TOPIC or CREATE STREAM ... AS SELECT against real production topics — direct data-plane impact, not just configuration risk. The same minimum bar (authentication + network restriction) applies, with less room for “it’s probably fine for now.”

12 Common pitfalls

• Assuming WITH (PARTITIONS = N) controls an existing topic’s partition count — it’s silently ignored for topics that already exist; only affects topics ksqlDB is creating fresh (§3, contrasted with §6)
• Forgetting that scaling ksqlDB is a Kafka Streams rebalance — pull queries can hit the same availability gap covered for raw interactive queries in Day 34 §9, and standby replicas need the same deliberate configuration (§9)
• Leaving exploratory persistent queries running indefinitely — each one is a real, resource-consuming Kafka Streams app; SHOW QUERIES and TERMINATE unused ones as a routine cleanup habit (§10)
• No authentication on the ksqlDB REST API — higher stakes than Connect’s or Schema Registry’s REST APIs, since ksqlDB SQL can directly drop topics and mutate data-plane state (§11)
• Treating “no Java required” as “no Kafka Streams operational concerns” — the SQL abstraction doesn’t remove the need to think about state store sizing, changelog topics, or rebalancing (§8)

Useful ksqlDB commands

-- Show all objects
SHOW STREAMS;
SHOW TABLES;
SHOW QUERIES;

-- Describe schema
DESCRIBE orders_stream EXTENDED;

-- Terminate a running push query
TERMINATE query_id;

-- Drop a stream (and optionally its underlying topic)
DROP STREAM IF EXISTS orders_stream DELETE TOPIC;

-- Explain a query plan
EXPLAIN SELECT * FROM orders_stream EMIT CHANGES;

Key Takeaways

• ksqlDB = SQL on Kafka — no Java, no Kafka Streams DSL required, but the underlying execution model IS Kafka Streams
• STREAM = append-only (KStream); TABLE = materialized latest state (KTable)
• Push query (EMIT CHANGES) runs forever — perfect for dashboards and alerts
• Pull query runs once — read current TABLE state like a database lookup, and can hit the same rebalance-availability gap as raw Kafka Streams interactive queries (Day 34 §9)
• CSAS (CREATE STREAM AS SELECT) writes filtered output to a new Kafka topic — and the WITH clause’s PARTITIONS setting only takes effect for genuinely new topics, not existing ones
• Every persistent query (CSAS/CTAS) is its own running Kafka Streams application — audit with SHOW QUERIES and TERMINATE what’s no longer needed
• KSQL_KSQL_SERVICE_ID is effectively the application-id — same consumer-group semantics, rebalancing, and standby-replica configuration as raw Kafka Streams
• The REST API needs authentication in anything beyond local dev — it can execute arbitrary SQL including topic-destructive statements

---

Support me through GitHub Sponsors.

Thank you for Reading !! See you in the next post.

Next

➡️ Day 41: ksqlDB advanced — windowed aggregations in SQL

Resources

• 📘 Kafka: The Definitive Guide — Chapter 11 (ksqlDB)
• 🌐 docs.ksqldb.io — ksqlDB reference
• 🌐 ksqlDB — High Availability for Pull Queries

Link to Medium blog