Stream Kafka topics into the grid

In this tutorial, you will build a live inventory grid fed by Apache Kafka. A small Node.js bridge reads two Kafka topics and streams them to the browser with Server-Sent Events. Handsontable DataGrid applies hundreds of updates a second without losing the user’s sort or filter. When the user edits a cell, the edit goes back to Kafka, so every open grid sees it.

Overview

Difficulty: Advanced
Time: ~40 minutes
Stack: Apache Kafka 4.3 (Docker), Node.js 22, @confluentinc/kafka-javascript 1.11, Server-Sent Events, Handsontable, Vite

Browsers can’t talk to Kafka directly: Kafka uses its own binary protocol over TCP. So the recipe puts a bridge in between. The bridge consumes the topics, keeps the latest record for every key in memory, and sends the browser a snapshot followed by batched changes.

Kafka stays the single source of truth. The grid never saves an edit locally. It sends the edit to the bridge, the bridge writes it to Kafka, and the change reaches every grid, including the one that made it, through the stream.

What you’ll build

• A local Kafka broker in Docker, with two compacted topics.
• A producer that simulates a warehouse system: 500 SKUs across 4 warehouses, 200 stock movements a second, and the occasional new or discontinued SKU.
• A bridge that serves a snapshot and batched changes over Server-Sent Events, and writes grid edits to Kafka.
• A grid that applies each animation frame’s changes in one call, highlights the cells that changed, flags low stock, and keeps the user’s sort and filter.
• An editable Reorder point column whose edits round-trip through Kafka.
• A stream that recovers on its own when the bridge restarts.

Before you begin

Requirement	Version
Docker	any current version, with Compose
Node.js	22 or later
handsontable	18.1 or later
@confluentinc/kafka-javascript	1.11.0
Vite	any current version

This recipe uses Confluent’s @confluentinc/kafka-javascript client, which wraps the librdkafka C library. It ships prebuilt binaries for common platforms, and its KafkaJS export follows the kafkajs API, so code written for kafkajs ports with small configuration changes.

Create a folder with two parts: server/ for Kafka, the producer, and the bridge, and a Vite app for the grid:

mkdir kafka-inventory && cd kafka-inventory
mkdir server && cd server
npm init -y
npm install --save-exact @confluentinc/kafka-javascript@1.11.0
cd ..
npm create vite@latest client -- --template vanilla-ts
cd client
npm install handsontable

How the pieces fit

producer.mjs ──► inventory.stock    (compacted, key = SKU)  ──┐
                                                              ├──► bridge.mjs ──SSE──► grid
grid edit ──POST──► bridge.mjs ──► inventory.settings (compacted, key = SKU) ──┘

Two topics, each owned by one writer:

• inventory.stock belongs to the warehouse system. Each record is the full stock state of one SKU.
• inventory.settings belongs to the grid. Each record holds the reorder point of one SKU.

If both writers sent full records to one compacted topic, the warehouse system’s next stock update for a SKU would overwrite the reorder point a user set a moment earlier. Splitting the topics by owner avoids that, and the bridge joins them into one row per SKU.

1. Start Kafka

   Create docker-compose.yml in the server/ folder:

   YML

   services:
     kafka:
       # Official Apache Kafka image: a single-node KRaft cluster, reachable at localhost:9092.
       image: apache/kafka:4.3.1
       ports:
         - "9092:9092"

   Start the broker:

   docker compose up -d

   What’s happening: the official apache/kafka image starts a single-node cluster in KRaft mode, with no ZooKeeper and no configuration. It listens on localhost:9092.

2. Share the client configuration and create the topics

   Create server/kafka.mjs:

   MJS

   import KafkaJavaScript from '@confluentinc/kafka-javascript';
   
   const { Kafka, logLevel } = KafkaJavaScript.KafkaJS;
   
   // Stock levels, owned by the warehouse system. Key: SKU. A null value removes the SKU.
   export const STOCK_TOPIC = 'inventory.stock';
   // Settings the grid edits, owned by the grid. Key: SKU. Value: { reorderPoint }.
   export const SETTINGS_TOPIC = 'inventory.settings';
   
   export const kafka = new Kafka({
     kafkaJS: {
       brokers: (process.env.KAFKA_BROKERS ?? 'localhost:9092').split(','),
       logLevel: logLevel.WARN,
     },
   });
   
   // Creates both topics as compacted topics: Kafka keeps at least the latest record per
   // key, so each topic always holds the current state of every SKU.
   export async function ensureTopics() {
     const admin = kafka.admin();
   
     await admin.connect();
   
     try {
       const existing = new Set(await admin.listTopics());
       const missing = [STOCK_TOPIC, SETTINGS_TOPIC].filter((topic) => !existing.has(topic));
   
       if (missing.length > 0) {
         await admin.createTopics({
           topics: missing.map((topic) => ({
             topic,
             numPartitions: 3,
             configEntries: [{ name: 'cleanup.policy', value: 'compact' }],
           })),
         });
       }
     } finally {
       await admin.disconnect();
     }
   }

   What’s happening:

   • The kafkaJS key selects the KafkaJS-style API. The Confluent client also accepts librdkafka property names ('bootstrap.servers'), but this recipe uses the kafkaJS block throughout.
   • Both topics are compacted. With cleanup.policy=compact, Kafka keeps at least the latest record for every key and discards older ones in the background. A compacted topic holds the current state of every SKU, so the bridge can rebuild its view by reading the topic from the start.
   • A null value is a tombstone. In a compacted topic, a record with a null value deletes its key. The producer uses tombstones to discontinue SKUs.
   • ensureTopics is safe to call repeatedly. It creates only the topics that don’t exist yet, so both the producer and the bridge call it on start.

