Create a Low Level Consumer

Warning

Consider using the low level consumer only if you need more flexible control over consuming messages.

Configure the Consumer

$conf = new \RdKafka\Conf();
$conf->set('bootstrap.servers', 'kafka:9092');
$conf->set('group.id', 'test');
$conf->set('log_level', (string) LOG_DEBUG);
$conf->set('debug', 'all');
$conf->setLogCb(
    function (\RdKafka\Consumer $consumer, int $level, string $facility, string $message): void {
        // Perform your logging mechanism here
    }
);

$conf->set('statistics.interval.ms', (string) 1000);
$conf->setStatsCb(
    function (\RdKafka\Consumer $consumer, string $json, int $jsonLength, $opaque): void {
        // Perform your stats mechanism here ...
    }
);

See configuration for more details.

Configure the topic

$topicConf = new \RdKafka\TopicConf();
$topicConf->set('enable.auto.commit', 'true');
$topicConf->set('auto.commit.interval.ms', (string) 100);
$topicConf->set('auto.offset.reset', 'earliest');

Create a consumer instance

$consumer = new \RdKafka\Consumer($conf);

Consume from topic

$topic = $consumer->newTopic('playground', $topicConf); // Our example topic name

Create a new Queue

$queue = $consumer->newQueue();

Consume Messages

E.g. consume messages from 3 partitions via the new queue

$topic->consumeQueueStart(0, RD_KAFKA_OFFSET_BEGINNING, $queue);
$topic->consumeQueueStart(1, RD_KAFKA_OFFSET_BEGINNING, $queue);
$topic->consumeQueueStart(2, RD_KAFKA_OFFSET_BEGINNING, $queue);
do {
    $message = $queue->consume(1000);

    if ($message === null) {
        continue;
    } elseif ($message->err === RD_KAFKA_RESP_ERR_NO_ERROR) {
        // process your message here
    } elseif ($message->err === RD_KAFKA_RESP_ERR__PARTITION_EOF) {
        // handle end of partition
    } elseif ($message->err === RD_KAFKA_RESP_ERR__TIMED_OUT) {
        // handle timeout
    } else {
        // handle other errors
        $topic->consumeStop(0);
        $topic->consumeStop(1);
        $topic->consumeStop(2);
        throw new \Exception($message->errstr(), $message->err);
    }

    // trigger callback queues
    $consumer->poll(1);
} while (true);

Finish the Consumption

$topic->consumeStop(0);
$topic->consumeStop(1);
$topic->consumeStop(2);

Run Example

If you want to test consuming messages with the high level consumer, you can take a look at the low level consumer example. Please take a look at how to prepare running the examples before.

Produce to the same topic that we created using the following command:

docker-compose run --rm php74 php examples/producer.php

Consume the messages produced to the same topic using the following command:

docker-compose run --rm php74 php examples/consumer-lowlevel.php

And you should see all messages consumed from topic playground.