SSE from Kafka

The sse-kafka binding fans a Kafka topic out to any number of browser clients over Server-Sent Events (SSE). This tutorial streams Hello, world messages from a Kafka topic to a small HTML page using the EventSource API, with no custom server code in between.

Prerequisites

Docker Compose

Set up the SSE Kafka proxy

Create zilla.yaml, docker-compose.yaml, and index.html in the same directory.

zilla.yaml

name: SSE-example
bindings:
  # Proxy service entrypoint
  north_tcp_server:
    type: tcp
    kind: server
    options:
      host: 0.0.0.0
      port: 7114
    exit: north_http_server
  north_http_server:
    type: http
    kind: server
    options:
      access-control:
        policy: cross-origin
    routes:
      - when:
          - headers:
              :scheme: http
              :authority: localhost:7114
              :path: /events
        exit: north_sse_server
      - when:
          - headers:
              :scheme: http
              :authority: localhost:7114
        exit: east_http_filesystem_mapping

  # UI html file server
  east_http_filesystem_mapping:
    type: http-filesystem
    kind: proxy
    routes:
      - when:
          - path: /{path}
        with:
          path: ${params.path}
        exit: east_filesystem_server
  east_filesystem_server:
    type: filesystem
    kind: server
    options:
      location: /var/www/

  # SSE Server With an exit to Kafka
  north_sse_server:
    type: sse
    kind: server
    exit: north_sse_kafka_mapping
  north_sse_kafka_mapping:
    type: sse-kafka
    kind: proxy
    routes:
      - when:
          - path: /events
        with:
          topic: events
        exit: north_kafka_cache_client

  # Kafka sync layer
  north_kafka_cache_client:
    type: kafka
    kind: cache_client
    exit: south_kafka_cache_server
  south_kafka_cache_server:
    type: kafka
    kind: cache_server
    options:
      bootstrap:
        - events
    exit: south_kafka_client

  # Connect to Kafka
  south_kafka_client:
    type: kafka
    kind: client
    options:
      servers:
        - ${{env.KAFKA_BOOTSTRAP_SERVER}}
    exit: south_kafka_tcp_client
  south_kafka_tcp_client:
    type: tcp
    kind: client

telemetry:
  exporters:
    stdout_logs_exporter:
      type: stdout

docker-compose.yaml

version: '3'
services:

  zilla:
    image: ghcr.io/aklivity/zilla:latest
    pull_policy: always
    depends_on:
      - kafka
    ports:
      - 7114:7114
    environment:
      KAFKA_BOOTSTRAP_SERVER: "kafka:29092"
    volumes:
      - ./zilla.yaml:/etc/zilla/zilla.yaml
      - ./index.html:/var/www/index.html
    command: start -v -e

  kafka:
    image: bitnami/kafka:3.5
    hostname: kafka
    ports:
      - 9092:9092
      - 29092:9092
    environment:
      ALLOW_PLAINTEXT_LISTENER: "yes"
      KAFKA_CFG_NODE_ID: "1"
      KAFKA_CFG_BROKER_ID: "1"
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: "1@127.0.0.1:9093"
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: "CLIENT:PLAINTEXT,INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT"
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_CFG_LOG_DIRS: "/tmp/logs"
      KAFKA_CFG_PROCESS_ROLES: "broker,controller"
      KAFKA_CFG_LISTENERS: "CLIENT://:9092,INTERNAL://:29092,CONTROLLER://:9093"
      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: "INTERNAL"
      KAFKA_CFG_ADVERTISED_LISTENERS: "CLIENT://localhost:9092,INTERNAL://kafka:29092"
      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "true"

  kafka-init:
    image: bitnami/kafka:3.5
    command:
        - /bin/sh
        - -c
        - |
          /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server kafka:29092 --create --if-not-exists --topic events
    depends_on:
      - kafka
    init: true

  kafka_messenger:
    image: confluentinc/cp-kafkacat:7.1.9
    command:
        - /bin/sh
        - -c
        - |
          while true; do echo "{ \"data\": \"Hello, world $(date)\" }" | kafkacat -P -b kafka:29092 -t events -k 1; echo "message sent, waiting 5 sec"; sleep 5; done
    depends_on:
      - kafka

index.html

<!DOCTYPE html>
<html lang="en">
  <head>
    <title>Welcome to Zilla!</title>
    <style type="text/css">
      .row {
        overflow: hidden;
        padding: 10px;
        width: 300px;
      }
      .col {
        float: left;
        width: 50%;
      }
    </style>
  </head>

  <body>
    <div class="col">
      <h3>Event Source</h3>

      <div class="row">
        <label for="location">Location</label>
        <input id="location" value="http://localhost:7114/events" style="width: 200px" />
      </div>
      <div class="row">
        <button onclick="javascript:attachEventSource()">Go</button>
      </div>

      <h3>Messages</h3>
      <div id="messages"></div>
    </div>
    <script>
        async function attachEventSource() {
          const location = document.getElementById("location");
          const messages = document.getElementById("messages");

          const printOutput = (text) => {
            var line = document.createElement("p");
            line.appendChild(document.createTextNode(text ?? ""));
            messages.insertBefore(line, messages.firstChild);
          }
 
          const es = new EventSource(`${location.value}`);
          es.onmessage = ({ data }) => {
            printOutput(data)
          };
          es.onopen = ({ type }) => {
            printOutput(type)
          };
          es.onerror = printOutput;
        }
      </script>
  </body>
</html>

The highlighted access-control option enables cross-origin requests so the browser page can open an EventSource connection to /events. The north_sse_kafka_mapping route maps that path onto the events Kafka topic. A second route, east_http_filesystem_mapping, serves index.html itself from the container's filesystem.

The docker-compose.yaml file also starts a kafka_messenger container that produces a Hello, world message to the events topic every five seconds, so there is always something to see on the page.

Run Zilla and Kafka

docker-compose up --detach

Open http://localhost:7114/index.html in a browser. With the location field set to http://localhost:7114/events, click Go to open the SSE connection. Messages produced by kafka_messenger stream in as they arrive:

...
message: Hello, world Wed May 10 14:25:45 UTC 2023

message: Hello, world Wed May 10 14:25:40 UTC 2023

open:

Remove the running containers

docker-compose down

Next Steps

Walk through the SSE from Kafka use case for more on streaming Kafka topics to browsers.
See the sse-kafka binding reference for every route option.
Try the sse.kafka.fanout example this tutorial is based on, plus sse.proxy.jwt for adding JWT auth in front of the SSE stream.