### Setup Docker Environment

Source: https://github.com/platformatic/kafka/blob/main/CONTRIBUTING.md

Start the Docker environment for local testing. This is optional and only needed if you plan to run tests locally.

```bash
docker compose up -d
```

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

### Clone and Setup Project

Source: https://github.com/platformatic/kafka/blob/main/CONTRIBUTING.md

Clone the repository and navigate into the project directory.

```bash
git clone https://github.com/platformatic/kafka.git
cd kafka
```

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

### Install Confluent Schema Registry Support

Source: https://github.com/platformatic/kafka/blob/main/docs/confluent-schema-registry.md

Install the main package for Confluent Schema Registry support. For Protocol Buffers, also install protobufjs.

```bash
npm install @platformatic/kafka
```

```bash
npm install protobufjs
```

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

### Start Kafka in CI

Source: https://github.com/platformatic/kafka/blob/main/REGRESSION-TEST-PLAN.md

This command is used in GitHub Actions to start Kafka for CI environments. It ensures a clean and up-to-date Kafka instance is running.

```bash
docker compose up --build --force-recreate -d --wait
```

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

### Install Dependencies

Source: https://github.com/platformatic/kafka/blob/main/CONTRIBUTING.md

Install project dependencies using pnpm.

```bash
pnpm install
```

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

### Start Kafka Dependencies

Source: https://github.com/platformatic/kafka/blob/main/regression/README.md

Use this command to start the necessary Kafka services for the regression suite. It assumes you have docker-compose configured.

```sh
npm run test:docker:up
```

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

### Install @platformatic/kafka

Source: https://github.com/platformatic/kafka/blob/main/README.md

Install the @platformatic/kafka package using npm.

```bash
npm install @platformatic/kafka
```

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

### Install @platformatic/kafka

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Uninstall KafkaJS and install @platformatic/kafka.

```bash
npm uninstall kafkajs
npm install @platformatic/kafka
```

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

### Create Kafka Clients (@platformatic/kafka)

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Example of creating @platformatic/kafka producer, consumer, and admin clients via direct instantiation.

```javascript
import { Producer, Consumer, Admin } from '@platformatic/kafka'

const producer = new Producer({
  clientId: 'my-app',
  bootstrapBrokers: ['localhost:9092']
})

const consumer = new Consumer({
  clientId: 'my-app',
  bootstrapBrokers: ['localhost:9092'],
  groupId: 'my-group'
})

const admin = new Admin({
  clientId: 'my-app',
  bootstrapBrokers: ['localhost:9092']
})
```

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

### Create Kafka Clients (KafkaJS)

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Example of creating Kafka producer, consumer, and admin clients using the KafkaJS factory pattern.

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

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

const producer = kafka.producer()
const consumer = kafka.consumer({ groupId: 'my-group' })
const admin = kafka.admin()
```

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

### Simple Transaction Example

Source: https://github.com/platformatic/kafka/blob/main/docs/transactions.md

Demonstrates the basic workflow of initiating a transaction, sending messages, and then committing or aborting the transaction.

```APIDOC
## Simple Transaction

### Description
Demonstrates the basic workflow of initiating a transaction, sending messages, and then committing or aborting the transaction.

### Code Example
```typescript
// Begin a transaction
const transaction = await producer.beginTransaction()

try {
  // Send messages within the transaction
  await transaction.send({
    messages: [
      { topic: 'orders', value: 'order-1' },
      { topic: 'inventory', value: 'reduce-stock-1' }
    ]
  })

  // Commit the transaction
  await transaction.commit()
} catch (error) {
  // Abort the transaction on error
  await transaction.abort()
}
```
```

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

### Generate kafka.keytab with ktutil

Source: https://github.com/platformatic/kafka/blob/main/docker/kerberos/README.md

Use this command sequence to create a Kerberos keytab file named `kafka.keytab`. Ensure you have `ktutil` installed and configured.

```bash
ktutil
addent -password -p admin/localhost@example.com -k 1 -e aes256-cts-hmac-sha1-96
write_kt kafka.keytab
quit
```

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

### Kafka Operation Tracing Channel Example

Source: https://github.com/platformatic/kafka/blob/main/docs/diagnostic.md

Illustrates the execution flow and event publishing for tracing Kafka operations using the Node.js Diagnostic Channel API. It shows how to publish start, end, error, and async events with context.

```javascript
function operation (options, callback) {
  const channel = tracingChannel('plt:kafka:example')
  const context = { operationId: 0n, options }

  try {
    channel.start.publish(context)

    doSomethingAsync((error, result) => {
      if (error) {
        context.error = error
        channel.error.publish(context)
        channel.asyncStart.publish(context)
        callback(error)
        channel.asyncEnd.publish(context)

        return
      }

      context.result = result
      channel.asyncStart.publish(context)
      callback(error)
      channel.asyncEnd.publish(context)
    })

    channel.end.publish(context)
  } catch (error) {
    context.error = error
    channel.error.publish(context)
    channel.end.publish(context)
  }
}
```

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

### Start Local Kafka Cluster

Source: https://github.com/platformatic/kafka/blob/main/README.md

Launch the Kafka cluster using Docker Compose. Ensure the .env file is configured correctly before running this command.

```bash
docker compose -f compose.yml up -d
```

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

### Manual Writes and Delivery Reports with Producer Streams

Source: https://github.com/platformatic/kafka/blob/main/docs/producer.md

This example demonstrates manual writes to a producer stream and handling delivery reports in `BATCH` mode. It also shows how to close the stream and producer, with an alternative for force closing the producer.

```typescript
import { Producer, ProducerStreamReportModes, stringSerializers } from '@platformatic/kafka'

