### Install node-rdkafka

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Install the node-rdkafka module using npm.

```bash
npm install node-rdkafka
```

--------------------------------

### producer.beginTransaction(callback)

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Starts a new transaction.

```APIDOC
## producer.beginTransaction(callback)

### Description
Starts a new transaction.

### Method

```
producer.beginTransaction(callback)
```

### Parameters
#### Path Parameters
- **callback** (function) - Optional - A callback function to be executed after starting the transaction.
```

--------------------------------

### Kafka Consumer Flow Example

Source: https://github.com/blizzard/node-rdkafka/blob/master/examples/consumer-flow.md

Connects to Kafka, subscribes to a topic, consumes messages, and commits offsets periodically. Ensure Kafka is running and accessible at 'localhost:9092' and the topic 'test' exists.

```javascript
/*
 * node-rdkafka - Node.js wrapper for RdKafka C/C++ library
 *
 * Copyright (c) 2016 Blizzard Entertainment
 *
 * This software may be modified and distributed under the terms
 * of the MIT license.  See the LICENSE.txt file for details.
 */

var Kafka = require('../');

var consumer = new Kafka.KafkaConsumer({
  //'debug': 'all',
  'metadata.broker.list': 'localhost:9092',
  'group.id': 'node-rdkafka-consumer-flow-example',
  'enable.auto.commit': false
});

var topicName = 'test';

//logging debug messages, if debug is enabled
consumer.on('event.log', function(log) {
  console.log(log);
});

//logging all errors
consumer.on('event.error', function(err) {
  console.error('Error from consumer');
  console.error(err);
});

//counter to commit offsets every numMessages are received
var counter = 0;
var numMessages = 5;

consumer.on('ready', function(arg) {
  console.log('consumer ready.' + JSON.stringify(arg));

  consumer.subscribe([topicName]);
  //start consuming messages
  consumer.consume();
});


consumer.on('data', function(m) {
  counter++;

  //committing offsets every numMessages
  if (counter % numMessages === 0) {
    console.log('calling commit');
    consumer.commit(m);
  }

  // Output the actual message contents
  console.log(JSON.stringify(m));
  console.log(m.value.toString());

});

consumer.on('disconnected', function(arg) {
  console.log('consumer disconnected. ' + JSON.stringify(arg));
});

//starting the consumer
consumer.connect();

//stopping this example after 30s
setTimeout(function() {
  consumer.disconnect();
}, 30000);

```

--------------------------------

### Transactional Producer and Consumer Example

Source: https://context7.com/blizzard/node-rdkafka/llms.txt

Demonstrates exactly-once semantics using transactional producers and consumers. Requires Kafka broker configuration for transactions and idempotence. Ensure 'input-topic' and 'output-topic' exist.

```javascript
const Kafka = require('node-rdkafka');

const producer = new Kafka.Producer({
  'metadata.broker.list': 'localhost:9092',
  'transactional.id': 'my-txn-producer',
  'enable.idempotence': true,
  'dr_cb': true,
});

const consumer = new Kafka.KafkaConsumer({
  'metadata.broker.list': 'localhost:9092',
  'group.id': 'txn-consumer-group',
  'enable.auto.commit': false,
}, { 'auto.offset.reset': 'earliest' });

producer.setPollInterval(100);

producer.connect({}, () => {
  consumer.connect({}, () => {
    consumer.subscribe(['input-topic']);

    // Initialize transactions once
    producer.initTransactions(10000, (err) => {
      if (err) throw err;

      consumer.consume(1, (err, msgs) => {
        if (err || msgs.length === 0) return;
        const msg = msgs[0];

        // Begin a transaction
        producer.beginTransaction((err) => {
          if (err) throw err;

          // Produce to output topic within the transaction
          producer.produce('output-topic', null, msg.value, null, Date.now());

          // Atomically commit the consumer offset with the transaction
          const positions = consumer.position();
          producer.sendOffsetsToTransaction(positions, consumer, 10000, (err) => {
            if (err) throw err;

            producer.commitTransaction(10000, (err) => {
              if (err) {
                // Abort on failure
                producer.abortTransaction(10000, () => {});
                throw err;
              }
              console.log('Transaction committed');
              producer.disconnect();
              consumer.disconnect();
            });
          });
        });
      });
    });
  });
});
```

--------------------------------

### High-Level Producer Setup and Message Production

Source: https://github.com/blizzard/node-rdkafka/blob/master/examples/high-level-producer.md

Instantiate a HighLevelProducer, configure key and value serializers, connect to the broker, and produce a message. The key serializer returns null, and the value serializer extracts the 'message' field. Delivery offsets are acknowledged.

