Laravel Kafka Queue

A Laravel queue connector backed by Apache Kafka-compatible brokers, including Redpanda. Existing Laravel jobs can continue to use dispatch(), Queue::push(), queue:work, retry handling, and failed_jobs.

The default mode needs only Kafka. Redis is optional and is used only when delayed jobs, positive retry backoff, or durable crash-attempt counters are required.

Requirements

• PHP 8.2 or newer
• Laravel 10, 11, or 12
• ext-rdkafka 6.x
• Apache Kafka or a Kafka-compatible broker such as Redpanda

The package does not create topics. Provision topics, ACLs, retention, replication factor, and partitions before starting producers or workers.

Installation

composer require posapp-vn/laravel-kafka-queue

Laravel package discovery registers the kafka queue driver and an opt-in connection named kafka. It does not change QUEUE_CONNECTION.

The minimum configuration is:

KAFKA_QUEUE_BROKERS=127.0.0.1:9092
KAFKA_QUEUE_TOPIC_PREFIX=posapp.local.queue.
KAFKA_QUEUE_GROUP_PREFIX=posapp.admin-api.local

The default queue is default, so the example above uses:

• Topic: posapp.local.queue.default
• Consumer group: posapp.admin-api.local.default

Publish the configuration when you need to customize all options:

php artisan vendor:publish --tag=kafka-queue-config

An explicit entry in config/queue.php overrides the package-provided connection:

'connections' => [
    'kafka' => [
        'driver' => 'kafka',
        'brokers' => env('KAFKA_QUEUE_BROKERS', '127.0.0.1:9092'),
        'queue' => env('KAFKA_QUEUE_DEFAULT', 'default'),
        'topic_prefix' => env('KAFKA_QUEUE_TOPIC_PREFIX', 'app.production.queue.'),
        'group_id_prefix' => env('KAFKA_QUEUE_GROUP_PREFIX', 'app.production'),
    ],
],

Missing connection options are filled from config/kafka-queue.php.

Dispatching jobs

Migrate one job at a time without changing the application's default queue:

ReportJob::dispatch($reportId)
    ->onConnection('kafka')
    ->onQueue('default');

The connection can also be declared on the job:

public string $connection = 'kafka';

Or used directly:

Queue::connection('kafka')->push(new ReportJob($reportId));

Run a normal Laravel worker:

php artisan queue:work kafka --queue=high,default

Each queue name maps to a separate topic and consumer group. Laravel polls comma-separated queues using strict priority: each loop checks high before default. A continuously busy high queue can therefore starve default. Use dedicated workers (or separate worker pools) when lower-priority queues must have guaranteed capacity. Useful worker concurrency is bounded by the partition count of each topic.

Kafka-only mode

Kafka-only mode is the default:

KAFKA_QUEUE_STATE_STORE=null

It supports:

• Immediate dispatch and consumption
• Synchronous offset commit on delete()
• Immediate retry through release(0)
• Laravel failed_jobs for failures handled by the worker
• Approximate size() based on consumer lag

Clean retries are republished with a laravel_prior_attempts header before the original offset is committed. If the publish or commit fails, the original offset remains uncommitted.

Kafka has no visibility timeout or mutable delivery counter. If a worker is killed before it releases or commits a message, the message can be redelivered with the same attempt number. Jobs must be idempotent.

Delayed dispatch without a state store fails synchronously in the producer with StateStoreRequiredException, so no job is partially enqueued:

ReportJob::dispatch($reportId)->delay(now()->addMinute()); // requires state

If a running job requests a positive retry backoff without a state store, Laravel's worker has already caught the job exception and calls release($delay). Throwing from that method would leave the offset uncommitted and redeliver the same record forever. The connector therefore logs a warning, republishes the job as an immediate retry with an incremented laravel_prior_attempts header, and only then commits the original offset:

public function backoff(): int
{
    return 30; // retried immediately unless a state store is configured
}

