Kafka

KafkaActor integrates with Apache Kafka. Wraps kafkajs internally; you don’t import it directly.

import { ActorSystem, KafkaActor, KafkaOptions } from 'actor-ts';

const system = ActorSystem.create('my-app');

const kafkaOptions = KafkaOptions.create()
  .withBrokers(['kafka-1:9092', 'kafka-2:9092'])
  .withClientId('my-app')
  // Consumer-actor target is configured here, once.
  .withConsumer({ groupId: 'my-app-orders' })
  .withTopics(['orders', 'payments'])
  .withTarget(orderHandler);
const kafka = system.spawn(
  () => new KafkaActor(
    kafkaOptions,
  ),
  'kafka',
);

// Add another topic at runtime (the constructor-time `topics` covers
// the common case; `subscribe` is for late-binding additions).
kafka.tell({ kind: 'subscribe', topic: 'audit' });

// Publish:
kafka.tell({ kind: 'publish', publish: { topic: 'audit', value: 'something' } });

Settings

interface KafkaOptionsType extends BrokerCommonOptionsType {
  brokers:    string[] | string;           // bootstrap servers
  clientId?:  string;
  sasl?:      {
    mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512';
    username:  string;
    password:  string;
  };
  ssl?:        boolean;
  producer?: {
    idempotent?:               boolean;
    allowAutoTopicCreation?:   boolean;
  };
  consumer?: {
    groupId?:        string;
    fromBeginning?:  boolean;
    commitMode?:     'auto' | 'manual';
    commitTimeoutMs?: number;
  };
  target?:  ActorRef<KafkaRecord>;     // consumer-actor for inbound records
  topics?:  ReadonlyArray<string>;
}

The BrokerCommonOptionsType parent gives you reconnect, buffer, breaker — see BrokerActor base.

Producing

kafka.tell({
  kind: 'publish',
  publish: {
    topic:   'orders',
    key:     orderId,        // optional — partitions by key
    value:   JSON.stringify(order),
    headers: { 'x-trace': traceId },
  },
});

Buffered when disconnected; flushed in order on reconnect.

For strict ordering within a partition, set producer.idempotent: true and provide consistent key for related records — kafkajs deduplicates retries within the producer session.

Consuming

// The consumer-actor target is wired up via settings (see above);
// `subscribe` only adds further topics at runtime if needed.
kafka.tell({ kind: 'subscribe', topic: 'orders' });

class OrderHandler extends Actor<KafkaRecord> {
  override onReceive(record: KafkaRecord): void {
    // record.topic / partition / offset / key / value / timestamp / headers
    const order = JSON.parse(new TextDecoder().decode(record.value!));
    this.process(order);
  }
}

The actor delivers every consumed record to target. For Kafka’s consumer-group work-sharing, use multiple actor instances with the same groupId — each instance owns a subset of partitions.

topics and the runtime subscribe command feed the same desired set, held by the BrokerActor base rather than by the consumer: a topic added at runtime is re-subscribed after a reconnect, and a subscribe sent while the actor is disconnected is applied on the next connect instead of being dropped. Configured topics inherit consumer.fromBeginning; a runtime addition starts at the current offset.

There is no unsubscribe — kafkajs cannot drop a single topic from a running consumer. Stop the actor (or spawn a second one with a different groupId) if a consumer needs a genuinely different topic set.

Commit modes

consumer: { commitMode: 'auto' }    // at-least-once
consumer: { commitMode: 'manual' }  // exactly-once-with-processing

auto (default): kafkajs commits after the handler returns. Crash between handler and commit → re-delivery on next start. At-least-once. Handlers must be idempotent.

manual: the consumer pauses on each record and waits for an explicit commit message:

override async onReceive(record: KafkaRecord): Promise<void> {
  await processIdempotently(record);
  this.kafka.tell({
    kind:      'commit',
    topic:     record.topic,
    partition: record.partition,
    offset:    record.offset,
  });
}

Gives exactly-once-with-processing semantics if your processing is itself transactional. commitTimeoutMs (default 30 s) caps how long the pump waits before giving up and triggering a rebalance.

Auth + TLS

const kafkaOptions = KafkaOptions.create()
  .withBrokers(['kafka-1:9093'])
  .withSsl(true)
  .withSasl({
    mechanism: 'scram-sha-512',
    username:  process.env.KAFKA_USER!,
    password:  process.env.KAFKA_PASS!,
  });
new KafkaActor(
  kafkaOptions,
);

Both TLS (ssl: true) and SASL credentials pass through to kafkajs. Use env vars for secrets; don’t hard-code in application.conf.

Peer dependency

npm install kafkajs
# or: bun add kafkajs

kafkajs is a peer dependency — the framework doesn’t bundle it. You install when using KafkaActor.

When to use Kafka

Three primary use cases:

1. Durable event streaming — multiple consumers reading the same stream with different offsets; replayable history.
2. Decoupling producers and consumers — high-volume downstream that shouldn’t pace the producer.
3. Bridging existing Kafka infrastructure — connecting actor-ts to a system already using Kafka.

For in-cluster pub/sub, prefer DistributedPubSub — no broker dependency.

Consumer group rebalancing

// Adding a new consumer-group member triggers rebalance

Kafka rebalances cause brief unavailability (typically seconds). Existing handlers complete in-flight work; new partitions reassign. Bound your handler latency to within Kafka’s session timeout (30 s default) to avoid being kicked out of the group mid-handler.

Slow handlers fall behind

Kafkajs caps the in-flight rate — if your handler takes 500 ms per record, the consumer processes ~2 records/sec. For high-throughput consumption, parallelize inside the handler (Promise.all batches) or scale by adding consumer-group members.

Where to next

• I/O overview — the bigger picture.
• BrokerActor base — the shared lifecycle.
• DistributedPubSub — in-cluster alternative when you don’t need Kafka.
• NATS / MQTT — alternative brokers.