Skip to content

Kafka clients on Queen through the Kafka port

Turn on the Kafka port and point unchanged Kafka producers and consumers at Queen: a topic is a queue, a key is a partition, and consumer groups are Queen's groups.

Updated View as Markdown

Kafka producers, consumers and admin tools can use Queen as they are: change bootstrap.servers, and nothing else. A topic is a Queen queue and a Kafka consumer group is a Queen consumer group, so Kafka clients and Queen clients read and write the same data.

Why
moving away from Kafka should not mean rewriting every service on the same day. With the Kafka port, you move one service at a time, and the services not moved yet keep working on the same topics.

Turn it on

docker run -d --name queen -p 6632:6632 -p 127.0.0.1:9092:9092 \
  -v /var/lib/queen:/var/lib/queen/raft \
  -e QUEEN_KAFKA_EMBEDDED=true \
  -e QUEEN_KAFKA_ADVERTISED_ADDR=127.0.0.1:9092 \
  ghcr.io/queen-mq/queen:2.6.0

QUEEN_KAFKA_ADVERTISED_ADDR is the address Kafka clients come back to after their first request, so they must be able to reach it. It has no default: without it the node refuses to start. Send three keyed records with kcat, and read them back:

kcat -P -b 127.0.0.1:9092 -t orders -K: <<'EOF'
customer-123:{"orderId":8891,"total":42}
customer-456:{"orderId":8892,"total":17}
customer-123:{"orderId":8893,"total":5}
EOF
kcat -C -b 127.0.0.1:9092 -t orders -o beginning -e -q -f '%t [%p] offset %o  key=%k  %s\n'
orders [307] offset 0  key=customer-123  {"orderId":8891,"total":42}
orders [307] offset 1  key=customer-123  {"orderId":8893,"total":5}
orders [400] offset 0  key=customer-456  {"orderId":8892,"total":17}

Nobody created orders: kcat’s first request did, 1,024 partitions wide. kcat hashed each key onto a partition, and both orders of customer-123 landed in partition 307, in order. The HTTP API reads the same records. A Kafka key and value are bytes, so each message carries them in base64 (k, v), with the producer’s timestamp (t):

curl -s 'localhost:6632/api/v1/pop/queue/orders/partition/307?batch=10' \
  | jq -c '.messages[].data'
{"k":"Y3VzdG9tZXItMTIz","t":1792837397814,"v":"eyJvcmRlcklkIjo4ODkxLCJ0b3RhbCI6NDJ9"}
{"k":"Y3VzdG9tZXItMTIz","t":1792837397814,"v":"eyJvcmRlcklkIjo4ODkzLCJ0b3RhbCI6NX0="}

One consumer group for both

Run a Kafka group consumer, and stop it once it has printed the three records:

kcat -b 127.0.0.1:9092 -G billing -o beginning -q -f 'billing [%p] offset %o key=%k\n' orders
billing [400] offset 0 key=customer-456
billing [307] offset 0 key=customer-123
billing [307] offset 1 key=customer-123

billing is a Queen consumer group now, with its lag in the dashboard (/api/v1/consumer-groups lists it with "kind":"kafka"). A Queen consumer of billing starts where the Kafka consumer stopped:

curl -s -X POST localhost:6632/api/v1/push -H 'content-type: application/json' -d '{"items":[
  {"queue":"orders","partition":"307","payload":{"orderId":8894,"total":99}}]}' >/dev/null
Q=localhost:6632/api/v1/pop/queue/orders/partition/307
curl -s "$Q?consumerGroup=billing&batch=10" | jq -c '.messages[] | {offset, data}'
{"offset":2,"data":{"orderId":8894,"total":99}}

And a Kafka consumer reads what a Queen client pushed: a record with no key, whose value is the JSON.

How Kafka maps onto Queen

Kafka Queen
Topic A queue of the same name, created by CreateTopics or by the first request that names it
Partition n The partition named "n"
Partition count What CreateTopics asked for, else 1,024. A topic never gets narrower
Offset, committed offset The partition’s offset; the group’s position on that partition
acks=1, acks=all One path: answered when the push is committed in the raft log
Replication factor The raft voters: every voter holds every partition
Idempotent producer Queen’s dedup: producer id, epoch and sequence are the transactionId
Transaction Kept in the leader’s memory, committed with its offsets as one raft entry
retention.ms The queue’s retention, set when the topic is created
cleanup.policy=compact Retention off: every record is kept
Left, what kcat sees: topic orders with 1,024 partitions; the key customer-123 hashes to partition 307; the consumer group billing commits offsets. Right, what Queen stores: the queue orders, the partition named 307, each record one message whose payload holds the key, value and timestamp in base64, and the Queen consumer group billing, whose position is the committed offset. HTTP clients read and write the same data.Left, what kcat sees: topic orders with 1,024 partitions; the key customer-123 hashes to partition 307; the consumer group billing commits offsets. Right, what Queen stores: the queue orders, the partition named 307, each record one message whose payload holds the key, value and timestamp in base64, and the Queen consumer group billing, whose position is the committed offset. HTTP clients read and write the same data.
What a Kafka client sees, and what Queen stores.

On a cluster every node runs the Kafka port, and together they are one Kafka cluster. The raft leader leads every partition and coordinates every group; a client sent to another node is told to go to the leader. Give each node its own QUEEN_KAFKA_ADVERTISED_ADDR, and never put the nodes behind one load-balanced address: a client sent to the leader would land on another node, and loop. A load balancer is fine in bootstrap.servers, which a client uses once.

A client finds the cluster through bootstrap.servers, which may be a load balancer, used once. The metadata it gets back lists each node at its own advertised address and names the raft leader, node 2, as the leader of every partition, so the client then talks to node 2 directly. If every node advertised the load balancer's address, a client sent to the leader would land on another node, be told to go to the leader, and loop.A client finds the cluster through bootstrap.servers, which may be a load balancer, used once. The metadata it gets back lists each node at its own advertised address and names the raft leader, node 2, as the leader of every partition, so the client then talks to node 2 directly. If every node advertised the load balancer's address, a client sent to the leader would land on another node, be told to go to the leader, and loop.
One Kafka cluster: bootstrap once, then straight to the raft leader.

Five things to know

  • Kafka consumers take no lease. Queen’s retries and dead-letter queue do not apply to them: a Kafka consumer handles its own failures, as on Kafka. Kafka and Queen consumers share a group’s position, not its work, so hand a group from one kind to the other, and do not run both at once.
  • Committed offsets never expire. Kafka forgets an empty group after seven days; here a group keeps its position until you delete it.
  • Transactions live in the leader’s memory, up to 8 MiB, 50,000 records and 1,000 partitions each. A leader change aborts the transactions that were open.
  • The Kafka port has no TLS in 2.6, and SASL passwords travel in clear: keep it on a private network.
  • Advertise an address the client can reach, as it will try it. On macOS, localhost is tried as ::1 first, so a port published on 127.0.0.1 only fails with localhost: use 127.0.0.1.

See also

Edit this page
Navigation

Type to search…

↑↓ navigate↵ selectEsc close