### Create a Basic Kafka Producer

Source: https://github.com/tulios/kafkajs/blob/master/docs/Producing.md

Instantiate a producer using the Kafka client. This is the simplest way to get started.

```javascript
const producer = kafka.producer()
```

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

### Start Local Development Environment

Source: https://github.com/tulios/kafkajs/blob/master/docs/ContributionGuide.md

Bring up the local development environment using docker-compose. This command is part of the setup for local testing.

```sh
./scripts/dockerComposeUp.sh
```

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

### JavaScript Producer Example

Source: https://github.com/tulios/kafkajs/blob/master/docs/ProducerExample.md

This snippet shows a basic Kafka.js producer setup in JavaScript. It connects to Kafka, sends random messages at intervals, and includes graceful shutdown handling for errors and signals.

```javascript
const ip = require('ip')

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

const host = process.env.HOST_IP || ip.address()

const kafka = new Kafka({
  logLevel: logLevel.DEBUG,
  brokers: [`${host}:9092`],
  clientId: 'example-producer',
})

const topic = 'topic-test'
const producer = kafka.producer()

const getRandomNumber = () => Math.round(Math.random(10) * 1000)
const createMessage = num => ({
  key: `key-${num}`,
  value: `value-${num}-${new Date().toISOString()}`,
})

const sendMessage = () => {
  return producer
    .send({
      topic,
      compression: CompressionTypes.GZIP,
      messages: Array(getRandomNumber())
        .fill()
        .map(_ => createMessage(getRandomNumber()))
    })
    .then(console.log)
    .catch(e => console.error(`[example/producer] ${e.message}`, e))
}

const run = async () => {
  await producer.connect()
  setInterval(sendMessage, 3000)
}

run().catch(e => console.error(`[example/producer] ${e.message}`, e))

const errorTypes = ['unhandledRejection', 'uncaughtException']
const signalTraps = ['SIGTERM', 'SIGINT', 'SIGUSR2']

errorTypes.forEach(type => {
  process.on(type, async () => {
    try {
      console.log(`process.on ${type}`)
      await producer.disconnect()
      process.exit(0)
    } catch (_) {
      process.exit(1)
    }
  })
})

signalTraps.forEach(type => {
  process.once(type, async () => {
    try {
      await producer.disconnect()
    } finally {
      process.kill(process.pid, type)
    }
  })
})
```

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

### Start Kafka Cluster and Create Credentials

Source: https://github.com/tulios/kafkajs/blob/master/docs/DevelopmentEnvironment.md

Run these commands from the repository root to start the multi-broker Kafka cluster and generate SCRAM credentials.

```sh
# This will run a Kafka cluster configured with your current IP
./scripts/dockerComposeUp.sh
./scripts/createScramCredentials.sh
```

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

### Start Development Server

Source: https://github.com/tulios/kafkajs/blob/master/website/README.md

Run the development server to preview the Docusaurus website locally.

```sh
# Start the site
$ yarn start
```

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

### Install and Configure Snappy Compression

Source: https://github.com/tulios/kafkajs/blob/master/docs/Producing.md

Install the 'kafkajs-snappy' package and configure it to use Snappy compression with KafkaJS.

```sh
npm install --save kafkajs-snappy
# yarn add kafkajs-snappy
```

```javascript
const {  CompressionTypes, CompressionCodecs } = require('kafkajs')
const SnappyCodec = require('kafkajs-snappy')

CompressionCodecs[CompressionTypes.Snappy] = SnappyCodec
```

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

### Install KafkaJS with NPM

Source: https://github.com/tulios/kafkajs/blob/master/docs/GettingStarted.md

Install the KafkaJS library using the NPM package manager.

```bash
npm install kafkajs
```

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

### Install KafkaJS with Yarn

Source: https://github.com/tulios/kafkajs/blob/master/docs/GettingStarted.md

Install the KafkaJS library using the Yarn package manager.

```bash
yarn add kafkajs
```

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

### Install KafkaJS

Source: https://github.com/tulios/kafkajs/blob/master/README.md

Install the KafkaJS package using npm or yarn.

```sh
npm install kafkajs
# yarn add kafkajs
```

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

### Install Dependencies

Source: https://github.com/tulios/kafkajs/blob/master/website/README.md

Install the necessary dependencies for the Docusaurus website using yarn.

```sh
# Install dependencies
$ yarn
```

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

### Subscribe with fromBeginning Option

Source: https://github.com/tulios/kafkajs/blob/master/docs/Consuming.md

Configure whether to start consuming from the earliest or latest offset when subscribing to a topic. The default behavior is to use the latest offset.

```javascript
await consumer.subscribe({ topics: ['test-topic'], fromBeginning: true })
await consumer.subscribe({ topics: ['other-topic'], fromBeginning: false })
```

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

### Install and Configure LZ4 Compression

Source: https://github.com/tulios/kafkajs/blob/master/docs/Producing.md

Install the 'kafkajs-lz4' package and configure it to use LZ4 compression with KafkaJS. Options can be passed for granular control.

```sh
npm install --save kafkajs-lz4
# yarn add kafkajs-lz4
```

```javascript
const { CompressionTypes, CompressionCodecs } = require('kafkajs')
const LZ4 = require('kafkajs-lz4')

CompressionCodecs[CompressionTypes.LZ4] = new LZ4().codec
```

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

### Kafka.js Consumer Setup and Run

Source: https://github.com/tulios/kafkajs/blob/master/docs/ConsumerExample.md

This snippet shows how to initialize a Kafka client, create a consumer, connect to Kafka, subscribe to a topic, and process incoming messages. It also includes error handling and graceful shutdown logic for unhandled rejections, uncaught exceptions, and termination signals.

