KafkaActor

Defined in: src/io/broker/KafkaActor.ts:166

Kafka producer + consumer in one actor, backed by kafkajs. When consumer.groupId is set, a consumer is started after connectImpl and consumed records are delivered to target. When a producer is the only goal, leave consumer and topics empty.

Offset-commit semantics.

commitMode: 'auto' (default) — kafkajs commits after each handler returns successfully → at-least-once. Cheap; OK for idempotent handlers.
commitMode: 'manual' — pump pauses on each message and waits for an explicit commit command from the handler (#2). The handler is responsible for sending exactly one commit (or nack) per delivered record. If neither arrives within commitTimeoutMs the pump rejects internally, kafkajs treats the partition as failed, and re-delivery happens on rebalance. Produces exactly-once-with-processing: a message that successfully passed through commit is committed; a crash or nack re-delivers.

const kafka = system.actorOf(Props.create(() => new KafkaActor({ brokers: [‘kafka:9092’], consumer: { groupId: ‘orders’, commitMode: ‘manual’ }, topics: [‘orders’], target: orderProcessor, })));

class OrderProcessor extends Actor { constructor(private readonly kafka: ActorRef) { super(); } async onReceive(rec: KafkaRecord) { try { await db.insertOrder(JSON.parse(rec.value!.toString())); this.kafka.tell({ kind: ‘commit’, topic: rec.topic, partition: rec.partition, offset: rec.offset }); } catch (e) { this.kafka.tell({ kind: ‘nack’, topic: rec.topic, partition: rec.partition, offset: rec.offset }); } } }