const producer = new Producer<string, string, string, string>({
  clientId: 'my-producer',
  bootstrapBrokers: ['localhost:9092'],
  serializers: stringSerializers
})

const stream = producer.asStream({
  batchSize: 100,
  batchTime: 50,
  reportMode: ProducerStreamReportModes.BATCH
})

stream.on('delivery-report', report => {
  // In BATCH mode: { batchId, count, result }
  console.log(report)
})

stream.on('flush', info => {
  // { batchId, count, duration, result }
  console.log('flushed', info.count)
})

stream.write({ topic: 'events', key: 'user-1', value: 'login' })
stream.write({ topic: 'events', key: 'user-2', value: 'logout' })

await stream.close()
await producer.close()

// Alternative shutdown:
// await producer.close(true)
```

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

### Kafka Producer with Compatibility Partitioner

Source: https://github.com/platformatic/kafka/blob/main/docs/producer.md

Configure a Kafka producer to use a compatibility partitioner for Java/kafkajs compatibility. This example sets up serializers and the specific partitioner.

```typescript
import { Producer, stringSerializers, compatibilityPartitioner } from '@platformatic/kafka'

const producer = new Producer({
  clientId: 'my-producer',
  bootstrapBrokers: ['localhost:9092'],
  serializers: stringSerializers,
  partitioner: compatibilityPartitioner
})
```

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

### Producer with Confluent Schema Registry

Source: https://github.com/platformatic/kafka/blob/main/docs/producer.md

Example demonstrating how to initialize a producer with Confluent Schema Registry for automatic serialization. Ensure the registry URL and authentication are correctly configured. Schema IDs for key and value must be provided in the message metadata.

```typescript
import { Producer } from '@platformatic/kafka'
import { ConfluentSchemaRegistry } from '@platformatic/kafka/registries'

// Create a schema registry instance
const registry = new ConfluentSchemaRegistry({
  url: 'http://localhost:8081',
  auth: {
    username: 'user',
    password: 'password'
  }
})

// Create a producer with the registry
const producer = new Producer({
  clientId: 'my-producer',
  bootstrapBrokers: ['localhost:9092'],
  registry // Registry handles serialization automatically
})

// Send messages with schema IDs in metadata
await producer.send({
  messages: [
    {
      topic: 'events',
      key: { id: 123 },
      value: { name: 'John', action: 'login' },
      metadata: {
        schemas: {
          key: 1, // Schema ID for key
          value: 2 // Schema ID for value
        }
      }
    }
  ]
})

await producer.close()
```

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

### KafkaJS Consumer Offset Modes

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

KafkaJS's `fromBeginning: true` option in `subscribe` controls starting from the earliest offset.

```javascript
await consumer.subscribe({ topic: 'my-topic', fromBeginning: true })
// or
await consumer.subscribe({ topic: 'my-topic' }) // latest by default
```

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

### Message Compression (KafkaJS)

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Example of sending a compressed message using KafkaJS with CompressionTypes.GZIP.

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

await producer.send({
  topic: 'my-topic',
  compression: CompressionTypes.GZIP,
  messages: [{ value: 'compressed' }]
})
```

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

### get(broker[, callback])

Source: https://github.com/platformatic/kafka/blob/main/docs/other.md

Retrieves an existing connection for a given broker or creates a new one if it does not exist.

```APIDOC
## get(broker[, callback])

### Description
Returns an existing connection for the broker or creates a new one.

### Method
get

### Parameters
#### Path Parameters
- **broker** (object) - Required - The broker details.
- **callback** (function) - Optional - A callback function to handle the result.
```

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

### Send Batch Messages (KafkaJS)

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Example of sending messages to multiple topics using KafkaJS's sendBatch method.

```javascript
await producer.sendBatch({
  topicMessages: [
    { topic: 'topic-a', messages: [{ value: 'msg1' }] },
    { topic: 'topic-b', messages: [{ value: 'msg2' }] }
  ]
})
```

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

### Send Messages (KafkaJS)

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Example of sending messages to a specific topic using KafkaJS.

```javascript
await producer.send({
  topic: 'my-topic',
  messages: [
    { key: 'key1', value: 'value1' },
    { key: 'key2', value: 'value2' }
  ]
})
```

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

### Kafka Producer with Prometheus Metrics

Source: https://github.com/platformatic/kafka/blob/main/docs/metrics.md

Example of creating a Kafka producer with Prometheus metrics enabled. Ensure you provide a `prom-client` instance and a `Registry`. This snippet demonstrates sending a message and then retrieving all registered metrics as JSON.

```javascript
import * as client from 'prom-client'
import { Producer, stringSerializer } from '@platformatic/kafka'

const registry = new client.Registry()

// Create a producer with string serialisers
const producer = new Producer({
  clientId: 'my-producer',
  bootstrapBrokers: ['localhost:9092'],
  serializers: stringSerializers,
  metrics: { client, registry }
})

// Send messages
await producer.send({
  messages: [
    {
      topic: 'events',
      key: 'user-123',
      value: JSON.stringify({ name: 'John', action: 'login' }),
      headers: { source: 'web-app' }
    }
  ]
})

// Close the producer when done
await producer.close()

console.log(JSON.stringify(await registry.getMetricsAsJSON(), null, 2))
```

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

