Apache Kafka is a distributed streaming platform capable of handling trillions of events a day. It enables developers to publish and subscribe to streams of records, similar to a message queue or enterprise messaging system. Among various programming languages that can interface with Kafka, PHP can be utilized through the php-rdkafka extension, which provides a low-level Kafka consumer and producer client.

Understanding Message Consumption in Kafka

In Kafka, messages are stored in topics which can be split into partitions to allow for parallel consumption. Consumers read messages from these partitions. One crucial aspect of Kafka consumption is message acknowledgment, often referred to as committing offsets. An offset is a unique identifier for a record within a partition. By committing an offset, the consumer is essentially informing Kafka that it has successfully processed all prior messages up to that offset. Failure to properly commit offsets can lead to message duplication or loss.

Installing php-rdkafka

Before you can consume messages with Kafka using PHP, you need to set up the php-rdkafka extension. You can typically install it via PECL:

pecl install rdkafka

You'll then need to add the extension to your PHP configuration:

echo "extension=rdkafka.so" >> /path/to/php.ini

Setting Up the Consumer

Consuming messages from Kafka involves setting up a Consumer object, configuring it, subscribing to topics, and then continually polling the broker for new messages. Here’s a basic setup:

1$conf = new RdKafka\Conf();
2$conf->set('group.id', 'myConsumerGroup');
3$conf->set('metadata.broker.list', 'localhost:9092');
4$conf->set('auto.offset.reset', 'smallest');
5
6$rk = new RdKafka\Consumer($conf);
7$topicConf = new RdKafka\TopicConf();
8$topicConf->set('auto.commit.interval.ms', 100);
9
10$topic = $rk->newTopic("myTopic", $topicConf);
11$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

Consuming Messages

To consume messages, you can use a loop that will keep running to receive new messages as follows:

1while (true) {
2    $message = $topic->consume(0, 120*1000);
3    switch ($message->err) {
4        case RD_KAFKA_RESP_ERR_NO_ERROR:
5            // Process the message
6            processMessage($message);
7            break;
8        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
9            echo "No more messages; will wait for more\n";
10            break;
11        case RD_KAFKA_RESP_ERR__TIMED_OUT:
12            echo "Timed out\n";
13            break;
14        default:
15            throw new Exception($message->errstr(), $message->err);
16    }
17}

Acknowledging Messages

To ensure that messages are acknowledged (offsets are committed), you can either enable auto-commit or manually control when offsets are committed. The choice often depends on the processing requirements and the desired level of control over the message acknowledgment.

Auto-commit

This is enabled by default. The auto.commit.interval.ms setting in TopicConf determines how frequently offsets are committed. This is a hands-off approach but may lead to duplicated processing if your consumer fails between commits.

Manual Commit

If you need more control or want to ensure that messages are only committed after certain conditions are met (e.g., after successfully processing a message), you can manually commit offsets:

1$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
2
3while (true) {
4    $message = $topic->consume(0, 120*1000);
5    if ($message->err === RD_KAFKA_RESP_ERR_NO_ERROR) {
6        processMessage($message);
7
8        // Manually committing offset
9        $topic->offsetStore($message->partition, $message->offset);
10    }
11}

Summary Table

Here is a table summarizing the key points related to message consumption and acknowledgment in Kafka using php-rdkafka:

Conclusion

Using php-rdkafka to consume and acknowledge messages from Kafka requires understanding of Kafka's offset behavior and a careful consideration of the commit strategy. Depending on the application's needs, developers can choose between higher throughput with potential message duplication (auto-commit) or lower throughput with more strict processing guarantees (manual commit).