Kafka

Tomaj\Hermes\Driver\KafkaDriver sends messages to an Apache Kafka topic and consumes them in a consumer group. It uses ext-rdkafka 6, the PHP binding of librdkafka. It is tested against Kafka 4.3 in KRaft mode.

public function __construct(
    RdKafka\Producer $producer,
    ?RdKafka\KafkaConsumer $consumer = null,
    string $topic = 'hermes',
    int $consumeTimeout = 1000,
    int $flushTimeout = 10000
)

• $producer sends messages. The worker needs one too, because the dispatcher sends retries back to the topic.
• $consumer receives messages in wait(). The emitter can leave it out. Without it, wait() throws KafkaDriverException.
• $consumeTimeout is how many milliseconds wait() blocks for the next message before it checks shutdown again.
• $flushTimeout is how many milliseconds send() waits for Kafka to confirm the message.

An empty topic, or a timeout below 1, throws InvalidArgumentException.

Install

pecl install rdkafka          # needs librdkafka, e.g. apt install librdkafka-dev

ext-rdkafka is a suggest of tomaj/hermes, so add it to your own composer.json ("ext-rdkafka": "^6.0"). The pure-PHP Kafka clients are not maintained any more, which is why the driver uses the extension. See the FAQ.

Emit and handle

This is examples/kafka. It confirms orders, and make up starts the same Kafka image as the Hermes tests.

Runnable project

Every file below is imported from examples/kafka, which has a Makefile and an e2e.sh that runs the whole flow. Clone the repository, cd examples/kafka, and run make install up.

Key files

• src/Hermes.php
• src/OrderHandler.php
• bin/emit.php
• bin/worker.php
• bin/shutdown.php
• docker-compose.yml
• e2e.sh

View source on GitHub

One class builds the drivers for both sides:

<?php

declare(strict_types=1);

namespace App;

use RdKafka\Conf;
use RdKafka\KafkaConsumer;
use RdKafka\Producer;
use Tomaj\Hermes\Driver\KafkaDriver;

final class Hermes
{
    public const string TOPIC = 'orders';

    public static function brokers(): string
    {
        return getenv('KAFKA_BROKERS') ?: '127.0.0.1:9092';
    }

    /**
     * The web process only sends, so it needs no consumer.
     */
    public static function emitterDriver(): KafkaDriver
    {
        return new KafkaDriver(self::producer(), null, self::TOPIC);
    }

    /**
     * Every worker with the same group.id shares the partitions of the topic.
     */
    public static function workerDriver(): KafkaDriver
    {
        $conf = new Conf();
        $conf->set('bootstrap.servers', self::brokers());
        $conf->set('group.id', getenv('KAFKA_GROUP') ?: 'order-workers');
        // Hermes commits each message after the handler; never let librdkafka commit on its own
        $conf->set('enable.auto.commit', 'false');
        // a new group starts at the oldest message instead of only new ones
        $conf->set('auto.offset.reset', 'earliest');

        // the producer is for retries, which the dispatcher sends back to the topic
        return new KafkaDriver(self::producer(), new KafkaConsumer($conf), self::TOPIC);
    }

    private static function producer(): Producer
    {
        $conf = new Conf();
        $conf->set('bootstrap.servers', self::brokers());
        return new Producer($conf);
    }
}

The consumer must have enable.auto.commit=false. Otherwise librdkafka commits messages before the handler finished, and a crash loses them. It also needs a group.id.

<?php

declare(strict_types=1);

use App\Hermes;
use Tomaj\Hermes\Emitter;
use Tomaj\Hermes\Message;

require __DIR__ . '/../vendor/autoload.php';

// php bin/emit.php <order number>
$order = $argv[1] ?? '1001';

// emit() returns after Kafka confirmed the message
new Emitter(Hermes::emitterDriver())->emit(new Message('order.placed', ['order' => $order]));

echo "placed order {$order}\n";

<?php

declare(strict_types=1);

use App\Hermes;
use App\OrderHandler;
use Tomaj\Hermes\Dispatcher;
use Tomaj\Hermes\Shutdown\SharedFileShutdown;

require __DIR__ . '/../vendor/autoload.php';

$var = getenv('APP_VAR') ?: __DIR__ . '/../var';
@mkdir($var, 0o775, true);

