Kafka Laravel Package

Getting Started

Minimal Setup

1. Installation:

   composer require ecotone/kafka
   

   Ensure you also have the core Ecotone package:

   composer require ecotone/ecotone
   

2. Configure Kafka Transport: Add the Kafka transport to your Ecotone container in bootstrap/app.php or a service provider:

   use Ecotone\Kafka\KafkaTransport;
   use Ecotone\Kafka\KafkaTransportConfig;
   
   $container->addTransport(
       new KafkaTransport(
           new KafkaTransportConfig(
               bootstrapServers: ['localhost:9092'],
               clientId: 'laravel-app',
               groupId: 'laravel-consumer-group'
           )
       )
   );
   

3. First Use Case: Publish an Async Message Define a message class and handler:

   #[Asynchronous]
   class OrderCreated {
       public function __construct(public string $orderId) {}
   }
   
   #[CommandHandler]
   class OrderCreatedHandler {
       public function __invoke(OrderCreated $orderCreated) {
           // Business logic
       }
   }
   

   Publish the message from a controller or command:

   $container->getBus()->dispatch(new OrderCreated('order-123'));
   

4. Verify Kafka Topics: Check if the topic ecotone.order_created (auto-generated from the message class) exists in Kafka.

Implementation Patterns

Workflows

1. Message Routing:

   • Use #[Asynchronous] attribute to auto-route messages to Kafka topics (topic name derived from the message class).
   • Customize topic names with #[RouteToTopic("custom.topic")].

2. Consumer Groups:

   • Configure groupId in KafkaTransportConfig to manage consumer groups for parallel processing.
   • Scale consumers by running multiple instances with the same groupId.

3. Partition Awareness:

   • Leverage Kafka partitions for parallel processing. Ecotone automatically handles partition assignment.
   • Use #[PartitionKey] to control message partitioning:

     #[Asynchronous]
     #[PartitionKey('orderId')]
     class OrderCreated { ... }
     

4. Error Handling:

   • Integrate with Ecotone’s retry, outbox, and dead letter queue (DLQ) mechanisms:

     $container->addTransport(
         new KafkaTransport(
             new KafkaTransportConfig(
                 // ...
                 retryConfig: new RetryConfig(maxRetries: 3, delay: 1000),
                 deadLetterTopic: 'dlq.order_created'
             )
         )
     );
     

5. Event Sourcing:

   • Store events in Kafka and replay them using Ecotone’s event sourcing:

     #[EventSourcedAggregateRoot]
     class Order {
         #[Event]
         public function create(string $orderId) {
             $this->recordThat(new OrderCreated($orderId));
         }
     }
     

6. Sagas:

   • Coordinate distributed transactions across services using Kafka:

     #[Saga]
     class OrderSaga {
         public function __invoke(OrderCreated $orderCreated) {
             // Orchestrate steps
         }
     }
     

Integration Tips

• Laravel Integration: Bind the Ecotone container to Laravel’s service container in AppServiceProvider:

  public function register() {
      $this->app->singleton(Ecotone::class, fn() => new Ecotone());
  }
  

• Configuration Management: Store Kafka config in .env:

  KAFKA_BOOTSTRAP_SERVERS=localhost:9092
  KAFKA_CLIENT_ID=laravel-app
  KAFKA_GROUP_ID=laravel-consumer-group
  

  Load it dynamically:

  $config = new KafkaTransportConfig(
      bootstrapServers: env('KAFKA_BOOTSTRAP_SERVERS'),
      // ...
  );
  

• Testing: Use Ecotone’s testing utilities with Kafka:

  use Ecotone\Testing\TestContainer;
  
  public function testOrderCreated() {
      $container = new TestContainer();
      $container->addTransport(new KafkaTransport(new KafkaTransportConfig(
          bootstrapServers: ['localhost:9092'],
          // Use a test group ID
          groupId: 'test-group'
      )));
  
      $container->getBus()->dispatch(new OrderCreated('test-123'));
      // Assertions...
  }
  

Gotchas and Tips

Pitfalls

