laravel-kafka

By default, the library uses committers provided by DefaultCommitterFactory. To implement custom logic for how messages are committed (e.g., only committing on success or handling DLQ logic), you must create a class that implements the Committer interface.

The Committer interface provides hooks for both automatic and manual commit operations, ensuring that whether the consumer is performing automatic background commits or your code is calling manual commit methods, the logic remains consistent.

use Junges\Kafka\Contracts\Committer;
use RdKafka\Message;

class MyCommitter implements Committer
{
    public function commitMessage(Message $message, bool $success) : void {
        // Logic for automatic commits
    }
    
    public function commitDlq(Message $message) : void {
        // Logic for dead letter queue commits
    }

    public function commit(mixed $messageOrOffsets = null): void {
        // Logic for manual synchronous commits
    }

    public function commitAsync(mixed $messageOrOffsets = null): void {
        // Logic for manual asynchronous commits
    }
}

To enrich DLQ messages with custom metadata (like IDs or correlation keys), throw an exception that implements the Junges\Kafka\Contracts\ContextAware interface.

When such an exception is thrown, the consumer merges the following into the message headers:

1. Original message headers.
2. Throwable headers (kafka_throwable_message, kafka_throwable_code, kafka_throwable_class_name).
3. The array returned by the exception's getContext() method.

Note: Header values must be strings. Arrays, objects, numbers, or empty string keys are ignored.

use Junges\Kafka\Contracts\ContextAware;
use RuntimeException;
use Throwable;

class OrderProcessingException extends RuntimeException implements ContextAware
{
    public function __construct(
        private array $context,
        string $message = 'Order processing failed',
        int $code = 0,
        ?Throwable $previous = null,
    ) {
        parent::__construct($message, $code, $previous);
    }

    public function getContext(): array
    {
        return $this->context;
    }
}

// Usage in a handler
$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->subscribe('orders')
    ->withDlq()
    ->withHandler(function($message) {
        $payload = $message->getBody();

        throw new OrderProcessingException([
            'x-order-id' => (string)($payload['order_id'] ?? 'unknown'),
            'x-user-id' => (string)($payload['user_id'] ?? 'unknown'),
        ]);
    })
    ->build();

$consumer->consume();

You can control when message offsets are committed to Kafka using two modes:

Auto Commit (Default)

Messages are automatically committed after the handler successfully processes them. This is the simplest mode.

Manual Commit

Provides full control over when offsets are committed. This is recommended for guaranteed processing or complex error handling. When using manual commit, the handler receives the $consumer instance, which provides the following methods:

• commit(): Commit current assignment offsets (synchronous).
• commit($message): Commit a specific message offset (synchronous).
• commitAsync(): Commit current assignment offsets (asynchronous).
• commitAsync($message): Commit a specific message offset (asynchronous).

// Manual Commit Example
$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->withManualCommit()
    ->withHandler(function($message, $consumer) {
        try {
            processMessage($message);
            $consumer->commit($message); // Synchronous commit
        } catch (Exception $e) {
            // Handle error
        }
    })
    ->build();

$consumer->consume();

You can offload Kafka message processing to the Laravel queue system by implementing the Illuminate\\Contracts\\Queue\\ShouldQueue interface in your handler class. When a message is received by the Kafka consumer, a new job is dispatched to your Laravel queue for each message.

Important Limitation: Because queued handlers run asynchronously via Laravel's queue workers, they do not have access to a MessageConsumer instance. You cannot perform actions that require a live connection to the Kafka consumer within the __invoke method of a queued handler.

use Illuminate\Contracts\Queue\ShouldQueue;
use Junges\Kafka\Contracts\Handler as HandlerContract;
use Junges\Kafka\Contracts\KafkaConsumerMessage;

class Handler implements HandlerContract, ShouldQueue
{
    public function __invoke(KafkaConsumerMessage $message): void
    {
        // Handle the consumed message.
    }
}

Kafka consumers in this package are implemented as standard Laravel Command classes. You define the Kafka topic, brokers, and the message handling logic within the handle() method of your command.

To handle messages, use ->withHandler() which provides a ConsumerMessage instance and a MessageConsumer instance. If you need manual control over message commits, use ->withManualCommit() and call $consumer->commit($message) inside the handler.

<?php declare(strict_types=1);

namespace App\Console\Commands\Consumers;

use Illuminate\Console\Command;
use Junges\Kafka\Facades\Kafka;
use Junges\Kafka\Contracts\MessageConsumer;
use Junges\Kafka\Contracts\ConsumerMessage;

class MyTopicConsumer extends Command
{
    protected $signature = "consume:my-topic";

    protected $description = "Consume Kafka messages from 'my-topic'."

    public function handle()
    {
        $consumer = Kafka::consumer(['my-topic'])
            ->withBrokers('localhost:8092')
            ->withAutoCommit()
            ->withHandler(function(ConsumerMessage $message, MessageConsumer $consumer) {
                // Handle your message here
                // For manual commit control, use ->withManualCommit() and call $consumer->commit($message)
            })
            ->build();
            
        $consumer->consume();
    }
}