```javascript
var Kafka = require('../');

var producer = new Kafka.HighLevelProducer({
  'metadata.broker.list': 'localhost:9092',
});

// Throw away the keys
producer.setKeySerializer(function(v) {
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      resolve(null);
    }, 20);
  });
});

// Take the message field
producer.setValueSerializer(function(v) {
  return Buffer.from(v.message);
});

producer.connect(null, function() {
  producer.produce('test', null, {
    message: 'alliance4ever',
  }, null, Date.now(), function(err, offset) {
    // The offset if our acknowledgement level allows us to receive delivery offsets
    setImmediate(function() {
      producer.disconnect();
    });
  });
});
```

--------------------------------

### OAuthBearer Authentication Setup

Source: https://context7.com/blizzard/node-rdkafka/llms.txt

Configures SASL_SSL authentication using OAUTHBEARER tokens for Kafka clients. Tokens can be set initially and refreshed dynamically.

```javascript
const Kafka = require('node-rdkafka');

let token = 'initial_oauth_token';

// Producer + KafkaConsumer: use setOauthBearerToken
const producer = new Kafka.Producer({
  'metadata.broker.list': 'localhost:9093',
  'security.protocol': 'SASL_SSL',
  'sasl.mechanisms': 'OAUTHBEARER',
}).setOauthBearerToken(token);

producer.connect();

// Refresh when token is rotated
producer.setOauthBearerToken(getNewToken());

// AdminClient: pass token as second argument to create()
const admin = Kafka.AdminClient.create({
  'metadata.broker.list': 'localhost:9093',
  'security.protocol': 'SASL_SSL',
  'sasl.mechanisms': 'OAUTHBEARER',
}, token);

admin.refreshOauthBearerToken(getNewToken());
admin.disconnect();

// KafkaConsumerStream: pass initOauthBearerToken in stream options
const stream = Kafka.KafkaConsumer.createReadStream(
  { 'metadata.broker.list': 'localhost:9093', 'group.id': 'grp',
    'security.protocol': 'SASL_SSL', 'sasl.mechanisms': 'OAUTHBEARER' },
  {},
  { topics: 'my-topic', initOauthBearerToken: token }
);

stream.refreshOauthBearerToken(getNewToken());

function getNewToken() { return 'refreshed_token'; }
```

--------------------------------

### Kafka Consumer Flowing Mode Example

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Illustrates how to consume messages continuously using the flowing mode. Ensure the consumer is connected and subscribed to topics before calling consume without a callback.

```javascript
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. By not specifying a callback (or specifying
    // only a callback) we get messages as soon as they are available.
    consumer.consume();
  })
  .on('data', (data) => {
    // Output the actual message contents
    console.log(data.value.toString());
  });
```

--------------------------------

### Kafka Producer Setup and Message Production

Source: https://github.com/blizzard/node-rdkafka/blob/master/examples/producer.md

Configure and use a Kafka producer to send messages. Ensure the producer is ready before sending and poll to receive delivery reports. Disconnect when finished.

```javascript
/*
 * node-rdkafka - Node.js wrapper for RdKafka C/C++ library
 *
 * Copyright (c) 2016 Blizzard Entertainment
 *
 * This software may be modified and distributed under the terms
 * of the MIT license.  See the LICENSE.txt file for details.
 */

var Kafka = require('../');

var producer = new Kafka.Producer({
  //'debug' : 'all',
  'metadata.broker.list': 'localhost:9092',
  'dr_cb': true  //delivery report callback
});

var topicName = 'test';

//logging debug messages, if debug is enabled
producer.on('event.log', function(log) {
  console.log(log);
});

//logging all errors
producer.on('event.error', function(err) {
  console.error('Error from producer');
  console.error(err);
});

//counter to stop this sample after maxMessages are sent
var counter = 0;
var maxMessages = 10;

producer.on('delivery-report', function(err, report) {
  console.log('delivery-report: ' + JSON.stringify(report));
  counter++;
});

//Wait for the ready event before producing
producer.on('ready', function(arg) {
  console.log('producer ready.' + JSON.stringify(arg));

  for (var i = 0; i < maxMessages; i++) {
    var value = Buffer.from('value-' +i);
    var key = "key-"+i;
    // if partition is set to -1, librdkafka will use the default partitioner
    var partition = -1;
    var headers = [
      { header: "header value" }
    ]
    producer.produce(topicName, partition, value, key, Date.now(), "", headers);
  }

  //need to keep polling for a while to ensure the delivery reports are received
  var pollLoop = setInterval(function() {
      producer.poll();
      if (counter === maxMessages) {
        clearInterval(pollLoop);
        producer.disconnect();
      }
    }, 1000);

});

producer.on('disconnected', function(arg) {
  console.log('producer disconnected. ' + JSON.stringify(arg));
});

//starting the producer
producer.connect();

```

--------------------------------

### AdminClient Create, Expand, and Delete Topic

Source: https://context7.com/blizzard/node-rdkafka/llms.txt