```javascript
const ip = require('ip')

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

const host = process.env.HOST_IP || ip.address()

const kafka = new Kafka({
  logLevel: logLevel.INFO,
  brokers: [`${host}:9092`],
  clientId: 'example-consumer',
})

const topic = 'topic-test'
const consumer = kafka.consumer({ groupId: 'test-group' })

const run = async () => {
  await consumer.connect()
  await consumer.subscribe({ topic, fromBeginning: true })
  await consumer.run({
    // eachBatch: async ({ batch }) => {
    //   console.log(batch)
    // },
    eachMessage: async ({ topic, partition, message }) => {
      const prefix = `${topic}[${partition} | ${message.offset}] / ${message.timestamp}`
      console.log(`- ${prefix} ${message.key}#${message.value}`)
    },
  })
}

run().catch(e => console.error(`[example/consumer] ${e.message}`, e))

const errorTypes = ['unhandledRejection', 'uncaughtException']
const signalTraps = ['SIGTERM', 'SIGINT', 'SIGUSR2']

errorTypes.forEach(type => {
  process.on(type, async e => {
    try {
      console.log(`process.on ${type}`)
      console.error(e)
      await consumer.disconnect()
      process.exit(0)
    } catch (_) {
      process.exit(1)
    }
  })
})

signalTraps.forEach(type => {
  process.once(type, async () => {
    try {
      await consumer.disconnect()
    } finally {
      process.kill(process.pid, type)
    }
  })
})

```

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

### TypeScript Kafka Consumer Example

Source: https://github.com/tulios/kafkajs/blob/master/docs/ConsumerExample.md

This snippet shows a complete TypeScript class for a Kafka consumer. It includes methods to start consuming individual messages or batches, handle connection and subscription, and shut down the consumer. It requires an `ExampleMessageProcessor` to be provided.

```typescript
import { Consumer, ConsumerSubscribeTopics, EachBatchPayload, Kafka, EachMessagePayload } from 'kafkajs'

export default class ExampleConsumer {
  private kafkaConsumer: Consumer
  private messageProcessor: ExampleMessageProcessor

  public constructor(messageProcessor: ExampleMessageProcessor) {
    this.messageProcessor = messageProcessor
    this.kafkaConsumer = this.createKafkaConsumer()
  }

  public async startConsumer(): Promise<void> {
    const topic: ConsumerSubscribeTopics = {
      topics: ['example-topic'],
      fromBeginning: false
    }

    try {
      await this.kafkaConsumer.connect()
      await this.kafkaConsumer.subscribe(topic)

      await this.kafkaConsumer.run({
        eachMessage: async (messagePayload: EachMessagePayload) => {
          const { topic, partition, message } = messagePayload
          const prefix = `${topic}[${partition} | ${message.offset}] / ${message.timestamp}`
          console.log(`- ${prefix} ${message.key}#${message.value}`)
        }
      })
    } catch (error) {
      console.log('Error: ', error)
    }
  }

  public async startBatchConsumer(): Promise<void> {
    const topic: ConsumerSubscribeTopics = {
      topics: ['example-topic'],
      fromBeginning: false
    }

    try {
      await this.kafkaConsumer.connect()
      await this.kafkaConsumer.subscribe(topic)
      await this.kafkaConsumer.run({
        eachBatch: async (eachBatchPayload: EachBatchPayload) => {
          const { batch } = eachBatchPayload
          for (const message of batch.messages) {
            const prefix = `${batch.topic}[${batch.partition} | ${message.offset}] / ${message.timestamp}`
            console.log(`- ${prefix} ${message.key}#${message.value}`) 
          }
        }
      })
    } catch (error) {
      console.log('Error: ', error)
    }
  }

  public async shutdown(): Promise<void> {
    await this.kafkaConsumer.disconnect()
  }

  private createKafkaConsumer(): Consumer {
    const kafka = new Kafka({
      clientId: 'client-id',
      brokers: ['example.kafka.broker:9092']
    })
    const consumer = kafka.consumer({ groupId: 'consumer-group' })
    return consumer
  }
}

```

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

### Install and Configure ZSTD Compression

Source: https://github.com/tulios/kafkajs/blob/master/docs/Producing.md

Install the '@kafkajs/zstd' package and configure it to use ZSTD compression with KafkaJS. Configuration options can be passed to control levels.

```sh
npm install --save @kafkajs/zstd
# yarn add @kafkajs/zstd
```

```javascript
const { CompressionTypes, CompressionCodecs } = require('kafkajs')
const ZstdCodec = require('@kafkajs/zstd')

CompressionCodecs[CompressionTypes.ZSTD] = ZstdCodec()
```

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

### Run Tests with Yarn

Source: https://github.com/tulios/kafkajs/blob/master/docs/ContributionGuide.md

Execute the test suite using the yarn test command. Ensure you have yarn installed.

```sh
yarn test
```

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

### Describe Kafka Resource Configurations

Source: https://github.com/tulios/kafkajs/blob/master/docs/Admin.md

Get the configuration for specified resources. Use to query all or specific configurations.

```javascript
await admin.describeConfigs({
  includeSynonyms: <boolean>,
  resources: <ResourceConfigQuery[]> 
})
```

```javascript
{
    type: <ConfigResourceType>,
    name: <String>,
    configNames: <String[]>
}
```

```javascript
const { ConfigResourceTypes } = require('kafkajs')

