PHPackages micromus/kafka-bus-commiter - 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. [Utility & Helpers](/categories/utility) 4. / 5. micromus/kafka-bus-commiter Abandoned → [kafka-bus/commiter](/?search=kafka-bus%2Fcommiter)Library[Utility & Helpers](/categories/utility) micromus/kafka-bus-commiter =========================== This is my package kafka-bus-repeater v1.1.2(1mo ago)01271MITPHPPHP ^8.2CI failing Since Jan 5Pushed 1mo agoCompare [ Source](https://github.com/micromus/kafka-bus-commiter)[ Packagist](https://packagist.org/packages/micromus/kafka-bus-commiter)[ Docs](https://github.com/micromus/kafka-bus-commiter)[ GitHub Sponsors](https://github.com/Micromus)[ RSS](/packages/micromus-kafka-bus-commiter/feed)WikiDiscussions 1.x Synced 1w ago READMEChangelog (4)Dependencies (10)Versions (7)Used By (1) Kafka Bus Commiter for PHP ========================== [](#kafka-bus-commiter-for-php) [![Latest Version on Packagist](https://camo.githubusercontent.com/d91c9b96e0607fa8fc90935112018131c11f2c1e8f04b435b7ef7c8b7d381ab5/68747470733a2f2f696d672e736869656c64732e696f2f7061636b61676973742f762f6d6963726f6d75732f6b61666b612d6275732d636f6d6d697465722e7376673f7374796c653d666c61742d737175617265)](https://packagist.org/packages/micromus/kafka-bus-commiter)[![GitHub Tests Action Status](https://camo.githubusercontent.com/71db84137c2f3a5bd7f74b58f48173ac2afad30355ef3f16ad63ecf5370c2ddc/68747470733a2f2f696d672e736869656c64732e696f2f6769746875622f616374696f6e732f776f726b666c6f772f7374617475732f6d6963726f6d75732f6b61666b612d6275732d636f6d6d697465722f72756e2d74657374732e796d6c3f6272616e63683d312e78266c6162656c3d7465737473267374796c653d666c61742d737175617265)](https://github.com/micromus/kafka-bus-commiter/actions?query=workflow%3Arun-tests+branch%3A1.x)[![GitHub Code Style](https://camo.githubusercontent.com/9c61d10e8d94820aced301de56d27e6bc7f4843dfdf1fe9b72b2c7d51b501f16/68747470733a2f2f696d672e736869656c64732e696f2f6769746875622f616374696f6e732f776f726b666c6f772f7374617475732f6d6963726f6d75732f6b61666b612d6275732d636f6d6d697465722f7068702d636f64652d7374796c652e796d6c3f6272616e63683d312e78266c6162656c3d636f64652d7374796c65267374796c653d666c61742d737175617265)](https://github.com/micromus/kafka-bus-commiter/actions?query=workflow%3Acode-style+branch%3A1.x)[![GitHub PHPStan](https://camo.githubusercontent.com/58344aad3312227ed9048a0508c7dd6737afc2524f879ab33cddab844a8eea85/68747470733a2f2f696d672e736869656c64732e696f2f6769746875622f616374696f6e732f776f726b666c6f772f7374617475732f6d6963726f6d75732f6b61666b612d6275732d636f6d6d697465722f7068707374616e2e796d6c3f6272616e63683d312e78266c6162656c3d7068707374616e267374796c653d666c61742d737175617265)](https://github.com/micromus/kafka-bus-commiter/actions?query=workflow%3Aphpstan+branch%3A1.x)[![Total Downloads](https://camo.githubusercontent.com/8a33bb872ea7e99f6d48ec8223ffbe99de822c61a8216f4be04364c884ae7aba/68747470733a2f2f696d672e736869656c64732e696f2f7061636b61676973742f64742f6d6963726f6d75732f6b61666b612d6275732d636f6d6d697465722e7376673f7374796c653d666c61742d737175617265)](https://packagist.org/packages/micromus/kafka-bus-commiter) A middleware package for [micromus/kafka-bus](https://github.com/micromus/kafka-bus) that provides idempotent Kafka message processing. It tracks which messages have already been handled, prevents duplicate processing, and allows limiting the maximum number of read attempts. How It Works ------------ [](#how-it-works) `ConsumerCommiterMiddleware` is inserted into the consumer pipeline and handles four outcomes: 1. **Already committed** — if the message was already successfully processed (`commitedAt` is not `null`), the middleware logs a warning and stops the pipeline. 2. **Max attempts exceeded** — if `maxAttempt` is set and the attempt count has exceeded it, the middleware logs an error and stops the pipeline. 3. **Successful processing** — if both checks pass, the message continues down the pipeline; once handled, `commit()` is called to record it as processed. 4. **Handler error** — if downstream processing throws an exception, the middleware calls `failed()` and rethrows the exception. Installation ------------ [](#installation) ``` composer require micromus/kafka-bus-commiter ``` Usage ----- [](#usage) ### Basic Example [](#basic-example) Implement `RepositorySourceInterface` to persist message state (for example in a database or Redis), then pass the middleware into your worker options: ``` use Micromus\KafkaBus\Bus; use Micromus\KafkaBus\Consumers\Router\ConsumerRoutesBuilder; use Micromus\KafkaBus\Consumers\Router\RouteInfo; use Micromus\KafkaBus\Topics\Topic; use Micromus\KafkaBus\Topics\TopicRegistry; use Micromus\KafkaBusCommiter\Middleware\ConsumerCommiterMiddleware; $topicRegistry = (new TopicRegistry()) ->add(new Topic('production.fact.products.1', 'products')); $consumerRoutes = ConsumerRoutesBuilder::make($topicRegistry) ->add(new RouteInfo('products', new YourMessageHandler())) ->build(); $workerRegistry = (new Bus\Listeners\Workers\MemoryWorkerRegistry()) ->add( new Bus\Listeners\Workers\Worker( name: 'default-listener', routes: $consumerRoutes, options: new Bus\Listeners\Workers\Options( middleware: [ new ConsumerCommiterMiddleware(new NativeMessageRepository(new DatabaseRepositorySource)) ] ) ) ); ``` ### Middleware Options [](#middleware-options) ``` new ConsumerCommiterMiddleware( repository: $repository, // required logger: $logger, // PSR-3 logger, defaults to NullLogger maxAttempt: 3, // max attempts, -1 = unlimited ) ``` ParameterTypeDefaultDescription`repository``RepositorySourceInterface`—Storage for message state`logger``LoggerInterface``NullLogger`PSR-3 compatible logger`maxAttempt``int``-1`Maximum number of processing attempts. `-1` means unlimited### Implementing the Repository [](#implementing-the-repository) You need to provide your own implementation of `RepositorySourceInterface`: ``` use Micromus\KafkaBusCommiter\Attempt; use Micromus\KafkaBusCommiter\Interfaces\RepositorySourceInterface; class DatabaseRepositorySource implements RepositorySourceInterface { /** * Returns the current attempt for a given key. */ public function get(string $key): ?Attempt { // ... } /** * Increments the number of failed read attempts for a key. */ public function increment(string $key): void { // ... } /** * Marks the message key as successfully processed. */ public function commit(string $key): void { // ... } } ``` The middleware reads/writes processing state through `RepositorySourceInterface`. If you need key derivation from message data, use `IdempotencyMessageRepository` as an adapter that maps `ConsumerMessageInterface` to repository keys. Idempotency Keys ---------------- [](#idempotency-keys) Out of the box the consumer uses Kafka's `msgId()` (a combination of topic, partition and offset) as the storage key. That works as long as the same physical message is never replayed under a different offset. As soon as you have retries, producer-side resends, or cross-cluster mirroring, the same logical event can arrive with a different `msgId()` and slip past the duplicate check. An **idempotency key** is a stable identifier that the producer attaches to a message so the consumer can recognize duplicates regardless of where or how they arrive. This package transports the key through the `x-idempotency-key` Kafka header (`IdempotencyMessageRepository::HEADER_NAME`). ### Producing: `HasIdempotency` + `ProducerIdempotencyMiddleware` [](#producing-hasidempotency--produceridempotencymiddleware) On the producer side you mark your message class with the `HasIdempotency` interface and return the stable key. The `ProducerIdempotencyMiddleware` reads that key and writes it into the `x-idempotency-key` header before the message hits the broker. ``` use Micromus\KafkaBus\Interfaces\Producers\Messages\ProducerMessageInterface; use Micromus\KafkaBusCommiter\Interfaces\HasIdempotency; final readonly class ProductCreated implements ProducerMessageInterface, HasIdempotency { public function __construct( private string $productId, private string $payload, ) { } public function toPayload(): string { return $this->payload; } public function getIdempotencyKey(): string { return $this->productId; } } ``` Wire the middleware into the publisher route for that message class: ``` use Micromus\KafkaBus\Bus\Publishers\Router\Options; use Micromus\KafkaBus\Bus\Publishers\Router\PublisherRoutesBuilder; use Micromus\KafkaBusCommiter\Middleware\ProducerIdempotencyMiddleware; $publisherRoutes = PublisherRoutesBuilder::make($topicRegistry) ->add( ProductCreated::class, 'products', new Options(middleware: [new ProducerIdempotencyMiddleware()]) ) ->build(); ``` Middleware is registered per publisher route, so you opt individual message classes into the header. Messages that do not implement `HasIdempotency` pass through the middleware untouched — no header is added. ### Consuming: `IdempotencyMessageRepository` [](#consuming-idempotencymessagerepository) On the consumer side, plug `IdempotencyMessageRepository` into `ConsumerCommiterMiddleware`. It reads the `x-idempotency-key` header and builds the storage key as `"{header}-{topicName}"`, so the same idempotency key in two different topics is still treated as two distinct events. If the header is missing, it falls back to `msgId()` so legacy producers keep working. ``` use Micromus\KafkaBusCommiter\Middleware\ConsumerCommiterMiddleware; use Micromus\KafkaBusCommiter\Repositories\IdempotencyMessageRepository; $repository = new IdempotencyMessageRepository(new DatabaseRepositorySource()); new ConsumerCommiterMiddleware($repository, maxAttempt: 3); ``` ### Picking a key [](#picking-a-key) Pick something that uniquely identifies the **business event**, not the transport. Good choices are an aggregate id plus a version (`order-42-v3`), an outbox row id, or any value the upstream system already treats as unique. Avoid values that change on retry (timestamps, random UUIDs generated per send attempt) — they defeat the whole mechanism. Testing ------- [](#testing) ``` composer test ``` Changelog --------- [](#changelog) Please see [CHANGELOG](CHANGELOG.md) for more information on what has changed recently. Contributing ------------ [](#contributing) Please see [CONTRIBUTING](CONTRIBUTING.md) for details. Security Vulnerabilities ------------------------ [](#security-vulnerabilities) Please review [our security policy](../../security/policy) on how to report security vulnerabilities. Credits ------- [](#credits) - [Kirill Popkov](https://github.com/popkovkirill) - [All Contributors](../../contributors) License ------- [](#license) The MIT License (MIT). Please see [License File](LICENSE.md) for more information. ### Health Score 44 — FairBetter than 90% of packages Maintenance90 Actively maintained with recent releases Popularity12 Limited adoption so far Community13 Small or concentrated contributor base Maturity52 Maturing project, gaining track record Bus Factor2 2 contributors hold 50%+ of commits How is this calculated?**Maintenance (25%)** — Last commit recency, latest release date, and issue-to-star ratio. Uses a 2-year decay window. **Popularity (30%)** — Total and monthly downloads, GitHub stars, and forks. Logarithmic scaling prevents top-heavy scores. **Community (15%)** — Contributors, dependents, forks, watchers, and maintainers. Measures real ecosystem engagement. **Maturity (30%)** — Project age, version count, PHP version support, and release stability. ### Release Activity Cadence Every ~2 days Total 5 Last Release 50d ago ### Community Maintainers ![](https://www.gravatar.com/avatar/920047b57e71b2959cf0934a7d6e45a3a18b639a3bc297cab6e0f0ccb488d535?d=identicon)[kEERill](/maintainers/kEERill) --- Top Contributors [![popkovkirill](https://avatars.githubusercontent.com/u/6718174?v=4)](https://github.com/popkovkirill "popkovkirill (11 commits)")[![keerill](https://avatars.githubusercontent.com/u/74487982?v=4)](https://github.com/keerill "keerill (6 commits)")[![dependabot[bot]](https://avatars.githubusercontent.com/in/29110?v=4)](https://github.com/dependabot[bot] "dependabot[bot] (5 commits)")[![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?v=4)](https://github.com/github-actions[bot] "github-actions[bot] (3 commits)") --- Tags inboxmicromuskafka-bus-commitercommiter ### Code Quality Static AnalysisPHPStan Code StylePHP CS Fixer Type Coverage Yes ### Embed Badge ![Health badge](/badges/micromus-kafka-bus-commiter/health.svg) ``` [![Health](https://phpackages.com/badges/micromus-kafka-bus-commiter/health.svg)](https://phpackages.com/packages/micromus-kafka-bus-commiter) ``` ### Alternatives [multicaret/laravel-inbox Laravel messages and inbox system 301.0k](/packages/multicaret-laravel-inbox)[qalainau/filament-inbox Email-like inbox messaging plugin for Filament v5 — threads, starring, trash, forwarding, read receipts, and multi-tenancy out of the box. 171.5k](/packages/qalainau-filament-inbox)[shipsaas/laravel-inbox-process Inbox pattern process implementation for your Laravel Applications 132.5k](/packages/shipsaas-laravel-inbox-process) PHPackages © 2026 [Directory](/)[Categories](/categories)[Trending](/trending)[Changelog](/changelog)[Analyze](/analyze)