Kafka

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

Editorial Team

60-Day Kafka 4 Learning Plan · Week 6 — Kafka Connect & ksqlDB Sources: Kafka: The Definitive Guide Ch.9 · redis.io/docs/manual/pubsub · github.com/redis-field-engineering/redis-kafka-connect

Goal

Bridge Kafka topics to Redis Pub/Sub channels using the Redis Sink connector, then subscribe in labs-socket to broadcast events to WebSocket clients in real time — without any custom consumer code on the Kafka side — and understand the message-loss trade-off this architecture accepts in exchange for simple multi-instance fan-out.

1. End-to-end pipeline

labs.events (Kafka topic)
  → Redis Sink Connector
  → PUBLISH labs.events (Redis channel)
  → labs-socket (SUBSCRIBE labs.*)
  → WebSocket broadcast → browser clients

Why this architecture?

  • Kafka holds the durable, replayable event log
  • Redis Pub/Sub is the low-latency fan-out layer between Kafka and WebSocket servers
  • labs-socket instances all receive every event — no sticky sessions needed
  • Multiple labs-socket pods scale horizontally, all subscribing to the same Redis channel

This is the production fix for the exact gap flagged in Day 20 §7. There, a single-instance SimpMessagingTemplate broker meant scaling labs-socket horizontally silently dropped messages for users connected to a different instance than the one that consumed their Kafka partition. Day 20 solved it with a STOMP broker relay (RabbitMQ); this architecture solves the same problem with Redis Pub/Sub instead — every labs-socket instance subscribes to the same channels, so it no longer matters which instance holds a given user’s WebSocket connection.

2. Why Redis Pub/Sub alongside Kafka?

Pattern: Kafka is the source of truth. Redis is the last-mile delivery bus to WebSocket servers.

3. Redis Sink connector config

Install the plugin

docker exec connect confluent-hub install redis-field-engineering/redis-kafka-connect:latest --no-prompt
docker-compose restart connect

Deploy the connector

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "labs-redis-sink",
    "config": {
      "connector.class": "com.redis.kafka.connect.RedisSinkConnector",
      "topics": "labs.events,labs.notifications,labs.order-stats",
      "redis.uri": "redis://redis:6379",
      "redis.command": "PUBLISH",
      "redis.key": "${topic}",
      "tasks.max": "2",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter.schemas.enable": "false",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter"
    }
  }'

Key config properties

Redis commands reference

# Verify connector is working — subscribe manually
redis-cli SUBSCRIBE labs.events

# Check how many subscribers are on each channel
redis-cli PUBSUB NUMSUB labs.events labs.notifications labs.order-stats

# Monitor all Pub/Sub traffic (dev only)
redis-cli PSUBSCRIBE "labs.*"

4. Topic → Redis channel mapping

With redis.key: "${topic}", each Kafka topic maps 1:1 to a Redis channel:

Use PatternTopic("labs.*") in Spring to subscribe to all current and future labs.* channels with one registration.

5. labs-socket: subscribe Redis → broadcast WebSocket

Maven dependency

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

Redis subscriber config

// RedisSubscriberConfig.java — labs-socket Spring Boot 3
@Configuration
@RequiredArgsConstructor
public class RedisSubscriberConfig {

    private final RedisConnectionFactory factory;
    private final SimpMessagingTemplate websocket;
    private final ObjectMapper objectMapper;

    @Bean
    RedisMessageListenerContainer container() {
        var container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);

        // Subscribe to all labs.* channels with one pattern
        container.addMessageListener(
            orderListener(),
            new PatternTopic("labs.*")
        );
        return container;
    }

    @Bean
    MessageListenerAdapter orderListener() {
        return new MessageListenerAdapter(
            (MessageListener) (message, pattern) -> {
                String channel = new String(message.getChannel());   // e.g. "labs.events"
                String body    = new String(message.getBody());

                // Map channel → WebSocket topic
                String wsTopic = "/topic/" + channel.replace("labs.", "");
                // labs.events → /topic/events
                // labs.notifications → /topic/notifications

                websocket.convertAndSend(wsTopic, body);
                log.debug("Redis [{}] → WS [{}]", channel, wsTopic);
            }
        );
    }
}

