Download the PHP package kafka-bus/laravel-bridge without Composer

Kafka Bus for Laravel

Laravel integration for kafka-bus — a configuration-driven Apache Kafka client built on top of ext-rdkafka. The package wires producers, consumer workers, topic routing, and middleware into the framework, and ships an optional Commiter component for idempotent message handling backed by the database.

Requirements

• PHP ^8.2
• Laravel
• ext-rdkafka

Installation

Install the package via Composer:

Publish the main configuration file:

To use the Commiter component (idempotency and commit tracking), additionally publish its configuration and migrations:

Configuration

The main configuration lives in config/kafka-bus.php and is split into four sections:

• connections — Kafka broker connections and driver-specific options.
• topics — logical topic keys mapped to physical Kafka topic names.
• consumers — workers, topic-to-handler bindings, middleware, and consumer options.
• producers — message-to-topic routes, middleware, and producer options.

Connections

Each connection is selected by a driver and a set of options passed straight to librdkafka. The default key picks the active connection by name.

The null driver is useful for tests — calls to the bus succeed without touching a real broker.

Topics

Topic names usually depend on the environment. The bus prepends topic_prefix to every physical topic name, and the topics map binds a short logical key to that physical name.

With APP_ENV=production, products resolves to production.fact.products.1.

Producers

A producer route binds a message class to a logical topic key. The shortest form maps the class directly to a topic key:

The verbose form lets you override timeouts, append per-route middleware, and pass driver options:

Publish a message through the bus:

Or via the KafkaBus facade:

Consumers

Workers are the units consumed by the artisan command. Each worker subscribes to one or more topics and dispatches incoming messages to handler classes.

Each worker resolves options in this order: additional_options, auto_commit, consume_timeout, and middleware are taken from the worker entry, then merged with the global consumers.* defaults.

Run a worker:

Artisan commands

Inspecting workers and routes

Inspecting and resetting offsets

Reset all partitions of a topic to the earliest available offset:

Move a single partition to an explicit numeric offset:

Jump every partition to the high-water mark (skip backlog):

The command prints the resulting offsets:

The worker must not be running while you reset its offsets — otherwise the active consumer group will overwrite the new position on its next commit.

Commiter

The Commiter component (powered by micromus/kafka-bus-commiter) provides:

• Consumer idempotency — every incoming message is tracked in the kafka_bus_commits table; duplicates are skipped, retries are counted, and a configurable max-attempt threshold can stop poison messages.
• Producer idempotency keys — outgoing messages implementing HasIdempotency automatically receive an x-idempotency-key header, which the consumer side uses as the dedup key.

It is registered automatically by CommiterServiceProvider (loaded via package auto-discovery).

Configuration

• connection — Laravel database connection name; null uses the default connection.
• table — name of the commits table created by the published migration.
• repository — strategy for deriving the dedup key:
  • idempotency — reads the x-idempotency-key header combined with the topic name, falling back to the raw Kafka message id if the header is missing.
  • native — uses the raw Kafka message id only.
• repositories — registry of repository implementations; add your own class here and reference it via KAFKA_COMMITER_REPOSITORY.

Enabling the consumer middleware

Add ConsumerCommiterMiddleware to the consumer middleware stack — either globally for every worker, or only for specific workers/topics:

For each message the middleware:

1. Resolves a dedup key via the configured repository.
2. If the key was already committed — the message is skipped and a warning is logged.
3. If the per-key attempt count exceeds maxAttempt (when configured) — the message is skipped and an error is logged.
4. Otherwise the pipeline is executed; on success the key is committed, on failure the attempt counter is incremented and the exception is re-thrown.

Producing idempotent messages

Implement HasIdempotency on the producer message and enable ProducerIdempotencyMiddleware:

The middleware adds the x-idempotency-key header to every outgoing message; consumers running ConsumerCommiterMiddleware with the idempotency repository will use it as the dedup key.

Testing

KafkaBus::fake()

The KafkaBus facade ships a first-class fake that works exactly like Event::fake() or Mail::fake(). Call KafkaBus::fake() at the start of a test to replace the real BusInterface binding with an in-memory FakeBus. From that point on every call to the facade is forwarded to the fake — publish() calls are intercepted and stored, consumer pipelines can be triggered directly, and the full set of assertion methods becomes available.

Asserting producer messages

Assert with a callback to inspect the serialised ProducerMessage (after the full producer pipeline, including middleware). The callback receives a Micromus\KafkaBus\Producers\Messages\ProducerMessage instance:

Other available assertions:

Retrieve the published messages directly for custom assertions:

Dispatching and asserting consumer messages

addMessage() queues an RdKafka\Message into the fake connection. Once queued, call listen() to run the full consumer path — ConnectionFaker → ConsumerFaker → ConsumerStream → consumer middleware → route middleware → handler → commit — without touching a real broker.

Queue multiple messages before triggering listen():

Asserting committed messages

After listen() each successfully processed message is committed into the fake connection. Use the commit assertions to verify that your handler ran and the offset was acknowledged:

Retrieve committed messages directly for custom assertions:

Changelog

Please see CHANGELOG for more information on what has changed recently.

Contributing

Please see CONTRIBUTING for details.

Security Vulnerabilities

Please review our security policy on how to report security vulnerabilities.

Credits

• Kirill Popkov
• All Contributors

License

The MIT License (MIT). Please see License File for more information.