### Send Messages (@platformatic/kafka)

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Example of sending messages to specified topics using @platformatic/kafka. The topic is defined per-message.

```javascript
import { stringSerializers } from '@platformatic/kafka'

const producer = new Producer({
  clientId: 'my-app',
  bootstrapBrokers: ['localhost:9092'],
  serializers: stringSerializers
})

const result = await producer.send({
  messages: [
    { topic: 'my-topic', key: 'key1', value: 'value1' },
    { topic: 'my-topic', key: 'key2', value: 'value2' }
  ]
})

// result.offsets is TopicWithPartitionAndOffset[]
// Each offset is a bigint, not a string
```

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

### Producer with Custom Serializer

Source: https://github.com/platformatic/kafka/blob/main/docs/other.md

Configure a `Producer` with a custom serialization function for keys and values. This example uses `JSON.stringify` and `Buffer.from`.

```typescript
import { Producer } from '@platformatic/kafka'

function serialize (source: YourType): Buffer {
  return Buffer.from(JSON.stringify(source))
}

const producer = new Producer({
  clientId: 'my-producer',
  bootstrapBrokers: ['localhost:9092'],
  serializers: {
    key: serialize,
    value: serialize,
    headerKey: serialize,
    headerValue: serialize
  }
})
```

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

### Get Connections for Messages

Source: https://github.com/platformatic/kafka/blob/main/docs/producer.md

Prepares the broker connections needed to send the provided messages. Use this method to warm up producer connections before calling send().

```typescript
const connections = await producer.getSendConnections({
  messages: [{ topic: 'events', key: Buffer.from('user-1'), value: Buffer.from('login') }]
})

console.log(connections['events:0'])
```

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

### Adding Kafka ACL Permissions

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/docker-compose-sasl-plain.md

Use the kafka-acls.sh script with the admin configuration to add permissions for a specific user and operation on a topic. This command-line example demonstrates how to secure access to Kafka resources.

```bash
/opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --add --allow-principal User:client --topic temp --operation all --command-config admin.config

```

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

### Consume with Confluent Schema Registry

Source: https://github.com/platformatic/kafka/blob/main/docs/consumer.md

Example demonstrating how to create a consumer that uses Confluent Schema Registry for automatic deserialization. Ensure the registry is configured with the correct URL and authentication if needed. Messages are automatically deserialized based on schema IDs embedded in the message.

```typescript
import { Consumer } from '@platformatic/kafka'
import { ConfluentSchemaRegistry } from '@platformatic/kafka/registries'

// Create a schema registry instance
const registry = new ConfluentSchemaRegistry({
  url: 'http://localhost:8081',
  auth: {
    username: 'user',
    password: 'password'
  }
})

// Create a consumer with the registry
const consumer = new Consumer({
  groupId: 'my-consumer-group',
  clientId: 'my-consumer',
  bootstrapBrokers: ['localhost:9092'],
  registry // Registry handles deserialization automatically
})

// Create a consumer stream
const stream = await consumer.consume({
  topics: ['events'],
  autocommit: true
})

// Messages are automatically deserialized according to their schemas
for await (const message of stream) {
  console.log('Key:', message.key) // Deserialized key object
  console.log('Value:', message.value) // Deserialized value object
}

await consumer.close()
```

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

### Custom Retry Delay Function Example

Source: https://github.com/platformatic/kafka/blob/main/docs/base.md

Implement custom retry strategies like exponential backoff or jitter by providing a function to the `retryDelay` option. This function receives details about the retry attempt and must return the delay in milliseconds.

```javascript
const producer = new Producer({
  clientId: 'my-producer',
  bootstrapBrokers: ['localhost:9092'],
  serializers: stringSerializers,
  tls: {
    rejectUnauthorized: false,
    cert: await readFile(resolve(import.meta.dirname, './ssl/client.pem')),
    key: await readFile(resolve(import.meta.dirname, './ssl/client.key'))
  }
})
```

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

### Copy Environment Sample

Source: https://github.com/platformatic/kafka/blob/main/README.md

Create a local .env file by copying the provided .env.sample. This file will store your environment-specific configurations.

```bash
cp .env.sample .env
```

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

### Admin Create Topics

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Creating topics with @platformatic/kafka simplifies configuration by sharing `partitions` and `replicas` across all topics in a single call.

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

```javascript
const created = await admin.createTopics({
  topics: ['topic-a', 'topic-b'],
  partitions: 3,
  replicas: 1
})
// Returns CreatedTopic[] with { id, name, partitions, replicas, configuration }
```

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

### beginTransaction([options, callback])

Source: https://github.com/platformatic/kafka/blob/main/docs/producer.md

Begins a new transaction and returns a `Transaction` object. Requires the producer to be configured with `idempotent: true`.

