Kafka Symfony Transport Laravel Package

Getting Started

Minimal Setup

1. Installation

   composer require arkemlar/kafka-symfony-transport
   

   Ensure ext-rdkafka is enabled in your PHP environment.

2. Configuration Add to config/packages/messenger.yaml:

   framework:
       messenger:
           transports:
               kafka:
                   dsn: '%env(KAFKA_DSN)%'
                   options:
                       topic: 'your_topic_name'
                       group_id: 'your_consumer_group'
                   retry_strategy:
                       max_retries: 3
                       delay: 1000
                       multiplier: 2
   

3. First Use Case Dispatch a message:

   use Symfony\Component\Messenger\MessageBusInterface;
   
   $bus->dispatch(new YourMessage());
   

   Start the worker:

   php bin/console messenger:consume kafka -vv
   

Implementation Patterns

Core Workflows

1. Producer Workflow

   • Use the default Symfony MessageBus to dispatch messages to Kafka.
   • Configure topics dynamically via message metadata:

     $message = new YourMessage();
     $message->setMetadata(['kafka_topic' => 'dynamic-topic']);
     $bus->dispatch($message);
     

2. Consumer Workflow

   • Subscribe to topics via messenger.yaml or dynamically:

     transports:
         kafka_dynamic:
             dsn: '%env(KAFKA_DSN)%'
             options:
                 topic: 'dynamic-topic'
     

   • Handle messages in a handler:

     use Symfony\Component\Messenger\Attribute\AsMessageHandler;
     
     #[AsMessageHandler]
     public function __invoke(YourMessage $message) {
         // Process logic
     }
     

3. Error Handling

   • Leverage Symfony’s retry strategy for transient failures.
   • Log failures via PSR-3 logger (injected into handlers).

Integration Tips

• Batching: Use symfony/messenger’s batching middleware for high-throughput scenarios.
• Serialization: Ensure messages implement Serializable or use Symfony’s Serializer component.
• Async Processing: Combine with symfony/messenger’s async transport for hybrid workflows.

Gotchas and Tips

Pitfalls

1. RDKafka Extension

   • Ensure ext-rdkafka is compiled with SSL/SASL support if using secure clusters.
   • Debug missing extension errors with:

     php -m | grep rdkafka
     

2. Topic Configuration

   • Hardcoding topics in messenger.yaml limits flexibility. Prefer dynamic topic resolution via message metadata.
   • Validate topic names (e.g., no spaces, max length) to avoid Kafka errors.

3. Consumer Groups

   • Reassigning group_id requires rebalancing. Use kafka-consumer-groups CLI tool to manage offsets:

     kafka-consumer-groups --bootstrap-server localhost:9092 --group your_group --reset-offsets --to-earliest --execute
     

Debugging

• Worker Logs: Increase verbosity with -vvv for detailed RDKafka logs.
• Message Dumping: Use symfony/messenger’s debug:message command to inspect dispatched messages:

  php bin/console debug:message YourMessage
  

Extension Points

1. Custom Transport Factory Override the default factory for advanced configurations:

   services:
       Symfony\Component\Messenger\Transport\Serialization\SerializerInterface: '@custom_serializer'
       Symfony\Component\Messenger\Transport\TransportInterface: '@custom_kafka_transport'
   

2. Middleware Integration Add custom middleware to the Kafka transport:

   transports:
       kafka:
           dsn: '%env(KAFKA_DSN)%'
           middleware:
               - 'your_custom_middleware'
   

3. Event Listeners Subscribe to MessageBusEvents for pre/post-dispatch hooks:

   use Symfony\Component\Messenger\Event\SentMessageEvent;
   
   $eventDispatcher->addListener(MessageBusEvents::SENT, function (SentMessageEvent $event) {
       if ($event->getEnvelope()->getTransportName() === 'kafka') {
           // Custom logic
       }
   });