3. Simulate the warehouse system

   Create server/producer.mjs:

   MJS

   import { kafka, ensureTopics, STOCK_TOPIC } from './kafka.mjs';
   
   // Simulates a warehouse system: stock movements for 500 SKUs in 4 warehouses.
   const RATE = Number(process.env.RATE ?? 200); // records per second
   const TICK_MS = 50;
   const WAREHOUSES = ['Berlin', 'Chicago', 'Singapore', 'Sao Paulo'];
   const PRODUCTS = ['Laptop', 'Monitor', 'Dock', 'Keyboard', 'Headset', 'Webcam', 'Switch', 'Router', 'Cable', 'Charger'];
   const SKU_COUNT = 500;
   
   const items = new Map();
   
   function newItem(index) {
     const sku = `SKU-${String(index).padStart(5, '0')}`;
   
     return {
       sku,
       name: `${PRODUCTS[index % PRODUCTS.length]} ${String.fromCharCode(65 + (index % 26))}${index % 97}`,
       warehouse: WAREHOUSES[index % WAREHOUSES.length],
       onHand: 50 + Math.floor(Math.random() * 400),
       reserved: Math.floor(Math.random() * 40),
       updatedAt: new Date().toISOString(),
     };
   }
   
   const messageOf = (item) => ({ key: item.sku, value: JSON.stringify(item) });
   const randomSku = () => [...items.keys()][Math.floor(Math.random() * items.size)];
   
   await ensureTopics();
   
   const producer = kafka.producer();
   
   await producer.connect();
   
   for (let index = 1; index <= SKU_COUNT; index++) {
     const item = newItem(index);
   
     items.set(item.sku, item);
   }
   
   await producer.send({ topic: STOCK_TOPIC, messages: [...items.values()].map(messageOf) });
   console.log(`Seeded ${items.size} SKUs, now sending ${RATE} records/s`);
   
   let nextIndex = SKU_COUNT + 1;
   
   setInterval(async () => {
     const messages = [];
   
     for (let i = 0; i < Math.max(1, Math.round((RATE * TICK_MS) / 1000)); i++) {
       const roll = Math.random();
   
       if (roll < 0.001 && items.size > 1) {
         // Discontinue a SKU: a tombstone (null value) removes it from the compacted topic.
         const sku = randomSku();
   
         items.delete(sku);
         messages.push({ key: sku, value: null });
       } else if (roll < 0.002) {
         const item = newItem(nextIndex++);
   
         items.set(item.sku, item);
         messages.push(messageOf(item));
       } else {
         const item = items.get(randomSku());
   
         item.onHand = Math.max(0, item.onHand + Math.floor(Math.random() * 21) - 10);
         item.reserved = Math.min(item.onHand, Math.max(0, item.reserved + Math.floor(Math.random() * 7) - 3));
         item.updatedAt = new Date().toISOString();
         messages.push(messageOf(item));
       }
     }
   
     await producer.send({ topic: STOCK_TOPIC, messages });
   }, TICK_MS);

   What’s happening: the producer sends a full record for all 500 SKUs, then sends a batch of stock movements every 50 ms. Each record’s key is its SKU. Kafka routes records with the same key to the same partition, so the changes to one SKU always arrive in the order they were sent. About one record in a thousand adds a new SKU, and about one in a thousand is a tombstone that removes one. Set the RATE environment variable to change the number of records per second.

   Run it in its own terminal:

   cd server
   node producer.mjs