Manage Kafka topics using the AdminClient. This example shows how to create a new topic with specific configurations, add partitions to an existing topic, and delete a topic. Requires administrative privileges on the Kafka cluster.

```javascript
const Kafka = require('node-rdkafka');

const admin = Kafka.AdminClient.create({
  'client.id': 'my-admin-client',
  'metadata.broker.list': 'localhost:9092',
});

// Create a topic with 3 partitions and replication factor 1
admin.createTopic(
  { topic: 'new-topic', num_partitions: 3, replication_factor: 1,
    config: { 'retention.ms': '86400000' } },
  10000, // timeout ms
  (err) => {
    if (err) console.error('Create failed:', err);
    else console.log('Topic created');

    // Expand the topic to 10 partitions
    admin.createPartitions('new-topic', 10, 5000, (err) => {
      if (err) console.error('Expand failed:', err);
      else console.log('Now has 10 partitions');

      // Delete the topic
      admin.deleteTopic('new-topic', 5000, (err) => {
        if (err) console.error('Delete failed:', err);
        else console.log('Topic deleted');

        admin.disconnect();
      });
    });
  }
);
```

--------------------------------

### producer.sendOffsetsToTransaction(offsets, consumer, timeout, callback)

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Sends consumed topic-partition-offsets to the broker, which will get committed along with the transaction.

```APIDOC
## producer.sendOffsetsToTransaction(offsets, consumer, timeout, callback)

### Description
Sends consumed topic-partition-offsets to the broker, which will get committed along with the transaction.

### Method

```
producer.sendOffsetsToTransaction(offsets, consumer, timeout, callback)
```

### Parameters
#### Path Parameters
- **offsets** (object) - Required - An object containing topic-partition-offsets.
- **consumer** (object) - Required - The consumer instance.
- **timeout** (number) - Optional - The timeout in milliseconds.
- **callback** (function) - Optional - A callback function to be executed after sending offsets.
```

--------------------------------

### Query Watermark Offsets with KafkaJS

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Use queryWatermarkOffsets to get the latest and earliest offsets for a topic partition. An error is returned if the client is not connected or the request times out.

```javascript
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;
});
```

--------------------------------

### Dockerfile for node-rdkafka on Alpine

Source: https://github.com/blizzard/node-rdkafka/blob/master/examples/docker-alpine.md

This Dockerfile installs node-rdkafka on an Alpine Linux base image. It includes essential build dependencies and runtime libraries required for node-rdkafka to function correctly.

```dockerfile
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
```

--------------------------------

### Kafka Producer - Standard API Example

Source: https://context7.com/blizzard/node-rdkafka/llms.txt

Send messages to Kafka using the Kafka.Producer class. Requires explicit connect() and manual or interval-based polling for delivery reports. Enable delivery-report events using 'dr_cb': true.

```javascript
const Kafka = require('node-rdkafka');

const producer = new Kafka.Producer({
  'metadata.broker.list': 'localhost:9092',
  'dr_cb': true,                        // enable delivery-report events
  '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,
});

// Log debug/error events
producer.on('event.log', (log) => console.log(log));
producer.on('event.error', (err) => console.error('Producer error:', err));

// Delivery report: fires after each message is acknowledged by Kafka
producer.on('delivery-report', (err, report) => {
  if (err) console.error('Delivery failed:', err);
  else console.log('Delivered:', JSON.stringify(report));
  // report => { topic, partition, offset, key, value, size, opaque, timestamp }
});

producer.on('ready', () => {
  const headers = [{ myHeader: 'headerValue' }];

  for (let i = 0; i < 10; i++) {
    producer.produce(
      'my-topic',          // topic (required, string)
      null,                // partition (-1 = use librdkafka partitioner)
      Buffer.from('msg-' + i), // message (Buffer)
      'key-' + i,          // key (string | Buffer | null)
      Date.now(),          // timestamp (number | null)
      null,                // opaque token (any, passed back in delivery-report)
      headers              // optional message headers
    );
  }

  // Poll until all delivery reports are received, then disconnect
  producer.setPollInterval(100);
});

producer.on('disconnected', (info) => console.log('Disconnected:', info));

producer.connect();
```

--------------------------------

### Kafka Consumer Non-Flowing Mode Example

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Demonstrates consuming messages one at a time at a set interval using non-flowing mode. This is achieved by using `setInterval` to call `consumer.consume(1)` periodically.

