Quick Start

Producer

Create a producer, send keyed messages for consistent partition routing, then flush and destroy.

uses librdkafka;

var prod := NewProducer('broker1:9092,broker2:9092');

{ Send messages with a key for partitioning }
ProduceKeyed(prod, 'orders', 'user-42', '{"id":1001,"total":99.90}', -1);
ProduceKeyed(prod, 'orders', 'user-17', '{"id":1002,"total":14.50}', -1);

{ Wait for all deliveries }
Flush(prod, 5000);
DestroyProducer(prod);

Consumer Loop

Create a consumer, subscribe to one or more topics, then poll in a loop and commit offsets after processing.

uses librdkafka;

var cons := NewConsumer('broker1:9092', 'my-group');
Subscribe(cons, ['orders', 'events']);

while True do
begin
  var msg := Poll(cons, 1000);
  if not msg.IsError then
  begin
    WriteLn('Topic: ' + msg.Topic + '  Key: ' + msg.Key);
    WriteLn('Value: ' + msg.Value);
    CommitMessage(cons, msg);
  end;
end;

Producer API

Consumer API

Admin / Metadata

uses librdkafka;

var prod := NewProducer('broker1:9092');

{ Inspect cluster topology }
var topics := ListTopics(prod, 5000);
for t in topics do
  WriteLn(t.Name + ' — ' + IntToStr(t.PartitionCount) + ' partitions');

{ Check how much data is waiting on a partition }
var wm := GetWatermarks(prod, 'orders', 0);
WriteLn('orders[0] lag window: ' + IntToStr(wm.High - wm.Low));

DestroyProducer(prod);

Key Concepts

Delivery Reports

Poll delivery reports after producing to confirm each message reached the broker and capture its final offset and partition assignment.

uses librdkafka;

var prod := NewProducer('broker1:9092,broker2:9092');

{ Enqueue a batch of messages }
ProduceKeyed(prod, 'payments', 'txn-001', '{"amount":250.00}', -1);
ProduceKeyed(prod, 'payments', 'txn-002', '{"amount":75.50}', -1);
ProduceKeyed(prod, 'payments', 'txn-003', '{"amount":1200.00}', -1);

{ Drive the event loop until all messages are delivered }
while OutQueueLen(prod) > 0 do
begin
  PollProducer(prod, 100);
  var reports := GetDeliveryReports(prod);
  for r in reports do
  begin
    if r.Error = '' then
      WriteLn('Delivered key=' + r.Key +
              ' partition=' + IntToStr(r.Partition) +
              ' offset=' + IntToStr(r.Offset))
    else
      WriteLn('FAILED key=' + r.Key + ' error: ' + r.Error);
  end;
end;

DestroyProducer(prod);

Batch Consumer with Manual Commit

Use PollBatch and commit only after the entire batch is processed for higher throughput with controlled at-least-once guarantees.

uses librdkafka;

var cons := NewConsumer('broker1:9092', 'analytics-group');
Subscribe(cons, ['payments']);

while True do
begin
  var batch := PollBatch(cons, 100, 500);
  if batch.Count > 0 then
  begin
    WriteLn('Processing ' + IntToStr(batch.Count) + ' messages');
    for msg in batch do
    begin
      if not msg.IsError then
        ProcessPayment(msg.Value);
    end;
    { Commit the entire batch in one round-trip }
    CommitSync(cons);
  end;
end;

Partition Seek & Replay

Rewind a partition to reprocess historical messages or skip ahead past a known bad range.

uses librdkafka;

var cons := NewConsumer('broker1:9092', 'replay-group');

{ Manually assign partition 0 of 'events' }
Assign(cons, [TopicPartition('events', 0)]);

{ Seek to a specific offset to replay from there }
SeekPartition(cons, 'events', 0, 8000);

var msg := Poll(cons, 2000);
WriteLn('Replaying from offset: ' + IntToStr(msg.Offset));

{ Check available range }
var wm := GetWatermarks(cons, 'events', 0);
WriteLn('Available range: [' + IntToStr(wm.Low) + ', ' + IntToStr(wm.High) + ']');

DestroyConsumer(cons);