Confluent JavaScript Client for Apache Kafka

The Confluent JavaScript Client is a client library for Apache Kafka®, developed and maintained by Confluent, that produces messages to and consumes messages from Kafka topics. It wraps librdkafka, a C client library for Kafka, to provide high performance and reliability.

The client offers these APIs, which are covered in this guide:

The promisified API exposes methods that return promises.
The callback-based API exposes methods that use callbacks and events.

To get started, choose one of the following topics based on your scenario:

For a new deployment, use the promisified API.
If you’re migrating from KafkaJS, use the KafkaJS migration guide.
If you’re migrating from node-rdkafka, use the node-rdkafka migration guide.

For more API details, see the JavaScript Client API reference.

Install the JavaScript Client

You can install the JavaScript Client using the prebuilt binaries on supported platforms or using C/C++ compilation for platforms that don’t support binaries.

Install with binary

The following configurations support the JavaScript Client binary:

Node.js: the LTS versions 18 and 20, plus the latest versions 21 and 22.
Linux on x64, arm64, and s390x: both glibc and musl/alpine. s390x is the CPU architecture for IBM Z mainframes and LinuxONE servers.
Apple macOS: arm64/m1. Confluent supports macOS on Intel on a best-effort basis.
Windows: x64.

Confluent offers experimental supports for Yarn and Performant Node Package Manager (pnpm).

To install the JavaScript Client on a supported platform, run the following command:

npm install @confluentinc/kafka-javascript

Install with C/C++ compilation

To install the JavaScript Client on a system that doesn’t support binaries, you need to use a C/C++ compiler. The node-gyp build tool requires Python. For details, see Configuring Python Dependency

To install the client with a compiler:

Install a recent version of Python.
Install a C++ compiler that supports C++20 or later, such as gcc or clang.
Install librdkafka. The following example installs librdkafka on Debian and Ubuntu platforms:
```
sudo mkdir -p /etc/apt/keyrings
wget -qO - https://packages.confluent.io/deb/7.8/archive.key | gpg --dearmor | sudo tee /etc/apt/keyrings/confluent.gpg > /dev/null
sudo apt-get update
sudo apt install librdkafka-dev
```
It is easiest to install librdkafka from your package manager based on your target machine. If necessary, you can also build it from the source

For detailed installation instructions, see the librdkafka repository.
Set the environment variables. The following example is for a Unix-like platform:
```
export CKJS_LINKING=dynamic
export BUILD_LIBRDKAFKA=0
```
Run npm install to build the library and link it against your librdkafka installation:
```
npm install @confluentinc/kafka-javascript
```
Note

If npm install takes more than a couple of minutes on macOS, see GitHub issue 103.

If you encounter other issues, search for similar issues in the JavaScript repository or file one.

Promisified API

The promisified API exposes methods that return promises. Import it in your code:

const {
  Kafka,
  ErrorCodes, CompressionTypes, // You can specify additional optional properties for further configuration.
} = require('@confluentinc/kafka-javascript').KafkaJS;

Configure the promisified API

The JavaScript Client supports many librdkafka configuration options.

JavaScript Client only supports the rebalance_cb callback for the consumer and the oauthbearer_token_refresh_cb callback for all clients.

You can use a kafkaJS configuration key to specify an object whose keys are a limited set of KafkaJS properties. Use this key only if you’re migrating from KafkaJS. For more information, see the KafkaJS migration guide.

Produce messages with the promisified API

A Producer sends messages to Kafka. The following example creates an instance and sends a message:

const producer = new Kafka().producer({
    'bootstrap.servers': '<fill>',
});

await producer.connect();

const deliveryReports = await producer.send({
    topic: 'test-topic',
    messages: [
      { value: 'v1', key: 'x' },
    ]
});

console.log({deliveryReports});
await producer.disconnect();

The send method sends messages to the Kafka cluster and returns an array of delivery reports, one for each message sent. Before using the send method, connect the producer.

A single send can send many messages to a topic. Besides the value, a message can contain other fields. For example:

const message = {
  // An optional key for the message. If a key is provided, all messages with the same key will be sent to the same partition. It can be a string, a Buffer, or null.
  key: null,
  // The value of the message. This can be a Buffer or a string, and must be set.
  value: Buffer.from('Hello'),
  // The partition to send the message to. It may be a number, or be unset. If not set, the partition will be chosen by the partitioner based on the key.
  partition: 1,
  // Headers to be sent along with the message. May be unset.
  headers: {
    // Header keys are strings.
    // Header values can be strings.
    'header-key-0': 'header-value',
    // Header values can be buffers, too.
    'header-key-1': Buffer.from('header-value'),
    // One or more values can be sent for a single key.
    'header-key-3': [Buffer.from('header-value-0'), Buffer.from('header-value-1')],
  }
  // The timestamp of the message in milliseconds since the Unix epoch. If not set, the broker will set the timestamp.
  timestamp?: string
}

Batch, send, and flush messages

The library batches messages internally on a per-partition basis. The following properties control batching:

linger.ms: Sets the maximum time the library waits to accumulate messages into a single batch.
batch.num.messages or batch.size: Sets the upper limit for the size of a batch.

