@platformatic/kafka Documentation

The library provides experimental support for Confluent Schema Registry with AVRO, Protobuf, and JSON Schema. By passing a ConfluentSchemaRegistry instance to the Producer or Consumer constructor, serialization and deserialization are handled automatically using schema IDs provided in message metadata.

Warning: This API is experimental, does not follow semver, and may change in minor or patch releases.

import { Producer, 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'
  }
})

// Producer with schema registry
const producer = new Producer({
  clientId: 'schema-producer',
  bootstrapBrokers: ['localhost:9092'],
  registry // Automatic serialization with schemas
})

// Send messages with schema IDs
await producer.send({
  messages: [
    {
      topic: 'users',
      value: { id: 1, name: 'Alice' },
      metadata: {
        schemas: {
          value: 100 // Schema ID in the registry
        }
      }
    }
  ]
})

// Consumer with schema registry
const consumer = new Consumer({
  groupId: 'schema-consumers',
  clientId: 'schema-consumer',
  bootstrapBrokers: ['localhost:9092'],
  registry // Automatic deserialization with schemas
})

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

// Messages are automatically deserialized
for await (const message of stream) {
  console.log('User:', message.value) // Typed object
}

You can provide custom serializers and deserializers for different parts of a Kafka message: key, value, headerKey, and headerValue. If no serializers are provided, the client defaults to no-operation (identity) functions, meaning all parts must be Buffers.

To customize, pass an object containing the desired functions to the serializers option (for Producer) or the deserializers option (for Consumer).

import {
  Consumer,
  jsonDeserializer,
  jsonSerializer,
  ProduceAcks,
  Producer,
  stringDeserializer,
  stringSerializer
} from '@platformatic/kafka'

type Strings = string[]

const producer = new Producer({
  clientId: 'my-producer',
  bootstrapBrokers: ['localhost:9092'],
  serializers: {
    key: stringSerializer,
    value: jsonSerializer<Strings>
  }
})

const consumer = new Consumer({
  groupId: 'my-consumer-group',
  clientId: 'my-consumer',
  bootstrapBrokers: ['localhost:9092'],
  deserializers: {
    key: stringDeserializer,
    value: jsonDeserializer<Strings>
  },
  maxWaitTime: 1000,
  autocommit: 100
})

// Produce some messages
let i = 0
const timer = setTimeout(() => {
  producer.send({
    messages: [{ topic: 'temp', key: `key-${i++}`, value: ['first', 'second'] }],
    acks: ProduceAcks.LEADER
  })

  if (i < 3) {
    timer.refresh()
  }
}, 1000)

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

for await (const message of stream) {
  console.log(message)

  if (message.key === 'key-2') {
    break
  }
}

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

Platformatic Kafka provides instrumentation via the Node.js Diagnostic Channel API. You can subscribe to specific tracing channels to monitor operations, performance, and errors across different Kafka components (Admin, Producer, Consumer, etc.).

Execution Model

Tracing follows a specific lifecycle. For asynchronous operations, the sequence of events is:

1. channel.start.publish(context)
2. channel.asyncStart.publish(context)
3. channel.asyncEnd.publish(context)

If an error occurs, the error event is published before the asyncStart and asyncEnd events.

const { diagnosticChannel } = require('node:diagnostics_channel');

// Example of subscribing to a channel
diagnosticChannel('plt:kafka:producer:sends').subscribe(({ context }) => {
  console.log('Producer send operation:', context.operationId, context.result);
});

To achieve exactly-once semantics when consuming and producing, follow these rules:

1. Consumer Configuration: Set autocommit to false (or omit it) and use isolationLevel: FetchIsolationLevels.READ_COMMITTED when consuming to ensure you only read messages from committed transactions.
2. Transaction Integration: Use transaction.addConsumer(consumer) to register the consumer group with the transaction.
3. Offset Management: Do not use message.commit(). Instead, use transaction.addOffset(message) to include the consumer offset in the atomic transaction. The offsets are committed only when transaction.commit() is called.

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()
}