Download the PHP package posapp-vn/laravel-kafka-queue without Composer

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

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:

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:

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

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:

The connection can also be declared on the job:

Or used directly:

Run a normal Laravel worker:

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:

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:

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:

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:

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:

Set the implementation class in the queue connection:

The class is resolved through Laravel's service container.

Security

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

Connection loss and recovery

librdkafka maintains broker connections, metadata, producer retries, and consumer group heartbeats in background threads. For transient network or broker failures it normally reconnects automatically, and a consumer can rejoin its group and continue from the committed offset without rebuilding the queue connection. Authentication, authorization, DNS, routing, and advertised-listener errors still require configuration or infrastructure changes.

The package leaves librdkafka's reconnect, socket, session, and heartbeat settings at their version-specific defaults. The behavior described here was verified against librdkafka 2.14.2; check the installed version's configuration and reliability documentation before tuning them. session.timeout.ms applies to the classic consumer group protocol; newer consumer protocols may move that setting to the broker. max.poll.interval.ms remains a separate upper bound on job processing time between calls to consume().

The default KAFKA_QUEUE_PRODUCER_TIMEOUT_MS=10000 is both the local message delivery deadline and the maximum flush() wait. A short outage may recover within that window. A longer outage makes dispatch throw, but a timeout does not always prove that the broker did not persist an in-flight message. Callers must treat retrying a timed-out dispatch as potentially duplicating the record.

Multiple bootstrap brokers are already supported as a comma-separated list:

The connector reports librdkafka client errors through Laravel's logger with the client ID, producer/consumer role, error code, name, and redacted reason. Detailed client statistics are disabled by default. They can be emitted as structured debug logs when an operations pipeline is ready to consume them:

Laravel's after_commit option only waits until the database transaction has committed before publishing. It does not make the database commit and Kafka publish atomic. Applications that must guarantee a job after a committed business change need an application-level transactional outbox or equivalent durable retry mechanism.

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

Broker integration tests use:

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

The CI matrix also stops and restarts the running broker to verify that the same producer and consumer instances recover. Run the same scenario after starting one of the provided Docker Compose brokers and creating the topic:

License

The MIT License. See LICENSE.md.