If you await send, the library batches together only the messages in that send.

For the best throughput, call send without awaiting each call, then await all the calls together. You can also increase linger.ms to accumulate more messages in each batch, which reduces network and CPU overhead:

const reportPromises = [];
for (const topicMessages of topicsMessages) {
  reportPromises.push(
    producer.send({topic: topicMessages.topic, messages: topicMessages.messages}));
}

await reportPromises;

For the lowest latency for each message, await each send individually and set linger.ms to 0 or 1:

const producer = new Kafka().producer({
    'bootstrap.servers': '<fill>',
    'linger.ms': 1,
});

await producer.connect();
await producer.send(/* ... */);

To send messages immediately when you use a larger linger.ms, call flush, which sends all messages in the internal buffer without waiting for linger.ms to expire:

 // Unawaited send.
const reportPromise = producer.send(/*...*/);

// Try to send everything, waiting up to 1s. If we weren't able to finish sending in 1s, throw an error.
await producer.flush({ timeout: 1000 });

Warning

If you increase the value of linger.ms, don’t await your send calls, because each await can take up to linger.ms to resolve. If you need the delivery report, call flush after sending.

For further details, see compression and retries/reliability.

Promisified producer methods

The promisified API supports the following producer methods:

Method	Description
`async connect()`	Connects to the broker. Call this method one time before using the producer.
`async disconnect()`	Flushes the producer, disconnects the producer, and cleans up connections and resources.
`async send()`	Sends messages to a Kafka topic. Returns the delivery reports when you await it.
`async sendBatch()`	Sends messages to different Kafka topics. Returns the delivery reports when you await it.
`async flush()`	Flushes any messages buffered in the library and attempts to send them immediately.
`async transaction()`	Starts a new transaction. See Transactional producer and exactly-once semantics.
`async sendOffsets()`	Commits offsets within a transaction. See Transactional producer and exactly-once semantics.
`async commit()`	Commits an ongoing transaction. See Transactional producer and exactly-once semantics.
`async abort()`	Cancels an ongoing transaction. See Transactional producer and exactly-once semantics.
`dependentAdmin()`	Returns an Admin client, which uses this producer for its underlying connections.
`setSaslCredentials()`	Sets the SASL username and password for the next login attempt. Use it for SASL PLAIN or SASL SCRAM authentication.

See the JavaScript Client API reference documentation for more details.

Consume messages with the promisified API

A consumer receives messages from Kafka.

You must connect the consumer before calling run. The run method starts the consumer loop, which polls the cluster and calls the eachMessage callback for every message the consumer gets from the cluster.

The following example creates an instance and starts consuming messages:

const consumer = new Kafka().consumer({
    'bootstrap.servers': '<fill>',
    'group.id': 'test-group', // Mandatory property for a consumer - the consumer group id.
});

await consumer.connect();
await consumer.subscribe({ topics: ["test-topic"] });

consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    console.log({
      topic,
      partition,
      headers: message.headers,
      offset: message.offset,
      key: message.key?.toString(),
      value: message.value.toString(),
    });
  }
});

// Whenever we're done consuming, maybe after user input or a signal:
await consumer.disconnect();

The message can contain other fields besides the value. For example:

{
  // Key of the message - may not be set.
  key: Buffer.from('key'),
  // Value of the message - will be set.
  value: Buffer.from('value'),
  // The timestamp set by the producer or the broker in milliseconds since the Unix epoch.
  timestamp: '1734008723000',
  // The current epoch of the leader for this partition.
  leaderEpoch: 2,
  // Size of the message in bytes.
  size: 6,
  // Offset of the message on the partition.
  offset: '42',
  // Headers that were sent along with the message.
  headers: {
    'header-key-0': ['header-value-0', 'header-value-1'],
    'header-key-1': Buffer.from('header-value'),
  }
}

The consumer considers a message processed successfully when eachMessage for that message runs to completion without throwing an error. If eachMessage throws an error, the consumer marks the message unprocessed and calls eachMessage with the same message again.

Commit offsets

By default, the library commits offsets automatically. Any consumer in the same group can resume from the last committed offset after a crash or restart.

Every 5,000 ms by default, the library commits the last successfully processed offset for each partition to the broker, which is the offset for which eachMessage ran to completion. Disconnecting the consumer or a rebalance also commits offsets. To change the interval, set the auto.commit.interval.ms property.

To commit offsets manually, set enable.auto.commit to false:

const consumer = new Kafka().consumer({
    'bootstrap.servers': '<fill>',
    'group.id': 'test-group',
    'enable.auto.commit': false,
});

consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    console.log(message);
    if (isItTimeToCommit) {
      await consumer.commitOffsets([
        {
          topic,
          partition,
          // We are committing offset + 1, not offset.
          // Kafka expects the offsets as 'the next offset to read', not 'the last offset processed'.
          offset: (parseInt(message.offset, 10) + 1).toString(),
        },
      ]);
    }
  },
});