4. Build the bridge

   Create server/bridge.mjs:

   MJS

   import { randomUUID } from 'node:crypto';
   import http from 'node:http';
   import { kafka, ensureTopics, SETTINGS_TOPIC, STOCK_TOPIC } from './kafka.mjs';
   
   const PORT = Number(process.env.PORT ?? 3001);
   const FLUSH_MS = 50;
   const HEARTBEAT_MS = 15_000;
   const DEFAULT_REORDER_POINT = 50;
   
   // The materialized view: the latest stock record and settings for every SKU.
   const stock = new Map();
   const settings = new Map();
   // SKUs changed since the last flush. Several changes to one SKU collapse into one.
   const changed = new Set();
   const clients = new Set();
   
   // settingsRevision is the Kafka offset of the SKU's latest settings record (-1 for none).
   // One key always lands in one partition, so a higher offset means a later edit, and a grid
   // can tell that its own edit was applied even when a later edit replaced the value.
   function toRow(sku) {
     const record = stock.get(sku);
     const setting = settings.get(sku);
   
     return record && {
       ...record,
       reorderPoint: setting?.reorderPoint ?? DEFAULT_REORDER_POINT,
       settingsRevision: setting?.revision ?? -1,
     };
   }
   
   function send(res, event, data) {
     res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
   }
   
   // Sends the changes of the last FLUSH_MS as one event, instead of one event per record.
   setInterval(() => {
     if (changed.size === 0 || clients.size === 0) {
       changed.clear();
   
       return;
     }
   
     const upserts = [];
     const deletes = [];
   
     for (const sku of changed) {
       const row = toRow(sku);
   
       if (row) {
         upserts.push(row);
       } else {
         deletes.push(sku);
       }
     }
   
     changed.clear();
   
     for (const res of clients) {
       send(res, 'changes', { upserts, deletes });
     }
   }, FLUSH_MS);
   
   await ensureTopics();
   
   // Every start reads both topics from the beginning and rebuilds the view; compaction keeps
   // that replay short. A new group per process gets its partitions at once: with a fixed
   // group, a restarted bridge waits until Kafka times out the stopped one. The group never
   // commits offsets, so Kafka removes it once it is empty.
   const consumer = kafka.consumer({
     kafkaJS: { groupId: `inventory-grid-bridge-${randomUUID()}`, fromBeginning: true, autoCommit: false },
   });
   
   await consumer.connect();
   await consumer.subscribe({ topics: [STOCK_TOPIC, SETTINGS_TOPIC] });
   await consumer.run({
     eachMessage: async ({ topic, message }) => {
       const sku = message.key.toString();
       const value = message.value ? JSON.parse(message.value.toString()) : null;
       const target = topic === STOCK_TOPIC ? stock : settings;
   
       if (value === null) {
         target.delete(sku);
       } else if (topic === SETTINGS_TOPIC) {
         target.set(sku, { ...value, revision: Number(message.offset) });
       } else {
         target.set(sku, value);
       }
   
       changed.add(sku);
     },
   });
   
   const producer = kafka.producer();
   
   await producer.connect();
   
   function readJson(req) {
     return new Promise((resolve, reject) => {
       let body = '';
   
       req.on('data', (chunk) => {
         body += chunk;
       });
       req.on('end', () => {
         try {
           resolve(JSON.parse(body));
         } catch (error) {
           reject(error);
         }
       });
       req.on('error', reject);
     });
   }
   
   const server = http.createServer(async (req, res) => {
     const url = new URL(req.url, 'http://localhost');
   
     // The live stream: a snapshot of every row, then batched changes.
     if (req.method === 'GET' && url.pathname === '/api/stream') {
       res.writeHead(200, {
         'Content-Type': 'text/event-stream',
         'Cache-Control': 'no-cache',
         Connection: 'keep-alive',
       });
       send(res, 'snapshot', { rows: [...stock.keys()].map(toRow) });
       clients.add(res);
   
       // A regular event, not an SSE comment: the browser can see it, so a client that hears
       // nothing for a while knows the stream is dead, even when a proxy keeps it open.
       const heartbeat = setInterval(() => send(res, 'ping', {}), HEARTBEAT_MS);
   
       req.on('close', () => {
         clearInterval(heartbeat);
         clients.delete(res);
       });
   
       return;
     }
   
     // An edit from the grid. It is written to Kafka, not to the view: the change reaches
     // every open grid, this one included, through the stream.
     const match = url.pathname.match(/^\/api\/settings\/([^/]+)$/);
   
     if (req.method === 'POST' && match) {
       const sku = decodeURIComponent(match[1]);
   
       try {
         const { reorderPoint } = await readJson(req);
   
         if (!stock.has(sku) || !Number.isInteger(reorderPoint) || reorderPoint < 0) {
           res.writeHead(400, { 'Content-Type': 'application/json' });
           res.end(JSON.stringify({ message: 'Unknown SKU or invalid reorder point' }));
   
           return;
         }
   
         const [{ baseOffset }] = await producer.send({
           topic: SETTINGS_TOPIC,
           messages: [{ key: sku, value: JSON.stringify({ reorderPoint }) }],
         });
   
         // The record's offset is its revision: the grid waits for a row at least this new.
         res.writeHead(202, { 'Content-Type': 'application/json' });
         res.end(JSON.stringify({ revision: Number(baseOffset) }));
       } catch (error) {
         res.writeHead(500, { 'Content-Type': 'application/json' });
         res.end(JSON.stringify({ message: error.message }));
       }
   
       return;
     }
   
     res.writeHead(404).end();
   });
   
   server.listen(PORT, () => console.log(`Bridge listening on http://localhost:${PORT}`));
   
   // Leave the consumer group and close connections on Ctrl+C or a stop signal.
   for (const signal of ['SIGINT', 'SIGTERM']) {
     process.once(signal, async () => {
       server.close();
       await Promise.allSettled([consumer.disconnect(), producer.disconnect()]);
       process.exit(0);
     });
   }

   What’s happening:

   • The bridge keeps a materialized view. It consumes both topics into two maps, stock and settings, and toRow joins them into the row the grid shows. A SKU with no settings record gets the default reorder point. Each row also carries settingsRevision, the Kafka offset of the SKU’s latest settings record.
   • Every connection starts with a snapshot. GET /api/stream first sends every row, then the changes. A browser that reconnects gets a fresh snapshot, so it can’t miss an update while it was offline, and the bridge needs no per-client history.
   • Changes go out in batches. At 200 records a second, sending one event per record wastes bandwidth and work. The bridge collects changed SKUs for 50 ms, keeps only the latest state of each, and sends one changes event.
   • A new consumer group per start. The bridge never commits offsets, so every start reads both topics from the beginning and rebuilds the view. A group name that changes on every start matters for restarts: with a fixed name, Kafka keeps a stopped bridge in the group until its session times out, and the new bridge waits for it. A group that never commits is removed once it is empty. The signal handler at the end leaves the group cleanly on Ctrl+C.
   • The heartbeat is a real event. Every 15 s the bridge sends a ping event. A proxy between the bridge and the browser can hold a connection open after the bridge stops, and an SSE comment line is invisible to the browser’s EventSource. A ping event lets the client notice the silence. See Step 5.
   • Edits go to Kafka, not to the view. POST /api/settings/:sku checks the SKU and the value, writes a record to inventory.settings, and returns 202 Accepted with the record’s offset as revision. Records with the same key go to the same partition, so for one SKU a higher offset is always a later edit. The bridge’s own consumer then reads the record back and streams it to every grid.

   Run it in a second terminal:

   cd server
   node bridge.mjs