```APIDOC
## beginTransaction([options, callback])

### Description
Begins a new transaction and returns a `Transaction` object. The producer must be configured with `idempotent: true` to use transactions. Only one transaction can be active at a time per producer.

### Parameters
#### Request Body
- **options** - Optional - Accepts all options from the constructor except `serializers`.

### Response
#### Success Response
- **Transaction** - The Transaction object.

### Example
```typescript
const producer = new Producer({
  clientId: 'my-producer',
  bootstrapBrokers: ['localhost:9092'],
  idempotent: true,
  transactionalId: 'my-transaction-id'
})

const transaction = await producer.beginTransaction()
await transaction.send({ messages: [{ topic: 'my-topic', value: 'message' }] })
await transaction.commit()
```
```

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

### Build Project

Source: https://github.com/platformatic/kafka/blob/main/CLAUDE.md

Use this command to compile the project's TypeScript code into JavaScript.

```bash
# Build the project
npm run build
```

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

### Import CA and Signed Certificate into Server Keystore

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/ssl.md

Imports both the CA certificate and the newly signed server certificate into the server's keystore.

```bash
keytool -storepass 12345678 -keypass 12345678 -noprompt -keystore data/ssl/server.keystore.jks -alias ca -importcert -file data/ssl/ca.cert
```

```bash
keytool -storepass 12345678 -keypass 12345678 -noprompt -keystore data/ssl/server.keystore.jks -alias server -importcert -file data/ssl/server.pem
```

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

### @platformatic/kafka Consumer Stream Consumption

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Use `consumer.consume` to get a Node.js Readable stream of messages. Deserializers should be configured for the consumer.

```javascript
import { Consumer, stringDeserializers, MessagesStreamModes } from '@platformatic/kafka'

const consumer = new Consumer({
  clientId: 'my-app',
  bootstrapBrokers: ['localhost:9092'],
  groupId: 'my-group',
  deserializers: stringDeserializers
})

const stream = await consumer.consume({
  topics: ['my-topic'],
  mode: MessagesStreamModes.EARLIEST
})

for await (const message of stream) {
  console.log({
    key: message.key,
    value: message.value,
    offset: message.offset,  // bigint
    topic: message.topic,
    partition: message.partition,
    headers: message.headers  // Map
  })
}
```

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

### Run Development Commands

Source: https://github.com/platformatic/kafka/blob/main/CONTRIBUTING.md

Common commands for the development workflow, including running tests, building the package, and linting.

```bash
# Development workflow
pnpm test              # Run all tests
pnpm run build         # Build the package
pnpm run lint          # Lint the package
```

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

### Navigate to Docker Directory

Source: https://github.com/platformatic/kafka/blob/main/README.md

Change the current directory to the 'docker' folder to access environment and configuration files.

```bash
cd docker
```

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

### JSON Schema Registry Response

Source: https://github.com/platformatic/kafka/blob/main/docs/confluent-schema-registry.md

Example of a schema registry response object for a JSON schema, including the schema type and the schema definition.

```json
{
  "schemaType": "JSON",
  "schema": "{\"type\":\"object\",\"properties\":{\"id\":{\"type\":\"number\"},\"name\":{\"type\":\"string\"}},\"required\":[\"id\",\"name\"]}"
```

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

### AVRO Schema Registry Response

Source: https://github.com/platformatic/kafka/blob/main/docs/confluent-schema-registry.md

Example of a schema registry response object for an AVRO schema, including the schema type and the schema definition.

```json
{
  "schemaType": "AVRO",
  "schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"name\",\"type\":\"string\"}]}"
```

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

### Admin List and Describe Groups

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Listing and describing consumer groups differs in return types. @platformatic/kafka returns Maps for groups and members.

```javascript
const { groups } = await admin.listGroups()
// groups: [{ groupId: string, protocolType: string }]

const { groups: described } = await admin.describeGroups(['group-id'])
// described: [{ groupId, members, state, ... }]
```

```javascript
const groups = await admin.listGroups()
// Returns Map<string, GroupBase> with { id, state, groupType, protocolType }

const described = await admin.describeGroups({ groups: ['group-id'] })
// Returns Map<string, Group> with { id, state, protocol, members, ... }
// members is a Map<string, GroupMember>
```

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

### Generate Client Keystore

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/ssl.md

Creates a Java keystore for the client, including a self-signed certificate with a Subject Alternative Name (SAN).

```bash
keytool -storepass 12345678 -keypass 12345678 -noprompt -keystore data/ssl/client.keystore.jks -alias client -genkey -keyalg RSA -validity 365 -dname CN=client -ext SAN=DNS:client
```

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

### Invoke Basic Consumer Client

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/playground.md

Consumes messages from a topic using both 'on("data")' and 'for-await-of' methods. Requires admin topics script to be run first.

```bash
node playground/clients/consumer.ts
```

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

### Protocol Buffers Schema Registry Response

Source: https://github.com/platformatic/kafka/blob/main/docs/confluent-schema-registry.md

Example of a schema registry response object for a Protocol Buffers schema, including the schema type and the schema definition.

```json
{
  "schemaType": "PROTOBUF",
  "schema": "syntax = \"proto3\";\n\nmessage User {\n  int32 id = 1;\n  string name = 2;\n}"
```

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

### createTopics

Source: https://github.com/platformatic/kafka/blob/main/docs/admin.md

Creates one or more topics with specified configurations such as partitions, replicas, assignments, and custom configs.