// commitOffsets can also be called without an argument to commit all the successfully
// processed offsets up to that point in time.
setInterval(() => {
  consumer.commitOffsets();
}, 1000);

Warning

If you commit offsets manually, set up a rebalance_cb and commit offsets on ERR__REVOKE_PARTITIONS.

Pause, resume, and seek partitions

You can pause consumption of owned partitions. The library makes no more requests to the broker to fetch messages for a paused partition until you resume it:

// Pause topic, partitions 0 and 1.
const resumeMethod = consumer.pause([{ topic, partitions: [0, 1] }]);

// Resume - after a while.
resumeMethod();

// Can also resume manually specifying partitions.
consumer.resume([{ topic, partitions: [1] }]);

Pausing a partition that isn’t in the assignment has no effect.

You can seek to specific offsets of a partition, which is useful for reprocessing or skipping messages:

consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    if (restartFromEarliest) {
      // offset can be a positive number (in a string), or be '-2' representing the 'earliest' offset available on the broker.
      consumer.seek({topic, partition, offset: '-2'});
    }
  },
});

If a partition belongs to the assignment, the seek takes effect immediately. Otherwise, the library stores the seek and applies it if a later rebalance assigns that partition to the consumer.

If enable.auto.commit is true, which is the default, seeking to an offset also commits that offset to the broker.

Configure concurrent and batch consumption

The consumer can consume messages concurrently. If your message processing involves asynchronous operations, pass a number to the partitionsConsumedConcurrently option of the run method. The library dispatches up to that number of eachMessage calls at the same time. For a single partition, the library guarantees message order, so at most one message from a partition triggers eachMessage at a time.

The consumer can also consume messages in batches. Provide an eachBatch callback to the run method instead of the eachMessage callback. The batch size has a fixed upper limit of 32. The batches passed to the callback are logical batches that the library creates by accumulating messages from the broker. They don’t correspond to the actual batches of messages on the broker.

The library can’t tell when the consumer processes individual messages in a batch, so it commits offsets for the entire batch when the eachBatch callback runs to completion. If the callback throws an error, the library commits no offsets and dispatches the callback again for the same batch.

For finer-grained control, the argument passed to eachBatch contains a resolve method. Calling this method lets the library commit offsets for the messages processed up to that point, even if an error occurs while processing later messages in the batch. To turn off auto-resolution and use only the resolve method, pass eachBatchAutoResolve: false to the run method. The partitionsConsumedConcurrently option also applies to eachBatch:

consumer.run({
    partitionsConsumedConcurrently,
    eachBatchAutoResolve: isAutoResolve,
    eachBatch: async event => {
        console.log(event.batch.messages);

        // If we're not auto-resolving, we need to resolve the offsets manually.
        // Note that unlike commit, we don't need to add 1 here to the message offset.
        if (!isAutoResolve)
            event.resolveOffset(event.batch.messages[event.batch.messages.length - 1].message.offset);
    }
});

Promisified consumer methods

Method	Description
`async connect()`	Connects to the broker. Call this method one time before using the consumer.
`async disconnect()`	Disconnects consumer, triggering one final rebalance, and cleans up connections and resources.
`async subscribe()`	Subscribes the consumer to one or more topics. Can add to or replace the current subscription.
`async run()`	Starts the main consume loop, polling the broker, and triggering message process callbacks.
`assignment()`	Gets the current assignment of partitions for this consumer.
`async commitOffsets()`	Commits offsets to the broker, up to the last processed offset, or custom ranges specified in the argument.
`async committed()`	Fetches committed offsets for the assigned topic partitions, or a custom topic partition list.
`pause()`	Pauses consumption for a subset of assigned partitions.
`resume()`	Resumes consumption for paused partitions.
`paused()`	Allows checking which partitions are currently paused.
`seek()`	Seeks to a specific offset. For assigned partitions, the seek takes effect immediately. For unassigned partitions, the seek applies when a rebalance assigns the partition.
`dependentAdmin()`	Returns an Admin client which uses this consumer for its underlying connections.

See the JavaScript Client API reference documentation for more details.

Transactional producer and exactly-once semantics

The JavaScript Client supports idempotent producers, transactional producers, and exactly-once semantics (EOS).

To use an idempotent producer, set enable.idempotence to true:

const producer = new Kafka().producer({
    'bootstrap.servers': '<fill>',
    'enable.idempotence': true,
});

For the guarantees, limitations, and other configuration changes of an idempotent producer, see the librdkafka idempotent producer documentation.

To use a transactional producer, set a unique transactional.id:

const producer = new Kafka().producer({
    'bootstrap.servers': '<fill>',
    'transactional.id': 'my-transactional-id', // Must be unique for each producer instance.
});

await producer.connect();

// Start transaction.
await producer.transaction();
await producer.send({topic: 'topic', messages: [{value: 'message'}]});

// Commit transaction.
await producer.commit();

Specifying a transactional.id makes the producer transactional. The transactional.id must be unique for each producer instance. Connect the producer before starting a transaction with transaction(). You can’t use a transactional producer as a non-transactional producer, and every message must be within a transaction.

For the guarantees of a transactional producer, see the librdkafka transactional producer documentation.

