---
title: "Kafka in, Queen out"
description: "Produce with a Kafka client, consume with the Queen SDK, against one set of rows. No connector, no mirror-maker, no second copy."
---

> Queen MQ documentation, for AI agents
> Complete self-contained summary of Queen MQ: https://queenmq.com/llms-brief.txt
> Fetch that first when the question is about the product rather than about this page.
> Index of all pages: https://queenmq.com/llms.txt

# Kafka in, Queen out

Most systems that speak more than one protocol speak them over separate data. A connector copies
rows from one place to another, and from then on you own two copies and the lag between them.

This pair does not copy anything. The producer writes over the Kafka wire protocol on port 9092.
The consumer reads over Queen's own API on port 6632. Both address **the same rows**, at the same
time, and neither knows the other exists.

## Why it works

Two translations, and both of them are the facade's, not yours.

**A Kafka topic is a Queen queue.** The Kafka partition *index* is the Queen partition *name*,
written out in decimal. Topic `orders` partition 7 is queue `orders`, partition `"7"`. Nothing
converts and nothing is mapped at read time: they are one row in one table.

**A record is stored in an envelope.** A Queen payload is JSON and a Kafka key and value are
arbitrary bytes, so the facade wraps each record and base64s the byte fields:

```json
{ "k": "<base64 key, or null>",
  "v": "<base64 value, or null>",
  "h": [{ "k": "header-name", "v": "<base64>" }],
  "t": 1788264581846 }
```

`h` and `t` are **omitted entirely** when the record carries no headers and no producer
timestamp, so a reader has to treat both as optional rather than assume the shape. The consumer
below prints the raw payload once, before it decodes anything, so the translation is visible
rather than asserted.

## The code

### Producer (Kafka wire)

```js title="examples/cross-protocol/producer.mjs"
// Cross-protocol example, PRODUCER half.
//
//   npm install
//   node producer.mjs                  # foreground
//   node producer.mjs &                # detached from this shell
//
// Produces to the Kafka facade on 9092. Its partner, consumer.mjs, reads the
// same rows through Queen's own API on 6632. Nothing replicates between them:
// a Kafka topic IS a Queen queue, and both halves address the same rows.
//
//   TOPIC=orders PARTITIONS=32 node producer.mjs
//   BROKER=localhost:9092 TOPIC=orders node producer.mjs
//
// Split into its own process on purpose: run together, produce and consume
// compete for one Node event loop and each one's number is really a measure of
// the pair. Separately, each is its own measurement.

// Setup the broker like:
// docker run -d --name queen --network queen -e PG_HOST=queen-pg -e PG_PASSWORD=postgres 
// -e QUEEN_SERVER=http://localhost:6632 -e QUEEN_KAFKA_EMBEDDED=true -e QUEEN_KAFKA_ADVERTISED_ADDR=localhost:9092 
// -e QUEEN_KAFKA_DEFAULT_PARTITIONS=16 -e QUEEN_KV_WRITE_RATE=2000 -e QUEEN_KV_WRITE_BURST=4000 
// -p 6632:6632 -p 9092:9092 ghcr.io/queen-mq/queen:1.4.1

import Confluent from '@confluentinc/kafka-javascript'

const TOPIC = process.env.TOPIC ?? 'cross-topic-big'
const BROKER = process.env.BROKER ?? 'localhost:9092'

// The width to declare for TOPIC. Since M7 the facade stores a create's
// numPartitions as that topic's own width FLOOR: it is advertised at
// max(live lanes, this) instead of max(live lanes,
// QUEEN_KAFKA_DEFAULT_PARTITIONS).
//
// TWO THINGS THAT WILL BITE:
//
// 1. The floor is set ONCE, AT CREATE. A create against a topic that already
//    exists is answered TOPIC_ALREADY_EXISTS and changes nothing -- there is no
//    alter for it. To change a width you delete the topic and make it again. So
//    this has to run BEFORE anything produces to TOPIC, or the producer's own
//    auto-create wins and makes it at the broker default with no floor.
// 2. The floor REPLACES the broker default rather than being compared against
//    it. Declaring 4 where QUEEN_KAFKA_DEFAULT_PARTITIONS is 8 gives a 4-wide
//    topic, not an 8-wide one. It is a floor under the LIVE LANE COUNT, not
//    under the default.
const PARTITIONS = Number(process.env.PARTITIONS ?? 1000)

const { KafkaJS } = Confluent

const kafka = new KafkaJS.Kafka({
  'bootstrap.servers': BROKER,
  kafkaJS: { clientId: 'cross-producer' },
})

// Declare the width before a single record exists, then report what the broker
// actually advertises -- which is the only number that matters, and is what the
// consumer half will see as Queen partition NAMES "0".."N-1".
async function declareWidth () {
  const admin = kafka.admin()
  await admin.connect()
  try {
    const created = await admin.createTopics({
      topics: [{ topic: TOPIC, numPartitions: PARTITIONS, replicationFactor: 1 }],
    })
    // An ARRAY, not kafkajs's `{ topics: [...] }` — one more divergence to know
    // about when porting: this client returns the topic list directly.
    const [described] = await admin.fetchTopicMetadata({ topics: [TOPIC] })
    const width = described.partitions.length
    console.log(
      created
        ? `created ${TOPIC} declaring ${PARTITIONS} partitions -> advertised at ${width}`
        : `${TOPIC} already existed, so the declared ${PARTITIONS} was NOT applied -> still ${width}. ` +
          `Delete it, or set TOPIC=<new name>, to see a different width.`
    )
    return width
  } finally {
    await admin.disconnect()
  }
}

// One message per send, never awaited individually, so librdkafka's per-partition
// accumulator can batch across them. linger.ms is what makes that pay.
const producer = kafka.producer({
  kafkaJS: { acks: -1, idempotent: false, allowAutoTopicCreation: true },
  'linger.ms': 1,
})

const WINDOW = 10000
const LOG_EVERY = 10000

let sent = 0
let nextLog = LOG_EVERY
const startTime = Date.now()

process.on('SIGINT', async () => {
  console.log('flushing...')
  await producer.flush({ timeout: 10000 }).catch(() => {})
  await producer.disconnect().catch(() => {})
  process.exit(0)
})

await declareWidth()

await producer.connect()
console.log(`producing to ${TOPIC} via ${BROKER} (pid ${process.pid})`)

while (true) {
  const inflight = []
  for (let k = 0; k < WINDOW; k++) {
    inflight.push(producer.send({ topic: TOPIC, messages: [{ value: 'mex-' + (sent + k) }] }))
  }
  await Promise.all(inflight)
  sent += WINDOW

  // A threshold, never `sent % N === 0`: the counter advances in batch-sized
  // jumps and would step straight over the multiples.
  if (sent >= nextLog) {
    nextLog += LOG_EVERY
    const secs = (Date.now() - startTime) / 1000
    console.log(new Date().toISOString(), 'produced', sent, Math.round(sent / secs), 'msg/s avg')
  }
}
```
### Consumer (Queen SDK)