1. Topic Auto-Creation:

   • Ecotone auto-creates topics based on message class names, but ensure your Kafka broker allows auto-topic creation (auto.create.topics.enable=true).
   • Fix: Manually create topics with desired configurations (partitions, replication factor) if needed.

2. Consumer Lag:

   • High consumer lag may occur if messages are processed slower than they’re produced.
   • Solution: Monitor lag with Kafka tools (e.g., kafka-consumer-groups --bootstrap-server localhost:9092 --group laravel-consumer-group) and scale consumers.

3. Schema Evolution:

   • Kafka doesn’t natively support schema evolution. Use Avro or Protobuf with a schema registry (e.g., Confluent Schema Registry) for backward compatibility.
   • Workaround: Version your message classes (e.g., OrderCreatedV1, OrderCreatedV2).

4. Serialization:

   • Ecotone uses PHP’s serialize() by default, which may not be ideal for cross-language compatibility.
   • Tip: Override serialization with #[SerializedWith(MySerializer::class)]:

     use Ecotone\Serialization\Attribute\SerializedWith;
     use Ecotone\Serialization\JsonSerializer;
     
     #[Asynchronous]
     #[SerializedWith(JsonSerializer::class)]
     class OrderCreated { ... }
     

5. Offset Management:

   • Kafka consumers commit offsets manually by default. Ecotone commits offsets after successful handler execution.
   • Risk: If a handler throws an exception, the offset won’t be committed, causing reprocessing.
   • Solution: Use #[Transactional] to ensure atomic processing:

     #[CommandHandler]
     #[Transactional]
     class OrderCreatedHandler { ... }
     

6. Dependency Injection:

   • Avoid injecting Kafka producers/consumers directly. Use Ecotone’s Bus or MessageDispatcher for decoupled messaging.

Debugging

1. Log Levels: Enable verbose logging in KafkaTransportConfig:

   new KafkaTransportConfig(
       // ...
       logLevel: \Monolog\Logger::DEBUG
   );
   

2. Common Errors:

   • UnhandledMessageException: The message class isn’t registered as a command/event. Ensure all async messages have #[Asynchronous].
   • ConnectionException: Check Kafka broker connectivity and credentials.
   • SerializationException: Validate message class properties are serializable.

3. Tooling:

   • Use kafka-console-consumer to inspect topics:

     kafka-console-consumer --bootstrap-server localhost:9092 --topic ecotone.order_created --from-beginning
     

   • Monitor consumer groups with kafka-consumer-groups.

Extension Points

1. Custom Serializers: Implement Ecotone\Serialization\SerializerInterface for custom formats (e.g., JSON, Protobuf):

   class JsonSerializer implements SerializerInterface {
       public function serialize($data): string { ... }
       public function deserialize(string $data): mixed { ... }
   }
   

2. Interceptors: Add pre/post-processing logic with #[BeforeHandler] or #[AfterHandler]:

   #[BeforeHandler]
   class LogMessageInterceptor {
       public function __invoke(InvokeCommandHandlerContext $context) {
           \Log::info('Processing message', ['message' => $context->getCommand()]);
       }
   }
   

3. Dynamic Routing: Override topic routing logic by implementing Ecotone\Transport\TransportRouter:

   class CustomKafkaRouter implements TransportRouter {
       public function getTopicName(string $messageClass): string {
           return 'custom.' . strtolower($messageClass);
       }
   }
   

   Register it in KafkaTransportConfig:

   new KafkaTransportConfig(
       // ...
       router: new CustomKafkaRouter()
   );
   

4. Metrics: Integrate with Prometheus or other monitoring tools by extending Ecotone\Transport\Transport and emitting custom metrics.

5. Security:

   • Enable SSL/SASL for Kafka:

     new KafkaTransportConfig(
         // ...
         ssl: new SslConfig(
             caLocation: '/path/to/ca.pem',
             certificateLocation: '/path/to/client.pem',
             keyLocation: '/path/to/client.key'
         ),
         sasl: new SaslConfig(
             mechanism: 'SCRAM-SHA-256',
             username: 'user',
             password: 'pass'
         )
     );