A transactional producer also enables exactly-once semantics (EOS) in the specific case of consuming from a Kafka cluster, processing the message, and producing to another topic on the same cluster:

consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
        try {
            const transaction = await producer.transaction();
            await transaction.send({
                topic: 'produceTopic',
                messages: [
                    { value: 'consumed a message: ' + message.value.toString() },
                ]
            });

            await transaction.sendOffsets({
                consumer,
                topics: [
                    {
                        topic,
                        partitions: [
                            { partition, offset: String(Number(message.offset) + 1) },
                        ],
                    }
                ],
            });

            // The transaction assures that the message sent and the offset committed
            // are transactional, only reflecting on the broker on commit.
            await transaction.commit();
        } catch (e) {
            console.error(e);
            await transaction.abort();
        }
    },
});

Admin client for the promisified API

The JavaScript Client includes an admin client that provides methods to manage topics, groups, and other Kafka entities in the cluster. Create the admin client from configuration, or from a producer or consumer instance:

// An admin client can be created from configuration.
const admin = new Kafka().admin({
    'bootstrap.servers': '<fill>',
});

// Or from a producer or consumer instance.
const depAdmin = producer.dependentAdmin();

await admin.connect();
await depAdmin.connect();

For a complete list of admin client methods, see the JavaScript Client API reference documentation.

Handle errors in the promisified API

Methods from the library throw KafkaJSError when they encounter a problem. Each error object has a code property. Compare it against the ErrorCodes object exported by the library to decide how to handle the error:

const { Kafka, ErrorCodes, isKafkaJSError } = require('@confluentinc/kafka-javascript').KafkaJS;
try {
  await producer.send({ /*... */ });
} catch (e) {
  if (!isKafkaJSError()) throw e;
  if (err.code === ErrorCodes.ERR__MSG_TIMED_OUT) {
    // Some backoff and retry logic.
  }
}

After the client connects successfully, it stays usable despite any errors it throws. The client attempts to reconnect to the broker after a disconnection and continues to work if the cluster is reachable. Don’t stop clients on errors indiscriminately. Check the error and take specific action instead.

The error object has a fatal attribute. In rare cases, the error is fatal and the client becomes unusable. These cases relate to idempotent or transactional producers and static consumer group membership, which you configure yourself. When an error is fatal, stop the client and create a new one.

The error object also has retriable and abortable properties, which the library sets only when you use a transactional producer. The retriable property indicates that you can retry the failed operation. The abortable property indicates that you should call abort().

Configure logging

The library has a built-in logger, and you can also set a custom logger. The default logger logs to console:

const { logLevel } = require('@confluentinc/kafka-javascript').KafkaJS;

// You can use the built-in logger.
const libraryLogger = producer.logger();
libraryLogger.info('This is an info message');

// And modify the log level on the fly.
libraryLogger.setLogLevel(logLevel.INFO);

librdkafka logs most of the library’s internal workings at loglevel.DEBUG and uses a component-wise logging system.

To turn on debug logs, set the debug property in the configuration object to a comma-separated list of the components you want to debug, or to all to turn on all debug logs. For the list of available components, see the librdkafka configuration reference. Setting debug in the configuration also changes the built-in logger to start with a default log level of logLevel.DEBUG.

Note

You can’t change the debug configuration on the fly, so most debug logs are unavailable if you don’t set the debug property before connecting. Using libraryLogger.setLogLevel(logLevel.DEBUG) is therefore generally not useful for debugging library internals. Going from logLevel.DEBUG to logLevel.INFO or logLevel.ERROR is still useful.

A custom logger must conform to the following interface:

class MyLogger {
  constructor(); // You can freely change the signature of the constructor.
  setLogLevel(logLevel); // logLevel is a member of the logLevel enum.
  info(message, extra); // message is a string, extra is an object with additional information.
  error(message, extra);
  warn(message, extra);
  debug(message, extra);
}

// Usage
const customLogger = new MyLogger();
const producer = new Kafka().producer({
    'bootstrap.servers': '<fill>',
    'debug': 'all',
    "kafkaJS": {
        "logger": customLogger,
    }
});

Callback-based API

The callback-based API exposes methods that use callbacks and events. Import it in your code:

const Kafka = require('@confluentinc/kafka-javascript');

Configure the callback-based API

The JavaScript Client supports many librdkafka configuration options. If you pass an unsupported configuration option, the client throws an error.

Configuration keys with the suffix _cb are callbacks:

Some of these keys are informational, and you can opt in to them, for example dr_cb.
Others are callbacks designed to return a value, such as partitioner_cb.

The JavaScript Client currently supports the following callbacks:

partitioner_cb
dr_cb or dr_msg_cb
event_cb
rebalance_cb. See Handle rebalances.
offset_commit_cb. See Commit offsets.
oauthbearer_token_refresh_cb

Check librdkafka features and version

The JavaScript Client includes two utility functions that report the status of your installation. When you report issues, include their output:

To get the features that your librdkafka build supports, read the features variable on the root of the confluent-kafka-javascript object:

const Kafka = require('@confluentinc/kafka-javascript');
console.log(Kafka.features);

// #=> [ 'gzip', 'snappy', 'ssl', 'sasl', 'regex', 'lz4' ]

To get the librdkafka version:

const Kafka = require('@confluentinc/kafka-javascript');
console.log(Kafka.librdkafkaVersion);

// #=> 2.3.0

Produce messages with the callback-based API

Initialize a producer

To produce messages with the callback-based API, create a Producer by passing a configuration object to its constructor. For example:

const producer = new Kafka.Producer({
  'bootstrap.servers': 'kafka-host1:9092,kafka-host2:9092'
});

A Producer requires only bootstrap.servers, a comma-separated list of the Kafka brokers. For other configuration options, see the librdkafka configuration options.

The following example sets these librdkafka options:

const producer = new Kafka.Producer({
  'client.id': 'kafka',
  'bootstrap.servers': 'localhost:9092',
  'compression.codec': 'gzip',
  'retry.backoff.ms': 200,
  'message.send.max.retries': 10,
  'socket.keepalive.enable': true,
  'queue.buffering.max.messages': 100000,
  'queue.buffering.max.ms': 1000,
  'batch.num.messages': 1000000,
  'dr_cb': true
});

Produce with the stream API

You can use the Producer as a writable stream immediately after creation. For example:

// Our producer with its Kafka brokers
// This call returns a new writable stream to our topic 'topic-name'
const stream = Kafka.Producer.createWriteStream({
  'bootstrap.servers': 'kafka-host1:9092,kafka-host2:9092'
}, {}, {
  topic: 'topic-name',
  autoClose: false,
  pollInterval: 1000,
});

// Writes a message to the stream
const queuedSuccess = stream.write(Buffer.from('Awesome message'));

if (queuedSuccess) {
  // This only indicates that the message was queued, not sent to Kafka.
  console.log('We queued our message!');
} else {
  console.log('Too many messages in our queue already');
}

// NOTE: MAKE SURE TO LISTEN TO THIS IF YOU WANT THE STREAM TO BE DURABLE
// Otherwise, any error will bubble up as an uncaught exception.
stream.on('error', (err) => {
  // Here's where we'll know if something went wrong sending to Kafka
  console.error('Error in our producer stream');
  console.error(err);
})

Note

To prevent your code from crashing when an error occurs, add an error listener to the stream. Most errors aren’t fatal, but a fatal error immediately destroys the stream. If you use autoClose, the stream closes itself at the first sign of a problem.

Produce with the standard API

The standard API has higher performance, particularly for high message volumes, but it requires more manual setup. The following example shows its use:

const producer = new Kafka.Producer({
  'bootstrap.servers': 'localhost:9092',
  'dr_cb': true
});

// Connect to the broker manually
producer.connect();

// Wait for the ready event before proceeding
producer.on('ready', () => {
  try {
    producer.produce(
      // Topic to send the message to
      'topic',
      // optionally we can manually specify a partition for the message
      // this defaults to -1 - which will use librdkafka's default partitioner (consistent random for keyed messages, random for unkeyed messages)
      null,
      // Message to send. Must be a buffer
      Buffer.from('Awesome message'),
      // for keyed messages, we also specify the key - note that this field is optional
      'Stormwind',
      // you can send a timestamp here. If your broker version supports it,
      // it will get added. Otherwise, we default to 0
      Date.now(),
      // you can send an opaque token here, which gets passed along
      // to your delivery reports
    );
  } catch (err) {
    console.error('A problem occurred when sending our message');
    console.error(err);
  }
});

// Any errors we encounter, including connection errors
producer.on('event.error', (err) => {
  console.error('Error from producer');
  console.error(err);
})

// We must either call .poll() manually after sending messages
// or set the producer to poll on an interval (.setPollInterval).
// Without this, we do not get delivery events and the queue
// will eventually fill up.
producer.setPollInterval(100);

// You can also set up the producer to poll in the background thread which is
// spawned by the C code. It is more efficient for high-throughput producers.
// Calling this clears any interval set in setPollInterval.
producer.setPollInBackground(true);

For the available configuration options, see the librdkafka configuration options.

Callback-based producer methods

Method	Description
`connect()`	Connects to the broker. The `connect()` method emits the `ready` event when it connects successfully. If it doesn’t, the client passes the error to the callback.
`disconnect()`	Disconnects from the broker. The `disconnect()` method emits the `disconnected` event when it has disconnected. If it doesn’t, the client passes the error to the callback.
`poll()`	Polls the producer for delivery reports or other events and delivers them through the `emitter`. To get the events in the librdkafka queue to emit, call this method regularly or use one of the methods that automatically poll the producer for you.
`setPollInterval(interval)`	Polls the producer on this interval, handling disconnections and reconnection. Set it to `0` to turn it off.
`setPollInBackground(set)`	Polls the producer on the librdkafka background thread. Pass a truthy value to turn it on, a non-truthy value to turn it off.
`produce(topic, partition, msg, key, timestamp, opaque)`	Sends a message. The `produce()` method throws when produce would return an error. Ordinarily, this happens only if the queue is full.
`flush(timeout, callback)`	Flushes the librdkafka internal queue, sending all messages. The default timeout is `500ms`.
`initTransactions(timeout, callback)`	Initializes the transactional producer.
`beginTransaction(callback)`	Starts a new transaction.
`sendOffsetsToTransaction(offsets, consumer, timeout, callback)`	Sends consumed `topic-partition-offsets` to the broker. The broker commits them along with the transaction.
`abortTransaction(timeout, callback)`	Cancels the ongoing transaction.
`commitTransaction(timeout, callback)`	Commits the ongoing transaction.
`setSaslCredentials(username, password)`	Changes the SASL credentials that the client sends on the next authentication attempt. Applies only if you use SASL authentication.