```js title="examples/cross-protocol/consumer.mjs"
// Cross-protocol example, CONSUMER half.
//
//   npm install
//   node consumer.mjs                  # foreground
//   node consumer.mjs &                # detached from this shell
//
//   TOPIC=orders node consumer.mjs     # must match the producer's TOPIC
//
// Reads, through Queen's OWN API on 6632, the rows a Kafka producer wrote
// through the facade on 9092. Two translations make that work:
//
// 1. A Kafka topic IS a Queen queue, and the Kafka partition INDEX is the Queen
//    partition NAME in decimal. Nothing converts; they are the same rows.
//
// 2. The facade wraps each record in an envelope, base64 in the byte slots,
//    because a Queen payload is JSON and a Kafka key/value is arbitrary bytes:
//      { k: <base64 key|null>, v: <base64 value|null>,
//        h: [{ k: name, v: <base64> }],   // omitted when there are none
//        t: <producer timestamp ms> }     // omitted when unset

import { Queen } from 'queen-mq'

// Read from the environment, exactly as cross-producer.mjs does, so that ONE
// variable moves both halves. They were separately hardcoded before, which is a
// good way to spend ten minutes watching a consumer "hang" on an empty topic
// while the producer fills a different one.
const TOPIC = process.env.TOPIC ?? 'cross-topic'
const GROUP = process.env.GROUP ?? 'queen-side-group'
const QUEEN = process.env.QUEEN ?? 'http://localhost:6632'
const LOG_EVERY = 10000

const queen = new Queen(QUEEN)

function decode (data) {
  return {
    key: data.k == null ? null : Buffer.from(data.k, 'base64').toString(),
    value: data.v == null ? null : Buffer.from(data.v, 'base64').toString(),
    headers: (data.h ?? []).map(h => ({
      name: h.k,
      value: h.v == null ? null : Buffer.from(h.v, 'base64').toString(),
    })),
    timestamp: data.t ?? null,
  }
}

let count = 0
let nextLog = LOG_EVERY
let shownEnvelope = false
const startTime = Date.now()

console.log(`consuming ${TOPIC} as group ${GROUP} via ${QUEEN} (pid ${process.pid})`)

await queen.queue(TOPIC)
  .group(GROUP)
  // The Queen equivalent of fromBeginning: a new group otherwise starts at the
  // messages arriving from now on.
  .subscriptionMode('all')
  .concurrency(16)
  // consume() hands the handler an ARRAY and acks the whole batch in one call.
  // .each() switches to per-message, which is one ack round trip each.
  .consume(async (messages) => {
    // Show the raw envelope once, from the first REAL batch. Doing this with a
    // throwaway consumer group instead would register a cursor that never
    // advances -- and retention rule 2 is capped at MIN(committed) across a
    // partition's consumer rows, so that idle group would pin the whole backlog
    // from ever being reclaimed (server/sql/procedures/006_log_maintenance.sql).
    if (!shownEnvelope) {
      shownEnvelope = true
      console.log('raw Queen payload as stored by the Kafka facade:')
      console.log(' ', JSON.stringify(messages[0].data))
      console.log('  decoded:', JSON.stringify(decode(messages[0].data)), '\n')
    }

    count += messages.length

    // A threshold, never `count % N === 0`: the pop autopilot sizes each batch
    // from live queue state, so the counter advances in irregular jumps and
    // would step straight over the multiples -- going silent for minutes while
    // consuming perfectly well.
    if (count >= nextLog) {
      nextLog += LOG_EVERY
      const secs = (Date.now() - startTime) / 1000
      console.log(new Date().toISOString(), 'consumed', count, Math.round(count / secs), 'msg/s avg')
    }
  })
```