Every published record also receives a laravel_queue_expires_at header based on KAFKA_QUEUE_MESSAGE_TTL=604800 (seven days by default). When a job does not define its own retryUntil, this timestamp is used as a hard safety limit. It prevents a process-crashing poison job from being redelivered forever even when its attempt counter cannot advance. A job-defined retryUntil takes precedence, and queue:retry assigns a fresh package TTL.

This safety limit is age-based rather than attempt-based: a record that remains in a long backlog past the TTL can be moved to failed_jobs before its handler runs. Set the TTL above the longest expected backlog and outage window.

Optional Redis state

Enable Redis only for workloads that need delayed dispatch, positive backoff, or durable crash-attempt counting:

KAFKA_QUEUE_STATE_STORE=redis
KAFKA_QUEUE_REDIS_CONNECTION=default

The Redis adapter stores:

• Delayed payloads in a sorted set and hash
• A distributed promotion lock per queue, acquired with SET NX EX
• Attempt counters keyed by consumer group, topic, partition, and offset

When a delayed record becomes due, pop() publishes it to Kafka and removes the Redis record only after Kafka confirms delivery. A delayed retry is stored before the original offset is committed.

The promotion lock expires after KAFKA_QUEUE_PROMOTION_LOCK_SECONDS (30 seconds by default), so a worker crash cannot leave a permanent lock. Set this above the worst-case time required to publish one promotion batch. If promotion runs longer than the lock TTL, another worker can acquire the lock and publish the same delayed record; this is allowed by the package's at-least-once guarantee.

Attempt keys expire after KAFKA_QUEUE_ATTEMPT_TTL (seven days by default), and the TTL is refreshed on every delivery. If the gap between deliveries exceeds that TTL, the crash-attempt counter resets. Configure it above the longest expected backlog/outage window and normally at least as large as KAFKA_QUEUE_MESSAGE_TTL. The message TTL remains the final safety net when an attempt counter expires or a process dies before Redis can record the next delivery.

Redis failures are fail-closed for state-dependent operations. The queue does not commit an offset when durable retry state could not be written.

Custom backends can implement:

PosAppVN\LaravelKafkaQueue\Contracts\StateStore

Set the implementation class in the queue connection:

'state_store' => App\Queue\DatabaseStateStore::class,

The class is resolved through Laravel's service container.

Security

Plaintext is the default. SASL and TLS options are available:

KAFKA_QUEUE_SECURITY_PROTOCOL=SASL_SSL
KAFKA_QUEUE_SASL_MECHANISMS=SCRAM-SHA-256
KAFKA_QUEUE_SASL_USERNAME=app
KAFKA_QUEUE_SASL_PASSWORD=secret
KAFKA_QUEUE_SSL_CA_LOCATION=/run/secrets/kafka-ca.pem

Delivery guarantee

Delivery is at-least-once. Duplicate execution remains possible when:

• Application side effects finish but the offset commit fails.
• A retry message is published but committing the original offset fails.
• A worker terminates while handling a message.

Jobs dispatched to Kafka must therefore be idempotent.

Producer idempotence prevents duplicate records caused by producer transport retries; it does not deduplicate business side effects.

Deployment notes

• Use replication factor 1 only for a single-broker development cluster.
• Production replication factor must not exceed the number of available brokers; 3 is a common choice for a three-or-more-broker cluster.
• Worker concurrency above the topic partition count does not increase throughput because one partition is assigned to only one consumer in a consumer group.
• Ensure the broker advertises an address reachable from the PHP runtime. Container-internal addresses such as redpanda:9092 are not reachable from a PHP process running directly on the host.
• Keep max_poll_interval_ms greater than the longest job runtime.

Testing

composer test:unit
composer analyse
vendor/bin/pint --test

Broker integration tests use:

KAFKA_QUEUE_INTEGRATION_BROKERS=127.0.0.1:9092
KAFKA_QUEUE_INTEGRATION_TOPIC=laravel-kafka-queue-integration

Create the topic laravel-kafka-queue-integration.default before running:

composer test:integration

License

The MIT License. See LICENSE.md.