Producer events

Some configuration properties that end in _cb generate an event for that option. To receive the event, you have two options:

Set the value to true and react to the event
Provide a callback function directly

The following example uses an event:

const producer = new Kafka.Producer({
  'client.id': 'my-client', // Specifies an identifier to use to help trace activity in Kafka
  'bootstrap.servers': 'localhost:9092', // Connect to a Kafka instance on localhost
  'dr_cb': true // Specifies that we want a delivery-report event to be generated
});

// Poll for events every 100 ms
producer.setPollInterval(100);

producer.on('delivery-report', (err, report) => {
  // Report of delivery statistics here:
  //
  console.log(report);
});

The following table describes the event types:

Event	Description
disconnected	The client emits the `disconnected` event when the broker has disconnected, but only when you call `.disconnect`. Otherwise, the wrapper always tries to reconnect.
ready	The client emits the `ready` event when the Producer is ready to send messages.
event	The client emits the `event` event when librdkafka reports an event, if you opted in with the `event_cb` option.
event.log	The client emits the `event.log` event when logging events come in, if you opted in to logging with the `event_cb` option. To receive log information, set a value for `debug`.
event.stats	The client emits the `event.stats` event when librdkafka reports `stats`, if you opted in by setting `statistics.interval.ms` to a non-zero value.
event.error	The client emits the `event.error` event when librdkafka reports an error.
event.throttle	The client emits the `event.throttle` event when librdkafka reports throttling.
delivery-report	The client emits the `delivery-report` event when polling finds a delivery report. To use this event, set `request.required.acks` to `1` or `-1` in topic configuration, and set `dr_cb` to `true` in the `Producer` constructor options. To include the message payload in the report, set `dr_msg_cb` to `true` instead.

Use the high-level producer

The high-level producer is a variant of the producer that propagates callbacks to you when the client delivers a message. The rest of the class behavior is the same as the producer class:

const producer = new Kafka.HighLevelProducer({
  'bootstrap.servers': 'localhost:9092',
});

The high-level producer adds a callback to the produce call that tells you when the client delivers the message. You can’t specify opaque tokens with it:

producer.produce(topicName, null, Buffer.from('alliance4ever'), null, Date.now(), (err, offset) => {
  // The offset if our acknowledgment level allows us to receive delivery offsets
  console.log(offset);
});

You can also add serializers to change the key or value of a produce call before the client sends it to Kafka:

producer.setValueSerializer((value) => {
  return Buffer.from(JSON.stringify(value));
});

producer.setTopicValueSerializer((topic, value) => {
  // Process value based on topic it's being produced to.
  return processedValue;
});

Otherwise, the behavior of the class is the same.

Consume messages with the callback-based API

Initialize a consumer

To consume messages with the callback-based API, instantiate a KafkaConsumer object:

const consumer = new Kafka.KafkaConsumer({
  'group.id': 'kafka',
  'bootstrap.servers': 'localhost:9092',
}, {});

The first parameter is the global configuration.

A consumer requires the group.id and bootstrap.servers properties.
The second parameter is an optional topic configuration that gets applied to all subscribed topics.

For all supported configuration properties, see the librdkafka configuration reference. Look for the C and * keys.

Handle rebalances

By default, librdkafka manages rebalancing internally. To override this behavior, provide your own logic as a rebalance callback:

const consumer = new Kafka.KafkaConsumer({
  'group.id': 'kafka',
  'bootstrap.servers': 'localhost:9092',
  'rebalance_cb': (err, assignment) => {

    if (err.code === ErrorCodes.ERR__ASSIGN_PARTITIONS) {
      // Note: this can throw when you are disconnected. Take care and wrap it in
      // a try catch if that matters to you
      this.assign(assignment);
    } else if (err.code == ErrorCodes.ERR__REVOKE_PARTITIONS){
      // Same as above
      this.unassign();
    } else {
      // We had a real error
      console.error(err);
    }

  }
})

this refers to the KafkaConsumer you created. When you specify a rebalance_cb, you can also listen to the rebalance event. The internal librdkafka rebalancer doesn’t emit this event.

Commit offsets with the callback-based API

In confluent-kafka-javascript, the standard way to commit is to queue the commit request with the next librdkafka request to the broker. With this approach, you can’t know the result of the commit.

To get the result of the commit, listen to the offset_commit_cb callback:

