Kafka #

Apache Kafka is a distributed streaming platform designed for high throughput, low latency, and data durability — used by LinkedIn, Uber, Netflix, and thousands of other companies to move trillions of messages per day. Unlike traditional message brokers like RabbitMQ, Kafka stores all messages in a persistent, ordered log — consumers can re-read messages from any position. The kafka_dart package provides a complete Dart client for interacting with Kafka clusters. This article covers the core concepts, producers, consumers, error handling, and the most useful patterns in event-driven architecture.

Kafka Core Concepts #

flowchart LR
    P1["Producer\n(Dart App)"] -->|publish| T["Topic: order-events\nPartition 0: [msg0, msg1, msg2]\nPartition 1: [msg3, msg4]\nPartition 2: [msg5, msg6]"]
    T -->|subscribe| CG1["Consumer Group A\nConsumer 1 → P0\nConsumer 2 → P1, P2"]
    T -->|subscribe| CG2["Consumer Group B\nConsumer 3 → P0, P1, P2"]

Concept	Description
Topic	Message channel category/name — like a database table
Partition	Horizontal split of a topic — Kafka’s unit of parallelism
Offset	Message position within a partition — monotonically increasing from 0
Producer	Sender of messages to a topic
Consumer	Receiver of messages from a topic
Consumer Group	Several consumers sharing the load of reading a topic
Broker	A Kafka server that stores and serves messages
Retention	How long messages are kept (default 7 days) — can be re-read

Package Setup #

dart pub add kafka_dart

# pubspec.yaml
dependencies:
  kafka_dart: ^0.0.5

For production, make sure the Kafka cluster is running and reachable from the Dart machine. For local development, use Docker: docker run -p 9092:9092 apache/kafka. Or use Confluent Platform / Redpanda as easier-to-set-up alternatives.

Producer — Sending Messages #

import 'package:kafka_dart/kafka_dart.dart';
import 'dart:convert';

Future<void> main() async {
  // Create a producer
  final producer = KafkaProducer(
    brokers: ['localhost:9092'],
    config: ProducerConfig(
      // Delivery reliability
      acks: Acks.all,           // wait for confirmation from all replicas
      retries: 3,               // retry 3 times on failure
      retryBackoffMs: 100,      // delay between retries

      // Performance
      batchSize: 16384,         // accumulate up to 16KB before sending
      lingerMs: 5,              // wait up to 5ms for a larger batch
      compressionType: CompressionType.snappy, // compression for efficiency

      // Idempotency — ensure messages aren't duplicated
      enableIdempotence: true,
    ),
  );

  // Send a simple message
  await producer.send(
    ProducerRecord(
      topic: 'order-events',
      value: utf8.encode(jsonEncode({
        'tipe': 'order_dibuat',
        'idOrder': 'ORD-001',
        'idPengguna': 'USR-123',
        'total': 150_000,
        'timestamp': DateTime.now().toUtc().toIso8601String(),
      })),
    ),
  );

  print('Message sent');
  await producer.close();
}

Producer with Keys for Partitioning #

The key determines which partition a message goes to — messages with the same key always land in the same partition, guaranteeing per-entity ordering:

import 'package:kafka_dart/kafka_dart.dart';
import 'dart:convert';

class OrderEventProducer {
  late final KafkaProducer _producer;

  Future<void> inisialisasi(List<String> brokers) async {
    _producer = KafkaProducer(
      brokers: brokers,
      config: ProducerConfig(
        acks: Acks.all,
        enableIdempotence: true,
        retries: 5,
      ),
    );
  }

  // Key = idOrder → all events for the same order go to the same partition
  // This guarantees event ordering per order
  Future<void> kirimEvent({
    required String idOrder,
    required String tipeEvent,
    required Map<String, dynamic> data,
  }) async {
    final pesan = {
      'tipe': tipeEvent,
      'idOrder': idOrder,
      'data': data,
      'timestamp': DateTime.now().toUtc().toIso8601String(),
      'versi': '1.0',
    };

    final record = ProducerRecord(
      topic: 'order-events',
      key: utf8.encode(idOrder),    // key = idOrder for consistent partitioning
      value: utf8.encode(jsonEncode(pesan)),
      headers: {
        'tipe-event': utf8.encode(tipeEvent),
        'versi': utf8.encode('1.0'),
      },
    );

    final metadata = await _producer.send(record);
    print('Event "$tipeEvent" for order $idOrder → '
        'partition ${metadata.partition}, offset ${metadata.offset}');
  }