```javascript
consumer.connect();

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

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

--------------------------------

### Set Value Serializer

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Configure a serializer for the HighLevelProducer to automatically transform message values before sending them to Kafka. This example uses JSON stringification.

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

--------------------------------

### VS Code C++ IntelliSense Configuration

Source: https://github.com/blizzard/node-rdkafka/blob/master/CONTRIBUTING.md

Configure VS Code's C++ extension to resolve include paths for IntelliSense on macOS. This setup helps in navigating and understanding the C++ codebase.

```json
{
    "configurations": [
        {
            "name": "Mac",
            "includePath": [
                "${workspaceFolder}/**",
                "${workspaceFolder}",
                "${workspaceFolder}/src",
                "${workspaceFolder}/node_modules/nan",
                "${workspaceFolder}/deps/librdkafka/src",
                "${workspaceFolder}/deps/librdkafka/src-cpp",
                "/usr/local/include/node",
                "/usr/local/include/node/uv"
            ],
            "defines": [],
            "macFrameworkPath": [
                "/Library/Developer/CommandLineTools/SDKs/MacOSX10.14.sdk/System/Library/Frameworks"
            ],
            "compilerPath": "/usr/bin/clang",
            "cStandard": "c11",
            "cppStandard": "c++17",
            "intelliSenseMode": "clang-x64"
        }
    ],
    "version": 4
}
```

--------------------------------

### Consuming Messages with Stream API

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

The stream API provides an easy way to consume messages. This example shows how to create a read stream and process incoming data.

```javascript
// Read from the librdtesting-01 topic... note that this creates a new stream on each call!
const stream = KafkaConsumer.createReadStream(globalConfig, topicConfig, {
  topics: ['librdtesting-01']
});

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

--------------------------------

### Worker Process Kafka Producer Configuration and Sending

Source: https://github.com/blizzard/node-rdkafka/blob/master/examples/producer-cluster.md

Each worker process configures and initializes a Kafka producer. It then starts sending messages to a specified topic and manages delivery reports and errors.

```javascript
else {
  // Configure client
  var producer = new Kafka.Producer({
    'client.id': 'kafka',
    'metadata.broker.list': 'localhost:9092',
    'compression.codec': 'none',
    '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
  });

  producer.setPollInterval(100);

  var total = 0;
  var totalSent = 0;
  var max = 20000;
  var errors = 0;
  var started = Date.now();

  var sendMessage = function() {
    var ret = producer.sendMessage({
      topic: 'librdtesting-01',
      message: Buffer.from('message ' + total)
    }, function() {
    });
    total++;
    if (total >= max) {
    } else {
      setImmediate(sendMessage);
    }
  };

  var verified_received = 0;
  var exitNextTick = false;
  var errorsArr = [];

  var t = setInterval(function() {
    producer.poll();

    if (exitNextTick) {
      clearInterval(t);
      return setTimeout(function() {
        console.log('[%d] Received: %d, Errors: %d, Total: %d', process.pid, verified_received, errors, total);
        // console.log('[%d] Finished sending %d in %d seconds', process.pid, total, parseInt((Date.now() - started) / 1000));
        if (errors > 0) {
          console.error(errorsArr[0]);
          return process.exitCode = 1;
        }
        process.exitCode = 0;
        setTimeout(process.exit, 1000);
      }, 2000);
    }

    if (verified_received + errors === max) {
      exitNextTick = true;
    }

  }, 1000);
  producer.connect()
    .on('event.error', function(e) {
      errors++;
      errorsArr.push(e);
    })
    .on('delivery-report', function() {
      verified_received++;
    })
    .on('ready', sendMessage);

}
```

--------------------------------

### Producer Error Handling with Kafka.CODES.ERRORS

Source: https://context7.com/blizzard/node-rdkafka/llms.txt

This snippet demonstrates how to handle delivery reports and general events errors from the producer. It shows how to access error properties like code, message, and origin, and provides examples of checking for specific error codes like ERR__MSG_TIMED_OUT and ERR__QUEUE_FULL.

```javascript
const Kafka = require('node-rdkafka');

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

producer.on('delivery-report', (err, report) => {
  if (err) {
    // err is a LibrdKafkaError
    console.error('Code:', err.code);
    console.error('Message:', err.message);
    console.error('Origin:', err.origin);   // 'local' | 'kafka'
    console.error('Fatal:', err.isFatal);
    console.error('Retriable:', err.isRetriable);
    console.error('Abort txn:', err.isTxnRequiresAbort);

    if (err.code === Kafka.CODES.ERRORS.ERR__MSG_TIMED_OUT) {
      console.warn('Message timed out, consider retrying');
    }
    if (err.code === Kafka.CODES.ERRORS.ERR__QUEUE_FULL) {
      console.warn('Internal queue full — slow down producing');
    }
  }
});

producer.on('event.error', (err) => {
  console.error(`[${err.origin}] ${err.code}: ${err.message}`);
});

producer.on('ready', () => {
  try {
    producer.produce('my-topic', null, Buffer.from('msg'), null, Date.now());
  } catch (e) {
    // produce() can throw synchronously on queue full or disconnect
    if (e.code === Kafka.CODES.ERRORS.ERR__QUEUE_FULL) {
      producer.poll(); // drain the queue
    }
  }
});

producer.connect();
```

--------------------------------