const consumer = new Kafka.KafkaConsumer({
  'group.id': 'kafka',
  'bootstrap.servers': 'localhost:9092',
  'offset_commit_cb': (err, topicPartitions) => {

    if (err) {
      // There was an error committing
      console.error(err);
    } else {
      // Commit went through. Let's log the topic partitions
      console.log(topicPartitions);
    }

  }
})

this refers to the KafkaConsumer you created. When you specify an offset_commit_cb, you can also listen to the offset.commit event. The event receives an error and the list of topic partitions as arguments. The consumer doesn’t emit it unless you opt in.

Consumed message structure

Messages that the KafkaConsumer returns have the following structure:

{
  value:     // message contents as a Buffer,
  size:      // size of the message, in bytes,
  topic:     // topic the message comes from,
  offset:    // offset the message was read from,
  partition: // partition the message was on,
  key:       // key of the message if present,
  timestamp: // timestamp of message creation if present
}

For example:

{
  value: Buffer.from('hi'),
  size: 2,
  topic: 'librdtesting-01',
  offset: 1337,
  partition: 1,
  key: 'someKey',
  timestamp: 1510325354780
}

Consume with the stream API

The stream API is the easiest way to consume messages. The following example shows its use:

// Read from the librdtesting-01 topic... Note that this creates a new stream on each call, so call it once and store it
const stream = KafkaConsumer.createReadStream(globalConfig, topicConfig, {
  topics: ['librdtesting-01']
});

stream.on('data', (message) => {
  console.log('Got message');
  console.log(message.value.toString());
});

You can also get the consumer from the stream consumer to use consumer methods. For example:

stream.consumer.commit(); // Commits all locally stored offsets

Consume with the standard API

The standard API lets you manage callbacks and events yourself. It supports two modes for consuming messages: flowing mode and non-flowing mode.

Flowing mode. This mode reads all the messages it can by maintaining an infinite loop in a separate thread. It stops only when the consumer calls unsubscribe or disconnect. The consumer sends messages to you either through a callback that you provide to the consume method, or through the data event.

The following example shows flowing mode:

// Flowing mode
consumer.connect();

consumer
  .on('ready', () => {
    consumer.subscribe(['librdtesting-01']);

    // Consume from the librdtesting-01 topic. This is what determines
    // the mode we are running in.
    // In case consume() is called without an argument, or with only one argument
    // which is the callback function, it works in the flowing mode.

    consumer.consume();
  })
  .on('data', (data) => {
    // Output the actual message contents
    console.log(data.value.toString());
  });

Non-flowing mode. This mode reads up to a specific number of messages from Kafka in one call and doesn’t create a separate thread for fetching. It stops only when the consumer calls unsubscribe or disconnect.

The following example shows non-flowing mode:

// Non-flowing mode
consumer.connect();

consumer
  .on('ready', () => {
    // Subscribe to the librdtesting-01 topic
    // This makes subsequent consumes read from that topic.
    consumer.subscribe(['librdtesting-01']);

    // Read up to 1000 message every 1000 milliseconds
    setInterval(() => {
      consumer.consume(1);
    }, 1000);
  })
  .on('data', (data) => {
    console.log('Message found!  Contents below.');
    console.log(data.value.toString());
  });

Callback-based consumer methods

Method	Description
`consumer.connect()`	Connects to the broker. The `connect()` method emits the `ready` event when it connects successfully. If it doesn’t, the client passes the error to the callback.
`consumer.disconnect()`	Disconnects from the broker. The `disconnect()` method emits `disconnected` when it has disconnected. If it doesn’t, the client passes the error to the callback.
`consumer.subscribe(topics)`	Subscribes to an array of topics.
`consumer.unsubscribe()`	Unsubscribes from the currently subscribed topics. You cannot subscribe to different topics without calling the `unsubscribe()` method first.
`consumer.consume(cb?)`	Gets messages from the existing subscription as quickly as possible. If you specify `cb`, the method invokes `cb(err, message)`. This method keeps a background thread running to do the work. The number of worker threads in a Node.js process has a limit that `UV_THREADPOOL_SIZE` sets. It defaults to `4`. Using up all the threads blocks other parts of the application that need threads. If you need more than one consumer, increase `UV_THREADPOOL_SIZE` or use `consumer.consume(number, cb)` instead.
`consumer.consume(number, cb?)`	Gets `number` of messages from the existing subscription. If you specify `cb`, the method invokes `cb(err, message)`.
`consumer.commit()`	Commits all locally stored offsets.
`consumer.commit(topicPartition)`	Commits the offsets that the topic partition specifies.
`consumer.commitMessage(message)`	Commits the offsets that the message specifies.

Consumer events

