Manual Commit | Laravel Kafka [ K Laravel Kafka ](/) [Docs](/docs) [Blog](https://laravelkafka.com/blog) [GitHub](https://github.com/mateusjunges/laravel-kafka) Search ⌘K [ Login with GitHub Login ](https://laravelkafka.com/oauth/github/redirect) Docs for version selected v3.0 v2.13 v2.12 v2.11 v2.10 v2.9 v2.8 v1.13 - - [ Introduction ](/docs/v3.0/introduction) - [ Requirements ](/docs/v3.0/requirements) - [ Installation and Setup ](/docs/v3.0/installation-and-setup) - [ Questions and issues ](/docs/v3.0/questions-and-issues) - [ Changelog ](/docs/v3.0/changelog) - [ Upgrade guide ](/docs/v3.0/upgrade-guide) - [ Example docker-compose file ](/docs/v3.0/example-docker-compose) - Producing messages ------------------ - [ Producing messages ](/docs/v3.0/producing-messages/producing-messages) - [ Configuring your kafka producer ](/docs/v3.0/producing-messages/configuring-producers) - [ Configuring message payload ](/docs/v3.0/producing-messages/configuring-message-payload) - [ Custom serializers ](/docs/v3.0/producing-messages/custom-serializers) - [ Publishing to kafka ](/docs/v3.0/producing-messages/publishing-to-kafka) - Consuming messages ------------------ - [ Creating a kafka consumer ](/docs/v3.0/consuming-messages/creating-consumer) - [ Subscribing to kafka topics ](/docs/v3.0/consuming-messages/subscribing-to-kafka-topics) - [ Using regex to subscribe to kafka topics ](/docs/v3.0/consuming-messages/using-regex-to-subscribe-to-kafka-topics) - [ Assigning consumers to a topic partition ](/docs/v3.0/consuming-messages/assigning-partitions) - [ Consuming messages from specific offsets ](/docs/v3.0/consuming-messages/consuming-from-specific-offsets) - [ Consumer groups ](/docs/v3.0/consuming-messages/consumer-groups) - [ Partition Discovery and Dynamic Assignment ](/docs/v3.0/consuming-messages/partition-discovery) - [ Message handlers ](/docs/v3.0/consuming-messages/message-handlers) - [ Configuring consumer options ](/docs/v3.0/consuming-messages/configuring-consumer-options) - [ Handling failed messages ](/docs/v3.0/consuming-messages/handling-failed-messages) - [ Custom deserializers ](/docs/v3.0/consuming-messages/custom-deserializers) - [ Consuming messages ](/docs/v3.0/consuming-messages/consuming-messages) - [ Pausing partitions ](/docs/v3.0/consuming-messages/pausing-partitions) - [ Consumer classes ](/docs/v3.0/consuming-messages/class-structure) - [ Using consumers with Laravel Telescope ](/docs/v3.0/consuming-messages/laravel-telescope) - Advanced usage -------------- - [ Connections ](/docs/v3.0/advanced-usage/connections) - [ Replacing the default serializer/deserializer ](/docs/v3.0/advanced-usage/replacing-default-serializer) - [ Graceful shutdown ](/docs/v3.0/advanced-usage/graceful-shutdown) - [ Running consumers in production ](/docs/v3.0/advanced-usage/running-consumers-in-production) - [ SASL Authentication ](/docs/v3.0/advanced-usage/sasl-authentication) - [ Custom Committers ](/docs/v3.0/advanced-usage/custom-committers) - [ Manual Commit ](/docs/v3.0/advanced-usage/manual-commit) - [ Middlewares ](/docs/v3.0/advanced-usage/middlewares) - [ Stop consumer when there are no messages left ](/docs/v3.0/advanced-usage/stop-consumer-after-last-message) - [ Stop consumer on demand ](/docs/v3.0/advanced-usage/stopping-a-consumer) - [ Writing custom loggers ](/docs/v3.0/advanced-usage/custom-loggers) - [ Before and after callbacks ](/docs/v3.0/advanced-usage/before-callbacks) - [ Setting global configurations ](/docs/v3.0/advanced-usage/setting-global-configuration) - [ Sending multiple messages with the same producer ](/docs/v3.0/advanced-usage/sending-multiple-messages-with-the-same-producer) - [ Events ](/docs/v3.0/advanced-usage/events) - [ Consumer lag ](/docs/v3.0/advanced-usage/consumer-lag) - Testing ------- - [ Kafka fake ](/docs/v3.0/testing/fake) - [ Assert Published ](/docs/v3.0/testing/assert-published) - [ Assert published On ](/docs/v3.0/testing/assert-published-on) - [ Assert not published ](/docs/v3.0/testing/assert-not-published) - [ Assert nothing published ](/docs/v3.0/testing/assert-nothing-published) - [ Assert published times ](/docs/v3.0/testing/assert-published-times) - [ Assert published on times ](/docs/v3.0/testing/assert-published-on-times) - [ Mocking your kafka consumer ](/docs/v3.0/testing/mocking-your-kafka-consumer) Manual Commit =============== By default, consumers use auto commit mode. The offset of each message is stored after your handler processes it, and librdkafka commits the stored offsets in the background every `auto.commit.interval.ms`, 5 seconds by default, and when the consumer stops. Failed messages are never committed unless they are sent to a dead letter queue or [ skipped on purpose ](../consuming-messages/handling-failed-messages), so auto commit already gives you at least once delivery. Sponsorship Support Laravel Kafka by sponsoring me! Laravel Kafka is free and Open Source software, built to empower developers like you. Your support helps maintain and enhance the project. If you find it valuable, please consider sponsoring me on GitHub. Every contribution makes a difference and keeps the development going strong! Thank you! [ Become a Sponsor ](https://github.com/sponsors/mateusjunges) Want to hide this message? Sponsor at any tier of $10/month or more! If the consumer process crashes, the messages processed since the last background commit are consumed again. To make that window shorter, lower the interval: ``` $consumer = Kafka::consumer(['orders']) ->withOption('auto.commit.interval.ms', 1000) ->withHandler($handler); ``` With manual commit, the handler decides when offsets are committed. Use it when auto commit does not fit, for instance to: - Commit a message only after work that completes outside the handler, such as a batch written to a database. - Commit only after a group of messages was processed. - Wait for each commit to be acknowledged before handling the next message. [](#content-enabling-manual-commit "Permalink")Enabling manual commit --------------------------------------------------------------------- Call `withManualCommit()` when creating your consumer, and commit from your handler through the consumer it receives: ``` use Junges\Kafka\Contracts\Consumer; use Junges\Kafka\Contracts\ConsumerMessage; use Junges\Kafka\Facades\Kafka; $consumer = Kafka::consumer(['orders']) ->withManualCommit() ->withHandler(function (ConsumerMessage $message, Consumer $consumer) { processOrder($message->getBody()); $consumer->commit($message); }) ->build(); $consumer->consume(); ``` If `processOrder` throws, the commit is not reached. The message is then handled as failed: it is [ retried ](../consuming-messages/handling-failed-messages) if the consumer retries failed messages, and then sent to the dead letter queue, skipped, or stops the consumer. Always let the exception of a failed message propagate. Catching it without rethrowing makes the consumer move on as if the message was processed, and the next commit of the same partition moves past it, so the message is lost. [](#content-commit-methods "Permalink")Commit methods ----------------------------------------------------- The consumer passed to handlers has two commit methods. `commit()` waits until Kafka acknowledged the commit, and `commitAsync()` returns right away: ``` // Commit the offsets of the current assignment $consumer->commit(); // Commit the offset of a message $consumer->commit($message); // Commit the offsets of specific partitions $consumer->commit([$topicPartition1, $topicPartition2]); // Same, without waiting for Kafka to acknowledge the commit $consumer->commitAsync($message); ``` Both methods accept: - Nothing, to commit the offsets of the current assignment. - A `Junges\Kafka\Contracts\ConsumerMessage` or `RdKafka\Message`, to commit the offset right after that message. - An array of `RdKafka\TopicPartition`, to commit specific offsets. Asynchronous commits don't throw when they fail. Register an [ offset commit callback ](../consuming-messages/configuring-consumer-options#configuration-callbacks) with `onOffsetCommit()` to find out about failures. Messages published while handling a message are flushed before each commit, so they are delivered before the consumed message is committed. [](#content-committing-groups-of-messages "Permalink")Committing groups of messages ----------------------------------------------------------------------------------- Kafka commits offsets per partition, so committing a message also commits every message before it in the same partition. To commit less often, process every message as it arrives and commit every few messages: ``` $handled = 0; $consumer = Kafka::consumer(['page-views']) ->withManualCommit() ->withHandler(function (ConsumerMessage $message, Consumer $consumer) use (&$handled) { recordPageView($message->getBody()); if (++$handled % 100 === 0) { $consumer->commitAsync(); } }) ->onStopConsuming(function () use (&$consumer) { // Commit the messages handled since the last commit $consumer->commit(); }) ->build(); $consumer->consume(); ``` Calling `commit()` or `commitAsync()` without arguments commits the offsets of every assigned partition, up to the last message the consumer received, not only the partition of the last message. With manual commit, nothing is committed when the consumer stops unless you do it, so the example commits in the `onStopConsuming` callback. If the consumer crashes, the messages handled since the last commit are consumed again. [](#content-dead-letter-queues "Permalink")Dead letter queues ------------------------------------------------------------- Manual commit works with [ dead letter queues ](../consuming-messages/configuring-consumer-options#configuring-a-dead-letter-queue). A message whose handler throws is sent to the dead letter queue and committed, so the consumer can move on: ``` $consumer = Kafka::consumer(['orders']) ->withManualCommit() ->retryFailedMessages(3, backoffInMs: 1000) ->withDlq('orders-dlq') ->withHandler(function (ConsumerMessage $message, Consumer $consumer) { processOrder($message->getBody()); $consumer->commit($message); }); ``` [](#content-troubleshooting "Permalink")Troubleshooting ------------------------------------------------------- **Messages are consumed again after a restart:** the handler did not commit them before the consumer stopped. Make sure every code path that finishes processing a message commits it, or commits a later message of the same partition. **Failed messages are not consumed again:** the handler caught the exception instead of letting it propagate, and a later commit moved past the message. **Commits are slow:** use `commitAsync()`, or commit groups of messages instead of every message. Previous [ Custom Committers ](https://laravelkafka.com/docs/v3.0/advanced-usage/custom-committers) Next [ Middlewares ](https://laravelkafka.com/docs/v3.0/advanced-usage/middlewares) Sponsors [ version="1.0" encoding="UTF-8"? EasyCal ](https://easycal.app/) [ Search ⌘ K ](https://typesense.org/)