The Consumer API

The idea in one sentence

A Kafka consumer subscribes to topics and then runs a consume loop — KafkaJS calls your eachMessage handler with records from the partitions assigned to it, and that running loop is what keeps the consumer alive in its group.

Subscribe, then run

You create a consumer from a Kafka client with a groupId, connect(), subscribe() to one or more topics, and start run(). KafkaJS hands each record to your eachMessage callback. The key and value arrive as raw Buffers — you deserialize them yourself (e.g. .toString() or JSON.parse), the mirror of the producer serializing.

import { Kafka } from 'kafkajs';

const kafka = new Kafka({ clientId: 'order-processor', brokers: ['localhost:9092'] });
const consumer = kafka.consumer({ groupId: 'order-processor' }); // consumer group membership

async function main() {
  await consumer.connect();
  await consumer.subscribe({ topic: 'orders', fromBeginning: false }); // false = only new records (latest)

  // run() drives the poll loop for you and calls eachMessage per record
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      // key/value are Buffers — deserialize them yourself
      console.log(
        `p=${partition} off=${message.offset} key=${message.key?.toString()} val=${message.value?.toString()}`,
      );
    },
  });
}

main();

Why the run loop matters

run() does far more than fetch data. Under the hood it polls Kafka, sends heartbeats, participates in group coordination, and fetches partition assignments. If your eachMessage blocks too long — processing a record takes longer than the session timeout — the group coordinator assumes the consumer is dead and triggers a rebalance, reassigning its partitions elsewhere. The rule: keep per-message processing bounded.

flowchart LR
  sub["subscribe(topics)"] --> run["consumer.run(...)"]
  run --> each["eachMessage per record"]
  each --> proc["process the record"]
  proc --> run
  run -->|also| hb["send heartbeats + coordinate group"]

The consumer run loop

Each record carries its coordinates

Every message exposes its topic, partition, offset, key, value, timestamp, and headers. Those coordinates are what you use to commit progress and to reason about ordering — the consumer sees records in offset order within each partition, though records from different partitions may interleave.

🌊 The one-line takeaway

A consumer subscribe()s to topics and starts consumer.run({ eachMessage }), which drives the poll loop for you — heartbeats, group coordination, fetching assignments — so if your handler blocks too long the group rebalances your partitions away. Records arrive in offset order within a partition, each tagged with topic, partition, and offset; you deserialize key and value yourself.

What does consumer.run() (the poll loop) do besides returning records?

Nothing else — it only fetches data It also sends heartbeats, coordinates group membership, and fetches partition assignments It commits offsets to a database It creates new partitions

What happens if a consumer stops consuming (its handler blocks) for too long?

Kafka speeds up delivery The group coordinator assumes it is dead and rebalances its partitions to other consumers The consumer automatically scales up Nothing; consuming is optional

How do you get a typed key or value from a KafkaJS message?

Compress records before sending The key and value arrive as Buffers — you deserialize them yourself (e.g. toString or JSON.parse) Assign partitions manually Commit offsets first

In what order does a consumer see records?

Random order across the whole topic In offset order within each partition, though partitions may interleave across batches Newest first, always Sorted by key

Submit

Mark as complete