5. Read the stream in the browser

   In the Vite app, create src/stream.ts:

   TypeScript

   import type { StockRecord } from './grid';
   
   interface StreamHandlers {
     onSnapshot: (rows: StockRecord[]) => void;
     onChanges: (upserts: StockRecord[], deletes: string[]) => void;
     onStatus: (connected: boolean) => void;
   }
   
   // The bridge sends a `ping` every 15 s. Silence for longer than this means the stream is
   // dead, even if a proxy in between still holds the connection open.
   const SILENCE_MS = 35_000;
   
   // Reads the bridge's Server-Sent Events. Changes are buffered and handed to the grid once
   // per animation frame, with only the latest record kept for each SKU.
   export function connectStream(url: string, { onSnapshot, onChanges, onStatus }: StreamHandlers) {
     const upserts = new Map<string, StockRecord>();
     const deletes = new Set<string>();
     let frame = 0;
     let received = 0;
     let lastEventAt = Date.now();
     let source: EventSource;
   
     function flush() {
       frame = 0;
       onChanges([...upserts.values()], [...deletes]);
       upserts.clear();
       deletes.clear();
     }
   
     function open() {
       source = new EventSource(url);
       lastEventAt = Date.now();
   
       source.addEventListener('open', () => onStatus(true));
       // EventSource reconnects on its own after an error. The bridge sends a fresh snapshot
       // on every connection, so nothing is lost while the stream was down.
       source.addEventListener('error', () => onStatus(false));
       source.addEventListener('ping', () => {
         lastEventAt = Date.now();
       });
   
       source.addEventListener('snapshot', (event) => {
         lastEventAt = Date.now();
         upserts.clear();
         deletes.clear();
         onSnapshot(JSON.parse(event.data).rows);
       });
   
       source.addEventListener('changes', (event) => {
         const batch = JSON.parse(event.data) as { upserts: StockRecord[]; deletes: string[] };
   
         lastEventAt = Date.now();
   
         for (const record of batch.upserts) {
           upserts.set(record.sku, record);
           deletes.delete(record.sku);
         }
   
         for (const sku of batch.deletes) {
           deletes.add(sku);
           upserts.delete(sku);
         }
   
         received += batch.upserts.length + batch.deletes.length;
         frame ||= requestAnimationFrame(flush);
       });
     }
   
     open();
   
     // A stream that has gone silent is replaced with a new one.
     const watchdog = setInterval(() => {
       if (Date.now() - lastEventAt > SILENCE_MS) {
         onStatus(false);
         source.close();
         open();
       }
     }, 5_000);
   
     return {
       close: () => {
         clearInterval(watchdog);
         source.close();
       },
       // Records received since the last call, for the throughput readout.
       takeReceived: () => {
         const count = received;
   
         received = 0;
   
         return count;
       },
     };
   }

   What’s happening:

   • One grid update per frame. Each changes event goes into a buffer keyed by SKU, so a SKU that changes three times between frames is applied once. requestAnimationFrame hands the buffer to the grid right before the browser paints.
   • A hidden tab doesn’t fall behind. Browsers pause requestAnimationFrame in background tabs. The buffer keeps at most one record per SKU, so it stays small, and the grid catches up on the first frame after the tab becomes visible.
   • A silent stream is replaced. EventSource reconnects by itself when a connection fails. It can’t tell when a connection stays open but stops delivering, which is what happens behind a proxy when the bridge stops. The watchdog closes the stream after 35 s with no event and opens a new one, and the new connection’s snapshot resyncs the grid.