  Future<void> tutup() => _producer.close();
}

// Usage
Future<void> main() async {
  final producer = OrderEventProducer();
  await producer.inisialisasi(['localhost:9092']);

  // All ORDER-001 events go to the same partition → ordering preserved
  await producer.kirimEvent(
    idOrder: 'ORDER-001',
    tipeEvent: 'order_dibuat',
    data: {'total': 150_000, 'itemCount': 3},
  );

  await producer.kirimEvent(
    idOrder: 'ORDER-001',
    tipeEvent: 'pembayaran_dikonfirmasi',
    data: {'metodePembayaran': 'transfer', 'jumlah': 150_000},
  );

  await producer.kirimEvent(
    idOrder: 'ORDER-001',
    tipeEvent: 'order_dikirim',
    data: {'kurir': 'JNE', 'noResi': 'JNE123456789'},
  );

  await producer.tutup();
}

Consumer — Receiving Messages #

import 'package:kafka_dart/kafka_dart.dart';
import 'dart:convert';

Future<void> main() async {
  final consumer = KafkaConsumer(
    brokers: ['localhost:9092'],
    groupId: 'order-processor-group',   // consumer group ID
    config: ConsumerConfig(
      autoOffsetReset: AutoOffsetReset.earliest,  // start from the beginning if no offset exists
      enableAutoCommit: false,    // IMPORTANT: manual commits for full control
      sessionTimeoutMs: 30000,    // timeout if the consumer doesn't send heartbeats
      maxPollRecords: 100,        // maximum messages per poll
    ),
  );

  // Subscribe to a topic
  consumer.subscribe(['order-events']);

  print('Consumer running, waiting for messages...');

  // Message consumption loop
  try {
    while (true) {
      // Poll with a 1-second timeout
      final records = await consumer.poll(Duration(seconds: 1));

      if (records.isEmpty) continue;

      for (final record in records) {
        try {
          await prosesRecord(record);
          // Commit the offset AFTER successful processing — at-least-once delivery
          await consumer.commitSync();
        } catch (e) {
          print('Failed to process message at offset ${record.offset}: $e');
          // Don't commit — the message will be reprocessed after a restart
        }
      }
    }
  } finally {
    await consumer.close();
  }
}

Future<void> prosesRecord(ConsumerRecord record) async {
  final key = record.key != null ? utf8.decode(record.key!) : null;
  final value = jsonDecode(utf8.decode(record.value)) as Map<String, dynamic>;

  print('Partition: ${record.partition}, Offset: ${record.offset}');
  print('Key: $key');
  print('Event type: ${value['tipe']}');
  print('Data: ${value['data']}');

  // Process based on the event type
  switch (value['tipe'] as String) {
    case 'order_dibuat':
      await prosesOrderDibuat(value);
    case 'pembayaran_dikonfirmasi':
      await prosesPembayaran(value);
    case 'order_dikirim':
      await prosesOrderDikirim(value);
    default:
      print('Unknown event type: ${value['tipe']}');
  }
}

Consumer Groups and Parallelism #

Consumer groups let multiple consumer instances share the load of reading a topic — each partition is read by only one consumer in the group:

import 'package:kafka_dart/kafka_dart.dart';

// Run several workers in parallel with the same group
Future<void> jalankanWorker(
  String workerId,
  List<String> brokers,
  String groupId,
  List<String> topics,
) async {
  final consumer = KafkaConsumer(
    brokers: brokers,
    groupId: groupId,   // same group ID → Kafka splits partitions between workers
    config: ConsumerConfig(
      enableAutoCommit: false,
      maxPollRecords: 50,
    ),
  );

  consumer.subscribe(topics);
  print('Worker $workerId started');

  // Handler for rebalancing — when consumers join/leave the group
  consumer.onPartitionsAssigned = (partitions) {
    print('Worker $workerId got partitions: $partitions');
  };

  consumer.onPartitionsRevoked = (partitions) {
    print('Worker $workerId lost partitions: $partitions');
    // Commit offsets before the partition is taken over by another worker
    consumer.commitSync();
  };

  try {
    while (true) {
      final records = await consumer.poll(Duration(seconds: 1));
      for (final record in records) {
        await prosesRecord(record);
      }
      if (records.isNotEmpty) {
        await consumer.commitSync();
      }
    }
  } finally {
    await consumer.close();
    print('Worker $workerId finished');
  }
}

