Php Kafka Lib Laravel Package

Getting Started

jobcloud/php-kafka-lib is a thin, framework-agnostic Kafka client built on ext-rdkafka. In Laravel it fills the gap between the queue system and a real Kafka cluster: you get producers, consumers, subscriptions, Avro support, and commit control without pulling in a full framework.

Minimal setup:

pecl install rdkafka
composer require jobcloud/php-kafka-lib

Laravel does not auto-discover a service provider for this package, so register the builders yourself:

// app/Providers/KafkaServiceProvider.php
public function register(): void
{
    $this->app->singleton(KafkaProducerInterface::class, function () {
        return KafkaProducerBuilder::create()
            ->withAdditionalBroker(config('kafka.brokers'))
            ->withAdditionalConfig(config('kafka.producer_options', []))
            ->build();
    });
}

First use case: publish a domain event from a Laravel service and consume it in an artisan command.

$producer->produce(new KafkaMessage(
    json_encode($event->toArray()),
    key: (string) $event->orderId,
));
$producer->flush(10_000);

Look first at:

• JobCloud\Kafka\Producer\KafkaProducerBuilder / KafkaProducerInterface
• JobCloud\Kafka\Consumer\KafkaConsumerBuilder / KafkaConsumerInterface
• JobCloud\Kafka\Message\KafkaMessage
• JobCloud\Kafka\Subscription\KafkaSubscription
• JobCloud\Kafka\Config\ProducerConfig / ConsumerConfig

The current release, v2.0.2, is a patch release: a PHP deprecation fix only. No API changes, no new features. Safe to upgrade from any v2.0.x.

Implementation Patterns

Singleton producer, flush on terminate. Builders are cheap but producers hold broker connections. Bind the producer as a singleton and flush in terminate() or a finally block. Do not flush after every message unless you are producing a single critical event.

$this->app->terminating(fn () => app(KafkaProducerInterface::class)->flush(5_000));

Partition deliberately. Use the message key for ordering. If you rely on Kafka ordering without a key, messages for the same entity can land on different partitions. This is the most common source of “out of order” bugs in Laravel event pipelines.

Consumer command with explicit commit. Model consumers as artisan commands rather than queue workers. Disable auto-commit and commit only after your handler succeeds.

$consumer = KafkaConsumerBuilder::create()
    ->withAdditionalBroker(config('kafka.brokers'))
    ->withConsumerGroupId('orders-consumer')
    ->withAdditionalSubscription('orders')
    ->withAdditionalConfig([
        'auto.offset.reset' => 'earliest',
        'enable.auto.commit' => 'false',
    ])
    ->build();

$consumer->subscribe();

while ($this->running) {
    $message = $consumer->consume(1_000); // timeout ms

    if ($message === null) {
        continue;
    }

    try {
        $this->handler->handle($message);
        $consumer->commit();
    } catch (Throwable $e) {
        report($e);
        // decide: retry topic, DLQ, or halt
    }
}

Batch consume for throughput. For high-volume topics, use the consumer’s batch API and commit after the batch succeeds. This reduces round trips and makes back-pressure easier to reason about.

Avro at the edges. If your cluster uses a schema registry, wrap encode/decode in the package’s Avro components and keep your domain handlers working with plain arrays or DTOs. Centralise schema IDs in config, not in handlers.

Laravel integration pattern.

• Put broker lists and topic names in config/kafka.php.
• Bind one producer singleton and one consumer per command.
• Use Laravel’s Context / logging to attach topic, partition, offset, and key to every log line.
• Keep Kafka consumers out of the HTTP request lifecycle. They are long-running processes.
• Use pcntl signals or Laravel’s $this->trap() in commands for graceful shutdown, and commit the last processed offset before exiting.

Testing. Depend on KafkaProducerInterface and KafkaConsumerInterface, not concrete builders. Fake the producer in feature tests and assert on KafkaMessage payloads and keys. Reserve real broker tests for integration suites (Testcontainers or a CI Kafka service).

Gotchas and Tips

ext-rdkafka is non-negotiable. This package does not speak the Kafka protocol in pure PHP. If the extension is missing or compiled against an older librdkafka, you will get confusing connection and metadata errors. Pin both in your Docker image and CI.

v2.0.2 is a PHP deprecation fix, not a behaviour change. The PR addresses a deprecation notice (likely an implicit nullable parameter under newer PHP versions). It is safe to upgrade, but if you are still on PHP 7.x or early 8.x, verify your error reporting does not turn deprecations into exceptions in production.

Auto-commit hides failures. Leaving enable.auto.commit enabled means offsets can be committed before your handler finishes. If you care about at-least-once delivery, set it to false and commit explicitly. If you need exactly-once semantics, this library alone will not give it to you; you need idempotent handlers or transactional Kafka features.

consume() timeouts are milliseconds, not seconds. A timeout of 0 blocks indefinitely. A timeout that is too small spins the CPU in a tight loop. Use a 500–1000 ms poll timeout and check a shutdown flag between polls.

Rebalance handling is your responsibility. When partitions are reassigned, uncommitted offsets and in-flight messages can be reprocessed. Make handlers idempotent, or use a deduplication table keyed by message ID. Do not assume a consumer owns a partition for the lifetime of the process.

Producer errors are asynchronous. A successful produce() call only means the message was queued locally. Delivery failures surface via delivery reports or flush() exceptions. Always check the result of flush() and log delivery failures with the message key.

Config keys are passed through to librdkafka. Typos in withAdditionalConfig() are often silently ignored rather than rejected. Validate your config array against known librdkafka properties, especially security.protocol, sasl.*, and auto.offset.reset.

Message keys must be strings or null. Passing an integer key or an object can lead to unexpected partitioning or serialization errors. Cast keys explicitly.

Extension points to use before forking the package:

• Custom decoder / encoder for non-Avro formats.
• A message validator or middleware layer around your handler.
• Logger injection via the builder for consumer/producer diagnostics.
• Custom KafkaMessage subclasses if you need typed payloads.

Observability gap. The package is intentionally low-level; it will not emit Laravel events or metrics for you. Instrument produce/consume latency, lag, and commit failures yourself. Consumer lag is the metric that will page you at 3 a.m.