6. Build the grid

   Create src/grid.ts:

   TypeScript

   import Handsontable from 'handsontable/base';
   import { registerAllModules } from 'handsontable/registry';
   import type { CellChange, ChangeSource } from 'handsontable/settings';
   import 'handsontable/styles/handsontable.min.css';
   import 'handsontable/styles/ht-theme-main.min.css';
   
   registerAllModules();
   
   export interface InventoryRow {
     sku: string;
     name: string;
     warehouse: string;
     onHand: number;
     reserved: number;
     available: number;
     reorderPoint: number;
     updatedAt: string;
   }
   
   // A row as the bridge sends it. settingsRevision is the Kafka offset of the SKU's latest
   // reorder point record.
   export type StockRecord = Omit<InventoryRow, 'available'> & { settingsRevision: number };
   
   interface PendingEdit {
     value: number;
     // Tells this edit apart from a later edit of the same SKU.
     generation: number;
     // The Kafka offset of the edit's record, known once the bridge accepts it.
     revision?: number;
   }
   
   const LIVE_PROPS = ['onHand', 'reserved', 'available', 'reorderPoint', 'updatedAt'] as const;
   const FLASH_MS = 800;
   // Changes the grid makes itself. Only a user's edit goes back to Kafka.
   const IGNORED_SOURCES = new Set<string>(['kafka', 'loadData', 'updateData']);
   
   // The grid's data. Handsontable keeps a reference to this array, so a physical row index
   // is a position in it.
   let rows: InventoryRow[] = [];
   const indexBySku = new Map<string, number>();
   // When each SKU's cells last changed, for the highlight.
   const changedAt = new Map<string, number>();
   // The latest reorder point Kafka has confirmed for each SKU, and its offset.
   const confirmed = new Map<string, { reorderPoint: number; revision: number }>();
   // The user's latest edit for each SKU, until Kafka confirms it or a newer edit.
   const pendingEdits = new Map<string, PendingEdit>();
   let nextGeneration = 0;
   
   // Builds a grid row from a Kafka record. A reorder point the user edited keeps showing until
   // a record at least as new as the edit arrives, so a record that was already in flight can't
   // briefly undo it. The check compares offsets, not values: when another user edits the same
   // SKU first, the bridge sends only the latest value, and that value wins.
   function toRow(record: StockRecord): InventoryRow {
     const { settingsRevision, ...fields } = record;
     const row = { ...fields, available: fields.onHand - fields.reserved };
     const pending = pendingEdits.get(row.sku);
   
     confirmed.set(row.sku, { reorderPoint: row.reorderPoint, revision: settingsRevision });
   
     if (pending) {
       if (pending.revision !== undefined && settingsRevision >= pending.revision) {
         pendingEdits.delete(row.sku);
       } else {
         row.reorderPoint = pending.value;
       }
     }
   
     return row;
   }
   
   function reindex() {
     indexBySku.clear();
     rows.forEach((row, index) => indexBySku.set(row.sku, index));
   }
   
   export function createGrid(container: HTMLElement) {
     const hot = new Handsontable(container, {
       data: rows,
       colHeaders: ['SKU', 'Product', 'Warehouse', 'On hand', 'Reserved', 'Available', 'Reorder point', 'Updated'],
       columns: [
         { data: 'sku', readOnly: true, width: 100 },
         { data: 'name', readOnly: true, width: 150 },
         { data: 'warehouse', readOnly: true, width: 110 },
         { data: 'onHand', type: 'numeric', readOnly: true },
         { data: 'reserved', type: 'numeric', readOnly: true },
         { data: 'available', type: 'numeric', readOnly: true },
         {
           data: 'reorderPoint',
           type: 'numeric',
           validator: (value: unknown, callback: (valid: boolean) => void) => callback(Number.isInteger(value) && (value as number) >= 0),
           allowInvalid: false,
         },
         { data: 'updatedAt', readOnly: true, width: 200 },
       ],
       // `row` is a physical index here, so it points straight into `rows`.
       cells(row, _column, prop) {
         const item = rows[row];
   
         if (!item || !LIVE_PROPS.includes(prop as (typeof LIVE_PROPS)[number])) {
           return {};
         }
   
         const classes = [];
   
         if (performance.now() - (changedAt.get(item.sku) ?? -Infinity) < FLASH_MS) {
           classes.push('is-changed');
         }
   
         if (prop === 'available' && item.available < item.reorderPoint) {
           classes.push('is-low');
         }
   
         return { className: classes.join(' ') };
       },
       rowHeaders: true,
       columnSorting: true,
       filters: true,
       dropdownMenu: ['filter_by_condition', 'filter_operators', 'filter_by_condition2', 'filter_by_value', 'filter_action_bar'],
       notification: true,
       height: 600,
       stretchH: 'all',
       themeName: 'ht-theme-main',
       // licenseKey: 'your-license-key', // get a key at https://handsontable.com/pricing/
     });
   
     // Sorting and filtering run when the user applies them, not on every streamed value,
     // so rows don't move while someone reads them. After rows are added or removed,
     // re-apply both so new rows land in the right place.
     function reapplyView() {
       const filters = hot.getPlugin('filters');
       const sorting = hot.getPlugin('columnSorting');
       const sortConfig = sorting.getSortConfig();
   
       if (filters.exportConditions().length > 0) {
         filters.filter();
       }
   
       if (Array.isArray(sortConfig) && sortConfig.length > 0) {
         sorting.sort(sortConfig);
       }
     }
   
     function applySnapshot(records: StockRecord[]) {
       confirmed.clear();
       rows = records.map(toRow);
       reindex();
       hot.updateData(rows, 'kafka');
       reapplyView();
     }
   
     function applyChanges(upserts: StockRecord[], deletes: string[]) {
       const now = performance.now();
       const cellChanges: [number, string, unknown][] = [];
       let structural = deletes.length > 0;
   
       for (const record of upserts) {
         const next = toRow(record);
         const index = indexBySku.get(next.sku);
   
         if (index === undefined) {
           rows.push(next);
           structural = true;
           continue;
         }
   
         for (const prop of LIVE_PROPS) {
           if (rows[index][prop] !== next[prop]) {
             cellChanges.push([index, prop, next[prop]]);
           }
         }
   
         changedAt.set(next.sku, now);
       }
   
       // Physical indexes: sorting and filtering can't send a value to the wrong row.
       if (cellChanges.length > 0) {
         hot.setSourceDataAtCell(cellChanges, undefined, undefined, 'kafka');
       }
   
       if (structural) {
         const removed = new Set(deletes);
   
         deletes.forEach((sku) => {
           confirmed.delete(sku);
           pendingEdits.delete(sku);
         });
         rows = rows.filter((row) => !removed.has(row.sku));
         reindex();
         hot.updateData(rows, 'kafka');
         reapplyView();
       }
     }
   
     // Shows the reorder point Kafka last confirmed, once no edit of this SKU is pending.
     function showConfirmed(sku: string) {
       const index = indexBySku.get(sku);
       const server = confirmed.get(sku);
   
       // setSourceDataAtCell doesn't fire afterChange, so the value isn't sent again.
       if (index !== undefined && server && rows[index].reorderPoint !== server.reorderPoint) {
         hot.setSourceDataAtCell(index, 'reorderPoint', server.reorderPoint, 'kafka');
       }
     }
   
     // Sends a user's edit to the bridge, which writes it to Kafka. Each edit gets a generation,
     // so the response to an older request can't confirm or undo a newer edit of the same SKU.
     hot.addHook('afterChange', (changes: CellChange[] | null, source: ChangeSource) => {
       if (!changes || IGNORED_SOURCES.has(source)) {
         return;
       }
   
       for (const [visualRow, prop, oldValue, newValue] of changes) {
         if (prop !== 'reorderPoint' || oldValue === newValue) {
           continue;
         }
   
         const sku = rows[hot.toPhysicalRow(visualRow)].sku;
         const generation = ++nextGeneration;
         const isCurrent = () => pendingEdits.get(sku)?.generation === generation;
   
         pendingEdits.set(sku, { value: newValue as number, generation });
   
         fetch(`/api/settings/${encodeURIComponent(sku)}`, {
           method: 'POST',
           headers: { 'Content-Type': 'application/json' },
           body: JSON.stringify({ reorderPoint: newValue }),
         })
           .then(async (response) => {
             const body = await response.json().catch(() => null);
   
             if (!response.ok) {
               throw new Error(body?.message ?? `HTTP ${response.status}`);
             }
   
             if (!isCurrent()) {
               return;
             }
   
             // The stream may already have delivered the record, or a newer one.
             if ((confirmed.get(sku)?.revision ?? -1) >= body.revision) {
               pendingEdits.delete(sku);
               showConfirmed(sku);
             } else {
               pendingEdits.get(sku)!.revision = body.revision;
             }
           })
           .catch((error: Error) => {
             // Only the latest edit decides what the cell shows. If it failed, show Kafka's value.
             if (isCurrent()) {
               pendingEdits.delete(sku);
               showConfirmed(sku);
             }
   
             hot.getPlugin('notification').showMessage({
               variant: 'error',
               title: `Couldn't save ${sku}`,
               message: error.message,
             });
           });
       }
     });
   
     return { hot, applySnapshot, applyChanges };
   }

   What’s happening:

   • Values go in by physical row. setSourceDataAtCell takes physical row indexes, the positions in the rows array, so a sort or filter can’t send a value to the wrong SKU. It accepts all the frame’s changes as one array and renders once. It doesn’t fire afterChange, so streamed values never trigger the write-back.
   • New and removed SKUs replace the data. When the batch adds or removes rows, the grid passes the new array to updateData, which keeps the column order, cell state, and sort and filter settings.
   • Sorting and filtering run on structural changes, not on every value. ColumnSorting and Filters don’t re-run when values change through setSourceDataAtCell or updateData. That is useful here: at 200 records a second, rows that re-sorted on every value would jump while the user reads them. reapplyView re-runs the filter and the sort after rows are added or removed, so new rows land in the right place and hidden rows stay hidden. To re-sort on value changes too, call reapplyView() on a timer, for example every few seconds.
   • The highlight comes from cell meta. The cells function receives the physical row and adds the is-changed class for 800 ms after a SKU changes, and is-low when available stock falls below the reorder point. Because the class comes from cell meta, it survives scrolling and sorting, which a class added directly to a td element doesn’t.
   • Edits wait for Kafka. afterChange ignores the grid’s own changes and sends a user’s Reorder point edit to the bridge. Until a row with a settingsRevision at least as high as the edit’s revision arrives, toRow keeps the edit on screen, in change batches and in the snapshot after a reconnect alike, so a stale record that was already in flight can’t briefly overwrite it. The check compares offsets, not values: if another user edits the same SKU before the bridge flushes, the bridge sends only the newer value, and the grid accepts it instead of waiting forever for its own.
   • The latest edit wins. Each edit gets a generation number. A response for an older edit of the same SKU can’t confirm or roll back a newer one that is still in flight. If the latest edit fails, the grid shows the value Kafka last confirmed, using setSourceDataAtCell, which doesn’t fire afterChange, so the revert isn’t sent again. The Notification plugin shows the error.