$driver = Hermes::workerDriver();
$driver->setShutdown(new SharedFileShutdown("{$var}/shutdown"));

$dispatcher = new Dispatcher($driver);
$dispatcher->registerHandler('order.placed', new OrderHandler("{$var}/orders.log"));
$dispatcher->handle();

The worker uses SharedFileShutdown, so make e2e can stop it cleanly before it starts a second worker of the same group:

<?php

declare(strict_types=1);

use Tomaj\Hermes\Shutdown\SharedFileShutdown;

require __DIR__ . '/../vendor/autoload.php';

$var = getenv('APP_VAR') ?: __DIR__ . '/../var';
@mkdir($var, 0o775, true);
new SharedFileShutdown("{$var}/shutdown")->shutdown();
echo "workers asked to stop\n";

Send

send() produces the JSON to the topic, keyed by the message id, then calls flush($flushTimeout). It returns only after Kafka confirmed the message. If Kafka does not confirm in time, it throws KafkaDriverException with the librdkafka error code. A send costs one round trip to the broker, which is fine for a web request and a little slow for bulk imports.

flush() reports a broker that does not answer in time. A message the broker rejects for good, for example on a topic the client may not write to, also leaves the queue, and only librdkafka's delivery report shows the error. To fail loudly on those, set a delivery report callback on the producer's Conf with setDrMsgCb() and throw or log when $message->err is not 0.

The key decides the partition. Message ids are random, so messages spread over all partitions. Kafka keeps order only within a partition, so with more than one partition Hermes messages are not FIFO.

Consume and commit

wait() subscribes the consumer to the topic once, then loops. Each loop checks shutdown and max items, then calls consume($consumeTimeout):

• A message. The driver unserializes it, runs the callback, and then commits the offset synchronously. Delivery is at least once: a worker that dies before the commit leaves the message for the next worker of the group.
• The callback throws. The offset is not committed and the exception stops the worker. The next worker of the group starts at that message again. Handler errors and retries are handled by the dispatcher before they reach the driver.
• Invalid JSON or an empty value. The driver commits it and throws SerializeException. A message that fails on every delivery would otherwise block its partition for good.
• Timeout or end of partition. The driver checks shutdown and consumes again.
• Any other consumer error. The driver throws KafkaDriverException with the librdkafka message and code.

Workers that use the same group.id share the partitions. A topic with three partitions keeps up to three workers busy. More workers than partitions wait idle.

Limits

• No priorities. setupPriorityQueue() throws NotSupportedException, like RabbitMQ and SQS. send() ignores $priority and wait() ignores $priorities, so Dispatcher::handle([200]) consumes every message on the topic. Use one topic and driver per priority if you need lanes. See Priority.
• No delay. A message with a future executeAt is delivered at once, and so are retries. A failing handler retries until maxRetry(). See Delayed execution.
• Commit per message. Each handled message costs a synchronous commit. For very high throughput, batch outside Hermes.
• Topic settings. The driver does not create topics. Kafka creates one on first use when auto.create.topics.enable is on. In production, create the topic yourself with the partitions and retention you need.

FAQ

Why ext-rdkafka and not a pure-PHP client?

The pure-PHP clients, longlang/phpkafka and nmred/kafka-php, have had no release since 2023 and 2017. ext-rdkafka is maintained and wraps librdkafka, the client that most Kafka tools build on. It is a suggest, so you install it only if you use this driver.

Can several workers consume the same topic?

Yes. Give them the same group.id and Kafka assigns each partition to one of them. See Consume and commit.

What happens when a worker crashes?

The offset of the message it was handling is not committed, so the next worker of the group gets that message again. Make handlers idempotent. See Architecture.

Are delayed messages and priorities supported?

No. Kafka delivers in order, at once. Compare the drivers on Drivers.

Which Kafka versions are supported?

The integration tests run against Kafka 4.3.1 (apache/kafka, KRaft) with ext-rdkafka 6.0.5 and librdkafka 2. librdkafka also talks to older brokers.

Is there a full example?

Yes. examples/kafka has a docker-compose.yml for Kafka, and its make e2e checks that orders are handled once, that SharedFileShutdown stops the worker, and that a new worker of the group resumes after the last commit.