PHPackages                             gurento/kafka-consumer - PHPackages - PHPackages  [Skip to content](#main-content)[PHPackages](/)[Directory](/)[Categories](/categories)[Trending](/trending)[Leaderboard](/leaderboard)[Changelog](/changelog)[Analyze](/analyze)[Collections](/collections)[Log in](/login)[Sign up](/register)

1. [Directory](/)
2. /
3. [Queues & Workers](/categories/queues)
4. /
5. gurento/kafka-consumer

ActiveLibrary[Queues & Workers](/categories/queues)

gurento/kafka-consumer
======================

Laravel Kafka consumer module with topic mapping, retry tracking, and operational logs.

v1.1.3(3mo ago)1602MITPHPPHP ^8.2

Since Apr 22Pushed 1w agoCompare

[ Source](https://github.com/fglend/kafka-consumer)[ Packagist](https://packagist.org/packages/gurento/kafka-consumer)[ RSS](/packages/gurento-kafka-consumer/feed)WikiDiscussions main Synced 1w ago

READMEChangelogDependencies (6)Versions (6)Used By (2)

Kafka Consumer
==============

[](#kafka-consumer)

[![Latest Version on Packagist](https://camo.githubusercontent.com/9b65418b70d6fa34952ec57e6666184bc255afe4fd19fe3263da483abd291dae/68747470733a2f2f696d672e736869656c64732e696f2f7061636b61676973742f762f677572656e746f2f6b61666b612d636f6e73756d65722e7376673f7374796c653d666c61742d737175617265)](https://packagist.org/packages/gurento/kafka-consumer)[![Total Downloads](https://camo.githubusercontent.com/bb03d14e5758022a75da54de9b0ee0369dc6a3e645a779fcbadec1a9f07d879f/68747470733a2f2f696d672e736869656c64732e696f2f7061636b61676973742f64742f677572656e746f2f6b61666b612d636f6e73756d65722e7376673f7374796c653d666c61742d737175617265)](https://packagist.org/packages/gurento/kafka-consumer)[![License](https://camo.githubusercontent.com/4f4ae119c88ea9eb382bd2513d961f25004ddbea65c65c8ae568e2a80f409afa/68747470733a2f2f696d672e736869656c64732e696f2f6769746875622f6c6963656e73652f677572656e746f2f6b61666b612d636f6e73756d65723f7374796c653d666c61742d737175617265)](LICENSE)

A Laravel package for consuming Kafka messages and writing them into Eloquent models using **declarative, database-driven topic mappings** — no per-topic consumer classes required.

Built for operations teams that need:

- declarative topic-to-model mapping
- safe `updateOrCreate` upserts
- failed-message tracking and re-consume flows
- operational counters, health, and heartbeat metadata
- command-driven workflow for normal consume and replay

Ships with a default engine based on [`mateusjunges/laravel-kafka`](https://github.com/mateusjunges/laravel-kafka), so it works out of the box.

> **Admin UI available:** install [`gurento/kafka-consumer-filament`](https://packagist.org/packages/gurento/kafka-consumer-filament) for a full Filament panel — topic CRUD, live health monitoring, and one-click replay.

Features
--------

[](#features)

- Topic configuration in DB (`kafka_topics`) — no code deploys to add a topic
- Per-message logs in DB (`kafka_consume_logs`) with partition/offset/key metadata
- Payload-to-model field mapping with key exclusion
- Configurable upsert key per topic
- `BelongsToMany` relation syncing from nested payload arrays
- Automatic retry scheduling with per-topic backoff and attempt limits
- Re-consume command mode for failed messages
- Health status (`healthy` / `degraded` / `stalled` / `inactive`) from heartbeat + error state
- `KafkaMessageConsumed` / `KafkaMessageFailed` events for alerting and metrics

Requirements
------------

[](#requirements)

DependencyVersionPHP8.2+Laravel11 / 12 / 13ext-rdkafkaconfigured for your PHP runtimeKafka brokerreachable from the appInstallation
------------

[](#installation)

```
composer require gurento/kafka-consumer
php artisan vendor:publish --tag=kafka-consumer-config
php artisan vendor:publish --tag=kafka-consumer-migrations
php artisan migrate
```

This creates two tables:

- `kafka_topics` — topic config, counters, and health metadata
- `kafka_consume_logs` — per-message processing logs

### Environment

[](#environment)

Add your Kafka connection settings to `.env`:

```
KAFKA_BROKERS=localhost:9092
KAFKA_CONSUMER_GROUP_ID=app-consumer
KAFKA_DEBUG=false
```

- `KAFKA_BROKERS` — comma-separated broker list (host:port)
- `KAFKA_CONSUMER_GROUP_ID` — default consumer group used when `--group` is not passed
- `KAFKA_DEBUG` — set to `true` to enable verbose librdkafka debug output while troubleshooting

Quick Start
-----------

[](#quick-start)

1. Create a topic configuration (via seeder, tinker, or the Filament UI).
2. Start the consumer:

    ```
    php artisan gurento:kafka-consume
    ```
3. Inspect results in `kafka_topics.messages_consumed` and `kafka_consume_logs`.

Topic Configuration
-------------------

[](#topic-configuration)

Each `kafka_topics` row defines how a Kafka topic maps into your model.

**Core columns:**

ColumnPurpose`topic`Kafka topic name`model_class`Fully-qualified model class (e.g. `App\Models\Office`)`upsert_key`Unique model column used by `updateOrCreate``field_map`JSON array of `{from, to}` mapping pairs`exclude_keys`Optional payload keys to strip before mapping`relations`Optional `BelongsToMany` sync definitions`is_active`Whether the topic is consumed**Retry/health overrides** (fall back to config when `NULL`): `max_reconsume_attempts`, `retry_backoff_seconds`, `health_stale_after_seconds`.

**Operational columns** (auto-managed): `messages_consumed`, `messages_failed`, `messages_reconsumed`, `last_consumed_at`, `consumer_last_heartbeat_at`, `consumer_last_error`, `consumer_lag_seconds`.

### Field Mapping Example

[](#field-mapping-example)

Given this Kafka payload:

```
{
  "uuid": "off-001",
  "name": "Accounting Office",
  "meta": {"source": "hr"}
}
```

Use `field_map`:

```
[
  {"from": "uuid", "to": "id"},
  {"from": "name", "to": "name"}
]
```

And `exclude_keys`:

```
["meta"]
```

The package upserts by `upsert_key` (for example `id`).

Command Usage
-------------

[](#command-usage)

### Normal Consumption

[](#normal-consumption)

```
php artisan gurento:kafka-consume
```

OptionDescription`--topics=*`Consume only selected topics`--limit=`Process only N messages, then stop`--group=`Consumer group id (defaults to `kafka-consumer.default_group` config)`--from-beginning`Read from earliest offsets (replay mode)`--stop-on-empty`Stop after the last available message instead of polling foreverExamples:

```
php artisan gurento:kafka-consume --topics=HR_APP.LIVE.office
php artisan gurento:kafka-consume --topics=HR_APP.LIVE.office --limit=500
php artisan gurento:kafka-consume --from-beginning --stop-on-empty
```

### Re-consume Failed Logs

[](#re-consume-failed-logs)

Instead of polling Kafka, reprocess failed entries from `kafka_consume_logs`:

```
php artisan gurento:kafka-consume --reconsume-failed --reconsume-limit=100
```

Useful after fixing a wrong `field_map`, a wrong `model_class`, or a temporary DB/infrastructure failure.

How Replay Works
----------------

[](#how-replay-works)

`--from-beginning` sets offset reset to `earliest` and uses a unique consumer group (unless `--group` is provided) to avoid corrupting your normal consumer offsets.

Recommended pattern:

1. Run the normal consumer with a stable group for real-time operations.
2. Use `--from-beginning` only for backfills/replays.

Failure & Retry Behavior
----------------------------

[](#failure--retry-behavior)

On processing errors, the package logs `status=failed`, the `error` message, and retry metadata (`attempt_count`, `next_retry_at`, `retryable`). Retries back off linearly per attempt using the topic's `retry_backoff_seconds` and stop after `max_reconsume_attempts`. Re-consume mode marks replayed entries `reconsumed_success` on success.

Configuration
-------------

[](#configuration)

Values in `config/kafka-consumer.php` act as fallbacks when a topic row leaves the matching column `NULL`:

KeyDefaultPurpose`default_group``app-consumer`Consumer group id when `--group` is not passed`max_reconsume_attempts``3`Retry attempts before a failure becomes permanent`retry_backoff_seconds``60`Base backoff between retries`health_stale_after_seconds``300`Heartbeat age before a topic is considered stalledEvents
------

[](#events)

The service dispatches events on every processing outcome — hook them for alerting or metrics:

EventWhenPayload`Gurento\KafkaConsumer\Events\KafkaMessageConsumed`Message (or re-consume) succeeds`$topic`, `$log`, `$isReconsume``Gurento\KafkaConsumer\Events\KafkaMessageFailed`Processing fails`$topic`, `$log`, `$error`, `$willRetry````
use Gurento\KafkaConsumer\Events\KafkaMessageFailed;

Event::listen(KafkaMessageFailed::class, function (KafkaMessageFailed $event): void {
    if (! $event->willRetry) {
        // permanent failure — alert your on-call channel
    }
});
```

Custom Engine (Optional)
------------------------

[](#custom-engine-optional)

By default the package binds `Gurento\KafkaConsumer\Contracts\ConsumerEngine` to `LaravelKafkaConsumerEngine`. To use a different transport, override the binding in your host app:

```
$this->app->singleton(
    \Gurento\KafkaConsumer\Contracts\ConsumerEngine::class,
    YourCustomEngine::class,
);
```

Your engine must implement:

```
public function consume(array $topics, callable $handler, array $options = []): void;
```

Programmatic Consumption Hook
-----------------------------

[](#programmatic-consumption-hook)

If you already consume Kafka elsewhere, call the service directly:

```
app(\Gurento\KafkaConsumer\Services\KafkaConsumerService::class)
    ->handle($topicName, $payload, $metadata);
```

Production Recommendations
--------------------------

[](#production-recommendations)

- Run the consumer as a daemon (Supervisor/systemd)
- Monitor `messages_failed` and `consumer_last_error` (or listen to `KafkaMessageFailed`)
- Use separate groups for replay/backfill jobs
- Keep `field_map` explicit and reviewed
- Keep topic configs in source-controlled seeders where possible

Troubleshooting
---------------

[](#troubleshooting)

**`No active topics configured`** — ensure `kafka_topics` has rows with `is_active = true`.

**Command not behaving as expected** — run `php artisan optimize:clear && composer dump-autoload`.

**Kafka connection/auth issues** — verify the host app's `config/kafka.php` and environment values (`KAFKA_BROKERS`, auth settings).

**Replay not processing failed logs** — check that rows have `status = failed`, `retryable = true`, and `next_retry_at