WebSocket config (STOMP)

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/topic");
        registry.setApplicationDestinationPrefixes("/app");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws").setAllowedOriginPatterns("*").withSockJS();
    }
}

6. Docker Compose + application.yml

# docker-compose.yml — add Redis service
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    command: redis-server --save "" --appendonly no   # disable persistence for Pub/Sub use
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      retries: 5
# application.yml — labs-socket
spring:
  data:
    redis:
      host: redis
      port: 6379
      timeout: 2000ms
      lettuce:
        pool:
          max-active: 8
          max-idle: 4

JavaScript client (browser)

// Browser WebSocket client — subscribe to order events
const socket = new SockJS('/ws');
const stompClient = Stomp.over(socket);

stompClient.connect({}, () => {
    // Subscribe to order events broadcast from Kafka via Redis
    stompClient.subscribe('/topic/events', (msg) => {
        const order = JSON.parse(msg.body);
        console.log('New order:', order.orderId, order.status);
        renderOrderUpdate(order);
    });

    stompClient.subscribe('/topic/notifications', (msg) => {
        showNotification(JSON.parse(msg.body));
    });
});

7. The message-loss gap this architecture deliberately accepts

Redis Pub/Sub is fire-and-forget by design — a message is only delivered to clients subscribed at the exact moment it’s published. Combined with --appendonly no (§6’s config, disabling Redis persistence), this creates real, permanent message loss in a few common scenarios:

Scenario A — labs-socket instance restarting (deploy, crash, scale-down):
  Redis PUBLISHes an event → the restarting instance isn't subscribed yet
  → that instance's connected browser clients never receive it, ever
  → Kafka still has the event, but nothing re-delivers it through this path

Scenario B — browser reconnects after a network blip:
  Client disconnects → events published during the gap → client reconnects
  → those events are gone; there is no "catch-up" mechanism in Pub/Sub

Why this is an accepted trade-off, not an oversight: Kafka remains the durable source of truth (§2) — nothing is lost from the system’s perspective, only from this specific real-time delivery path’s perspective. For live UI updates (a notification badge, a live order status), a client that missed a Pub/Sub message will simply catch up on its next full data refresh (e.g. re-fetching current state via a REST call on reconnect) rather than needing gap-free delivery.

When this trade-off is NOT acceptable: if the WebSocket-delivered event is the only delivery mechanism for something that matters (not just a live-UI nicety, but the actual notification of an important state change), Redis Pub/Sub’s fire-and-forget nature is the wrong tool. Either pair it with a REST endpoint the client can poll/refetch on reconnect to reconcile missed state (recommended — keeps Redis Pub/Sub for what it’s good at), or use Redis Streams (§8) instead, which does support consumer groups and replay.

8. Redis Streams — the alternative when loss is unacceptable

Redis also offers Streams (a different data structure from Pub/Sub, despite the name similarity to Kafka Streams) that supports persistence, consumer groups, and replay — closing the exact gap §7 describes, at the cost of Pub/Sub’s sub-millisecond simplicity.

Redis Streams vs Redis Pub/Sub:
  Pub/Sub: PUBLISH/SUBSCRIBE — no persistence, no replay, simplest, fastest
  Streams: XADD/XREAD — persisted, consumer groups, replay from any point (like Kafka)
{
  "redis.command": "XADD"
}

When to reach for Streams instead: if the reconnect-gap scenario in §7 is a real product concern (not just a rare edge case you’re comfortable with), Redis Streams gives you Kafka-like delivery guarantees at the Redis layer, at the cost of losing Pub/Sub’s simplicity and raw speed. For most “live UI nicety” use cases, Pub/Sub + a client-side reconciliation fetch on reconnect (§7’s recommended mitigation) is simpler and sufficient — reach for Streams only when you’ve concluded the gap genuinely can’t be tolerated.

9. Securing Redis

The compose config in §6 has no authentication — redis-server --save "" --appendonly no with no requirepass or ACL means any client on the network can PUBLISH/SUBSCRIBE to any channel, including injecting fake events that labs-socket would broadcast as if they came from Kafka.

