Kafka

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

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

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 | TlsTransportOptionsType;   // true, or cert material
  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 and SASL credentials pass through to kafkajs. Use env vars for secrets; don’t hard-code in application.conf.

withSsl(true) verifies the broker against the system trust store. Behind a private CA — or when the broker asks for a client certificate — pass the material instead of the boolean:

const kafkaOptions = KafkaOptions.create()
  .withBrokers(['kafka-1:9093'])
  .withSsl({
    ca: fs.readFileSync('./tls/ca.crt', 'utf8'),
    cert: fs.readFileSync('./tls/client.crt', 'utf8'),
    key: fs.readFileSync('./tls/client.key', 'utf8'),
  });
new KafkaActor(kafkaOptions);

Every field carries the material itself, never a path to it — which is why the example reads the files. cert and key only mean anything together, and one without the other is rejected when the actor starts.

The ssl config key stays boolean-only on purpose: a private key does not belong in application.conf, so the material form is reachable from code only.

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.