Commit Strategies — At-Least-Once vs Exactly-Once #

The commit strategy choice largely determines the message delivery guarantees:

// AT-LEAST-ONCE (most common) — commit AFTER processing
// Messages can be processed more than once if a crash happens after processing but before the commit
for (final record in records) {
  await prosesRecord(record);     // process first
  await consumer.commitSync();    // then commit — at-least-once
}

// AT-MOST-ONCE — commit BEFORE processing (not recommended for critical data)
// Messages can be lost if a crash happens after the commit but before processing
for (final record in records) {
  await consumer.commitSync();    // commit first
  await prosesRecord(record);     // then process — at-most-once
}

// EXACTLY-ONCE — needs transactions or an idempotent consumer
// Store the offset together with the processing result in an atomic transaction
// Example: save the offset to the database together with the processing result
Future<void> prosesExactlyOnce(
  ConsumerRecord record,
  Pool pgPool,
) async {
  await pgPool.runTx((tx) async {
    // Check whether this offset has already been processed
    final sudahDiproses = await tx.execute(
      r'SELECT 1 FROM processed_offsets WHERE topic = $1 AND partition = $2 AND offset = $3',
      parameters: [record.topic, record.partition, record.offset],
    );

    if (sudahDiproses.isNotEmpty) {
      print('Offset ${record.offset} already processed, skipping');
      return;
    }

    // Process within the same transaction
    await prosesDalamTransaksi(tx, record);

    // Store the offset as proof of processing
    await tx.execute(
      r'INSERT INTO processed_offsets (topic, partition, offset, diproses_pada) VALUES ($1, $2, $3, NOW())',
      parameters: [record.topic, record.partition, record.offset],
    );
  });
}

Error Handling and Dead Letter Queues #

import 'package:kafka_dart/kafka_dart.dart';

class ResilientConsumer {
  final KafkaConsumer _consumer;
  final KafkaProducer _dlqProducer;  // Dead Letter Queue producer
  final int _maxRetry;

  ResilientConsumer({
    required KafkaConsumer consumer,
    required KafkaProducer dlqProducer,
    int maxRetry = 3,
  })  : _consumer = consumer,
        _dlqProducer = dlqProducer,
        _maxRetry = maxRetry;

  Future<void> jalankan(List<String> topics) async {
    _consumer.subscribe(topics);

    while (true) {
      final records = await _consumer.poll(Duration(seconds: 1));

      for (final record in records) {
        bool berhasil = false;

        // Try up to maxRetry times
        for (int percobaan = 1; percobaan <= _maxRetry; percobaan++) {
          try {
            await prosesRecord(record);
            berhasil = true;
            break;
          } catch (e) {
            print('Attempt $percobaan/$_maxRetry failed: $e');
            if (percobaan < _maxRetry) {
              // Exponential backoff
              await Future.delayed(Duration(milliseconds: 100 * (1 << percobaan)));
            }
          }
        }

        if (!berhasil) {
          // Send to the Dead Letter Queue for manual investigation
          await _kirimKeDLQ(record);
        }
      }

      if (records.isNotEmpty) {
        await _consumer.commitSync();
      }
    }
  }

  Future<void> _kirimKeDLQ(ConsumerRecord record) async {
    print('Sending failed message to DLQ: offset ${record.offset}');

    await _dlqProducer.send(ProducerRecord(
      topic: '${record.topic}.dlq',  // DLQ topic naming convention
      key: record.key,
      value: record.value,
      headers: {
        // Additional metadata for debugging
        'original-topic': utf8.encode(record.topic),
        'original-partition': utf8.encode(record.partition.toString()),
        'original-offset': utf8.encode(record.offset.toString()),
        'failed-at': utf8.encode(DateTime.now().toIso8601String()),
      },
    ));
  }
}

Admin — Topic Management #

import 'package:kafka_dart/kafka_dart.dart';