redis:
  image: redis:7-alpine
  command: redis-server --save "" --appendonly no --requirepass ${REDIS_PASSWORD}
spring:
  data:
    redis:
      host: redis
      port: 6379
      password: ${REDIS_PASSWORD}

For finer-grained control than a single shared password, Redis ACLs (ACL SETUSER) can restrict the Connect sink’s user to PUBLISH-only on labs.* channels, separate from labs-socket‘s SUBSCRIBE-only access — following the same least-privilege principle as any other credential (Day 37 §8’s connector credential externalization applies here too, via redis.uri).

Same minimum bar as every other exposed component in this pipeline (Connect REST API — Day 36 §11, webhook receiver — Day 38 §10): if Redis is reachable from anywhere beyond a fully trusted internal network, it needs authentication. An open Redis instance backing a real-time broadcast layer means anyone who can reach it can inject fabricated “Kafka events” straight to every connected browser client.

10. Monitoring Pub/Sub health

# A quick health check: are there actually any subscribers right now?
redis-cli PUBSUB NUMSUB labs.events
# 0 subscribers here means real-time delivery is silently broken,
# even if every other part of the pipeline reports healthy

Why NUMSUB matters as a distinct signal: unlike Kafka consumer lag (which shows up clearly in kafka-consumer-groups.sh --describe), a Redis Pub/Sub channel with zero subscribers doesn’t error anywhere — the connector keeps publishing successfully (from its perspective, PUBLISH always “succeeds” whether or not anyone’s listening), and nothing surfaces the fact that messages are being broadcast into a void. Treat NUMSUB as a first-class health check for this architecture, not an afterthought.

11. Common pitfalls

  • Treating Redis Pub/Sub as durable because Kafka upstream is — the durability guarantee stops at the Kafka→Redis hop; anything downstream of PUBLISH is genuinely fire-and-forget (§7)
  • Not providing a reconciliation path for reconnecting clients — without a REST fetch to catch up on missed state, a client that was briefly disconnected has no way to know what it missed
  • Choosing Pub/Sub when the use case actually needs replay/consumer-group semantics — that’s what Redis Streams (§8) is for; don’t force Pub/Sub to do a job it’s not designed for
  • No authentication on the Redis instance — allows both eavesdropping and, more seriously, injecting fabricated events that get broadcast to every connected client as if legitimate (§9)
  • Assuming a healthy connector means healthy real-time delivery — PUBSUB NUMSUB returning 0 is a silent failure mode invisible to standard Connect/Kafka monitoring (§10)

Key Takeaways

  • Kafka = durable log; Redis Pub/Sub = ephemeral broadcast — use both together
  • This architecture is the production fix for Day 20 §7’s multi-instance WebSocket gap — every labs-socket instance subscribes to the same Redis channels, so partition-to-instance assignment no longer matters for delivery
  • Redis Sink PUBLISH maps each Kafka topic to a Redis channel by name (${topic})
  • labs-socket subscribes with PatternTopic("labs.*") — catches all labs.* channels
  • Redis delivers to ALL subscribers instantly — fan-out across multiple socket instances
  • No message persistence in Redis Pub/Sub — clients that miss a message (instance restart, brief disconnect) cannot replay; pair with a client-side reconciliation fetch or switch to Redis Streams if that gap is unacceptable
  • An unauthenticated Redis instance lets anyone inject fabricated events broadcast as legitimate — secure it like any other exposed component in the pipeline
  • PUBSUB NUMSUB returning 0 is a silent failure mode that standard connector/Kafka health checks won’t surface
  • Kafka retains the full history — Redis is just the last-mile delivery bus

Support me through GitHub Sponsors.

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

Next

➡️ Day 40: ksqlDB intro — CREATE STREAM, SELECT, push queries

Resources

Link to Medium blog

, , ,

Related Posts

Kafka

60-Day Kafka 4 Learning Plan · Week 9 — Day 60 of 60

Editorial Team
Kafka

60-Day Kafka 4 Learning Plan · Week 9 — Day 59 of 60

Editorial Team