await admin.describeConfigs({
  includeSynonyms: false,
  resources: [
    {
      type: ConfigResourceTypes.TOPIC,
      name: 'topic-name'
    }
  ]
})
```

```javascript
const { ConfigResourceTypes } = require('kafkajs')

await admin.describeConfigs({
  includeSynonyms: false,
  resources: [
    {
      type: ConfigResourceTypes.TOPIC,
      name: 'topic-name',
      configNames: ['cleanup.policy']
    }
  ]
})
```

```javascript
{
    resources: [
        {
            configEntries: [{
                configName: 'cleanup.policy',
                configValue: 'delete',
                isDefault: true,
                configSource: 5,
                isSensitive: false,
                readOnly: false
            }],
            errorCode: 0,
            errorMessage: null,
            resourceName: 'topic-name',
            resourceType: 2
        }
    ],
    throttleTime: 0
}
```

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

### Configure SASL PLAIN/SCRAM Authentication with SSL

Source: https://github.com/tulios/kafkajs/blob/master/docs/Configuration.md

This example demonstrates how to configure SASL authentication using PLAIN or SCRAM mechanisms along with SSL enabled. It requires a username and password for authentication.

```javascript
new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  // authenticationTimeout: 10000,
  // reauthenticationThreshold: 10000,
  ssl: true,
  sasl: {
    mechanism: 'plain', // scram-sha-256 or scram-sha-512
    username: 'my-username',
    password: 'my-password'
  },
})
```

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

### Connect to Kafka Cluster with SSL and SASL

Source: https://github.com/tulios/kafkajs/blob/master/docs/DevelopmentEnvironment.md

Example JavaScript code to instantiate the Kafka client, connecting to a local cluster with SSL and SASL authentication.

```javascript
const fs = require('fs')
const ip = require('ip')

const { Kafka, CompressionTypes, logLevel } = require('./index')

const host = process.env.HOST_IP || ip.address()

const kafka = new Kafka({
  logLevel: logLevel.DEBUG,
  brokers: [`${host}:9094`, `${host}:9097`, `${host}:9100`],
  clientId: 'example-producer',
  ssl: {
    servername: 'localhost',
    rejectUnauthorized: false,
    ca: [fs.readFileSync('./testHelpers/certs/cert-signed', 'utf-8')],
  },
  sasl: {
    mechanism: 'plain',
    username: 'test',
    password: 'testtest',
  },
})
```

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

### Export Host IP and Start Docker Compose

Source: https://github.com/tulios/kafkajs/blob/master/docs/DockerLocal.md

These commands are used to set the HOST_IP environment variable, which is necessary for Kafka to advertise its correct network interface, and then to start the Kafka and ZooKeeper services defined in docker-compose.yml.

```sh
export HOST_IP=$(ifconfig | grep -E "([0-9]{1,3}\.){3}[0-9]{1,3}" | grep -v 127.0.0.1 | awk '{ print $2 }' | cut -f2 -d: | head -n1)
docker-compose up
```

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

### Start Kafka Cluster with Custom Compose File

Source: https://github.com/tulios/kafkajs/blob/master/docs/DevelopmentEnvironment.md

Specify a different compose file using the COMPOSE_FILE environment variable to run a different version of Kafka.

```sh
COMPOSE_FILE="docker-compose.2_3.yml" ./scripts/dockerComposeUp.sh
```

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

### Subscribe from Beginning of Topic

Source: https://github.com/tulios/kafkajs/blob/master/docs/Consuming.md

Configure the subscription to start consuming messages from the earliest available offset in the specified topic. Useful for reprocessing or initial data loading.

```javascript
await consumer.subscribe({ topics: ['topic-D'], fromBeginning: true })
```

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

### Install Latest Pre-release Version

Source: https://github.com/tulios/kafkajs/blob/master/docs/PreReleases.md

Use these commands to install the latest pre-release version of KafkaJS using Yarn or Npm. This is useful for testing upcoming features before they are officially released.

```sh
# Yarn
yarn add kafkajs@beta

# Npm
npm install --save kafkajs@beta
```

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

### TypeScript Producer with Batch Sending

Source: https://github.com/tulios/kafkajs/blob/master/docs/ProducerExample.md

This TypeScript example demonstrates creating a producer class that supports batch message sending. It includes connection management and a method to send multiple messages efficiently.

```typescript
import { Kafka, Message, Producer, ProducerBatch, TopicMessages } from 'kafkajs'

interface CustomMessageFormat { a: string }

export default class ProducerFactory {
  private producer: Producer

  constructor() {
    this.producer = this.createProducer()
  }

  public async start(): Promise<void> {
    try {
      await this.producer.connect()
    } catch (error) {
      console.log('Error connecting the producer: ', error)
    }
  }

  public async shutdown(): Promise<void> {
    await this.producer.disconnect()
  }

  public async sendBatch(messages: Array<CustomMessageFormat>): Promise<void> {
    const kafkaMessages: Array<Message> = messages.map((message) => {
      return {
        value: JSON.stringify(message)
      }
    })

    const topicMessages: TopicMessages = {
      topic: 'producer-topic',
      messages: kafkaMessages
    }

    const batch: ProducerBatch = {
      topicMessages: [topicMessages]
    }

    await this.producer.sendBatch(batch)
  }

