README

A Filament admin panel for gurento/kafka-consumer — manage Kafka topic mappings, monitor consumer health, and replay failed messages without leaving your admin panel.

Features

• Topic management — full CRUD for topic-to-model mappings, field maps, and relation syncs
• Live monitoring — auto-polling table with consumed/failed counters, failure rate, heartbeat, lag, and health badges
• Consume logs — per-message log viewer with payload inspection, Kafka partition/offset/key metadata, and status filters
• One-click operations — re-consume failed messages, mark topics healthy, and reset counters from the UI
• Fully customizable — navigation label, icon, group, badge, slug, labels, and polling are all configurable via a fluent plugin API

Requirements

Installation

1. Install the packages

This plugin is the UI layer for gurento/kafka-consumer, which does the actual consuming. Install both:

composer require gurento/kafka-consumer gurento/kafka-consumer-filament

2. Set up gurento/kafka-consumer

Publish the config and migrations, then migrate:

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-to-model mappings, counters, and health metadata (what this plugin manages)
• kafka_consume_logs — per-message processing logs (what the logs viewer reads)

Review config/kafka-consumer.php for defaults (consumer group, retry attempts, backoff, health thresholds). The consumer itself ships with a plug-and-play engine based on mateusjunges/laravel-kafka — make sure the rdkafka PHP extension is installed and your broker settings are configured in the host app's config/kafka.php.

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

3. Register the plugin

In your Filament panel provider:

use Gurento\KafkaConsumerFilament\Filament\Plugins\KafkaConsumerPlugin;

public function panel(Panel $panel): Panel
{
    return $panel
        ->plugins([
            KafkaConsumerPlugin::make(),
        ]);
}

That's it — a Kafka Topics resource appears in your panel's navigation.

4. Start consuming

Create a topic mapping in the UI, then run the consumer (as a daemon under Supervisor/systemd in production):

php artisan gurento:kafka-consume

See the gurento/kafka-consumer README for command options (--topics, --group, --from-beginning, --stop-on-empty, --reconsume-failed), replay patterns, events, and custom engines.

Customization

Every presentation element is dynamic. Configure it fluently on the plugin — each setter accepts a plain value or a Closure:

KafkaConsumerPlugin::make()
    ->navigationLabel('Event Streams')            // sidebar title
    ->navigationIcon('heroicon-o-queue-list')     // hero icon (string or BackedEnum)
    ->navigationGroup('Integrations')
    ->navigationSort(5)
    ->navigationBadge()                           // badge showing pending retries
    ->modelLabel('Stream')
    ->pluralModelLabel('Streams')
    ->slug('event-streams')                       // URL: /admin/event-streams
    ->tablePollInterval('30s')                    // null disables auto-refresh
    ->modelOptions(fn (): array => [              // restrict target-model dropdown
        \App\Models\Office::class => 'Office',
        \App\Models\Employee::class => 'Employee',
    ]),

Available options

All options are optional — KafkaConsumerPlugin::make() alone keeps the defaults above.

Creating a Topic Mapping

The create/edit form covers everything a topic needs to go from raw Kafka payload to an upserted Eloquent record — no code required.

Topic Configuration

Field Mapping (payload matching)

A repeater that maps payload fields to model columns — each row is one from → to pair:

Given {"uuid": "off-001", "name": "Accounting Office"}, the consumer writes id = off-001, name = Accounting Office and upserts by the configured upsert key. Unmapped fields are skipped, so mappings stay explicit and reviewable. Rows collapse to a readable uuid → id label.

Relation Syncs (relationships)

A repeater for syncing BelongsToMany relationships from nested arrays in the payload:

Example: with payload key office_controller and payload

{
  "uuid": "off-001",
  "office_controller": [
    {"uuid": "emp-001"},
    {"uuid": "emp-002"}
  ]
}

the consumer looks up each employee by uuid and calls $office->office_controller()->sync([...]). Items whose related record doesn't exist yet are skipped.

Typical Workflow

1. Open Kafka Topics in your panel.

2. Create a topic mapping: topic name, target model, upsert key, field mappings, optional relation syncs.

3. Run the consumer:

   php artisan gurento:kafka-consume

4. Monitor counters, health, and logs in the resource (the table auto-refreshes).

5. Re-consume failed messages from the row action or the logs relation manager when needed.

What It Registers

• KafkaTopicResource — list, create, view, and edit pages
• LogsRelationManager — read-only consume-log browser with payload inspection and per-log retry
• Row actions: Re-consume Failed, Mark Healthy, Reset Counters

Security & Access

This package ships UI classes only — it does not impose authorization. Define policies/permissions in your host app to control who can:

• edit topic mappings
• run replay actions
• inspect payload and error logs

Troubleshooting

Resource not visible — ensure the plugin is registered on the panel you're viewing, then run php artisan optimize:clear.

Class not found — confirm both packages are installed and run composer dump-autoload.

Customizations not applying — plugin options are read at runtime from the current panel; make sure you configure them on the same KafkaConsumerPlugin::make() instance passed to ->plugins([...]).

Changelog

See CHANGELOG.md for release history.

License

The MIT License (MIT). See LICENSE for details.