Confluent Kafka JavaScript Client

Concurrent Message Processing

To process messages concurrently, pass partitionsConsumedConcurrently to the run() method. The library will attempt to dispatch up to that number of eachMessage calls simultaneously. Note that for a single partition, messages are still processed sequentially to guarantee order.

Batch Processing

Instead of eachMessage, you can use eachBatch in the run() method. The library creates logical batches (up to a limit of 32).

If eachBatchAutoResolve is set to false, you must manually resolve offsets using event.resolveOffset(offset) within the callback. This allows you to commit offsets for messages processed even if a subsequent message in the same batch fails.

consumer.run({
    partitionsConsumedConcurrently: 5,
    eachBatchAutoResolve: false,
    eachBatch: async event => {
        // Process messages...
        // Manually resolve the offset of the last successfully processed message
        event.resolveOffset(event.batch.messages[event.batch.messages.length - 1].message.offset);
    }
});

Offsets are automatically committed by default every 5000ms (controlled by auto.commit.interval.ms). Disconnecting or rebalancing also triggers a commit.

Manual Offset Committing

To manage offsets manually, set enable.auto.commit: false. When committing, you must provide the next offset to read (current offset + 1).

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

consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    if (isItTimeToCommit) {
      await consumer.commitOffsets([
        {
          topic,
          partition,
          offset: (parseInt(message.offset, 10) + 1).toString(),
        },
      ]);
    }
  },
});

Warning: If using manual commits, you must implement a rebalance_cb to commit offsets during ERR__REVOKE_PARTITIONS, as automatic commits are disabled.

Properties for auto-committing, such as autoCommit and autoCommitInterval, are no longer passed to the run method. They must be moved to the kafkaJS configuration block in the consumer setup. autoCommitThreshold is no longer supported.

When autoCommit is enabled, messages are committed periodically based on the autoCommitInterval (default 5 seconds) rather than per-message.

Key changes in run behavior:

• heartbeat(): Users no longer need to call heartbeat() manually inside eachMessage or eachBatch callbacks; heartbeats are managed automatically by librdkafka.
• eachBatch compatibility: An API compatible version of eachBatch is available. The maximum batch size is configured via js.consumer.max.batch.size (defaults to 32).
• eachBatch properties: eachBatchAutoResolve is supported. Within the callback, offsetLag and offsetLagLow are supported in the returned batch, but uncommittedOffsets is unsupported.

// Before
const kafka = new Kafka({ /* ... */ });
const consumer = kafka.consumer({ /* ... */ });
await consumer.connect();
await consumer.subscribe({ topics: ["topic"] });
consumer.run({
  eachMessage: someFunc,
  autoCommit: true,
  autoCommitInterval: 5000,
});

// After
const kafka = new Kafka({ kafkaJS: { /* ... */ } });
const consumer = kafka.consumer({
  kafkaJS: {
    /* ... */,
    autoCommit: true,
    autoCommitInterval: 5000,
  },
});
await consumer.connect();
await consumer.subscribe({ topics: ["topic"] });
consumer.run({
  eachMessage: someFunc,
});

The library batches messages internally per partition. You can control this behavior using configuration:

High Throughput

To maximize throughput, increase linger.ms to allow more messages to accumulate in a batch. Crucially, do not await individual send() calls; instead, collect the promises and await them all at once to allow batching to work effectively.

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

await reportPromises;

Low Latency

To minimize latency, set linger.ms to 0 or 1 and await each send() call individually.

Manual Flushing

If you have a large linger.ms but need to send messages immediately, use flush() to clear the internal buffer.

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

// Try to send everything, waiting up to 1s.
await producer.flush({ timeout: 1000 });

The library has a built-in logger (defaults to console) and supports custom loggers.

Using the Built-in Logger

const { logLevel } = require('@confluentinc/kafka-javascript').KafkaJS;
const libraryLogger = producer.logger();
libraryLogger.info('Message');
libraryLogger.setLogLevel(logLevel.INFO);

Debugging Library Internals

To see internal librdkafka logs, set the debug property in your configuration to a comma-separated list of components or all. Note: This must be set before connecting, as it cannot be changed on the fly.

Custom Logger

Provide a custom logger via the kafkaJS.logger key in the configuration. The logger must implement setLogLevel(logLevel), info(message, extra), error(message, extra), warn(message, extra), and debug(message, extra).

class MyLogger {
  constructor() {}
  setLogLevel(logLevel) {}
  info(message, extra) {}
  error(message, extra) {}
  warn(message, extra) {}
  debug(message, extra) {}
}

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

To ensure messages are produced exactly once, use a transactional producer by providing a unique transactional.id in the configuration.

Note: A transactional producer cannot be used as a non-transactional producer; every message must be wrapped in a transaction.

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

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

EOS Pattern: Consume-Process-Produce

You can achieve Exactly-Once Semantics when consuming from one topic and producing to another by including the consumer offsets in the producer transaction.

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

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

            await transaction.commit();
        } catch (e) {
            await transaction.abort();
        }
    },
});