### Create Kafka Producer with Options

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Instantiate a Kafka Producer with various configuration options, including client ID, compression, retries, and callbacks.

```javascript
const producer = new Kafka.Producer({
  'client.id': 'kafka',
  'metadata.broker.list': '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
});
```

--------------------------------

### Create Basic Kafka Producer

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Instantiate a Kafka Producer with the minimum required configuration, specifying the Kafka brokers.

```javascript
const producer = new Kafka.Producer({
  'metadata.broker.list': 'kafka-host1:9092,kafka-host2:9092'
});
```

--------------------------------

### producer.connect()

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Connects to the broker. The `connect()` method emits the `ready` event when it connects successfully. If it does not, the error will be passed through the callback.

```APIDOC
## producer.connect()

### Description
Connects to the broker. The `connect()` method emits the `ready` event when it connects successfully. If it does not, the error will be passed through the callback.

### Method

```
producer.connect()
```
```

--------------------------------

### Get node-rdkafka Features

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Retrieve the features supported by your compiled librdkafka. This is useful for debugging and issue reporting.

```javascript
const Kafka = require('node-rdkafka');
console.log(Kafka.features);
```

--------------------------------

### Get librdkafka Version

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Retrieve the version of the librdkafka library that node-rdkafka is linked against. Useful for compatibility checks and issue reporting.

```javascript
const Kafka = require('node-rdkafka');
console.log(Kafka.librdkafkaVersion);
```

--------------------------------

### Create Topic with Admin Client

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Use the createTopic method of the AdminClient to create a new topic on the broker with specified partitions and replication factor. Optional timeout can be provided.

```javascript
client.createTopic({
  topic: topicName,
  num_partitions: 1,
  replication_factor: 1
}, (err) => {
  // Done!
});
```

--------------------------------

### Get Metadata

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Retrieve metadata from Kafka, including broker information and topic details. This method is available on both producer and consumer clients.

```APIDOC
## getMetadata

### Description
Retrieves metadata from Kafka brokers, including information about brokers and topics.

### Method
producer.getMetadata(opts, cb)
consumer.getMetadata(opts, cb)

### Parameters
- **opts** (object) - Options for fetching metadata.
  - **topic** (string) - Optional. The name of the topic to fetch metadata for. If not provided, metadata for all topics is returned.
  - **timeout** (number) - Optional. The timeout in milliseconds for the request.
- **cb** (function) - A callback function that receives an error and the metadata object.

### Response
- **metadata** (object) - An object containing metadata information.
  - **orig_broker_id** (number) - The ID of the originating broker.
  - **orig_broker_name** (string) - The name of the originating broker.
  - **brokers** (array) - An array of broker objects.
    - **id** (number) - Broker ID.
    - **host** (string) - Broker host.
    - **port** (number) - Broker port.
  - **topics** (array) - An array of topic objects.
    - **name** (string) - Topic name.
    - **partitions** (array) - An array of partition objects.
      - **id** (number) - Partition ID.
      - **leader** (number) - Leader broker ID.
      - **replicas** (array) - Array of replica broker IDs.
      - **isrs** (array) - Array of in-sync replica broker IDs.
```

--------------------------------

### consumer.connect()

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Connects the consumer to the Kafka broker. Emits the 'ready' event upon successful connection, or passes an error through a callback if connection fails.

```APIDOC
## consumer.connect()

### Description
Connects to the broker. The `connect()` emits the event `ready` when it has successfully connected. If it does not, the error will be passed through the callback.

### Method
APICALL

### Endpoint
N/A

### Parameters
None

### Request Example
None

### Response
#### Success Response (ready event)
Emits a 'ready' event upon successful connection.

#### Response Example
None
```

--------------------------------

### producer.poll()

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Polls the producer for delivery reports or other events to be transmitted via the emitter. In order to get the events in `librdkafka`'s queue to emit, you must call this regularly.

```APIDOC
## producer.poll()

### Description
Polls the producer for delivery reports or other events to be transmitted via the emitter. In order to get the events in `librdkafka`'s queue to emit, you must call this regularly.

### Method

```
producer.poll()
```
```

--------------------------------

### Instantiate Kafka Consumer

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Initialize a KafkaConsumer with global and topic configurations. 'group.id' and 'metadata.broker.list' are required properties.

```javascript
const consumer = new Kafka.KafkaConsumer({
  'group.id': 'kafka',
  'metadata.broker.list': 'localhost:9092',
}, {});
```

--------------------------------

### Get Kafka Metadata with node-rdkafka

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Retrieve metadata from Kafka using the getMetadata method on Producer or KafkaConsumer instances. This can be used to fetch metadata for all topics or a specific topic.

```javascript
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);
  }
});
```

--------------------------------

### KafkaConsumer Seek, Pause, and Resume

Source: https://context7.com/blizzard/node-rdkafka/llms.txt