```APIDOC
## createTopics(options[, callback])

### Description
Creates one or more topics.

### Method
`createTopics`

### Parameters
#### Query Parameters
- **options** (object) - Required - Options for creating topics.
  - **topics** (string[]) - Required - Topics to create.
  - **partitions** (number) - Required - Number of partitions for each topic.
  - **replicas** (number) - Required - Number of replicas for each topic.
  - **assignments** (BrokerAssignment[]) - Optional - Assignments of partitions. Each assignment is an object with `partition` and `brokers` properties.
  - **configs** (CreateTopicsRequestTopicConfig[]) - Optional - Topic configurations. Each configuration is an object with `name` and `value` properties.

#### Callback
- **callback** (function) - Optional - Callback function to handle the result.
```

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

### Get Brokers for Messages

Source: https://github.com/platformatic/kafka/blob/main/docs/producer.md

Returns the brokers that would receive the provided messages without sending them. This method uses getSendTopicPartitions() to resolve topic partitions before looking up their leaders.

```typescript
const brokers = await producer.getSendBrokers({
  messages: [{ topic: 'events', key: Buffer.from('user-1'), value: Buffer.from('login') }]
})

console.log(brokers['events:0'])
```

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

### Generate Server Keystore

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/ssl.md

Creates a Java keystore for the server, including a self-signed certificate with a Subject Alternative Name (SAN).

```bash
keytool -storepass 12345678 -keypass 12345678 -noprompt -keystore data/ssl/server.keystore.jks -alias server -genkey -keyalg RSA -validity 365 -dname CN=server -ext SAN=DNS:server
```

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

### getFirstAvailable(brokers[, callback])

Source: https://github.com/platformatic/kafka/blob/main/docs/other.md

Attempts to establish a connection using the provided list of brokers in order, returning the first successful connection.

```APIDOC
## getFirstAvailable(brokers[, callback])

### Description
Tries the provided brokers in order and returns the first successful connection.

### Method
getFirstAvailable

### Parameters
#### Path Parameters
- **brokers** (array) - Required - An array of broker details to try.
- **callback** (function) - Optional - A callback function to handle the result.
```

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

### Get Topic Partitions for Messages

Source: https://github.com/platformatic/kafka/blob/main/docs/producer.md

Synchronously returns the topic partitions that would be used for provided messages without sending them. Serializer errors are reported the same way as send().

```typescript
const topicPartitions = producer.getSendTopicPartitions({
  messages: [{ topic: 'events', key: Buffer.from('user-1'), value: Buffer.from('login') }]
})

console.log(topicPartitions.get('events'))
```

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

### Add ACL Permission for User

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/docker-compose-sasl-scram-sha.md

Grant specific permissions to a user for a topic using the Kafka ACL tool. This example allows all operations for the 'client' user on the 'temp' topic.

```bash
/opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --add --allow-principal User:client --topic temp --operation all --command-config admin.config
```

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

### Run Performance Tests with Custom Settings

Source: https://github.com/platformatic/kafka/blob/main/regression/README.md

Execute performance tests with specific configurations for messages, samples, warmups, and failure ratios. Ensure a baseline exists for comparison.

```sh
REGRESSION_REQUIRE_BASELINE=1 \
REGRESSION_PERFORMANCE_MESSAGES=5000 \
REGRESSION_PERFORMANCE_SAMPLES=5 \
REGRESSION_PERFORMANCE_WARMUPS=1 \
REGRESSION_THROUGHPUT_FAIL_RATIO=0.85 \
REGRESSION_LATENCY_FAIL_RATIO=1.2 \
REGRESSION_MEMORY_FAIL_RATIO=1.25 \
npm run test:performance
```

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

### Invoke Admin Topics Multiple Client

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/playground.md

Ensures the presence of 'temp1' and 'temp2' topics on multiple brokers, attempting specific partition assignments.

```bash
node playground/clients/admin-topics-multiple.ts
```

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

### Create Server Truststore

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/ssl.md

Imports the CA certificate into a truststore for the server. This allows the server to trust clients signed by this CA.

```bash
keytool -storepass 12345678 -keypass 12345678 -noprompt -keystore data/ssl/server.truststore.jks -alias ca -importcert -file data/ssl/ca.cert
```

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

### Get Cluster Metadata with @platformatic/kafka

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Retrieve cluster metadata using the admin client. The result includes cluster ID, broker information, controller ID, topic details, and the last update timestamp.

```javascript
const metadata = await admin.metadata({})
// Returns ClusterMetadata:
// {
//   id: string,
//   brokers: Map<number, { host, port, rack }>,
//   controllerId: number,
//   topics: Map<string, ClusterTopicMetadata>,
//   lastUpdate: number
// }
```

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

### Create and Send Messages with Kafka Producer

Source: https://github.com/platformatic/kafka/blob/main/README.md

Demonstrates how to create a Kafka producer with string serializers and send messages to a topic. Ensure the producer is closed when no longer needed.

```typescript
import { Producer, stringSerializers } from '@platformatic/kafka'

// Create a producer with string serialisers
const producer = new Producer({
  clientId: 'my-producer',
  bootstrapBrokers: ['localhost:9092'],
  serializers: stringSerializers
})

// Send messages
await producer.send({
  messages: [
    {
      topic: 'events',
      key: 'user-123',
      value: JSON.stringify({ name: 'John', action: 'login' }),
      headers: { source: 'web-app' }
    }
  ]
})

// Close the producer when done
// If you created producer streams with producer.asStream(), either close them first or use producer.close(true)
await producer.close()
```

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

