Kafka Driver

Introduction

Note: This driver is only available with Kipchak Enterprise.

The Kafka Driver provides a fully typed interface for producing and consuming messages via Apache Kafka. It wraps the native php-rdkafka extension — the industry-standard PHP binding for librdkafka — offering production-grade throughput and reliability.

Key features include:

• Multiple Connections: Configure separate clients for different Kafka clusters or environments.
• Typed API: Dedicated message objects and enums — your IDE guides every call.
• Producer: Publish single messages or batches, with configurable compression, acknowledgements, and delivery timeouts.
• Non-blocking produce: Queue a message on the request path in microseconds, with listeners for messages that could not be delivered and a flush when the worker exits.
• Consumer: Subscribe to topics with explicit consumer group management, configurable offset reset, and manual commit support.
• Security: SASL PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, and SSL/TLS — suitable for managed Kafka services (Confluent Cloud, MSK, Aiven, Redpanda, etc.).
• FrankenPHP Worker-Safe: The underlying producer is lazily created and reused across requests within the same worker process.

Requirements

• PHP 8.4+
• The ext-rdkafka PHP extension (wraps librdkafka)
• librdkafka >= 2.0 installed on the host system

The hardened Kipchak Enterprise image includes ext-rdkafka. The free Kipchak images do not, because the extension lengthens the image build considerably and increases the image size. To use the driver on a free image, add the extension in your API's Dockerfile:

FROM 1x.ax/mamluk/php/8.5:franken
RUN install-php-extensions rdkafka

Installation

To install this driver, you need access to the Enterprise Composer repository at https://php.pkgs.1x.ax/.

If you have an enterprise licence, please contact your account representative for access.

composer require kipchak/driver-kafka

Initialise the Driver

Add the following line to your drivers/drivers.php file:

\Kipchak\Driver\Kafka\Kafka::initialise($container);

Configuration

Create a config file named kipchak.kafka.php in your project's config/ directory.

Connection Settings

Define connections in the connections array. Each key (e.g. default, events) is a connection name.

Security Settings

Configure security when connecting to a secured cluster:

Producer Settings

Consumer Settings

Consumer settings define defaults applied when calling consumer(). The group_id is always supplied at call time.

Example Configuration

<?php

use function Kipchak\Core\env;

return [
    'connections' => [
        'default' => [
            'brokers'  => env('KAFKA_BROKERS', 'kafka:9092'),
            'security' => [
                'protocol' => env('KAFKA_SECURITY_PROTOCOL', 'SASL_SSL'),
                'sasl'     => [
                    'mechanism' => env('KAFKA_SASL_MECHANISM', 'SCRAM-SHA-256'),
                    'username'  => env('KAFKA_USERNAME', ''),
                    'password'  => env('KAFKA_PASSWORD', ''),
                ],
                'ssl' => [
                    'ca_location'      => env('KAFKA_CA_LOCATION', '/etc/ssl/certs/ca-certificates.crt'),
                    'verify_hostname'  => true,
                ],
            ],
            'producer' => [
                'acks'                => 'all',
                'compression'         => 'lz4',
                'linger_ms'           => 5,
                'delivery_timeout_ms' => 30000,
            ],
            'consumer' => [
                'auto_offset_reset'    => 'earliest',
                'enable_auto_commit'   => false,
                'session_timeout_ms'   => 30000,
                'max_poll_interval_ms' => 300000,
            ],
        ],
    ],
];

Environment Variables

KAFKA_BROKERS=broker1:9092,broker2:9092
KAFKA_SECURITY_PROTOCOL=SASL_SSL
KAFKA_SASL_MECHANISM=SCRAM-SHA-256
KAFKA_USERNAME=your-kafka-username
KAFKA_PASSWORD=your-kafka-password
KAFKA_CA_LOCATION=/etc/ssl/certs/ca-certificates.crt

Usage

Retrieve a connection by name. If no name is given, default is used.

use Kipchak\Driver\Kafka\Kafka;

$kafka = Kafka::get();           // 'default' connection
$kafka = Kafka::get('events');   // named connection