  private createProducer() : Producer {
    const kafka = new Kafka({
      clientId: 'producer-client',
      brokers: ['localhost:9092'],
    })

    return kafka.producer()
  }
}
```

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

### Consume Messages in Batches with Manual Resolve and Shutdown Handling

Source: https://github.com/tulios/kafkajs/blob/master/docs/Consuming.md

This example shows `eachBatch` with `eachBatchAutoResolve` set to false, requiring manual offset resolution. It includes checks for `isRunning` and `isStale` to ensure graceful shutdown and prevent processing stale messages.

```javascript
consumer.run({
    eachBatchAutoResolve: false,
    eachBatch: async ({ batch, resolveOffset, heartbeat, isRunning, isStale }) => {
        for (let message of batch.messages) {
            if (!isRunning() || isStale()) break
            await processMessage(message)
            resolveOffset(message.offset)
            await heartbeat()
        }
    }
})

```

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

### Configure Default Retry Mechanism

Source: https://github.com/tulios/kafkajs/blob/master/docs/Configuration.md

Customize the retry mechanism for connections and API calls. This example sets the initial retry time and the maximum number of retries.

```javascript
new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  retry: {
    initialRetryTime: 100,
    retries: 8
  }
})
```

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

### Custom Socket Factory for Proxy Support

Source: https://github.com/tulios/kafkajs/blob/master/docs/Configuration.md

Create a custom socket factory to enable proxy support for KafkaJS connections. This example uses the 'proxy-chain' library to establish a tunnel through an HTTP proxy.

```javascript
const tls = require('tls')
const net = require('net')
const { createTunnel, closeTunnel } = require('proxy-chain')

const socketFactory = ({ host, port, ssl, onConnect }) => {
  const socket = ssl ? new tls.TLSSocket() : new net.Socket()

  createTunnel(process.env.HTTP_PROXY, `${host}:${port}`)
    .then((tunnelAddress) => {
      const [tunnelHost, tunnelPort] = tunnelAddress.split(':')
      socket.setKeepAlive(true, 60000)
      socket.connect(
        Object.assign({ host: tunnelHost, port: tunnelPort, servername: host }, ssl),
        onConnect
      )

      socket.on('close', () => {
        closeTunnel(tunnelServer, true)
      })
    })
    .catch(error => socket.emit('error', error))

  return socket
}
```

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

### OAUTHBEARER Authentication with simple-oauth2

Source: https://github.com/tulios/kafkajs/blob/master/docs/Configuration.md

Implement a robust oauthBearerProvider using simple-oauth2 to handle token refreshing. This example shows how to configure the client and manage token expiration.

```typescript
import { AccessToken, ClientCredentials } from 'simple-oauth2'
interface OauthBearerProviderOptions {
  clientId: string;
  clientSecret: string;
  host: string;
  path: string;
  refreshThresholdMs: number;
}

const oauthBearerProvider = (options: OauthBearerProviderOptions) => {
  const client = new ClientCredentials({
    client: {
      id: options.clientId,
      secret: options.clientSecret
    },
    auth: {
      tokenHost: options.host,
      tokenPath: options.path
    }
  });

  let tokenPromise: Promise<string>;
  let accessToken: AccessToken;

  async function refreshToken() {
    try {
      if (accessToken == null) {
        accessToken = await client.getToken({})
      }

      if (accessToken.expired(options.refreshThresholdMs / 1000)) {
        accessToken = await accessToken.refresh()
      }

      const nextRefresh = accessToken.token.expires_in * 1000 - options.refreshThresholdMs;
      setTimeout(() => {
        tokenPromise = refreshToken()
      }, nextRefresh);

      return accessToken.token.access_token;
    } catch (error) {
      accessToken = null;
      throw error;
    }
  }

  tokenPromise = refreshToken();

  return async function () {
    return {
      value: await tokenPromise
    }
  }
};

const kafka = new Kafka({
  // ... other required options
  sasl: {
    mechanism: 'oauthbearer',
    oauthBearerProvider: oauthBearerProvider({
      clientId: 'oauth-client-id',
      clientSecret: 'oauth-client-secret',
      host: 'https://my-oauth-server.com',
      path: '/oauth/token',
      // Refresh the token 15 seconds before it expires
      refreshThreshold: 15000,
    }),
  },
})
```

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

### Basic KafkaJS Producer and Consumer Usage

Source: https://github.com/tulios/kafkajs/blob/master/README.md

Demonstrates how to create a Kafka client, connect a producer and consumer, send a message, and subscribe to a topic to consume messages.

```javascript
const { Kafka } = require('kafkajs')

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092']
})

const producer = kafka.producer()
const consumer = kafka.consumer({ groupId: 'test-group' })