Event	Description
data	When you use the standard API, the client emits consumed messages in this event.
partition.eof	When you use the standard API and set the `enable.partition.eof` configuration option, the client emits `partition.eof` events. The event contains `topic`, `partition`, and `offset` properties.
warning	The client emits this event on `UNKNOWN_TOPIC_OR_PART` or `TOPIC_AUTHORIZATION_FAILED` errors when consuming in flowing mode. Because the consumer keeps working while the error persists, the warning event reappears after the next metadata refresh. To control the metadata refresh rate, set the `topic.metadata.refresh.interval.ms` property. After you resolve the error, you can call `getMetadata` manually to speed up consumer recovery.
disconnected	The client emits the `disconnected` event when the broker disconnects, but only when you call `.disconnect`. Otherwise, the wrapper always tries to reconnect.
ready	The client emits the `ready` event when the Consumer is ready to read messages.
event	The client emits the `event` event when librdkafka reports an event, if you opted in with the `event_cb` option.
event.log	The client emits the `event.log` event when logging events occur, if you opted in to logging with the `event_cb` option. To receive log information, set a value for `debug`.
event.stats	The client emits the `event.stats` event when librdkafka reports stats, if you opted in by setting `statistics.interval.ms` to a non-zero value.
event.error	The client emits the `event.error` event when librdkafka reports an error.
event.throttle	The client emits the `event.throttle` event when librdkafka reports throttling.

Read current offsets from the broker for a topic

To query the latest and earliest offsets for a topic, call queryWatermarkOffsets on a connected producer or consumer:

const timeout = 5000, partition = 0;
consumer.queryWatermarkOffsets('my-topic', partition, timeout, (err, offsets) => {
  const high = offsets.highOffset;
  const low = offsets.lowOffset;
});

producer.queryWatermarkOffsets('my-topic', partition, timeout, (err, offsets) => {
  const high = offsets.highOffset;
  const low = offsets.lowOffset;
});

The call returns an error if the client isn’t connected or the request times out within the specified interval.

Retrieve cluster metadata

To retrieve metadata from Kafka, use the getMetadata method on Kafka.Producer or Kafka.KafkaConsumer.

When you fetch metadata for a specific topic and the topic reference doesn’t exist, the client creates one using the default configuration. To set configuration parameters, such as acks on a topic to produce messages to, see the documentation on Client.getMetadata.

The following example uses the getMetadata method:

const opts = {
  topic: 'librdtesting-01',
  timeout: 10000
};

producer.getMetadata(opts, (err, metadata) => {
  if (err) {
    console.error('Error getting metadata');
    console.error(err);
  } else {
    console.log('Got metadata');
    console.log(metadata);
  }
});

Metadata on any connection has the following structure:

{
  orig_broker_id: 1,
  orig_broker_name: "broker_name",
  brokers: [
    {
      id: 1,
      host: 'localhost',
      port: 40
    }
  ],
  topics: [
    {
      name: 'awesome-topic',
      partitions: [
        {
          id: 1,
          leader: 20,
          replicas: [1, 2],
          isrs: [1, 2]
        }
      ]
    }
  ]
}

Admin client for the callback-based API

The library provides an admin client with methods to manage topics, groups, and other Kafka entities in the cluster. The following code creates the AdminClient:

const Kafka = require('@confluentinc/kafka-javascript');

const client = Kafka.AdminClient.create({
  'client.id': 'kafka-admin',
  'bootstrap.servers': 'broker01'
});

// From an existing producer or consumer
const depClient = Kafka.AdminClient.createFrom(producer);

These calls create and connect the AdminClient, which lets you call the admin methods. For a complete list of methods, see the JavaScript Client API reference documentation.

OAuthbearer callback authentication

The JavaScript Client supports OAUTHBEARER token authentication for both the promisified and the callback-based APIs. You provide a callback that fetches the token. The library calls the callback at 80% of the token expiry time and uses the new token for the next login attempt:

async function token_refresh(oauthbearer_config /* string - passed from config */, cb /* can be used if function is not async */) {
    // Some logic to fetch the token, before returning it.
    return { tokenValue, lifetime, principal, extensions };
}

const producer = new Kafka().producer({
    'bootstrap.servers': '<fill>',
    'security.protocol': 'sasl_ssl', // or sasl_plain
    'sasl.mechanisms': 'OAUTHBEARER',
    'sasl.oauthbearer.config': 'someConfigPropertiesKey=value', // Just passed straight to token_refresh as a string, carries no other significance.
    'oauthbearer_token_refresh_cb': token_refresh,
});

For tokens that you fetch from an OIDC provider with the client_credentials grant type, the library provides a built-in callback. Set it through configuration only, with no custom function:

const producer = new Kafka().producer({
    'bootstrap.servers': '<fill>',
    'security.protocol': 'sasl_ssl', // or sasl_plain
    'sasl.mechanisms': 'OAUTHBEARER',
    'sasl.oauthbearer.method': 'oidc',
    'sasl.oauthbearer.token.endpoint.url': issuerEndpointUrl,
    'sasl.oauthbearer.scope': scope,
    'sasl.oauthbearer.client.id': oauthClientId,
    'sasl.oauthbearer.client.secret': oauthClientSecret,
    'sasl.oauthbearer.extensions': `logicalCluster=${kafkaLogicalCluster},identityPoolId=${identityPoolId}`
});

These examples use the promisified API, but the callback-based API accepts the same configuration settings.

JavaScript client API documentation

For the JavaScript Client API, see the JavaScript Client API documentation.