KafkaJS Documentation

KafkaJS is a modern Apache Kafka® client for Node.js, compatible with Kafka 0.10+ and offering native support for 0.11 features. Key features include:

• Producer: Send messages to topics.
• Consumer Groups: Support for pause, resume, and seek.
• Transactions: Transactional support for both producers and consumers.
• Message Headers: Support for metadata in messages.
• Compression: Native GZIP support, with pluggable codecs for Snappy, LZ4, and ZSTD.
• Security: Plain, SSL, and SASL_SSL implementations; support for SCRAM-SHA-256, SCRAM-SHA-512, and AWS IAM authentication.
• Admin Client: Manage Kafka cluster resources.

Kafka treats messages as raw byte sequences (key-value pairs). When using JSON, the producer must stringify the object into a Buffer, and the consumer must convert the Buffer back to a string before parsing it with JSON.parse().

Note: JSON is schemaless, meaning there are no guarantees regarding field presence or data types, which can lead to errors if producers change formats without notice.

await producer.send({
  topic,
  messages: [{
    key: 'my-key',
    value: JSON.stringify({ some: 'data' })
  }]
})

const eachMessage = async ({ /*topic, partition,*/ message }) => {
  // From Kafka's perspective, both key and value are just bytes
  // so we need to parse them.
  console.log({
    key: message.key.toString(),
    value: JSON.parse(message.value.toString())
  })

  /**
   * { key: 'my-key', value: { some: 'data' } }
   */
}

You can pause consumption to handle external pressure (e.g., a database being overloaded). Pausing a topic prevents it from being fetched in the next cycle.

Important: Calling pause or resume while the consumer is not running will throw an error.

await consumer.connect()
await consumer.subscribe({ topics: ['jobs'] })

await consumer.run({ eachMessage: async ({ topic, message }) => {
    try {
        await sendToDependency(message)
    } catch (e) {
        if (e instanceof TooManyRequestsError) {
            consumer.pause([{ topic }])
            setTimeout(() => consumer.resume([{ topic }]), e.retryAfter * 1000)
        }

        throw e
    }
}})

To use KafkaJS, create a new Kafka instance. You must provide at least one broker from your cluster. These are used as seed brokers to bootstrap the client and load initial metadata. You should also provide a clientId, which is a logical identifier for your application used for request tracing and quotas.

const { Kafka } = require('kafkajs')

// Create the client with the broker list
const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092']
})

The Admin Client is used for cluster operations like creating topics, managing partitions, and resetting offsets. You obtain an admin instance from a Kafka instance. Always remember to connect() before performing operations and disconnect() when finished.

const kafka = new Kafka(...)
const admin = kafka.admin()

// remember to connect and disconnect when you are done
await admin.connect()
await admin.disconnect()

To run the documentation website in development mode, ensure you have installed the dependencies using yarn and then start the dev server with yarn start.

# Install dependencies
$ yarn

# Start the site
$ yarn start

To use an authentication mechanism not supported out of the box by KafkaJS, you can provide a custom authenticationProvider within the sasl configuration object.

Your provider must return an Authenticator object containing an authenticate() method. The authenticationProvider function receives an object containing host, port, logger, and saslAuthenticate.

Use saslAuthenticate to perform the actual protocol handshake. This function requires a request object with an encode() method to prepare the auth_bytes. If the broker returns auth_bytes, you should provide a response object with decode() and parse() methods to handle the incoming data.

To improve IntelliSense and type-hinting in Visual Studio Code for the project, add a jsconfig.json file to the root directory. This allows the Javascript Language Service to resolve paths like testHelpers and include both src and testHelpers in the project scope.

{
  "compilerOptions": {
    "baseUrl": ".",
    "module": "commonjs",
    "target": "es6",
    "paths": {
      "testHelpers": ["./testHelpers"]
    }
  },
  "include": [
    "src",
    "testHelpers"
  ]
}

To connect to a Kafka broker requiring SSL and SASL authentication (e.g., using scram-sha-256), include the ssl and sasl configuration objects in the Kafka constructor.

• ssl: Use rejectUnauthorized: true for secure connections.
• sasl: Specify the mechanism (e.g., 'scram-sha-256'), username, and password.

const { Kafka, logLevel } = require('kafkajs')

const kafka = new Kafka({
  logLevel: logLevel.INFO,
  brokers: ['localhost:9094'],
  clientId: 'example-consumer',
  ssl: {
    rejectUnauthorized: true
  },
  sasl: {
    mechanism: 'scram-sha-256',
    username: 'test',
    password: 'testtest',
  },
})

const consumer = kafka.consumer({ groupId: 'test-group' })
// ... proceed with connect, subscribe, and run

The transactionalId is used by Kafka to fence out 'zombie' instances by rejecting writes from older producers sharing the same ID.

To ensure EoS in stream processing (the read-process-write cycle), the transactionalId must be consistent for a given input topic and partition. A recommended pattern is to encode the topic and partition into the ID, for example: "myapp-producer-" + topic + "-" + partition.

For finer control, you can pause or resume specific partitions of a topic. This allows other partitions in the same topic to continue processing even if one partition is being throttled due to a slow process or dependency error.

consumer.run({
    partitionsConsumedConcurrently: 3, // Default: 1
    eachMessage: async ({ topic, partition, message }) => {
      try {
            await sendToDependency(message)
        } catch (e) {
            if (e instanceof TooManyRequestsError) {
                consumer.pause([{ topic, partitions: [partition] }])
                setTimeout(() => {
                    consumer.resume([{ topic, partitions: [partition] }])
                }, e.retryAfter * 1000)
            }
            throw e
        }
    },
})