Demonstrates manual partition control for Kafka consumers. Use seek to jump to a specific offset, pause to temporarily stop consumption, and resume to continue. Requires a running Kafka broker and subscribed topic.

```javascript
const Kafka = require('node-rdkafka');

const consumer = new Kafka.KafkaConsumer({
  'metadata.broker.list': 'localhost:9092',
  'group.id': 'seek-example-group',
  'enable.auto.commit': false,
});

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

  // Seek partition 0 back to offset 1000
  consumer.seek({ topic: 'my-topic', partition: 0, offset: 1000 }, 1000, (err) => {
    if (err) console.error('Seek failed:', err);
    else console.log('Now reading from offset 1000');
  });

  // Pause consumption on partition 0
  consumer.pause([{ topic: 'my-topic', partition: 0 }]);

  // Resume after 5 seconds
  setTimeout(() => {
    consumer.resume([{ topic: 'my-topic', partition: 0 }]);
  }, 5000);

  consumer.consume();
});

consumer.on('data', (msg) => console.log(msg.offset, msg.value.toString()));
consumer.connect();
```

--------------------------------

### producer.initTransactions(timeout, callback)

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Initializes the transactional producer.

```APIDOC
## producer.initTransactions(timeout, callback)

### Description
Initializes the transactional producer.

### Method

```
producer.initTransactions(timeout, callback)
```

### Parameters
#### Path Parameters
- **timeout** (number) - Optional - The timeout in milliseconds.
- **callback** (function) - Optional - A callback function to be executed after initialization.
```

--------------------------------

### Connect and Produce Messages with Node-Rdkafka

Source: https://github.com/blizzard/node-rdkafka/blob/master/examples/metadata.md

This snippet shows how to initialize a Kafka producer, connect to brokers, and handle 'ready' and 'error' events. Ensure Kafka brokers are running at 'localhost:9092'.

```javascript
/*
 * node-rdkafka - Node.js wrapper for RdKafka C/C++ library
 *
 * Copyright (c) 2016 Blizzard Entertainment
 *
 * This software may be modified and distributed under the terms
 * of the MIT license.  See the LICENSE.txt file for details.
 */

var Kafka = require('../');

var producer = new Kafka.Producer({
  'metadata.broker.list': 'localhost:9092',
  'client.id': 'hey',
  'compression.codec': 'snappy'
});

producer.connect()
  .on('ready', function(i, metadata) {
    console.log(i);
    console.log(metadata);
  })
  .on('event.error', function(err) {
    console.log(err);
  });

```

--------------------------------

### Create Admin Client with node-rdkafka

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Instantiate an AdminClient for managing Kafka topics. This client can be used for creating, deleting, and scaling topics.

```javascript
const Kafka = require('node-rdkafka');

const client = Kafka.AdminClient.create({
  'client.id': 'kafka-admin',
  'metadata.broker.list': 'broker01'
});
```

--------------------------------

### Kafka.Producer - Standard API

Source: https://context7.com/blizzard/node-rdkafka/llms.txt

Demonstrates how to instantiate and use the Kafka.Producer class for sending messages. It includes setting up event listeners for delivery reports and errors, and connecting to Kafka brokers.

```APIDOC
## Kafka.Producer - Standard API

### Description
This section details the standard API for the `Kafka.Producer` class, which is the primary interface for sending messages to Kafka. It requires an explicit `connect()` call and handles message delivery reports via events.

### Method
`new Kafka.Producer(options)`

### Parameters
#### Options
- `'metadata.broker.list'` (string) - Required - Comma-separated list of Kafka broker addresses.
- `'dr_cb'` (boolean) - Optional - Enable delivery-report events.
- `'compression.codec'` (string) - Optional - Compression codec to use (e.g., 'gzip', 'snappy').
- `'retry.backoff.ms'` (number) - Optional - Backoff time in milliseconds for retries.
- `'message.send.max.retries'` (number) - Optional - Maximum number of retries for message sending.
- `'socket.keepalive.enable'` (boolean) - Optional - Enable socket keep-alive.
- `'queue.buffering.max.messages'` (number) - Optional - Maximum number of messages in the queue.
- `'queue.buffering.max.ms'` (number) - Optional - Maximum time in milliseconds to buffer messages.
- `'batch.num.messages'` (number) - Optional - Maximum number of messages to batch together.

### Events
- `event.log`: Emitted for log messages from librdkafka.
- `event.error`: Emitted when an error occurs.
- `delivery-report`: Emitted after each message is acknowledged by Kafka. The report object contains `topic`, `partition`, `offset`, `key`, `value`, `size`, `opaque`, and `timestamp`.
- `ready`: Emitted when the producer is ready to send messages.
- `disconnected`: Emitted when the producer disconnects.

### Producer Methods
- `produce(topic, partition, message, key, timestamp, opaque, headers)`: Sends a message.
  - `topic` (string): The topic to send the message to.
  - `partition` (number | null): The partition to send the message to (-1 for default).
  - `message` (Buffer): The message payload.
  - `key` (string | Buffer | null): The message key.
  - `timestamp` (number | null): The message timestamp.
  - `opaque` (any): An opaque token passed back in the delivery report.
  - `headers` (Array<Object>): Optional message headers.
- `setPollInterval(interval)`: Sets the interval for polling delivery reports.
- `connect()`: Connects the producer to the Kafka brokers.
- `disconnect()`: Disconnects the producer.
- `flush(timeout, callback)`: Flushes all messages from the queue.

### Request Example
```javascript
const Kafka = require('node-rdkafka');

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

