Message handlers

Now that you have created your kafka consumer, you must create a handler for the messages this consumer receives. By default, a consumer is any callable. You can use an invokable class or a simple callback. Use the withHandler method to specify your handler:

$consumer = \Junges\Kafka\Facades\Kafka::consumer();

// Using callback:
$consumer->withHandler(function(\Junges\Kafka\Contracts\ConsumerMessage $message, \Junges\Kafka\Contracts\MessageConsumer $consumer) {
    // Handle your message here
});

Or, using an invokable class:

class Handler
{
    public function __invoke(\Junges\Kafka\Contracts\ConsumerMessage $message, \Junges\Kafka\Contracts\MessageConsumer $consumer) {
        // Handle your message here
    }
}

$consumer = \Junges\Kafka\Facades\Kafka::consumer()->withHandler(new Handler)

The ConsumerMessage contract gives you some handy methods to get the message properties:

getKey(): Returns the Kafka Message Key
getTopicName(): Returns the topic where the message was published
getPartition(): Returns the kafka partition where the message was published
getHeaders(): Returns the kafka message headers
getBody(): Returns the body of the message
getOffset(): Returns the offset where the message was published

Manual Commit in Handlers

When using manual commit mode (withAutoCommit(false)), your handlers receive a $consumer parameter that provides commit methods. This allows you to control exactly when message offsets are committed:

$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->withAutoCommit(false)  // Enable manual commit mode
    ->withHandler(function(\Junges\Kafka\Contracts\ConsumerMessage $message, \Junges\Kafka\Contracts\MessageConsumer $consumer) {
        try {
            // Process your message
            $data = json_decode($message->getBody(), true);
            processBusinessLogic($data);
            
            // Commit the message after successful processing
            $consumer->commit($message);
            
        } catch (ValidationException $e) {
            // Don't commit invalid messages, send to DLQ or handle differently
            Log::warning('Invalid message format', ['message' => $message->getBody()]);
            
        } catch (Exception $e) {
            Log::error('Processing failed', ['error' => $e->getMessage()]);
            throw $e;
        }
    });

Available Commit Methods

The $consumer parameter provides these commit methods:

Synchronous commits (blocking):

$consumer->commit() - Commit current assignment offsets
$consumer->commit($message) - Commit specific message offset

Asynchronous commits (non-blocking, better performance):

$consumer->commitAsync() - Commit current assignment offsets
$consumer->commitAsync($message) - Commit specific message offset

Handler Classes

You can also create dedicated handler classes by implementing the Handler interface. Handler classes receive both the message and consumer parameters, just like closure handlers:

use Junges\Kafka\Contracts\Handler;
use Junges\Kafka\Contracts\ConsumerMessage;
use Junges\Kafka\Contracts\MessageConsumer;

class ProcessOrderHandler implements Handler
{
    public function __invoke(ConsumerMessage $message, MessageConsumer $consumer): void
    {
        try {
            $order = json_decode($message->getBody(), true);
            
            // Process the order
            $this->processOrder($order);
            
            // Manual commit after successful processing
            $consumer->commit($message);
            
        } catch (ValidationException $e) {
            // Don't commit invalid messages
            Log::warning('Invalid order data', ['message' => $message->getBody()]);
            
        } catch (Exception $e) {
            // Don't commit on processing errors
            Log::error('Order processing failed', ['error' => $e->getMessage()]);
            throw $e; // Re-throw to trigger DLQ handling if configured
        }
    }
    
    private function processOrder(array $order): void
    {
        // Your business logic here
    }
}

Using Handler classes with the consumer:

use Junges\Kafka\Facades\Kafka;

$consumer = Kafka::consumer(['orders'])
    ->withAutoCommit(false)  // Enable manual commit mode
    ->withHandler(new ProcessOrderHandler())
    ->build();

$consumer->consume();

Support Laravel Kafka by sponsoring me!

Do you find Laravel Kafka valuable and wanna support its development?

Laravel Kafka is free and Open Source software, built to empower developers like you. Your support helps maintain and enhance the project. If you find it valuable, please consider sponsoring me on GitHub. Every contribution makes a difference and keeps the development going strong! Thank you!

Become a Sponsor

Want to hide this message? Sponsor at any tier of $10/month or more!