node-rdkafka

The Standard API allows manual management of callbacks and events. You can operate in two modes:

1. Flowing mode: Messages flow continuously via an infinite loop in the event loop. This is triggered by calling consumer.consume() without a callback (or with only a callback).
2. Non-flowing mode: You manually request messages. This is triggered by calling consumer.consume(number, cb) where number is the amount of messages to fetch.

Important: consumer.consume() uses background threads. The number of threads is limited by UV_THREADPOOL_SIZE (default 4). If using multiple consumers, increase UV_THREADPOOL_SIZE or use the (number, cb) variant to avoid blocking the application.

// Flowing mode example
consumer.connect();

consumer
  .on('ready', () => {
    consumer.subscribe(['librdtesting-01']);
    consumer.consume();
  })
  .on('data', (data) => {
    console.log(data.value.toString());
  });

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

consumer
  .on('ready', () => {
    consumer.subscribe(['librdtesting-01']);
    setInterval(() => {
      consumer.consume(1);
    }, 1000);
  })
  .on('data', (data) => {
    console.log(data.value.toString());
  });

The Standard API is more performant for high-volume message handling but requires manual lifecycle management.

Key Requirements:

1. Call .connect() to connect to the broker.
2. Wait for the ready event before calling .produce().
3. Crucial: You must call .poll() regularly or set a polling interval using .setPollInterval(interval) to process delivery reports and handle reconnections. If you don't, the internal queue will eventually fill up and stop sending messages.

const producer = new Kafka.Producer({
  'metadata.broker.list': 'localhost:9092',
  'dr_cb': true
});

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

// Wait for the ready event before proceeding
producer.on('ready', () => {
  try
    // 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);

When using Alpine Linux in Docker, you must install specific library dependencies because Alpine uses musl instead of glibc. To successfully build and run node-rdkafka, you need to install build tools (g++, make, python3) and development headers for SASL, SSL, and compression.

It is recommended to use a multi-stage build or a single Dockerfile that installs both the runtime dependencies and the build-time dependencies (like gcc and zlib-dev) to keep the image size optimized.

FROM node:14-alpine

RUN apk --no-cache add \
      bash \
      g++ \
      ca-certificates \
      lz4-dev \
      musl-dev \
      cyrus-sasl-dev \
      openssl-dev \
      make \
      python3

RUN apk add --no-cache --virtual .build-deps gcc zlib-dev libc-dev bsd-compat-headers py-setuptools bash

# Create app directory
RUN mkdir -p /usr/local/app

# Move to the app directory
WORKDIR /usr/local/app

# Install node-rdkafka
RUN npm install node-rdkafka
# Copy package.json first to check if an npm install is needed

The Producer can be used as a Node.js writable stream. This is convenient for piping data into Kafka. Use Kafka.Producer.createWriteStream to create a stream for a specific topic.

Important: You must listen to the error event on the stream. If you don't, errors will bubble up as uncaught exceptions and potentially crash your process. Note that stream.write() returning false only indicates that the internal queue is full; it does not guarantee the message reached Kafka.

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

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

if (queuedSuccess) {
  console.log('We queued our message!');
} else {
  // Note that this only tells us if the stream's queue is full,
  // it does NOT tell us if the message got to Kafka!  See below...
  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 kafka stream');
  console.error(err);
})

You can consume Kafka messages as a Node.js Readable stream. This is useful for piping Kafka data into other systems and ensures backpressure is respected (you only read as fast as you write).

Note: You should generally not instantiate KafkaConsumerStream directly. Instead, use the KafkaConsumer.createReadStream() method provided by the KafkaConsumer class.

Key Characteristics

• Object Mode: By default, the stream operates in objectMode, emitting {Consumer~Message} objects. If objectMode: false is explicitly set, it will stream the message value (buffers).
• Automatic Connection: The stream detects if the consumer is already connected. If not, it will connect and begin reading once ready.
• Backpressure: Unlike continuous subscribe callbacks, the stream implementation ensures you read only as fast as you can process the data.
• Performance Trade-off: The stream implementation is slower than using the continuous subscribe callback. If performance is more critical than backpressure, use the callback method instead.