const run = async () => {
  // Producing
  await producer.connect()
  await producer.send({
    topic: 'test-topic',
    messages: [
      { value: 'Hello KafkaJS user!' },
    ],
  })

  // Consuming
  await consumer.connect()
  await consumer.subscribe({ topic: 'test-topic', fromBeginning: true })

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

run().catch(console.error)
```

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

### Admin: Create Topics with Specific Partition and Replication Factor

Source: https://github.com/tulios/kafkajs/blob/master/docs/MigrationGuide-2-0-0.md

When creating topics, you can now specify `numPartitions` and `replicationFactor` to override cluster defaults. If not provided and cluster defaults are not set, topic creation will fail.

```javascript
await admin.createTopics({
  topics: [{ topic: 'topic-name', numPartitions: 1, replicationFactor: 1 }]
})
```

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

### Initialize and Connect Admin Client

Source: https://github.com/tulios/kafkajs/blob/master/docs/Admin.md

Instantiate the KafkaJS Admin client and establish a connection. Remember to disconnect when operations are complete.

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

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

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

### Create Topics Configuration

Source: https://github.com/tulios/kafkajs/blob/master/docs/Admin.md

Defines the structure for configuring new topics, including partition count, replication factor, and custom configuration entries.

```javascript
await admin.createTopics({
    validateOnly: <boolean>,
    waitForLeaders: <boolean>
    timeout: <Number>,
    topics: <ITopicConfig[]>,
})
```

```javascript
{
    topic: <String>,
    numPartitions: <Number>,     // default: -1 (uses broker `num.partitions` configuration)
    replicationFactor: <Number>, // default: -1 (uses broker `default.replication.factor` configuration)
    replicaAssignment: <Array>,  // Example: [{ partition: 0, replicas: [0,1,2] }] - default: []
    configEntries: <Array>       // Example: [{ name: 'cleanup.policy', value: 'compact' }] - default: []
}
```

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

### Create Partitions for Kafka Topics

Source: https://github.com/tulios/kafkajs/blob/master/docs/Admin.md

This snippet demonstrates how to add new partitions to an existing Kafka topic. The `validateOnly` flag can be used to test the request without making changes.

```javascript
await admin.createPartitions({
    validateOnly: <boolean>,
    timeout: <Number>,
    topicPartitions: <TopicPartition[]>,
})
```

```javascript
{
    topic: <String>,
    count: <Number>,     // partition count
    assignments: <Array<Array<Number>>> // Example: [[0,1],[1,2],[2,0]]
}
```

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

### Create Partitions

Source: https://github.com/tulios/kafkajs/blob/master/docs/Admin.md

Creates new partitions for existing topics. The operation resolves on success and throws exceptions on errors.

```APIDOC
## Create Partitions

### Description
Creates new partitions for one or more existing topics. The method resolves in case of success and throws exceptions in case of errors. A `validateOnly` flag can be used to perform validation without actual creation.

### Method
`admin.createPartitions(options)

### Parameters
#### Request Body
- **topicPartitions** (TopicPartition[]) - Required - An array of topic partition definitions.
- **validateOnly** (boolean) - Optional - If true, the request is validated but no partitions are created. Defaults to `false`.
- **timeout** (Number) - Optional - The time in ms to wait for the operation to complete. Defaults to 5000.

`TopicPartition` structure:
```javascript
{
    topic: <String>,       // The name of the topic
    count: <Number>,       // The new partition count for the topic
    assignments: <Array<Array<Number>>> // Optional: Array of broker IDs for each new partition assignment
}
```

### Request Example
```javascript
await admin.createPartitions({
    topicPartitions: [
        {
            topic: 'my-topic',
            count: 3,
            assignments: [[0, 1], [1, 2], [2, 0]]
        }
    ],
    validateOnly: false,
    timeout: 10000
})
```

### Response
This method does not return a value on success, but will throw an exception on error.
```

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

### Winston Log Creator Implementation

Source: https://github.com/tulios/kafkajs/blob/master/docs/CustomLogger.md

Demonstrates how to create a custom log creator using the Winston logging library for Kafka.js.

```javascript
const { logLevel } = require('kafkajs')
const winston = require('winston')
const toWinstonLogLevel = level => {
    switch(level) {
        case logLevel.ERROR:
        case logLevel.NOTHING:
            return 'error'
        case logLevel.WARN:
            return 'warn'
        case logLevel.INFO:
            return 'info'
        case logLevel.DEBUG:
            return 'debug'
    }
}

const WinstonLogCreator = logLevel => {
    const logger = winston.createLogger({
        level: toWinstonLogLevel(logLevel),
        transports: [
            new winston.transports.Console(),
            new winston.transports.File({ filename: 'myapp.log' })
        ]
    })

    return ({ namespace, level, label, log }) => {
        const { message, ...extra } = log
        logger.log({
            level: toWinstonLogLevel(level),
            message,
            extra,
        })
    }
}
```

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

### Verify Kafka Cluster Status

Source: https://github.com/tulios/kafkajs/blob/master/docs/DevelopmentEnvironment.md

Check the status of the Kafka cluster after starting it, using the appropriate compose file.

```sh
$ docker-compose -f docker-compose.2_3.yml ps
WARNING: The HOST_IP variable is not set. Defaulting to a blank string.
      Name                    Command               State                                   Ports
----------------------------------------------------------------------------------------------------------------------------------
kafkajs_kafka1_1   start-kafka.sh                   Up      0.0.0.0:9092->9092/tcp, 0.0.0.0:9093->9093/tcp, 0.0.0.0:9094->9094/tcp
kafkajs_kafka2_1   start-kafka.sh                   Up      0.0.0.0:9095->9095/tcp, 0.0.0.0:9096->9096/tcp, 0.0.0.0:9097->9097/tcp
kafkajs_kafka3_1   start-kafka.sh                   Up      0.0.0.0:9098->9098/tcp, 0.0.0.0:9099->9099/tcp, 0.0.0.0:9100->9100/tcp
kafkajs_zk_1       /bin/sh -c /usr/sbin/sshd  ...   Up      0.0.0.0:2181->2181/tcp, 22/tcp, 2888/tcp, 3888/tcp
```

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

### Produce JSON Message

Source: https://github.com/tulios/kafkajs/blob/master/docs/KafkaIntro.md

Example of sending a JSON message using the KafkaJS producer. The value is stringified before sending.

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

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

### Consume and Parse JSON Message

Source: https://github.com/tulios/kafkajs/blob/master/docs/KafkaIntro.md

Example of consuming a message and parsing its key and value from byte buffers to string and JSON object respectively.

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

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

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

### Instantiate KafkaJS Client

Source: https://github.com/tulios/kafkajs/blob/master/docs/GettingStarted.md

Create a KafkaJS client instance by providing a client ID and a list of Kafka brokers.

```javascript
const { Kafka } = require('kafkajs')

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
})
```

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

### Producer with Default Partitioner (Java Compatible)

Source: https://github.com/tulios/kafkajs/blob/master/docs/MigrationGuide-2-0-0.md

This snippet shows how to rely on the new default partitioner, which is compatible with the Java client's behavior. Existing code using JavaCompatiblePartitioner will continue to work but should be updated.

```javascript
// Rely on the default partitioner being compatible with the Java partitioner
kafka.producer()
```

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

### Configure Auto-Commit Interval

Source: https://github.com/tulios/kafkajs/blob/master/docs/Consuming.md

Set `autoCommitInterval` to commit offsets to Kafka periodically, for example, every 5000 milliseconds. This helps with consumer recovery during group rebalancing.

```javascript
consumer.run({
  autoCommitInterval: 5000,
  // ...
})
```

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

### Configure SSL with Custom Certificates

Source: https://github.com/tulios/kafkajs/blob/master/docs/Configuration.md

Use this snippet to enable SSL and provide custom CA certificates, client keys, and client certificates for secure connections. Ensure the certificate and key files are correctly specified.

```javascript
const fs = require('fs')

new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  ssl: {
    rejectUnauthorized: false,
    ca: [fs.readFileSync('/my/custom/ca.crt', 'utf-8')],
    key: fs.readFileSync('/my/custom/client-key.pem', 'utf-8'),
    cert: fs.readFileSync('/my/custom/client-cert.pem', 'utf-8')
  },
})
```

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

### Configure Kafka Client with Custom Logger

Source: https://github.com/tulios/kafkajs/blob/master/docs/CustomLogger.md

Shows how to integrate a custom log creator, like the WinstonLogCreator, into the Kafka client configuration.

```javascript
const kafka = new Kafka({
    clientId: 'my-app',
    brokers: ['kafka1:9092', 'kafka2:9092'],
    logLevel: logLevel.ERROR,
    logCreator: WinstonLogCreator
})
```

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

### Create Kafka Client with Static Brokers

Source: https://github.com/tulios/kafkajs/blob/master/docs/Configuration.md

Instantiate a KafkaJS client with a list of seed brokers. This is the basic configuration for connecting to a Kafka cluster.

```javascript
const { Kafka } = require('kafkajs')

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

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

### Set Log Level via Environment Variable

Source: https://github.com/tulios/kafkajs/blob/master/docs/Configuration.md

The `KAFKAJS_LOG_LEVEL` environment variable can be used to set the log level, which takes precedence over code configuration. Example: set log level to 'info'.

```sh
KAFKAJS_LOG_LEVEL=info node code.js
```

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

### fetchOffsets

Source: https://github.com/tulios/kafkajs/blob/master/docs/Admin.md

Fetches the consumer group offsets for a list of topics. If `topics` is omitted, it retrieves offsets for all topics with committed offsets. The optional `resolveOffsets` flag can be used to resolve offsets without starting a consumer.

```APIDOC
## fetchOffsets

### Description
Returns the consumer group offset for a list of topics. Omit `topics` to get offsets for all topics with committed offsets. The optional `resolveOffsets` flag can be used to resolve offsets without starting a consumer.

### Method
`admin.fetchOffsets(config)

### Parameters
#### Arguments
- **config** (object) - Required - Configuration object for fetching offsets.
  - **groupId** (string) - Required - The ID of the consumer group.
  - **topics** (Array<string>) - Optional - An array of topic names. If omitted, offsets for all topics are fetched.
  - **resolveOffsets** (boolean) - Optional - If true, resolves offsets without starting a consumer. Defaults to false.

### Request Example
```javascript
// Fetch offsets for specific topics
await admin.fetchOffsets({ groupId: 'my-group', topics: ['topic1', 'topic2'] })

// Fetch offsets for all topics
await admin.fetchOffsets({ groupId: 'my-group' })

// Fetch offsets and resolve them without starting a consumer
await admin.fetchOffsets({ groupId: 'my-group', topics: ['topic1'], resolveOffsets: true })
```

### Response
#### Success Response
Returns an array of objects, where each object represents a topic and its partitions with their respective offsets.
```json
[
  {
    "topic": "topic1",
    "partitions": [
      { "partition": 0, "offset": "31004" },
      { "partition": 1, "offset": "54312" },
      { "partition": 2, "offset": "32103" },
      { "partition": 3, "offset": "28" },
    ]
  },
  {
    "topic": "topic2",
    "partitions": [
      { "partition": 0, "offset": "1234" },
      { "partition": 1, "offset": "4567" },
    ]
  }
]
```

#### Example with resolveOffsets: true
```json
[
  { "partition": 0, "offset": "31004" },
  { "partition": 1, "offset": "54312" },
  { "partition": 2, "offset": "32103" },
  { "partition": 3, "offset": "28" },
]
```
```

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

### Run Local Tests in Watch Mode

Source: https://github.com/tulios/kafkajs/blob/master/docs/ContributionGuide.md

Start the local test suite in watch mode using yarn test:local:watch. This command automatically re-runs tests when code changes are detected.

```sh
yarn test:local:watch
```

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

### Configure Producer with LegacyPartitioner

Source: https://github.com/tulios/kafkajs/blob/master/docs/MigrationGuide-2-0-0.md

Use this snippet to maintain the previous default partitioner behavior by explicitly configuring the producer with LegacyPartitioner.

```javascript
const { Partitioners } = require('kafkajs')
kafka.producer({ createPartitioner: Partitioners.LegacyPartitioner })
```

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

### Fetch Offsets with resolveOffsets Flag

Source: https://github.com/tulios/kafkajs/blob/master/docs/Admin.md

This snippet demonstrates fetching offsets with the `resolveOffsets` flag. Set `resolveOffsets` to `false` to get unresolved offsets (typically '-1') or `true` to resolve them, which is useful after calling `resetOffsets`.

```javascript
await admin.resetOffsets({ groupId, topic })
await admin.fetchOffsets({ groupId, topics: [topic], resolveOffsets: false })
// [
//   { partition: 0, offset: '-1' },
//   { partition: 1, offset: '-1' },
//   { partition: 2, offset: '-1' },
//   { partition: 3, offset: '-1' },
// ]
```

```javascript
await admin.resetOffsets({ groupId, topic })
await admin.fetchOffsets({ groupId, topics: [topic], resolveOffsets: true })
// [
//   { partition: 0, offset: '31004' },
//   { partition: 1, offset: '54312' },
//   { partition: 2, offset: '32103' },
//   { partition: 3, offset: '28' },
// ]
```

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

### Create a Kafka Producer with Options

Source: https://github.com/tulios/kafkajs/blob/master/docs/Producing.md

Configure the producer with specific options such as disabling auto topic creation and setting a transaction timeout. This allows for more control over producer behavior.

```javascript
const producer = kafka.producer({
    allowAutoTopicCreation: false,
    transactionTimeout: 30000
})
```

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

### Get List of Paused Topic Partitions

Source: https://github.com/tulios/kafkajs/blob/master/docs/Consuming.md

Retrieve a list of all currently paused topic partitions using the `paused` method. This can be used for monitoring or debugging purposes to understand the consumer's current state.

```javascript
const pausedTopicPartitions = consumer.paused()

for (const topicPartitions of pausedTopicPartitions) {
  const { topic, partitions } = topicPartitions
  console.log({ topic, partitions })
}

```

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

### Describe Configs

Source: https://github.com/tulios/kafkajs/blob/master/docs/Admin.md

Fetches the configuration details for specified resources. You can choose to include synonyms and specify particular configuration names.

```APIDOC
## Describe Configs

### Description
Get the configuration for the specified resources.

### Method
```javascript
await admin.describeConfigs({
  includeSynonyms: <boolean>,
  resources: <ResourceConfigQuery[]> 
})
```

`ResourceConfigQuery` structure:

```javascript
{
    type: <ConfigResourceType>,
    name: <String>,
    configNames: <String[]> 
}
```

### Request Example (All configs for a resource)
```javascript
const { ConfigResourceTypes } = require('kafkajs')

await admin.describeConfigs({
  includeSynonyms: false,
  resources: [
    {
      type: ConfigResourceTypes.TOPIC,
      name: 'topic-name'
    }
  ]
})
```

### Request Example (Specific configs for a resource)
```javascript
const { ConfigResourceTypes } = require('kafkajs')

await admin.describeConfigs({
  includeSynonyms: false,
  resources: [
    {
      type: ConfigResourceTypes.TOPIC,
      name: 'topic-name',
      configNames: ['cleanup.policy']
    }
  ]
})
```

### Response Example
```json
{
    "resources": [
        {
            "configEntries": [
                {
                    "configName": "cleanup.policy",
                    "configValue": "delete",
                    "isDefault": true,
                    "configSource": 5,
                    "isSensitive": false,
                    "readOnly": false
                }
            ],
            "errorCode": 0,
            "errorMessage": null,
            "resourceName": "topic-name",
            "resourceType": 2
        }
    ],
    "throttleTime": 0
}
```
```

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

### Explicitly Configure Producer with DefaultPartitioner

Source: https://github.com/tulios/kafkajs/blob/master/docs/MigrationGuide-2-0-0.md

Explicitly configure the producer to use the new default partitioner. This is recommended for clarity and to prepare for future removal of the JavaCompatiblePartitioner export.

```javascript
const { Partitioners } = require('kafkajs')
kafka.producer({ createPartitioner: Partitioners.DefaultPartitioner })
```

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

### Enable Debug Logging and Protocol Buffers

Source: https://github.com/tulios/kafkajs/blob/master/docs/Testing.md

Runs local tests in watch mode with debug logging enabled and protocol buffer values displayed.

```sh
KAFKAJS_LOG_LEVEL=debug KAFKAJS_DEBUG_PROTOCOL_BUFFERS=1 yarn test:local:watch
```

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

### Define a Custom Partitioner Function

Source: https://github.com/tulios/kafkajs/blob/master/docs/Producing.md

Create a custom partitioner by returning a function that selects a partition based on provided message and topic metadata. This function should return the desired partition number.

```javascript
const MyPartitioner = () => {
    return ({ topic, partitionMetadata, message }) => {
        // select a partition based on some logic
        // return the partition number
        return 0
    }
}
```

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

### Add New Docs Page

Source: https://github.com/tulios/kafkajs/blob/master/website/README.md

Create a new markdown file in the 'docs/' directory for a new documentation page. The frontmatter must include 'id' and 'title'.

```md
---
id: newly-created-doc
title: This Doc Needs To Be Edited
---

My new content here..
```

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

### List Partition Reassignments

Source: https://github.com/tulios/kafkajs/blob/master/docs/Admin.md

Call this method to list current partition reassignments. It accepts an optional topics array and timeout. If topics is null, all topics are returned. Defaults to a 5000ms timeout.

```javascript
await admin.listPartitionReassignments({
  topics: <TopicPartitions[]>,
  timeout: <Number>
})
```

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

### Add custom partition assigner to consumer configuration

Source: https://github.com/tulios/kafkajs/blob/master/docs/Consuming.md

Include your custom partition assigner in the `partitionAssigners` array when creating a consumer. It's recommended to also include the default `roundRobin` assigner to ensure compatibility with older consumers.

```javascript
const { PartitionAssigners: { roundRobin } } = require('kafkajs')

kafka.consumer({
    groupId: 'my-group',
    partitionAssigners: [
        MyPartitionAssigner,
        roundRobin
    ]
})
```

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

### Create topics

Source: https://github.com/tulios/kafkajs/blob/master/docs/Admin.md

`createTopics` will resolve to `true` if the topic was created successfully or `false` if it already exists. The method will throw exceptions in case of errors.

```APIDOC
## Create topics

### Description
Creates one or more topics. Resolves to `true` if the topic was created successfully or `false` if it already exists. The method will throw exceptions in case of errors.

### Method
`createTopics(topicConfig: CreateTopicsOptions)`

### Endpoint
N/A (SDK method)

### Parameters
#### Request Body
- **topics** (ITopicConfig[]) - Required - Topic definition.
- **validateOnly** (boolean) - Optional - If this is `true`, the request will be validated, but the topic won't be created. Defaults to `false`.
- **waitForLeaders** (boolean) - Optional - If this is `true` it will wait until metadata for the new topics doesn't throw `LEADER_NOT_AVAILABLE`. Defaults to `true`.
- **timeout** (Number) - Optional - The time in ms to wait for a topic to be completely created on the controller node. Defaults to `5000`.

`ITopicConfig` structure:
- **topic** (String) - Required - The name of the topic.
- **numPartitions** (Number) - Optional - The number of partitions for the topic. Defaults to broker `num.partitions` configuration (-1).
- **replicationFactor** (Number) - Optional - The replication factor for the topic. Defaults to broker `default.replication.factor` configuration (-1).
- **replicaAssignment** (Array) - Optional - Example: `[{ partition: 0, replicas: [0,1,2] }]`. Defaults to `[]`.
- **configEntries** (Array) - Optional - Example: `[{ name: 'cleanup.policy', value: 'compact' }]`. Defaults to `[]`.

### Request Example
```json
{
  "topics": [
    {
      "topic": "my-topic",
      "numPartitions": 1,
      "replicationFactor": 1,
      "configEntries": [
        { "name": "cleanup.policy", "value": "compact" }
      ]
    }
  ],
  "waitForLeaders": true,
  "timeout": 10000
}
```

### Response
#### Success Response
- **boolean**: `true` if the topic was created successfully, `false` if it already exists.

#### Response Example
```json
true
```
```

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

### Producer with New Default Partitioner

Source: https://github.com/tulios/kafkajs/blob/master/docs/MigrationGuide-2-0-0.md

Use this snippet when co-partitioning behavior from previous versions is not a concern. This configures the producer to use the new default partitioner.

```javascript
kafka.producer()
```

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

### Kafkajs Consumer Options

Source: https://github.com/tulios/kafkajs/blob/master/docs/Consuming.md

Instantiate a Kafkajs consumer with various configuration options. These options control group ID, partition assignment, timeouts, and more.

```javascript
kafka.consumer({
  groupId: <String>,
  partitionAssigners: <Array>,
  sessionTimeout: <Number>,
  rebalanceTimeout: <Number>,
  heartbeatInterval: <Number>,
  metadataMaxAge: <Number>,
  allowAutoTopicCreation: <Boolean>,
  maxBytesPerPartition: <Number>,
  minBytes: <Number>,
  maxBytes: <Number>,
  maxWaitTimeInMs: <Number>,
  retry: <Object>,
  maxInFlightRequests: <Number>,
  rackId: <String>
})
```

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

### Enable Extended Debug Logging and Fetch Response Buffers

Source: https://github.com/tulios/kafkajs/blob/master/docs/Testing.md

Runs local tests in watch mode with debug logging, protocol buffer values, and full Fetch response buffer values enabled.

```sh
KAFKAJS_LOG_LEVEL=debug KAFKAJS_DEBUG_PROTOCOL_BUFFERS=1 KAFKAJS_DEBUG_EXTENDED_PROTOCOL_BUFFERS=1 yarn test:local:watch
```

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

### Admin: Fetch Offsets for Single Topic (Before)

Source: https://github.com/tulios/kafkajs/blob/master/docs/MigrationGuide-2-0-0.md

This is the previous implementation for fetching offsets for a single topic using `fetchOffsets`.

```javascript
// Before
const partitions = await admin.fetchOffsets({ groupId, topic: 'topic-a' })
for (const { partition, offset } of partitions) {
    admin.logger().info(`${groupId} is at offset ${offset} of partition ${partition}`)
}
```

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

### Pre-release Package Information in package.json

Source: https://github.com/tulios/kafkajs/blob/master/docs/PreReleases.md

Pre-release versions of KafkaJS include additional metadata in the `package.json` file. This metadata contains the git SHA and a comparison URL to track changes introduced in the pre-release.

```json
{
  // package.json
  "kafkajs": {
    "sha": "43e325e18133b8d6c1c80f8e95ef8610c44ec631",
    "compare": "https://github.com/tulios/kafkajs/compare/v1.9.3...43e325e18133b8d6c1c80f8e95ef8610c44ec631"
  }
}
```

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

### Use Custom Partitioner with Producer

Source: https://github.com/tulios/kafkajs/blob/master/docs/Producing.md

To implement a custom partitioner, pass your partitioner function to the `createPartitioner` option when creating the KafkaJS producer.

```javascript
kafka.producer({ createPartitioner: MyPartitioner })
```