Zum Inhalt springen
actor-ts
Deutsch

Kafka

KafkaActor integriert mit Apache Kafka. Umschließt intern kafkajs; du importierst es nicht direkt.

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 wird einmalig hier konfiguriert.
  .withConsumer({ groupId: 'my-app-orders' })
  .withTopics(['orders', 'payments'])
  .withTarget(orderHandler);
const kafka = system.spawn(
  () => new KafkaActor(
    kafkaOptions,
  ),
  'kafka',
);


// Ein weiteres Topic zur Laufzeit hinzufügen (das `topics` aus dem
// Konstruktor deckt den Normalfall ab; `subscribe` ist für späte
// Ergänzungen).
kafka.tell({ kind: 'subscribe', topic: 'audit' });


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

Settings

Abschnitt betitelt „Settings“
interface KafkaOptionsType extends BrokerCommonOptionsType {
  brokers:    string[] | string;           // Bootstrap-Server
  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 für eingehende Records
  topics?:  ReadonlyArray<string>;
}

Der Parent BrokerCommonOptionsType gibt dir Reconnect, Buffer, Breaker — siehe BrokerActor-Basis.

Producen

Abschnitt betitelt „Producen“
kafka.tell({
  kind: 'publish',
  publish: {
    topic:   'orders',
    key:     orderId,        // optional — partitioniert nach Key
    value:   JSON.stringify(order),
    headers: { 'x-trace': traceId },
  },
});

Bei getrennter Verbindung gepuffert; beim Reconnect in Reihenfolge geflusht.

Für strikte Reihenfolge innerhalb einer Partition setze producer.idempotent: true und gib einen konsistenten key für zusammengehörige Records — kafkajs dedupliziert Retries innerhalb der Producer-Session.

Consumen

Abschnitt betitelt „Consumen“
// Das Consumer-Actor-Target wird über die Settings verdrahtet
// (siehe oben); `subscribe` ergänzt zur Laufzeit nur weitere
// Topics, wenn nötig.
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);
  }
}

Der Actor liefert jeden konsumierten Record an target. Für Kafkas Consumer-Group-Work-Sharing nimm mehrere Actor-Instanzen mit derselben groupId — jede Instanz übernimmt eine Teilmenge der Partitionen.

topics und das subscribe-Kommando zur Laufzeit speisen dieselbe gewünschte Menge, die von der BrokerActor-Basisklasse gehalten wird und nicht vom Consumer: ein zur Laufzeit hinzugefügtes Topic wird nach einem Reconnect neu subscribed, und ein subscribe, das gesendet wird während der Actor getrennt ist, wird beim nächsten Connect angewandt statt verworfen. Konfigurierte Topics erben consumer.fromBeginning; eine Ergänzung zur Laufzeit startet beim aktuellen Offset.

Ein unsubscribe gibt es nicht — kafkajs kann einem laufenden Consumer kein einzelnes Topic entziehen. Stoppe den Actor (oder starte einen zweiten mit einer anderen groupId), wenn ein Consumer wirklich eine andere Topic-Menge braucht.

Commit-Modi

Abschnitt betitelt „Commit-Modi“
consumer: { commitMode: 'auto' }    // at-least-once
consumer: { commitMode: 'manual' }  // exactly-once-with-processing

auto (Default): kafkajs committed, nachdem der Handler zurückkehrt. Crash zwischen Handler und Commit → erneute Auslieferung beim nächsten Start. At-least-once. Handler müssen idempotent sein.

manual: der Consumer pausiert bei jedem Record und wartet auf eine explizite Commit-Nachricht:

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

Liefert Exactly-once-with-processing-Semantik, wenn deine Verarbeitung selbst transaktional ist. commitTimeoutMs (Default 30 s) deckelt, wie lange die Pumpe wartet, bevor sie aufgibt und einen Rebalance auslöst.

Auth + TLS

Abschnitt betitelt „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,
);

Sowohl TLS (ssl: true) als auch SASL-Credentials werden an kafkajs durchgereicht. Nimm Env-Variablen für Secrets; hardcode sie nicht in application.conf.

Peer-Dependency

Abschnitt betitelt „Peer-Dependency“
Terminal-Fenster
npm install kafkajs
# oder: bun add kafkajs

kafkajs ist eine Peer-Dependency — das Framework bundelt es nicht. Du installierst es, wenn du KafkaActor nutzt.

Wann Kafka

Abschnitt betitelt „Wann Kafka“

Drei primäre Anwendungsfälle:

  1. Durables Event-Streaming — mehrere Consumer lesen denselben Stream mit verschiedenen Offsets; wiederabspielbare Historie.
  2. Producer und Consumer entkoppeln — High-Volume-Downstreams, die nicht das Tempo des Producers vorgeben sollten.
  3. Brücken zu bestehender Kafka-Infrastruktur — actor-ts an ein bereits Kafka nutzendes System anbinden.

Für clusterinternes Pub/Sub bevorzuge DistributedPubSub — keine Broker- Abhängigkeit.

Wohin als Nächstes

Abschnitt betitelt „Wohin als Nächstes“