Future<void> kelolaTopik(List<String> brokers) async {
  final admin = KafkaAdmin(brokers: brokers);

  // Create topics
  await admin.createTopics([
    TopicConfig(
      name: 'order-events',
      numPartitions: 6,        // 6 partitions for parallelism
      replicationFactor: 3,   // 3 replicas for reliability
      configs: {
        'retention.ms': '604800000',  // keep for 7 days
        'cleanup.policy': 'delete',    // delete old messages
        'compression.type': 'snappy',
      },
    ),
    TopicConfig(
      name: 'order-events.dlq',
      numPartitions: 1,
      replicationFactor: 3,
      configs: {
        'retention.ms': '2592000000',  // keep for 30 days for the DLQ
      },
    ),
  ]);

  // List topics
  final topics = await admin.listTopics();
  print('Topics: $topics');

  // Describe topics — partition and replica info
  final detail = await admin.describeTopics(['order-events']);
  for (final topic in detail) {
    print('Topic: ${topic.name}');
    for (final partition in topic.partitions) {
      print('  Partition ${partition.id}: leader=${partition.leader}, '
          'replicas=${partition.replicas}');
    }
  }

  // Delete topics
  await admin.deleteTopics(['topic-lama']);

  await admin.close();
}

Event-Driven Patterns with Kafka #

Event Sourcing — Store Every Event #

// All state changes are stored as a sequence of events
// Current state = the result of replaying all events

class OrderEventStore {
  final KafkaProducer _producer;

  OrderEventStore(this._producer);

  Future<void> simpanEvent(OrderEvent event) async {
    await _producer.send(ProducerRecord(
      topic: 'order-events',
      key: utf8.encode(event.idOrder),
      value: utf8.encode(jsonEncode(event.toJson())),
    ));
  }
}

// A consumer that rebuilds state from the event stream
class OrderProjection {
  final Map<String, OrderState> _state = {};

  Future<void> prosesEvent(ConsumerRecord record) async {
    final event = OrderEvent.dariJson(
      jsonDecode(utf8.decode(record.value)) as Map<String, dynamic>,
    );

    final stateSebelumnya = _state[event.idOrder] ?? OrderState.awal();
    _state[event.idOrder] = stateSebelumnya.terapkan(event);
  }

  OrderState? stateOrder(String idOrder) => _state[idOrder];
}

CQRS — Command Query Responsibility Segregation #

// Command: write to the database, send an event
Future<void> buatOrder(BuatOrderCommand cmd, Pool db, OrderEventStore eventStore) async {
  // Save to the database (write model)
  await db.execute(
    r'INSERT INTO orders (id, id_pengguna, total, status) VALUES ($1, $2, $3, $4)',
    parameters: [cmd.idOrder, cmd.idPengguna, cmd.total, 'pending'],
  );

  // Send the event (for the read model and notifications)
  await eventStore.simpanEvent(OrderDibuatEvent(
    idOrder: cmd.idOrder,
    idPengguna: cmd.idPengguna,
    total: cmd.total,
  ));
}

// Query: read from the read model built from events
// Another consumer can update Elasticsearch, Redis, or denormalized tables

Kafka vs RabbitMQ #

Aspect	Kafka	RabbitMQ
Model	Distributed log	Traditional message broker
Retention	Messages stored (default 7 days)	Messages deleted after being consumed
Throughput	Very high (millions/sec)	High (hundreds of thousands/sec)
Ordering	Per partition	Per queue
Consumer	Pull-based	Push-based
Replay	✓ Can re-read from an offset	✗ Can’t
Use case	Event streaming, logs, analytics	Task queues, RPC, complex routing
Setup	More complex	Simpler

Summary #

Keys for ordering — messages with the same key always land in the same partition, guaranteeing event ordering per entity (e.g. all events for one order go to the same partition).
Consumer groups split partitions between consumers — one partition is read by only one consumer in the group. Adding consumers = more parallelism, but limited by the partition count.
Manual commits (enableAutoCommit: false) are safer than auto-commit — commit offsets only after a message is successfully processed for at-least-once delivery.
Dead Letter Queues (DLQ) for messages that fail after several retries — store them in a .dlq topic with debug metadata for manual investigation.
onPartitionsRevoked — commit offsets before partitions are taken over during rebalancing, so messages aren’t reprocessed from scratch by a new consumer.
Retention policies — Kafka keeps messages even after consumption (default 7 days). New consumers can read from the beginning; crashed consumers can resume from their last offset.
Exactly-once — store the offset together with the processing result in one database transaction. On restart, check whether the offset was already processed before reprocessing.
Partition count = the parallelism limit — one consumer per partition per group. Plan partition counts based on future throughput needs.
Producer idempotency (enableIdempotence: true) ensures messages aren’t duplicated even with retries — combined with acks: all for maximum reliability.
The Admin API for programmatic topic management — useful for infrastructure-as-code and testing.

← Previous: Elasticsearch Next: RabbitMQ →