### describeConfigs

Source: https://github.com/platformatic/kafka/blob/main/docs/admin.md

Describes configuration parameters for specified resources. Returns an array of resource configurations.

```APIDOC
## describeConfigs(options)

### Description
Describes configuration parameters for specified resources.

The return value is an array of resource configurations, each containing the resource type, name, and configuration entries.

### Parameters
#### Request Body
- **resources** (DescribeConfigsRequestResource[]) - Required - Array of resources specifying the resource type, name, and configuration keys to describe.
- **includeSynonyms** (boolean) - Optional - Whether to include configuration synonyms in the response. Defaults to `false`.
- **includeDocumentation** (boolean) - Optional - Whether to include configuration documentation in the response. Defaults to `false`.
```

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

### Producer Connection Lifecycle (KafkaJS)

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Demonstrates explicit connect and disconnect calls for a KafkaJS producer.

```javascript
await producer.connect()
// ... use producer ...
await producer.disconnect()
```

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

### Connection.connect(host, port[, callback])

Source: https://github.com/platformatic/kafka/blob/main/docs/other.md

Establishes a low-level socket connection to a Kafka broker and performs SASL authentication if configured.

```APIDOC
## Connection.connect(host, port[, callback])

### Description
Opens the socket connection to a broker and performs SASL authentication if configured.

### Method
`connect(host, port[, callback])`

### Parameters
- **host** (string): The hostname or IP address of the Kafka broker.
- **port** (number): The port number of the Kafka broker.
- **callback** (function, optional): A function to be called upon completion of the connection attempt.
```

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

### Manage Kafka Topics with Admin Client

Source: https://github.com/platformatic/kafka/blob/main/README.md

Illustrates how to use the Kafka Admin client to create topics (with default or custom assignments), retrieve metadata, and delete topics. The admin client should be closed after use.

```typescript
import { Admin } from '@platformatic/kafka'

// Create an admin client
const admin = new Admin({
  clientId: 'my-admin',
  bootstrapBrokers: ['localhost:9092']
})

// Create topics
await admin.createTopics({
  topics: ['my-topic'],
  partitions: 3,
  replicas: 1
})

// Create topics with custom assignments
await admin.createTopics({
  topics: ['my-custom-topic'],
  partitions: -1, // Use assignments instead
  replicas: -1, // Use assignments instead
  assignments: [
    { partition: 0, brokers: [1] },
    { partition: 1, brokers: [2] },
    { partition: 2, brokers: [3] }
  ]
})

// Get metadata
const metadata = await admin.metadata({ topics: ['my-topic'] })
console.log(metadata)

// Delete topics
await admin.deleteTopics({ topics: ['my-topic'] })

// Close the admin client when done
await admin.close()
```

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

### Invoke Consumer Rebalance Client

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/playground.md

Verifies topic and partition assignments during consumer rebalancing scenarios. Requires admin topics script to be run first.

```bash
node playground/clients/consumer-rebalance.ts
```

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

### Connection.ready([callback])

Source: https://github.com/platformatic/kafka/blob/main/docs/other.md

Waits until the connection to the Kafka broker becomes usable or fails during the waiting period.

```APIDOC
## Connection.ready([callback])

### Description
Waits until the connection becomes usable or fails while waiting.

### Method
`ready([callback])`

### Parameters
- **callback** (function, optional): A function to be called when the connection is ready or fails.
```

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

### Import CA and Signed Certificate into Client Keystore

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/ssl.md

Imports both the CA certificate and the newly signed client certificate into the client's keystore for mTLS authentication.

```bash
keytool -storepass 12345678 -keypass 12345678 -noprompt -keystore data/ssl/client.keystore.jks -alias ca -importcert -file data/ssl/ca.cert
```

```bash
keytool -storepass 12345678 -keypass 12345678 -noprompt -keystore data/ssl/client.keystore.jks -alias server -importcert -file data/ssl/client.pem
```

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

### Invoke Basic Producer Client

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/playground.md

Performs a series of produces with multiple messages each. Requires admin topics script to be run first.

```bash
node playground/clients/producer.ts
```

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

### Export Client Private Key and Convert to PKCS12

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/ssl.md

Exports the client's private key and certificate into a PKCS12 file, then extracts the private key separately.

```bash
keytool -noprompt -importkeystore -srckeystore data/ssl/client.keystore.jks -srcalias client -srcstorepass 12345678 -deststorepass 12345678 -destkeypass 12345678 -destkeystore data/ssl/client.p12 -deststoretype PKCS12
```

```bash
openssl pkcs12 -nocerts -legacy -in data/ssl/client.p12 -out data/ssl/client.key -passin pass:12345678 -nodes
```

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

### Run All Tests

Source: https://github.com/platformatic/kafka/blob/main/CLAUDE.md

Execute all tests in the project using the Node.js test runner.

```bash
# Run all tests
npm test
```

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

### describeLogDirs

Source: https://github.com/platformatic/kafka/blob/main/docs/admin.md

Describes log directories for specified topics across all brokers.

```APIDOC
## describeLogDirs(options)

### Description
Describes log directories for specified topics across all brokers.

### Parameters
#### Request Body
- **topics** (DescribeLogDirsRequestTopic[]) - Required - Array of topics specifying the topics and partitions for which to describe logs.
```

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