7. Run the app

   Replace src/main.ts:

   TypeScript

   import { createGrid } from './grid';
   import { connectStream } from './stream';
   import './styles.css';
   
   const { hot, applySnapshot, applyChanges } = createGrid(document.querySelector('#grid')!);
   const status = document.querySelector('#status')!;
   const rate = document.querySelector('#rate')!;
   const count = document.querySelector('#rows')!;
   
   const stream = connectStream('/api/stream', {
     onSnapshot: applySnapshot,
     onChanges: applyChanges,
     onStatus: (connected) => {
       status.textContent = connected ? 'Live' : 'Reconnecting…';
       status.classList.toggle('is-offline', !connected);
     },
   });
   
   setInterval(() => {
     rate.textContent = new Intl.NumberFormat('en-US').format(stream.takeReceived());
     count.textContent = new Intl.NumberFormat('en-US').format(hot.countSourceRows());
   }, 1000);
   
   window.addEventListener('beforeunload', () => stream.close());

   Add the styles in src/styles.css:

   CSS

   body {
     margin: 0;
     font-family: Inter, system-ui, -apple-system, sans-serif;
   }
   
   main {
     max-width: 1200px;
     margin: 0 auto;
     padding: 20px 16px;
   }
   
   h1 {
     margin: 0 0 8px;
     font-size: 22px;
   }
   
   .toolbar {
     display: flex;
     gap: 20px;
     align-items: center;
     font-size: 14px;
   }
   
   .badge {
     padding: 2px 8px;
     border-radius: 999px;
     background: #e6f4ea;
     color: #1e6b34;
   }
   
   .badge.is-offline {
     background: #fdecea;
     color: #a12622;
   }
   
   /* A streamed change fades from yellow. */
   .handsontable td.is-changed {
     animation: cell-flash 800ms ease-out;
   }
   
   @keyframes cell-flash {
     from {
       background-color: #fff3b0;
     }
   }
   
   /* Available stock below the reorder point. */
   .handsontable td.is-low {
     color: #b3261e;
     font-weight: 600;
   }

   Replace index.html:

   <!doctype html>
   <html lang="en">
     <head>
       <meta charset="UTF-8" />
       <meta name="viewport" content="width=device-width, initial-scale=1.0" />
       <title>Live inventory from Kafka</title>
     </head>
     <body>
       <main>
         <h1>Live inventory from Kafka</h1>
         <p class="toolbar">
           <span id="status" class="badge">Connecting…</span>
           <span>Records/s <strong id="rate">0</strong></span>
           <span>SKUs <strong id="rows">0</strong></span>
         </p>
         <div id="grid"></div>
       </main>
       <script type="module" src="/src/main.ts"></script>
     </body>
   </html>

   Create vite.config.ts, so the dev server forwards /api to the bridge and the page and the stream share one origin:

   import { defineConfig } from 'vite';
   
   export default defineConfig({
     server: {
       proxy: { '/api': 'http://localhost:3001' },
     },
   });

   Start the app in a third terminal:

   cd client
   npm run dev

   Open the URL Vite prints. The grid fills from the snapshot, cells flash as stock moves, and the toolbar shows about 200 records a second. Sort by Available and filter Warehouse to one location: values keep updating in place, and new SKUs appear in sorted position. Edit a Reorder point, then reload the page or open it in a second window: the value came back from Kafka.

