60-Day Kafka 4 Learning Plan · Week 6 — Kafka Connect & ksqlDB Sources: Kafka: The Definitive Guide Ch.9 + 11 · kafka.apache.org/documentation/#connect · docs.ksqldb.io
Goal
Wire every Week 6 component into one production-style pipeline: PostgreSQL order changes flow through Kafka Connect into Kafka, ksqlDB filters confirmed orders into a dedicated topic, the Redis Sink publishes them to a Redis channel, and labs-socket broadcasts them to WebSocket clients — zero custom consumer code — then consolidate every hardening fix from Days 36–41 into one production-readiness pass over this exact pipeline.
Concept overview
Four different systems have to agree on one contract — a JSON payload representing a confirmed order — without any of them knowing the others’ implementation details.
Kafka Connect turns “poll a database” and “write to Redis” into declarative JSON configuration submitted to a REST API. A source connector produces records into Kafka; a sink connector consumes records out of Kafka. Neither requires a line of application code — connectors are plugins, workers are stateless JVM processes, and the REST API is how you create, inspect, and delete them.
ksqlDB compiles SQL into a Kafka Streams topology. CREATE STREAM ... AS SELECT (a CSAS) is not a view — it’s a persistent, continuously running query that reads a source topic and writes a derived topic, forever, until you TERMINATE it. That’s the mental model shift: SQL text becomes a long-lived process with its own consumer group, not a one-time report.
Redis Streams is an append-only log inside Redis, read via consumer groups — closer to a lightweight Kafka than to Pub/Sub. That’s not a stylistic choice this article made for effect: the current redis-kafka-connect sink connector has no Pub/Sub PUBLISH destination at all (its redis.type enum is HASH, JSON, TIMESERIES, STRING, STREAM, LIST, SET, ZSET — no PUBLISH), so Streams is the only way to get Kafka records into Redis with this connector, zero code. It turns out to be the better choice anyway: entries persist and are replayable, unlike Pub/Sub’s fire-and-forget delivery — a distinction the production-readiness section below treats as load-bearing, not a footnote.
The WebSocket bridge is the one place a small application is unavoidable. Nothing in the Connect or ksqlDB ecosystem terminates a browser WebSocket connection and multiplexes it across STOMP subscriptions — that’s stateful, per-client, long-lived TCP handling, which is a different problem than “move a record from A to B.” So the honest claim for this article is zero custom Kafka consumer code, not zero code, period.
What we’re building
PostgreSQL JDBC Source Kafka ksqlDB Redis Sink
(orders table) ───▶ (Connector) ───▶ (labs.orders) ───▶ (CSAS filter) ───▶ (Connector)
polls 5s 3 partitions WHERE status = XADD (STREAM)
'CONFIRMED' labs.confirmed-orders
│
▼
labs.confirmed-orders
(Kafka topic)
Redis labs-socket Browser
(Stream) ───▶ (Spring Boot) ───▶ (WebSocket / STOMP)
XREADGROUP one consumer /topic/orders
labs.confirmed-orders group per replica
Database modeling
The pipeline hinges on one table. Get its change-tracking columns wrong and everything downstream silently stops working — no exception, no log line, just a dashboard that never updates.
CREATE TABLE orders
(
id BIGSERIAL PRIMARY KEY,
order_id VARCHAR(36) NOT NULL UNIQUE,
user_id VARCHAR(36) NOT NULL,
amount NUMERIC(10, 2) NOT NULL CHECK (amount > 0),
status VARCHAR(20) NOT NULL DEFAULT 'PENDING'
CHECK (status IN ('PENDING', 'CONFIRMED', 'CANCELLED', 'REFUNDED')),
created_at TIMESTAMP NOT NULL DEFAULT now(),
updated_at TIMESTAMP NOT NULL DEFAULT now()
);
CREATE OR REPLACE FUNCTION set_updated_at()
RETURNS TRIGGER AS $$
BEGIN
NEW.updated_at = now();
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trg_orders_updated_at
BEFORE UPDATE ON orders
FOR EACH ROW EXECUTE FUNCTION set_updated_at();
CREATE INDEX idx_orders_updated_at ON orders (updated_at);
CREATE INDEX idx_orders_status ON orders (status);+----------------+
| orders |
+----------------+
| id (PK) |
| order_id (UK) |
| user_id |
| amount |
| status |
| created_at |
| updated_at |
+----------------+A single table, no foreign keys — this pipeline is deliberately scoped to one entity’s lifecycle. Three modeling decisions matter more than the schema looks like it deserves:
The updated_at trigger is not optional. The JDBC source connector runs in timestamp+incrementing mode, which polls with WHERE updated_at > ? OR (updated_at = ? AND id > ?). Postgres does not bump a column on UPDATE for you — without the trigger, an UPDATE orders SET status = 'CONFIRMED' never changes updated_at, the connector’s watermark never advances past it, and the row is invisible to Kafka forever. This is the single most common reason a “working” JDBC source connector silently stops picking up changes.
Both updated_at and id need indexes. The connector’s poll query filters and sorts on both every 5 seconds. Without indexes, a growing orders table turns each poll into a sequential scan — the connector still “works,” but DB load climbs with table size until someone notices Postgres CPU pinned at 100%.
JDBC polling cannot see deletes. A physical DELETE FROM orders WHERE id = ... leaves no row for the connector to poll — it just disappears from the source of truth with no corresponding Kafka event. If your domain needs delete propagation, either soft-delete (status = 'CANCELLED', which this schema already supports) or switch to log-based CDC (Debezium), covered in “Alternative Approaches.”
Seed data for local testing:
INSERT INTO orders (order_id, user_id, amount, status)
VALUES ('ord-1001', 'usr-501', 129.99, 'PENDING'),
('ord-1002', 'usr-502', 59.50, 'PENDING'),
('ord-1003', 'usr-501', 899.00, 'PENDING');Implementation
Step 0 — Docker Compose
The full stack: broker, Postgres, Connect, ksqlDB, Redis, and labs-socket. Kafka 4 dropped ZooKeeper entirely, so the broker runs KRaft in combined broker+controller mode using the official apache/kafka:4.0.0 image.
services:
broker:
image: apache/kafka:4.0.0
container_name: broker
hostname: broker
healthcheck:
test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server broker:9092 >/dev/null 2>&1"]
interval: 10s
timeout: 10s
retries: 10
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,PLAINTEXT_HOST://:9094
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:9094
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:9093
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
# Default log.dirs is /tmp/kraft-combined-logs -- fine for the image's
# own lifecycle, but pointing it at a named volume means topics,
# connector configs, and ksqlDB's command topic all survive a
# `docker compose up` after a config change, not just a plain restart.
KAFKA_LOG_DIRS: /var/lib/kafka/data
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
ports: ["9094:9094"]
volumes:
- broker-data:/var/lib/kafka/data
postgres:
image: postgres:16-alpine
container_name: postgres
healthcheck:
test: ["CMD-SHELL", "pg_isready -U labs -d labsdb"]
interval: 5s
timeout: 5s
retries: 10
environment:
POSTGRES_DB: labsdb
POSTGRES_USER: labs
POSTGRES_PASSWORD: secret
volumes:
- postgres-data:/var/lib/postgresql/data
- ./sql/schema.sql:/docker-entrypoint-initdb.d/1-schema.sql
- ./sql/seed.sql:/docker-entrypoint-initdb.d/2-seed.sql
ports: ["5433:5432"] # 5433 on the host -- many dev machines already have something on 5432
connect:
build: ./connect # Dockerfile bakes in the JDBC + Redis plugins, see below
container_name: connect
depends_on:
broker: { condition: service_healthy }
postgres: { condition: service_healthy }
ports: ["8083:8083"]
environment:
CONNECT_BOOTSTRAP_SERVERS: broker:9092
CONNECT_GROUP_ID: labs-connect-group
CONNECT_CONFIG_STORAGE_TOPIC: _connect-configs
CONNECT_OFFSET_STORAGE_TOPIC: _connect-offsets
CONNECT_STATUS_STORAGE_TOPIC: _connect-status
CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
CONNECT_REST_ADVERTISED_HOST_NAME: connect
CONNECT_PLUGIN_PATH: /usr/share/java,/usr/share/confluent-hub-components
# Required for jdbc-source.json's ${file:...} password indirection to
# actually resolve -- without a registered ConfigProvider, that string
# is passed to the JDBC driver literally and auth fails.
CONNECT_CONFIG_PROVIDERS: file
CONNECT_CONFIG_PROVIDERS_FILE_CLASS: org.apache.kafka.common.config.provider.FileConfigProvider
volumes:
- ./secrets/connect-secrets.properties:/opt/secrets/connect-secrets.properties
ksqldb:
image: confluentinc/ksqldb-server:0.29.0
container_name: ksqldb
depends_on: { broker: { condition: service_healthy } }
ports: ["8088:8088"]
environment:
KSQL_BOOTSTRAP_SERVERS: broker:9092
KSQL_LISTENERS: http://0.0.0.0:8088
KSQL_KSQL_SERVICE_ID: labs-ksqldb_
KSQL_KSQL_STREAMS_REPLICATION_FACTOR: 1
KSQL_KSQL_INTERNAL_TOPIC_REPLICAS: 1
ksqldb-cli:
image: confluentinc/ksqldb-cli:0.29.0
container_name: ksqldb-cli
depends_on: [ksqldb]
volumes: ["./ksql:/ksql"]
entrypoint: /bin/sh
tty: true
redis:
image: redis:7-alpine
container_name: redis
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 5s
retries: 10
command: redis-server --save "" --appendonly no
ports: ["6379:6379"]
labs-socket:
build: ./labs-socket
container_name: labs-socket
depends_on: { redis: { condition: service_healthy } }
ports: ["8090:8090"]
environment:
SPRING_DATA_REDIS_HOST: redis
SPRING_DATA_REDIS_PORT: 6379
volumes:
broker-data:
postgres-data:Three things in that file earn an explanation, each found the hard way while actually running this stack rather than just writing it down:
CONNECT_KEY_CONVERTER / CONNECT_INTERNAL_*_CONVERTER and the storage-topic replication factors are required for cp-kafka-connect to boot at all against a single-broker cluster — omit them and the worker fails at startup with a converter-not-found or under-replicated-topic error. Easy to drop when adapting a multi-broker production example into a single-node dev stack.
connect builds from a local Dockerfile instead of using the bare cp-kafka-connect image directly. A bare image has no connector plugins installed, and installing them at container startup via confluent-hub install (a common pattern in tutorials, including an earlier draft of this one) doesn’t survive a container recreate — which happens on every docker compose up after any config change, not just a full teardown. Bake plugins into the image at build time instead:
# connect/Dockerfile
FROM confluentinc/cp-kafka-connect:7.7.2
USER root
RUN confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:10.7.4
# redis-kafka-connect is not resolvable via `confluent-hub install <owner>/<name>:<version>` --
# it isn't published to the Confluent Hub registry under that coordinate scheme.
# It ships as a Confluent Hub-compatible zip on GitHub Releases instead;
# confluent-hub installs from a local archive path just as well.
ARG REDIS_KAFKA_CONNECT_VERSION=1.1.0
RUN curl -sL -o /tmp/redis-kafka-connect.zip \
"https://github.com/redis-field-engineering/redis-kafka-connect/releases/download/v${REDIS_KAFKA_CONNECT_VERSION}/redis-redis-kafka-connect-${REDIS_KAFKA_CONNECT_VERSION}.zip" \
&& confluent-hub install --no-prompt /tmp/redis-kafka-connect.zip \
&& rm /tmp/redis-kafka-connect.zip
USER appuserAnd, worth flagging honestly: Confluent has not certified cp-kafka-connect:7.7.2 or ksqldb-server:0.29.0 against an Apache Kafka 4.0 broker — both predate Kafka 4’s release. The standard Kafka client library’s backward-compatible wire protocol makes this combination work for the connectors used here (verified by actually running it end to end for this article), but validate it in a staging environment before treating it as a supported production combination, and revisit “Alternative Approaches” for actively-maintained options if that matters for your organization.
docker compose up -d --build
docker compose logs -f connect # wait for "Kafka Connect started"
Step 1 — JDBC source connector
{
"name": "labs-jdbc-source",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"connection.url": "jdbc:postgresql://postgres:5432/labsdb",
"connection.user": "labs",
"connection.password": "${file:/opt/secrets/connect-secrets.properties:postgres.password}",
"tasks.max": "1",
"mode": "timestamp+incrementing",
"timestamp.column.name": "updated_at",
"incrementing.column.name": "id",
"table.whitelist": "orders",
"topic.prefix": "labs.",
"poll.interval.ms": "5000",
"batch.max.rows": "500",
"numeric.mapping": "best_fit",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"errors.tolerance": "all",
"errors.log.enable": "true",
"errors.log.include.messages": "true"
}
}Two changes from a typical first-draft config, both worth explaining rather than just applying:
connection.password uses ${file:...} indirection into a secrets file mounted on the worker, not a plaintext string in the connector config. Connector configs are readable via the REST API (GET /connectors/labs-jdbc-source/config) by anyone who can reach port 8083 — a plaintext DB password there is one curl away from anyone on the network.
numeric.mapping is set to best_fit because the default JDBC source behavior maps NUMERIC(10,2) to Kafka Connect’s Decimal logical type, which serializes as Base64-encoded bytes in JSON — technically correct, unreadable to a human and to ksqlDB’s default JSON handling without extra schema plumbing. best_fit maps it to a plain DOUBLE instead, trading a sliver of precision for a payload every downstream consumer can read without a decoder.
Note what’s not here: dead-letter-queue configuration. errors.deadletterqueue.* is a sink connector feature — the DLQ receives records a sink connector failed to write to its external system. A source connector has no inbound Kafka records to fail on; it only has DB rows and converter errors, which errors.tolerance and errors.log.enable already cover. Configuring a DLQ on a source connector is a no-op at best and a validation error at worst, and it’s an easy copy-paste mistake to make when you’ve just finished hardening a sink connector and reach for the same block.
curl -s -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
-d @connectors/jdbc-source.json | jq .
curl -s http://localhost:8083/connectors/labs-jdbc-source/status | jq .Step 2 — ksqlDB: filter confirmed orders
SET 'auto.offset.reset' = 'earliest';
CREATE STREAM orders_stream (
id BIGINT,
order_id VARCHAR,
user_id VARCHAR,
amount DOUBLE,
status VARCHAR,
created_at VARCHAR,
updated_at VARCHAR
) WITH (
KAFKA_TOPIC = 'labs.orders',
VALUE_FORMAT = 'JSON'
);
CREATE STREAM confirmed_stream
WITH (
KAFKA_TOPIC = 'labs.confirmed-orders',
VALUE_FORMAT = 'JSON',
PARTITIONS = 3,
REPLICAS = 1
) AS
SELECT
order_id,
user_id,
amount,
UCASE(status) AS status,
updated_at,
ROWTIME AS event_ts
FROM orders_stream
WHERE status = 'CONFIRMED'
EMIT CHANGES;CREATE STREAM ... AS SELECT is a persistent query, not a report you run once. It starts a Kafka Streams application under the hood, with its own consumer group, that reads every row on labs.orders — pending, cancelled, refunded, all of it — and discards everything that doesn’t match WHERE status = 'CONFIRMED'. Size tasks.max on the source connector and partition counts for the full volume of order mutations, not the smaller confirmed-only volume that ends up downstream; the filtering happens after ingestion, not before.
Notice order_id is declared as a plain column, not VARCHAR KEY. It’s tempting to key the stream by order_id — it looks like the natural partitioning key — but the JDBC source connector here never sets a Kafka record key (key.converter is configured, but nothing populates a key, so every record key is null). Declaring order_id VARCHAR KEY on top of a topic whose physical keys are all null doesn’t extract order_id out of the JSON value into the key the way it looks like it should — it binds the stream’s key concept to that already-null physical key. SELECT order_id in the CSAS would then propagate a key column, and ksqlDB drops key columns from the JSON value entirely, so order_id would vanish from confirmed_stream‘s output — not deserialize wrong, just not be there. Downstream consumers would see every field except the one they probably need most. Keep it a plain value column unless you’ve also arranged for the source connector to populate a real key.
REPLICAS = 1 is a single-broker dev setting. A production Kafka cluster with 3+ brokers should use REPLICAS = 3 for the output topic, and KSQL_KSQL_STREAMS_REPLICATION_FACTOR for ksqlDB’s internal changelog/repartition topics — losing the broker holding the only replica of a ksqlDB internal topic loses in-flight aggregation state, not just data.
docker exec -it ksqldb-cli ksql http://ksqldb:8088 --file /ksql/pipeline.sqlStep 3 — Redis sink connector
The plan going into this section was redis.command: PUBLISH — a Pub/Sub channel, fed straight from the sink connector, zero code. It doesn’t work, and it’s worth showing why rather than skipping straight to the fix: PUBLISH isn’t a config value the connector recognizes at all. Confirm it yourself against a running worker before trusting any connector’s example config, this article’s included:
curl -s -X PUT http://localhost:8083/connector-plugins/com.redis.kafka.connect.RedisSinkConnector/config/validate \
-H "Content-Type: application/json" \
-d '{"connector.class":"com.redis.kafka.connect.RedisSinkConnector","name":"x","topics":"labs.confirmed-orders","redis.uri":"redis://redis:6379","redis.type":"PUBLISH","redis.keyspace":"${topic}"}'
# {"error_code":500,"message":"java.lang.IllegalArgumentException:
# No enum constant com.redis.kafka.connect.sink.RedisSinkConnector.RedisType.PUBLISH"}redis.command and redis.key — the config keys an earlier plan for this article used, copied from example configs that predate the connector’s current release — aren’t recognized keys either. Kafka Connect’s REST validation only flags missing required fields, not unrecognized extra ones, so a config with wrong key names doesn’t error at deploy time: it silently falls back to the connector’s actual defaults (redis.type=STREAM, redis.keyspace=${topic}) and writes to a Redis Stream instead of Pub/Sub, with no PUBLISH ever happening and no error anywhere in the logs. The failure mode is total silence downstream, not an exception — exactly the kind of bug a demo doesn’t catch until someone asks “why isn’t anything showing up.” The real, current config:
{
"name": "labs-redis-sink",
"config": {
"connector.class": "com.redis.kafka.connect.RedisSinkConnector",
"topics": "labs.confirmed-orders",
"tasks.max": "2",
"redis.uri": "redis://redis:6379",
"redis.type": "STREAM",
"redis.keyspace": "${topic}",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"errors.tolerance": "all",
"errors.log.enable": "true",
"errors.log.include.messages": "true",
"errors.deadletterqueue.topic.name": "labs.confirmed-orders.dlq",
"errors.deadletterqueue.topic.replication.factor": "1",
"errors.deadletterqueue.context.headers.enable": "true",
"errors.retry.timeout": "60000",
"errors.retry.delay.max.ms": "5000"
}
}redis.keyspace = "${topic}" maps the Kafka topic name directly to the Redis Stream key — labs.confirmed-orders in, XADD labs.confirmed-orders * ORDER_ID ord-1001 USER_ID usr-501 ... out. Each Kafka record’s JSON value fields become separate Stream entry fields, not a single JSON blob — that shapes how labs-socket has to read it in Step 4. DLQ configuration belongs here, unlike on the source connector: this connector consumes Kafka records and can fail to deliver them (Redis unreachable, auth failure), and errors.deadletterqueue.* routes those failures to labs.confirmed-orders.dlq instead of blocking the task or silently dropping records.
redis.uri is plaintext, no auth, matching the local Redis container’s default (no requirepass set). That’s acceptable for docker compose up on a laptop and nowhere else — see the production-readiness pass below.
curl -s -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
-d @connectors/redis-sink.json | jq .
docker exec redis redis-cli XRANGE labs.confirmed-orders - + COUNT 5Step 4 — labs-socket: Redis → WebSocket
This is the only application code in the pipeline. Four files.
WebSocketConfig.java — registers the STOMP endpoint and message broker:
package com.labs.socket.config;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
@Value("${labs.websocket.allowed-origins}")
private String[] allowedOrigins;
@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
registry.enableSimpleBroker("/topic");
registry.setApplicationDestinationPrefixes("/app");
}
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/ws")
.setAllowedOriginPatterns(allowedOrigins)
.withSockJS();
}
}allowedOrigins is externalized to config instead of hardcoded "*" — an open origin policy on a WebSocket endpoint lets any website embed a script that connects to your socket and reads live order data from a logged-in user’s browser session. Set it to your actual frontend origin(s) via LABS_WEBSOCKET_ALLOWED_ORIGINS.
RedisConfig.java — reads the Stream the sink connector writes to:
Redis Streams use consumer-group semantics: each entry is delivered to exactly one member of a group. That’s the right model for a work queue, and the wrong one for this service. labs-socket needs to broadcast — every replica must see every confirmed order so it can push it to its own locally connected browsers. If every replica shared one consumer group named, say, labs-socket, they’d compete for entries: replica A gets order 1, replica B gets order 2, and a client connected to replica B never hears about order 1 at all. The fix is to give each replica its own consumer group, so N replicas behave like N independent readers of the same Stream rather than N competing workers on the same queue:
package com.labs.socket.config;
import com.labs.socket.listener.OrderEventListener;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.Duration;
@Configuration
@Slf4j
public class RedisConfig {
private static final String STREAM_KEY = "labs.confirmed-orders";
private static final String CONSUMER_NAME = "labs-socket";
private final RedisConnectionFactory connectionFactory;
private final StringRedisTemplate redisTemplate;
private final OrderEventListener orderEventListener;
private final String consumerGroup;
public RedisConfig(RedisConnectionFactory connectionFactory,
StringRedisTemplate redisTemplate,
OrderEventListener orderEventListener) throws UnknownHostException {
this.connectionFactory = connectionFactory;
this.redisTemplate = redisTemplate;
this.orderEventListener = orderEventListener;
this.consumerGroup = "labs-socket-" + InetAddress.getLocalHost().getHostName();
}
@PostConstruct
void createConsumerGroup() {
try {
redisTemplate.opsForStream().createGroup(STREAM_KEY, ReadOffset.from("0"), consumerGroup);
} catch (RedisSystemException e) {
// BUSYGROUP: this instance's group already exists from a prior
// run against the same hostname -- expected, not an error.
if (e.getMessage() == null || !e.getMessage().contains("BUSYGROUP")) {
throw e;
}
}
}
@PreDestroy
void destroyConsumerGroup() {
try {
redisTemplate.opsForStream().destroyGroup(STREAM_KEY, consumerGroup);
} catch (Exception e) {
log.warn("Could not clean up consumer group {} on shutdown: {}", consumerGroup, e.getMessage());
}
}
@Bean(destroyMethod = "stop")
StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer() {
var options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
.builder()
.pollTimeout(Duration.ofSeconds(2))
.build();
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
StreamMessageListenerContainer.create(connectionFactory, options);
container.receiveAutoAck(
Consumer.from(consumerGroup, CONSUMER_NAME),
StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed()),
orderEventListener
);
container.start();
return container;
}
}The consumer group name is derived from the instance’s own hostname (the container ID in Docker, the pod name in Kubernetes) rather than a fixed string — that’s what makes N replicas fan out correctly instead of competing. createGroup(key, readOffset, group) passes Redis’s MKSTREAM option internally, so it’s safe to call even if the Stream doesn’t exist yet (a fresh deployment with no confirmed orders yet, for instance) — verified against the actual Spring Data Redis source, not assumed. ReadOffset.from("0") backfills everything already on the Stream at startup, which is a deliberate choice for this dashboard use case; a service that only cares about events from the moment it started should use ReadOffset.latest() at group-creation time instead. @PreDestroy tears the group down on graceful shutdown so Redis doesn’t accumulate one abandoned consumer group per container restart forever.
OrderEventListener.java — parses and broadcasts, without ever letting a bad entry take the listener down:
package com.labs.socket.listener;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.labs.socket.model.ConfirmedOrder;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class OrderEventListener implements StreamListener<String, MapRecord<String, String, String>> {
private static final String WEBSOCKET_DESTINATION = "/topic/orders";
private final SimpMessagingTemplate websocket;
private final ObjectMapper objectMapper;
private final Counter deliveredCounter;
private final Counter malformedCounter;
public OrderEventListener(SimpMessagingTemplate websocket, ObjectMapper objectMapper, MeterRegistry meterRegistry) {
this.websocket = websocket;
this.objectMapper = objectMapper;
this.deliveredCounter = Counter.builder("labs.socket.orders.delivered").register(meterRegistry);
this.malformedCounter = Counter.builder("labs.socket.orders.malformed").register(meterRegistry);
}
@Override
public void onMessage(MapRecord<String, String, String> record) {
ConfirmedOrder order;
try {
order = objectMapper.convertValue(record.getValue(), ConfirmedOrder.class);
} catch (IllegalArgumentException e) {
malformedCounter.increment();
log.warn("Discarding stream entry [{}]: {}", record.getId(), e.getMessage());
return;
}
log.debug("Redis stream [{}] entry {} -> WebSocket {} : order={}",
record.getStream(), record.getId(), WEBSOCKET_DESTINATION, order.orderId());
websocket.convertAndSend(WEBSOCKET_DESTINATION, order);
deliveredCounter.increment();
}
}A Stream entry arrives as Map<String, String> — each Kafka record’s JSON fields became separate hash fields via XADD, not one JSON blob — so ObjectMapper.convertValue maps it the same way Jackson would map a JSON object, coercing the numeric strings along the way. The try/catch matters more than it looks like it should: a connector restart mid-flush, a hand-added test entry, or a future SMT that reshapes the payload can all put something on this Stream that doesn’t map onto ConfirmedOrder. receiveAutoAck means a discarded entry won’t be redelivered either way, so failing loudly here would only lose it more dramatically, not more safely — catch, log, count, move on, and let the malformed counter be the signal something upstream needs attention.
The ConfirmedOrder model carries a detail that will bite you exactly once, silently, before you find it:
package com.labs.socket.model;
import com.fasterxml.jackson.annotation.JsonAlias;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import java.math.BigDecimal;
@JsonIgnoreProperties(ignoreUnknown = true)
public record ConfirmedOrder(
@JsonAlias("ORDER_ID") String orderId,
@JsonAlias("USER_ID") String userId,
@JsonAlias("AMOUNT") BigDecimal amount,
@JsonAlias("STATUS") String status,
@JsonAlias("UPDATED_AT") String updatedAt,
@JsonAlias("EVENT_TS") Long eventTs
) {
}ksqlDB uppercases unquoted column identifiers, so confirmed_stream‘s order_id column shows up as the field ORDER_ID on the Redis Stream entry, not order_id or orderId. The first version of this mapping used @JsonProperty("ORDER_ID") instead of @JsonAlias — it compiles, it looks right, and it’s wrong in a way that only shows up once you inspect the actual WebSocket frame: @JsonProperty is bidirectional. It fixes deserialization (accepting ORDER_ID on the way in) but also forces ORDER_ID back onto serialization (the way out), so the browser would receive {"ORDER_ID": "...", "USER_ID": "...", ...} instead of clean camelCase — ksqlDB’s naming convention leaking all the way to the frontend. @JsonAlias is read-only: it accepts ORDER_ID as an alternate name on input while serialization still uses the record component’s own name, orderId, on output. Verified against a real WebSocket frame, not just a passing test — the difference between “a message arrived” and “the right message arrived” is exactly the field names in that frame.
websocket.convertAndSend(WEBSOCKET_DESTINATION, order) sends the typed record, not the raw Redis payload — Spring’s Jackson message converter serializes it to clean camelCase JSON for the browser:
{"orderId":"ord-6001","userId":"usr-111","amount":88.88,"status":"CONFIRMED","updatedAt":"1786528417605","eventTs":1786528418282}Browser client
const client = new StompJs.Client({
webSocketFactory: () => new SockJS('/ws'),
reconnectDelay: 5000,
onConnect: () => {
client.subscribe('/topic/orders', (message) => {
const order = JSON.parse(message.body);
console.log('Confirmed order:', order.orderId, order.amount);
updateOrdersTable(order);
});
},
});
client.activate();Production-readiness pass
This is where Days 36–41’s hardening lessons apply directly to this pipeline, joint by joint.
Connect REST API (Day 36). Port 8083 has no authentication by default and exposes every connector’s configuration — including any secret that isn’t behind ${file:...} indirection — to anyone who can reach it. Put it behind a network boundary (internal-only, VPN, or service mesh), and enable org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension at minimum. The same applies to ksqlDB’s REST API on 8088.
Polling vs. CDC (Day 37). timestamp+incrementing mode is the simplest connector to reason about, but it polls every 5 seconds regardless of whether anything changed, can’t see deletes, and depends entirely on the updated_at trigger discussed earlier. Debezium’s log-based CDC reads the Postgres WAL directly — it sees every insert, update, and delete with no polling overhead and no trigger dependency, at the cost of enabling logical replication on the database and a heavier connector. If delete propagation or sub-second latency matters for your domain, that tradeoff is usually worth making; see “Alternative Approaches.”
Retry, DLQ, SMT (Day 38). Applied above: the Redis sink connector has errors.tolerance=all, a DLQ topic, and bounded retry with backoff. Verify the DLQ actually receives something by intentionally breaking the connector (stop Redis, watch labs.confirmed-orders.dlq fill) before trusting it in an incident.
Redis’s actual delivery guarantee here (Day 39). Day 39 covered Pub/Sub, but the connector that’s actually available writes to a Stream instead — which turns out to be the better guarantee, not a downgrade. Stream entries persist until trimmed and a consumer group tracks exactly which entries it has and hasn’t acknowledged, so a labs-socket instance that’s down for a minute during a deploy picks up exactly where it left off on restart (ReadOffset.from("0") at group creation, then ReadOffset.lastConsumed() on every poll after). That’s strictly stronger than Pub/Sub’s fire-and-forget model, where a down subscriber simply misses whatever was published while it was down, permanently. The tradeoff moved elsewhere: get the consumer-group topology wrong (one shared group instead of one per replica, covered in Step 4) and you get silent partial delivery instead of guaranteed loss — arguably a worse failure mode, because it looks like it’s working for whichever replica happens to receive each entry.
ksqlDB operational notes (Days 40–41). Persistent queries resume automatically from their last committed offset when a ksqlDB server restarts — you don’t lose the filter logic, but you should still monitor query state with SHOW QUERIES and alert on a query dropping into ERROR. This pipeline doesn’t use windowing, but if you extend it toward aggregations, revisit GRACE PERIOD semantics from Day 41 before trusting any windowed output as final.
Everything above, plus:
labs-socketruns withspring.threads.virtual.enabled: true(Java 21 virtual threads) so a large number of concurrent WebSocket connections stays cheap on threads.server.shutdown: gracefulandspring.lifecycle.timeout-per-shutdown-phase: 20slet in-flight WebSocket sends finish before a rolling deploy kills the pod.- Actuator exposes
health,metrics, andprometheus, including thelabs.socket.orders.delivered/labs.socket.orders.malformedcounters fromOrderEventListener— the fastest signal that the Redis→WebSocket hop stopped working isdeliveredflatlining while the Redis sink connector’s own metrics show it’s still publishing. - Redis itself: the demo config has no
requirepassand no TLS. Production Redis needs both, plusredis.urion both connector andlabs-socketsides updated torediss://:password@host:port. - The Stream has no trimming configured, so
labs.confirmed-ordersgrows without bound in Redis memory forever. Checking the connector’s full config list (PUT /connector-plugins/.../config/validatewith a minimal body, same technique used above to catch thePUBLISHmistake) turns up noMAXLEN-equivalent option in this release at all — the closest isredis.key.ttl, which expires the entire Stream key after N seconds of no writes, not a sliding window during active use. Production needs an external scheduledXTRIM labs.confirmed-orders MAXLEN ~ 100000(or similar) run separately; the connector won’t do it for you.
Testing
Two levels: the connectors, and the one piece of custom code.
Connector-level: verify each connector reached RUNNING state, and that its task isn’t silently failing:
curl -s http://localhost:8083/connectors/labs-jdbc-source/status | jq '.connector.state, .tasks[].state'
curl -s http://localhost:8083/connectors/labs-redis-sink/status | jq '.connector.state, .tasks[].state'Pipeline-level: confirm.sh flips one seeded order to CONFIRMED and you watch it propagate:
./confirm.sh ord-1001
docker exec redis redis-cli XRANGE labs.confirmed-orders - + COUNT 5
Application-level: an integration test that proves the Redis→WebSocket bridge works without standing up the entire Kafka stack — Testcontainers gives a real Redis, a real embedded Spring Boot server, and a real STOMP client:
@Testcontainers
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class OrderBroadcastIntegrationTest {
@Container
static GenericContainer<?> redis = new GenericContainer<>(DockerImageName.parse("redis:7-alpine"))
.withExposedPorts(6379);
@DynamicPropertySource
static void redisProperties(DynamicPropertyRegistry registry) {
registry.add("spring.data.redis.host", redis::getHost);
registry.add("spring.data.redis.port", () -> redis.getMappedPort(6379));
}
@Test
void confirmedOrderPublishedToRedisArrivesOnWebSocketTopic() throws Exception {
// subscribe a STOMP client to /topic/orders, PUBLISH a payload
// shaped like the Redis Sink connector's actual output, assert
// it arrives deserialized. Full listing in the demo repository.
}
}This test is what catches the Jackson case-sensitivity bug described above in CI, before it reaches a demo or a production dashboard — publish a payload with uppercase keys exactly as ksqlDB would produce it, and assert the received ConfirmedOrder fields are non-null, not just that a message arrived.
Performance considerations
JDBC poll interval vs. database load. Halving poll.interval.ms doubles query frequency against orders. At 5 seconds and a properly indexed table, this is negligible; at 500ms on an unindexed table under write load, it competes with your application’s own traffic. If you need sub-second propagation, that’s a strong signal to switch to Debezium’s log-based CDC rather than polling harder.
ksqlDB throughput scales with source partitions. The CSAS’s internal Kafka Streams topology parallelizes across labs.orders‘s partition count — 3 partitions caps you at 3-way parallelism for the filter step regardless of how much traffic arrives. Size partitions for projected peak order-mutation volume, not confirmed-order volume.
Redis Pub/Sub has no backpressure. Redis pushes to subscribers as fast as it can; a slow consumer doesn’t throttle the publisher, it just accumulates in that client’s output buffer until client-output-buffer-limit pubsub kicks in and Redis forcibly disconnects it. labs-socket‘s listener does synchronous, non-blocking work per message (parse + convertAndSend), so it should never be the slow consumer in practice — but it’s worth knowing what happens if it ever is.
Horizontal scaling of labs-socket. Redis Pub/Sub fans out to every subscriber, so running N replicas of labs-socket means all N receive every message and each broadcasts to its own locally connected WebSocket clients — correct behavior, no deduplication needed. The requirement that comes with it: your load balancer needs WebSocket support and sticky sessions (or SockJS’s fallback transports), since a client’s long-lived connection has to stay pinned to the instance it opened it on.
End-to-end latency budget
Each hop adds its own latency, and the pipeline’s real-world responsiveness is the sum, not just whichever single hop looks fastest in isolation.
PostgreSQL write
→ up to poll.interval.ms (5000ms worst case) → JDBC Source picks it up
→ Kafka produce/consume (~5-20ms typical) → ksqlDB CSAS processes it
→ Kafka produce/consume (~5-20ms typical) → Redis Sink connector picks it up
→ Redis PUBLISH (sub-millisecond) → labs-socket receives it
→ WebSocket broadcast (~1-5ms typical) → browser rendersThe JDBC polling interval dominates this entire budget. Every other hop combined is on the order of tens of milliseconds;
poll.interval.ms: 5000alone means a real order confirmation can take up to 5 seconds to appear in the browser, regardless of how fast every downstream component is. If sub-second end-to-end latency actually matters for this use case, the fix is upstream (Debezium’s WAL streaming, Day 37 §6, cuts this dominant term to sub-second) — optimizing Redis or WebSocket configuration further would be solving the wrong part of the budget entirely.
Common pitfalls
- Trusting an example connector config over the installed plugin’s actual schema —
redis.command/redis.key(a Pub/Sub-era config) aren’t recognized keys by the currentredis-kafka-connectrelease, and Kafka Connect’s REST validation doesn’t flag unknown keys, only missing required ones. The connector silently falls back to its real defaults (redis.type=STREAM) instead of erroring. Always confirm withPUT /connector-plugins/<class>/config/validateagainst the worker you’re actually running before trusting any config, including this article’s. - Declaring a ksqlDB stream’s key column over a topic with null physical keys —
order_id VARCHAR KEYon a topic the JDBC source connector never keyed looks harmless and silently dropsorder_idfrom every downstream JSON value instead of erroring. - Missing
updated_attrigger — the JDBC source connector’s watermark never advances past anUPDATEthat doesn’t touch the timestamp column; rows go dark with no error anywhere. - Assuming JDBC polling sees deletes — it structurally can’t. Soft-delete or switch to CDC.
- DLQ config on a source connector —
errors.deadletterqueue.*only applies to sinks; on a source it’s dead configuration, not a safety net. - One shared Redis Streams consumer group across all
labs-socketreplicas — turns a broadcast fan-out into a competing-consumer work queue; each replica’s WebSocket clients only see a fraction of confirmed orders, with no error to indicate why. @JsonPropertywhere@JsonAliaswas needed —@JsonProperty("ORDER_ID")fixes deserializing ksqlDB’s uppercase keys but also forcesORDER_IDback onto the outbound WebSocket JSON, leaking an internal naming convention to the browser. Only visible by inspecting an actual frame, not by checking whether a message arrived.- Deleting and recreating a Kafka topic a connector is actively consuming — the connector’s consumer group can keep stale partition-count metadata and committed offsets that no longer correspond to real data, silently under-consuming (or missing) records until something forces a full rebalance. Restart the connector’s task, don’t just pause/resume, after recreating a topic underneath it.
- Exposing Connect/ksqlDB REST APIs without auth — both default to no authentication and readable configs, including any secret not routed through
${file:...}
Week 6 complete — what you built
Key Takeaways
- Kafka Connect and ksqlDB eliminate consumer code for every hop except one: nothing off-the-shelf terminates a browser WebSocket connection, so a small bridge application is the honest, unavoidable exception to “zero code.”
- The plan for this pipeline was Redis Pub/Sub; the connector that actually ships doesn’t support it. Validate a dependency’s real config schema before designing around it — an unrecognized config key doesn’t error, it silently falls back to a default and fails hundreds of miles downstream, in this case as a completely silent WebSocket that never receives anything.
- Most of what makes a pipeline like this production-ready lives at its boundaries: a database trigger the connector depends on, DLQ configuration on the right connector type, a consumer-group topology that matches broadcast semantics instead of competing-consumer semantics, and a JSON field-mapping contract that fails silently in either direction if you get the annotation wrong.
The complete source code is available on GitHub.
Support me through GitHub Sponsors.
Thank you for Reading !! See you in the next post.
Next
➡️ Week 7: Security & Monitoring (Day 43–49)
Resources
- 📘 Kafka: The Definitive Guide — Chapters 9 + 11
- 🌐 kafka.apache.org/documentation/#connect
- 🌐 docs.ksqldb.io