producer.on('event.log', (log) => console.log(log));
producer.on('event.error', (err) => console.error('Producer error:', err));

producer.on('delivery-report', (err, report) => {
  if (err) console.error('Delivery failed:', err);
  else console.log('Delivered:', JSON.stringify(report));
});

producer.on('ready', () => {
  const headers = [{ myHeader: 'headerValue' }];
  for (let i = 0; i < 10; i++) {
    producer.produce(
      'my-topic',
      null,
      Buffer.from('msg-' + i),
      'key-' + i,
      Date.now(),
      null,
      headers
    );
  }
  producer.setPollInterval(100);
});

producer.on('disconnected', (info) => console.log('Disconnected:', info));

producer.connect();
```
```

--------------------------------

### Connect to Kafka Consumer with Stream

Source: https://github.com/blizzard/node-rdkafka/blob/master/examples/consumer.md

Use KafkaConsumer.createReadStream to establish a connection to a Kafka broker. Configure broker list, group ID, and topic. Ensure auto commit is disabled if manual control is needed. Errors are handled via the 'error' event.

```javascript
/*
 * node-rdkafka - Node.js wrapper for RdKafka C/C++ library
 *
 * Copyright (c) 2016 Blizzard Entertainment
 *
 * This software may be modified and distributed under the terms
 * of the MIT license.  See the LICENSE.txt file for details.
 */

var Transform = require('stream').Transform;

var Kafka = require('../');

var stream = Kafka.KafkaConsumer.createReadStream({
  'metadata.broker.list': 'localhost:9092',
  'group.id': 'librd-test',
  'socket.keepalive.enable': true,
  'enable.auto.commit': false
}, {}, {
  topics: 'test',
  waitInterval: 0,
  objectMode: false
});

stream.on('error', function(err) {
  if (err) console.log(err);
  process.exit(1);
});

stream
  .pipe(process.stdout);

stream.on('error', function(err) {
  console.log(err);
  process.exit(1);
});

stream.consumer.on('event.error', function(err) {
  console.log(err);
})
```

--------------------------------

### Produce with Delivery Report Event

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Instantiate a Kafka Producer and configure it to emit delivery-report events. Set the poll interval and register a listener for the 'delivery-report' event.

```javascript
const producer = new Kafka.Producer({
  'client.id': 'my-client', // Specifies an identifier to use to help trace activity in Kafka
  'metadata.broker.list': '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);
});
```

--------------------------------

### Admin Client Operations

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Provides methods for administrative tasks such as creating and deleting topics, and creating partitions.

```APIDOC
## Admin Client

### Description
The Admin Client allows for administrative operations on Kafka, such as topic management.

### Creating an Admin Client
```javascript
const Kafka = require('node-rdkafka');

const client = Kafka.AdminClient.create({
  'client.id': 'kafka-admin',
  'metadata.broker.list': 'broker01'
});
```

### Methods

#### client.disconnect()

##### Description
Destroy the admin client, making it invalid for further use.

#### client.createTopic(topic, timeout, cb)

##### Description
Create a topic on the broker with the given configuration.

##### Parameters
- **topic** (object) - Configuration object for the topic. See JS doc for more on structure.
- **timeout** (number) - Optional. The timeout in milliseconds for the request.
- **cb** (function) - Callback function.

#### client.deleteTopic(topicName, timeout, cb)

##### Description
Delete a topic of the given name.

##### Parameters
- **topicName** (string) - The name of the topic to delete.
- **timeout** (number) - Optional. The timeout in milliseconds for the request.
- **cb** (function) - Callback function.

#### client.createPartitions(topicName, desiredPartitions, timeout, cb)

##### Description
Create partitions until the topic has the desired number of partitions.

##### Parameters
- **topicName** (string) - The name of the topic.
- **desiredPartitions** (number) - The desired number of partitions.
- **timeout** (number) - Optional. The timeout in milliseconds for the request.
- **cb** (function) - Callback function.
```

--------------------------------

### KafkaConsumer Flowing Mode

Source: https://context7.com/blizzard/node-rdkafka/llms.txt

Read messages from Kafka in flowing mode using `consumer.consume()`. This starts a background loop that emits `data` events as fast as possible. Configure event listeners for logs, errors, rebalances, and offset commits. Commit messages periodically using `commitMessage`.