### Consumer Integration with Transactions

Source: https://github.com/platformatic/kafka/blob/main/docs/transactions.md

Explains how to integrate consumers with transactions for exactly-once read-process-write patterns, emphasizing manual offset management.

```APIDOC
## Consumer Integration

### Description
Explains how to integrate consumers with transactions for exactly-once read-process-write patterns, emphasizing manual offset management.

### Important Notes
When using a consumer within a transaction, you must manually manage offset commits through the transaction itself. Do not use the consumer's `autocommit` option or the message's `commit()` method. Offsets should be committed via `transaction.addOffset()` instead. The transaction will atomically commit all offsets when `transaction.commit()` is called.

### Reading Committed Messages
Consumers should use `READ_COMMITTED` isolation level to only read messages from committed transactions:

```typescript
const consumer = new Consumer({
  groupId: 'my-group',
  clientId: 'my-client',
  bootstrapBrokers: ['localhost:9092'],
  deserializers: stringDeserializers
})

const stream = await consumer.consume({
  topics: ['input-topic'],
  isolationLevel: FetchIsolationLevels.READ_COMMITTED
})
```

### Adding Consumer Offsets to Transaction
When processing messages from a consumer, you can commit the consumer offsets as part of the transaction:

```typescript
const transaction = await producer.beginTransaction()

try {
  // Add the consumer to the transaction
  await transaction.addConsumer(consumer)

  // Process messages from the consumer
  stream.on('data', async message => {
    // Process the message and produce results
    await transaction.send({
      messages: [{ topic: 'output-topic', value: processedValue }]
    })

    // Add the message offset to the transaction
    await transaction.addOffset(message)
  })

  // Commit both the produced messages and consumer offsets
  await transaction.commit()
} catch (error) {
  await transaction.abort()
}
```
```

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

### connectToBrokers

Source: https://github.com/platformatic/kafka/blob/main/docs/base.md

Establishes a connection to one or more brokers in the cluster. It accepts an optional array of node IDs to connect to. Invalid IDs are ignored.

```APIDOC
## connectToBrokers

### Description
Establishes a connection to one or more brokers in the cluster.

### Method
`connectToBrokers([nodeIds][, callback])`

### Parameters
#### Path Parameters
- **nodeIds** (number[]) - Optional - The nodes to connect to. Valid IDs can be obtained via the `metadata` method and invalid IDs are ignored.

### Response
- Returns a `Map<number, Connection>` object.
```

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

### Run Regular Tests Plus All Regression Tests

Source: https://github.com/platformatic/kafka/blob/main/regression/README.md

Combines the execution of regular tests with the full regression suite, including performance tests. This is a comprehensive command for full validation.

```sh
npm run test:all
```

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

### Basic AVRO Schema Usage with Producer and Consumer

Source: https://github.com/platformatic/kafka/blob/main/docs/confluent-schema-registry.md

Demonstrates creating a ConfluentSchemaRegistry instance and using it with Kafka Producer and Consumer for AVRO messages. Ensure schemas are registered in Confluent Schema Registry.

```typescript
import { Producer, Consumer } from '@platformatic/kafka'
import { ConfluentSchemaRegistry } from '@platformatic/kafka/registries'

// Create registry instance
const registry = new ConfluentSchemaRegistry({
  url: 'http://localhost:8081'
})

// Producer
const producer = new Producer({
  clientId: 'avro-producer',
  bootstrapBrokers: ['localhost:9092'],
  registry
})

await producer.send({
  messages: [
    {
      topic: 'users',
      key: { id: 123 },
      value: { name: 'John Doe', age: 30 },
      metadata: {
        schemas: {
          key: 1, // AVRO schema ID for key
          value: 2 // AVRO schema ID for value
        }
      }
    }
  ]
})

// Consumer
const consumer = new Consumer({
  groupId: 'avro-consumers',
  clientId: 'avro-consumer',
  bootstrapBrokers: ['localhost:9092'],
  registry
})

const stream = await consumer.consume({
  topics: ['users']
})

for await (const message of stream) {
  // Automatically deserialized from AVRO
  console.log('User:', message.value)
}
```

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

### Create Server Certificate Signing Request (CSR)

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/ssl.md

Generates a CSR from the server's keystore. This request is then signed by the CA.

```bash
keytool -storepass 12345678 -keypass 12345678 -noprompt -keystore data/ssl/server.keystore.jks -alias server -certreq -file data/ssl/server.csr
```

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

### Invoke Idempotent Producer Client

Source: https://github.com/platformatic/kafka/blob/main/docs/internals/playground.md

Performs multiple idempotent produces, demonstrating offset management. Requires admin topics script to be run first.

```bash
node playground/clients/producer-idempotent.ts
```

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

### Consume Messages with Kafka Consumer

Source: https://github.com/platformatic/kafka/blob/main/README.md

Shows how to create a Kafka consumer with string deserializers and consume messages from topics using event-based, async iterator, or concurrent processing. Remember to close the consumer when done.

