README

Overview

Wrapper around the PHP rdkafka extension (Apache Kafka). Provides a registry of producers/consumers, a configuration builder, a safe shutdown-flush, and integration with Tereta\Logger for diagnostics.

Requirements

PHP 8.4+
ext-rdkafka extension
tereta/core, tereta/logger

Configuration

Registration is done in .config.php:

use Tereta\Kafka\Builders\Config as KafkaConfigBuilder;
use Tereta\Kafka\Services\Registry as KafkaRegistry;

KafkaRegistry::singleton()->register(
    KafkaConfigBuilder::factory()->create()
        ->set('metadata.broker.list', 'kafka:9092')
        ->set('group.id', 'workers')
);

Tereta\Kafka\Builders\Config::set() accepts any librdkafka option. When building the config for a producer/consumer, a separate whitelist of relevant options is applied.

Usage

Producer

use Tereta\Kafka\Services\Registry as KafkaRegistry;

$producer = KafkaRegistry::singleton()->getProducer();
$producer->newTopic('events')->produce(RD_KAFKA_PARTITION_UA, 0, 'payload');
$producer->flush();

flush() is synchronous and throws RuntimeException on failure. On process termination the destructor calls flush() itself so that pending messages aren't lost.

Consumer

use Tereta\Kafka\Services\Registry as KafkaRegistry;

$consumer = KafkaRegistry::singleton()->getConsumer();
$consumer->subscribeTopic('events');
$consumer->waitAssignment();

while (true) {
    $payload = $consumer->getMessage(maxWaitSeconds: 15, pollTimeoutMs: 1000);
    if ($payload === null) {
        continue;
    }
    // handle the message
    $consumer->commit();
}

Transient librdkafka errors are logged into the error channel and do not break the loop.

Author and License

Author: Tereta Alexander
Website: tereta.dev
License: Apache License 2.0. See LICENSE.

 www.████████╗███████╗██████╗ ███████╗████████╗ █████╗
     ╚══██╔══╝██╔════╝██╔══██╗██╔════╝╚══██╔══╝██╔══██╗
        ██║   █████╗  ██████╔╝█████╗     ██║   ███████║
        ██║   ██╔══╝  ██╔══██╗██╔══╝     ██║   ██╔══██║
        ██║   ███████╗██║  ██║███████╗   ██║   ██║  ██║
        ╚═╝   ╚══════╝╚═╝  ╚═╝╚══════╝   ╚═╝   ╚═╝  ╚═╝
                                                      .dev

tereta / kafka

Maintainers

Package info

Statistics

Security