@tamasha/kafka-connection-v2

Minimal KafkaJS helper that focuses on the essentials: connect quickly, enforce per-action schemas, and subscribe with clean teardown semantics. Inspired by the original @tamasha/kafka-connection, this version keeps only the primitives needed for simple services.

Highlights

• Lightweight wrapper around kafkajs with lazy producer and consumer creation.
• Built-in topic schema registry keyed by action (one topic per microservice, many actions per topic).
• Generated TypeScript payload unions for every topic/action combination.
• Explicit lifecycle methods (connect, disconnect) to simplify graceful shutdown.
• No external dependencies beyond kafkajs.

Installation

npm install @tamasha/kafka-connection-v2

Quick start

import {
  KafkaConnectionV2,
  USER_EVENTS_TOPIC,
  UserEventMessage,
} from "@tamasha/kafka-connection-v2";

const connection = new KafkaConnectionV2({
  clientId: "example-service",
  brokers: ["localhost:9092"],
});

await connection.connect();

const loginEvent: UserEventMessage = {
  payload: {
    action: "login",
    userId: "u-1",
    timestamp: Date.now(),
    service: "auth-service",
    sessionId: "session-xyz",
  },
};

await connection.produce({
  topic: USER_EVENTS_TOPIC,
  messages: [
    {
      value: JSON.stringify(loginEvent),
      // key will automatically be set to loginEvent.payload.userId if omitted
    },
  ],
});

await connection.subscribe(
  { groupId: "example-service-consumer" },
  USER_EVENTS_TOPIC,
  async ({ message }) => {
    const payload = JSON.parse(message.value?.toString("utf-8") ?? "{}") as UserEventMessage;
    console.log("Received:", {
      key: message.key?.toString(),
      payload,
    });
  }
);

// Later, during shutdown
await connection.disconnect();

API

new KafkaConnectionV2(config)

Creates a new connection manager. config is forwarded to the underlying Kafka constructor, with brokers required. Built-in schemas are loaded automatically; use getSchemaRegistry() if you need introspection.

connect()

Creates the internal Kafka clients (producer and consumer) lazily and verifies the brokers can be reached. Calling produce or subscribe will auto-connect if needed, but connect is useful for startup health checks.

produce(record)

Sends a single ProducerRecord. Ensures the producer is connected before sending and validates each message payload against the configured topic schema.

subscribe(consumerConfig, topics, handler, options?)

Initialises a consumer (one per connection) and starts consuming the provided topics. The handler receives the standard EachMessagePayload. Set options.useBatch = true to receive EachBatchPayload instead. Incoming messages are validated prior to invoking the handler.

disconnect()

Stops and disconnects the managed consumer and producer.

Built-in schemas & types

• DEFAULT_TOPIC_SCHEMAS exposes the runtime registry used for validation.
• UserEventMessage, OrderEventMessage, and KafkaEventMessage describe the allowed JSON payloads per topic/action.
• connection.getSchemaRegistry() returns the TopicSchemaRegistry instance, useful for debugging or listing available topics/actions.
• Partition keys derive automatically from each topic's schema (payload.userId, payload.orderId, etc.).

License

MIT © Tamasha Team