```typescript
import { Consumer, stringDeserializers } from '@platformatic/kafka'
import { forEach } from 'hwp'

// Create a consumer with string deserialisers
const consumer = new Consumer({
  groupId: 'my-consumer-group',
  clientId: 'my-consumer',
  bootstrapBrokers: ['localhost:9092'],
  deserializers: stringDeserializers
})

// Create a consumer stream
const stream = await consumer.consume({
  autocommit: true,
  topics: ['my-topic'],
  sessionTimeout: 10000,
  heartbeatInterval: 500
})

// Option 1: Event-based consumption
stream.on('data', message => {
  console.log(`Received: ${message.key} -> ${message.value}`)
})

// Option 2: Async iterator consumption
for await (const message of stream) {
  console.log(`Received: ${message.key} -> ${message.value}`)
  // Process message...
}

// Option 3: Concurrent processing
await forEach(
  stream,
  async message => {
    console.log(`Received: ${message.key} -> ${message.value}`)
    // Process message...
  },
  16
) // 16 is the concurrency level

// Close the consumer when done
await consumer.close()
```

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

### Run Single Test File

Source: https://github.com/platformatic/kafka/blob/main/CLAUDE.md

Execute a specific test file using the Node.js test runner.

```bash
# Run a single test file
node --test 'test/path/to/file.test.ts'
```

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

### startLagMonitoring

Source: https://github.com/platformatic/kafka/blob/main/docs/consumer.md

Initiates periodic consumer lag monitoring at a specified interval. Lag data is emitted via `consumer:lag` events.

```APIDOC
## `startLagMonitoring(options, interval)`

### Description
Initiates periodic consumer lag monitoring at the specified `interval` in milliseconds. Consumer lag data is automatically emitted through `consumer:lag` events at each monitoring cycle. Monitoring continues until explicitly stopped with `stopLagMonitoring` or automatically terminates when the consumer closes.

### Method
`startLagMonitoring`

### Parameters
- **options** (`object`) - Accepts the same configuration as `getLag`.
- **interval** (`number`) - The monitoring interval in milliseconds.
```

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

### Begin Kafka Transaction

Source: https://github.com/platformatic/kafka/blob/main/docs/producer.md

Begins a new transaction and returns a Transaction object. The producer must be configured with idempotent: true to use transactions. Only one transaction can be active at a time per producer.

```typescript
const producer = new Producer({
  clientId: 'my-producer',
  bootstrapBrokers: ['localhost:9092'],
  idempotent: true,
  transactionalId: 'my-transaction-id'
})

const transaction = await producer.beginTransaction()
await transaction.send({ messages: [{ topic: 'my-topic', value: 'message' }] })
await transaction.commit()
```

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

### createPartitions

Source: https://github.com/platformatic/kafka/blob/main/docs/admin.md

Adds partitions to existing topics, with an option to validate the request without applying changes.

```APIDOC
## createPartitions(options[, callback])

### Description
Creates additional partitions for existing topics.

### Method
`createPartitions`

### Parameters
#### Query Parameters
- **options** (object) - Required - Options for creating partitions.
  - **topics** (CreatePartitionsRequestTopic[]) - Required - Topics to create partitions for with information about partition count and assignments for each topic.
  - **validateOnly** (boolean) - Optional - Whether to only validate the request without applying changes. Defaults to `false`.

#### Callback
- **callback** (function) - Optional - Callback function to handle the result.
```

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

### Run Single Test File with CI Coverage (Node Built-in)

Source: https://github.com/platformatic/kafka/blob/main/CLAUDE.md

Execute a specific test file and generate code coverage using Node.js's experimental built-in coverage feature. This is recommended for Node.js 26 and later.

```bash
# CI uses Node's built-in coverage (works on Node 26, where c8/yargs breaks):
# node --test --experimental-test-coverage --test-coverage-include='src/**' 'test/path/to/file.test.ts'
```

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

### Run Single Test File with Local Coverage (c8)

Source: https://github.com/platformatic/kafka/blob/main/CLAUDE.md

Run a specific test file and generate local code coverage using c8. This is suitable for Node.js versions 22 and 24.

```bash
# Run a single test file with coverage (local, c8 — Node 22/24)
c8 -c test/config/c8-local.json node --test 'test/path/to/file.test.ts'
```

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

### Producer Connection Lifecycle (@platformatic/kafka)

Source: https://github.com/platformatic/kafka/blob/main/migration/README.md

Illustrates lazy auto-connection and explicit close() for @platformatic/kafka producers. No explicit connect() is needed.

```javascript
// No connect() needed - connections are established lazily on first operation
// ... use producer ...
await producer.close()
```

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

### getSendConnections<Key, Value, HeaderKey, HeaderValue>(options[, callback])

Source: https://github.com/platformatic/kafka/blob/main/docs/producer.md

Prepares and returns the broker connections needed to send the provided messages. This is useful for warming up producer connections before sending.

```APIDOC
## getSendConnections<Key, Value, HeaderKey, HeaderValue>(options[, callback])

### Description
Prepares the broker connections needed to send the provided messages. The return value is a record keyed by `topic:partition`, where each value is the open connection to the broker responsible for that topic partition. Use this method to warm up the producer connections for a set of messages before calling `send()`.

### Parameters
#### Request Body
- **messages** (MessageToProduce<Key, Value, HeaderKey, HeaderValue>[]) - Required - The messages to send.

Options: accepts the same options as `send()`.

### Request Example
```typescript
const connections = await producer.getSendConnections({
  messages: [{ topic: 'events', key: Buffer.from('user-1'), value: Buffer.from('login') }]
})

console.log(connections['events:0'])
```
```