```javascript
const Kafka = require('node-rdkafka');

const consumer = new Kafka.KafkaConsumer({
  'metadata.broker.list': 'localhost:9092',
  'group.id': 'my-consumer-group',
  'enable.auto.commit': false,
  'rebalance_cb': true,           // emit 'rebalance' events
  'offset_commit_cb': true,       // emit 'offset.commit' events
});

consumer.on('event.log', (log) => console.log(log));
consumer.on('event.error', (err) => console.error('Consumer error:', err));

consumer.on('rebalance', (err, assignments) => {
  console.log('Rebalance event:', err.code, assignments);
});

consumer.on('offset.commit', (err, offsets) => {
  if (err) console.error('Commit failed:', err);
  else console.log('Committed offsets:', offsets);
});

let msgCount = 0;

consumer.on('ready', () => {
  consumer.subscribe(['my-topic']);
  consumer.consume(); // start flowing mode

  // Stop after 30 seconds
  setTimeout(() => consumer.disconnect(), 30000);
});

consumer.on('data', (msg) => {
  msgCount++;
  console.log(`[${msg.topic}:${msg.partition}@${msg.offset}]`, msg.value.toString());
  console.log('Key:', msg.key, 'Timestamp:', msg.timestamp);

  // Commit every 5 messages
  if (msgCount % 5 === 0) {
    consumer.commitMessage(msg); // commits offset + 1
  }
});

consumer.on('disconnected', (info) => console.log('Consumer disconnected:', info));

consumer.connect();
```

--------------------------------

### Instantiate High-Level Producer

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Create an instance of the HighLevelProducer. This producer variant allows callbacks upon message delivery.

```javascript
const producer = new Kafka.HighLevelProducer({
  'metadata.broker.list': 'localhost:9092',
});
```

--------------------------------

### Produce Message with Callback

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Use the HighLevelProducer to send a message and receive a callback upon delivery. The callback provides the message offset if delivery acknowledgments are configured.

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

--------------------------------

### KafkaConsumer - Seek, Pause, and Resume

Source: https://context7.com/blizzard/node-rdkafka/llms.txt

Provides manual partition control methods for advanced consumption patterns, allowing users to seek to specific offsets and pause/resume partition consumption.

```APIDOC
## KafkaConsumer — Seek, Pause, and Resume

Manual partition control methods for advanced consumption patterns: seek to a specific offset, and pause/resume partition consumption.

### Example
```javascript
const Kafka = require('node-rdkafka');

const consumer = new Kafka.KafkaConsumer({
  'metadata.broker.list': 'localhost:9092',
  'group.id': 'seek-example-group',
  'enable.auto.commit': false,
});

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

  // Seek partition 0 back to offset 1000
  consumer.seek({ topic: 'my-topic', partition: 0, offset: 1000 }, 1000, (err) => {
    if (err) console.error('Seek failed:', err);
    else console.log('Now reading from offset 1000');
  });

  // Pause consumption on partition 0
  consumer.pause([{ topic: 'my-topic', partition: 0 }]);

  // Resume after 5 seconds
  setTimeout(() => {
    consumer.resume([{ topic: 'my-topic', partition: 0 }]);
  }, 5000);

  consumer.consume();
});

consumer.on('data', (msg) => console.log(msg.offset, msg.value.toString()));
consumer.connect();
```
```

--------------------------------

### producer.produce(topic, partition, msg, key, timestamp, opaque)

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Sends a message. The `produce()` method throws when produce would return an error. Ordinarily, this is just if the queue is full.

```APIDOC
## producer.produce(topic, partition, msg, key, timestamp, opaque)

### Description
Sends a message. The `produce()` method throws when produce would return an error. Ordinarily, this is just if the queue is full.

### Method

```
producer.produce(topic, partition, msg, key, timestamp, opaque)
```

### Parameters
#### Path Parameters
- **topic** (string) - Required - The topic to send the message to.
- **partition** (number) - Optional - Optionally we can manually specify a partition for the message. This defaults to -1, which will use librdkafka's default partitioner.
- **msg** (Buffer) - Required - Message to send. Must be a buffer.
- **key** (string) - Optional - For keyed messages, we also specify the key. Note that this field is optional.
- **timestamp** (number) - Optional - You can send a timestamp here. If your broker version supports it, it will get added. Otherwise, we default to 0.
- **opaque** (any) - Optional - You can send an opaque token here, which gets passed along to your delivery reports.
```

--------------------------------

### Produce Messages with Standard API

Source: https://github.com/blizzard/node-rdkafka/blob/master/README.md

Use this snippet to send messages using the standard producer API. Ensure you connect to the broker and wait for the 'ready' event before producing. Manual polling or setting an interval is required for delivery reports.

```javascript
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 {
    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);
```