How it works — complete flow

1. The producer writes a stock record for a SKU to inventory.stock, keyed by the SKU.
2. The bridge’s consumer reads it, updates the stock map, and marks the SKU as changed.
3. Within 50 ms, the bridge sends a changes event with the joined row of every changed SKU to every connected browser.
4. The browser adds the rows to its buffer. On the next animation frame, the grid writes the changed values with one setSourceDataAtCell call, or replaces the data with updateData when SKUs were added or removed.
5. The user edits a Reorder point. afterChange sends POST /api/settings/:sku, and the bridge writes a record to inventory.settings.
6. The bridge’s consumer reads that record, and the change reaches every grid, including the one that made the edit, through steps 3 and 4.

Production considerations

• Authenticate both endpoints. The stream and the edit endpoint are open in this recipe. Put them behind the same authentication as the rest of your app, and check that the user may edit the SKU before writing to Kafka.
• Connect to a managed cluster. To use a hosted cluster such as Confluent Cloud, set KAFKA_BROKERS to its bootstrap servers and add its SSL and SASL settings to the client configuration in kafka.mjs. See the client’s configuration reference.
• Size the view. Each bridge holds every row in memory and sends all of them in each snapshot. That suits thousands or tens of thousands of keys. For millions of rows, keep the data in a database and use the dataProvider plugin, with Kafka only pushing change notifications.
• Run more than one bridge. Every bridge builds the full view on its own, so you can run several behind a load balancer. A client that reconnects to a different bridge gets that bridge’s snapshot.
• Wait for the replay after a restart. A bridge accepts connections while it is still replaying the topics. A browser that connects in that window gets a partial snapshot, and the rest of the rows arrive as changes within a second or two. To serve only complete snapshots, compare the consumer’s position with the topics’ end offsets before you start the HTTP server.

What you learned

• Why a browser needs a bridge to read Kafka, and why Server-Sent Events suit a one-way stream.
• How compacted topics keyed by ID hold current state, and how tombstones delete keys.
• Why each writer should own its own topic when several systems update the same records.
• How to apply a high-rate stream to Handsontable: one setSourceDataAtCell call per frame, by physical row, and updateData for structural changes.
• Why sorting and filtering don’t re-run on streamed values, and how to re-apply them when rows are added or removed.
• How to detect a stream that a proxy keeps open after the server stops.

Next steps

• Real-time updates via WebSocket — the same update technique with a simulated feed and no server.
• Saving data — the hooks and sources behind the write-back.
• Server-side data — page through datasets that don’t fit in memory.
• Kafka log compaction — how Kafka keeps the latest record per key.
• Confluent’s JavaScript client — the client’s API, configuration, and migration guide from kafkajs.