## Run it

Against a broker from [the quickstart](/start/quickstart), with the Kafka facade listening (see
[deploy/kafka](/deploy/kafka)):

```bash
cd examples/cross-protocol
npm install
TOPIC=orders PARTITIONS=32 node producer.mjs &
TOPIC=orders node consumer.mjs &
```

`TOPIC` drives both halves. They are separate processes on purpose: run together in one Node
process, the producer and the consumer compete for one event loop, and each number you read is
really a measurement of the pair.

The consumer prints the stored envelope once, then the decoded records:

```text
raw Queen payload as stored by the Kafka facade:
  {"k":null,"t":1788264581846,"v":"bWV4LTYwMDAw"}
  decoded: {"key":null,"value":"mex-60000","headers":[],"timestamp":1788264581846}
```

## What the dashboard shows, and why it differs

Watch **Consumer Groups** while both halves run, and the difference between the two protocols is
visible in the counters.

A Kafka consumer reads through `POST /api/v1/fetch`, which takes no lease, claims nothing and
moves no cursor. Its progress lives in committed offsets in `queen.kv`. So a topic consumed only
by Kafka clients shows `0.0 POP`, `0.0 ACK` and a `pending` count that never falls, because only
retention removes those messages. That is correct, not a stall.

The consumer here is a native one, so it pops and acks. The same queue, read the other way, moves
the pop and ack counters and drains `pending` as it goes.

> **Three things that will catch you**
>
> **`TOPIC` has to match both halves.** They are separate processes with separate defaults. A
> consumer pointed at a topic nobody is producing to long-polls an empty queue and looks stalled.
>
> **A width is declared once, at create.** `PARTITIONS` reaches the broker as CreateTopics
> `numPartitions`, which is stored as that topic's width floor. A create against a topic that
> already exists is answered `TOPIC_ALREADY_EXISTS` and changes nothing, and there is no alter for
> it. The producer therefore declares the width before it connects, because its own auto-create
> would otherwise win and make the topic at the broker default. To change a width, use a new topic.
>
> **The floor replaces the broker default, it does not compete with it.** Declaring 4 where
> `QUEEN_KAFKA_DEFAULT_PARTITIONS` is 8 gives a topic 4 lanes wide, not 8. It is a floor under the
> live lane count, not under the default.

## Going the other way

This example proves Kafka in, Queen out. The reverse direction, a native push read by a Kafka
consumer, is the same two translations applied backwards, and the record decoder is written to
cope with a payload that is not an envelope at all. It is not measured here, so treat it as
untested rather than as working.

Source: https://queenmq.com/use/full-examples/cross-protocol/index.mdx