Kafka Skill

Producing messages

async function produce(topic: string, messages: Array<{ key?: string; value: string }>) {
  const producer = kafka.producer();
  await producer.connect();
  await producer.send({
    topic,
    messages
  });
  await producer.disconnect();
}

// usage
await produce('orders', [{ key: 'order-1', value: JSON.stringify(order) }]);

• You can reuse a producer instance instead of connecting/disconnecting on every send for higher throughput.
• send returns a record metadata object describing partition offsets.
• To guarantee delivery, wrap send in try/catch and optionally enable idempotence in broker config.

Consuming messages

async function consume(topic: string, groupId: string) {
  const consumer = kafka.consumer({ groupId });
  await consumer.connect();
  await consumer.subscribe({ topic, fromBeginning: true });
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log({
        key: message.key?.toString(),
        value: message.value?.toString(),
        headers: message.headers
      });
    }
  });
}

// usage
await consume('orders', 'order-service');

• Consumers in the same group share partitions for horizontal scaling.
• fromBeginning controls whether to start at earliest or latest offset.
• eachBatch is available if you need batch processing.

TypeScript usage

kafkajs exports types such as Kafka, Producer, Consumer, Message, ISerializer etc. You can annotate your wrappers accordingly:

import { Producer, Message } from 'kafkajs';

async function sendMessage(producer: Producer, topic: string, msg: Message) {
  await producer.send({ topic, messages: [msg] });
}

Common pitfalls

• Not handling rebalances - consumers receive consumer.on('consumer.group_join', ...) events when partitions move; ensure state is checkpointed.
• High memory usage - avoid calling await producer.send in a tight loop without backpressure, or use the high-level sendBatch.
• Missing error handlers - attach producer.on('producer.crash', ...) and consumer.on('consumer.crash', ...) to log problems.
• Order assumptions across partitions - Kafka only guarantees order within a single partition keyed by message key.
• Schema management - real applications typically use Avro/Protobuf with a schema registry; handle serialization accordingly.

Advanced features

• Transactions: const tx = await producer.transaction(); then use tx.send, tx.commit(), tx.abort() for atomic multi-topic writes.
• Admin client: const admin = kafka.admin(); await admin.connect(); allows topic creation, metadata queries, and ACL management.
• Compression: set compression option per message to CompressionTypes.GZIP, CompressionTypes.Snappy, or CompressionTypes.LZ4.
• Batch consuming: use consumer.run({ eachBatch }) for lower overhead when processing many messages at once.

SSL & SASL authentication

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['kafka:9093'],
  ssl: true,
  sasl: {
    mechanism: 'plain',       // 'scram-sha-256' | 'scram-sha-512' | 'aws'
    username: process.env.KAFKA_USER!,
    password: process.env.KAFKA_PASSWORD!,
  },
});

For mTLS pass ssl: { ca, cert, key } instead of SASL.

Environment variables

KAFKA_BROKERS=kafka1:9092,kafka2:9092
KAFKA_CLIENT_ID=my-service
KAFKA_GROUP_ID=my-service-group
KAFKA_USER=myuser
KAFKA_PASSWORD=secret

const kafka = new Kafka({
  clientId: process.env.KAFKA_CLIENT_ID || 'app',
  brokers: (process.env.KAFKA_BROKERS || 'localhost:9092').split(','),
});

Graceful shutdown

Always disconnect producers and consumers on process signals to flush in-flight messages and release partition assignments:

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

errorTypes.forEach(type => process.on(type, async () => {
  await consumer.disconnect();
  process.exit(0);
}));

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

Example

(async () => {
  const kafka = new Kafka({ clientId: 'app', brokers: ['localhost:9092'] });
  const producer = kafka.producer();
  await producer.connect();
  await producer.send({ topic: 'test', messages: [{ value: 'hello kafka' }] });
  await producer.disconnect();

  const consumer = kafka.consumer({ groupId: 'test-group' });
  await consumer.connect();
  await consumer.subscribe({ topic: 'test', fromBeginning: true });
  await consumer.run({ eachMessage: async ({ message }) => {
    console.log(message.value?.toString());
  } });
})();