Publishing a Single Message

// Publish with automatic partition assignment
$kafka->publish(
    topic:   'orders',
    payload: json_encode(['id' => 42, 'status' => 'placed']),
    key:     '42',
);

// Publish with custom headers
$kafka->publish(
    topic:   'orders',
    payload: json_encode(['id' => 43, 'status' => 'shipped']),
    key:     '43',
    headers: ['trace-id' => 'abc-123', 'source' => 'api'],
);

publish() waits until the broker has acknowledged the message, up to delivery_timeout_ms.

Producing Without Waiting

produce() queues the message and returns at once. librdkafka delivers it from its own background thread, so the call costs microseconds rather than a round trip to the broker, and a slow or unreachable cluster does not slow the request. Use it for messages sent on the request path, such as an event per request.

$kafka->produce(
    topic:   'audit',
    payload: json_encode($event),
    key:     $tenantId,
);

Because the call does not wait, a failed delivery is reported later rather than thrown. Register a listener for messages that could not be delivered, and record them somewhere they can be replayed from:

$kafka->onDeliveryFailure(function (\RdKafka\Message $message) use ($logger): void {
    $logger->error('Kafka delivery failed', [
        'topic'   => $message->topic_name,
        'error'   => $message->errstr(),
        'payload' => $message->payload,
    ]);
});

• Listeners run when delivery reports are served, inside produce(), poll() and flush(). They must not throw. A listener receives failures for every topic on the connection.
• A message is reported as failed once delivery_timeout_ms passes without delivery. A shorter outage loses nothing: queued messages are sent when the brokers return.
• If the brokers are unreachable long enough for the local queue to fill (queue_max_messages), produce() throws RdKafka\Exception with the code RD_KAFKA_RESP_ERR__QUEUE_FULL.
• Messages still queued when the process ends are flushed by a shutdown function, which waits at most shutdown_flush_ms. In FrankenPHP worker mode this happens when the worker stops gracefully. A process that is killed outright (SIGKILL, out of memory) loses what was still queued, so keep linger_ms low.
• outstanding() returns the number of messages not yet acknowledged, and flush() waits for all of them.

Publishing a Batch

Batching minimises round-trips to the broker. All messages are enqueued before a single flush.

use Kipchak\Driver\Kafka\Message\ProducerMessage;

$kafka->publishBatch([
    new ProducerMessage(topic: 'orders', payload: json_encode(['id' => 1]), key: '1'),
    new ProducerMessage(topic: 'orders', payload: json_encode(['id' => 2]), key: '2'),
    new ProducerMessage(topic: 'invoices', payload: json_encode(['order_id' => 1])),
]);

Consuming Messages

Consuming with the driver is designed for use in dedicated background worker processes, not inside HTTP request handlers. Create a consumer, poll in a loop, and commit offsets manually.

$consumer = $kafka->consumer(
    groupId: 'order-processor',
    topics:  ['orders'],
);

try {
    while (true) {
        $message = $consumer->consume(timeoutMs: 1000);

        if ($message === null) {
            // Timed out or end of partition — poll again
            continue;
        }

        // Process the message
        $data = json_decode($message->payload, true);
        processOrder($data);

        // Commit the offset after successful processing
        $consumer->commit();
    }
} finally {
    // Always close the consumer to release group membership
    $consumer->close();
}

Type Reference

Enums

ProducerMessage Properties

KafkaMessage Properties

Error Handling

publish() and publishBatch() throw \RuntimeException when the producer fails to deliver messages within delivery_timeout_ms:

[Kafka] Producer flush failed: Local: Message timed out

KafkaConsumer::consume() throws \RuntimeException on consumer errors other than timeouts and end-of-partition signals:

[Kafka] Consumer error: Broker: Unknown topic or partition

produce() throws RdKafka\Exception only when a message cannot be queued; delivery failures are reported to onDeliveryFailure() listeners instead.

An \InvalidArgumentException is thrown at construction time when brokers is missing from the connection config.

Source

The source code for this driver is hosted internally.