Handling failed messages | 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/v2.12/introduction) - [ Requirements ](/docs/v2.12/requirements) - [ Installation and Setup ](/docs/v2.12/installation-and-setup) - [ Questions and issues ](/docs/v2.12/questions-and-issues) - [ Changelog ](/docs/v2.12/changelog) - [ Upgrade guide ](/docs/v2.12/upgrade-guide) - [ Example docker-compose file ](/docs/v2.12/example-docker-compose) - Producing messages ------------------ - [ Producing messages ](/docs/v2.12/producing-messages/producing-messages) - [ Configuring your kafka producer ](/docs/v2.12/producing-messages/configuring-producers) - [ Configuring message payload ](/docs/v2.12/producing-messages/configuring-message-payload) - [ Custom serializers ](/docs/v2.12/producing-messages/custom-serializers) - [ Publishing to kafka ](/docs/v2.12/producing-messages/publishing-to-kafka) - Consuming messages ------------------ - [ Creating a kafka consumer ](/docs/v2.12/consuming-messages/creating-consumer) - [ Subscribing to kafka topics ](/docs/v2.12/consuming-messages/subscribing-to-kafka-topics) - [ Using regex to subscribe to kafka topics ](/docs/v2.12/consuming-messages/using-regex-to-subscribe-to-kafka-topics) - [ Assigning consumers to a topic partition ](/docs/v2.12/consuming-messages/assigning-partitions) - [ Consuming messages from specific offsets ](/docs/v2.12/consuming-messages/consuming-from-specific-offsets) - [ Consumer groups ](/docs/v2.12/consuming-messages/consumer-groups) - [ Partition Discovery and Dynamic Assignment ](/docs/v2.12/consuming-messages/partition-discovery) - [ Message handlers ](/docs/v2.12/consuming-messages/message-handlers) - [ Configuring consumer options ](/docs/v2.12/consuming-messages/configuring-consumer-options) - [ Handling failed messages ](/docs/v2.12/consuming-messages/handling-failed-messages) - [ Custom deserializers ](/docs/v2.12/consuming-messages/custom-deserializers) - [ Consuming messages ](/docs/v2.12/consuming-messages/consuming-messages) - [ Class structure ](/docs/v2.12/consuming-messages/class-structure) - [ Queueable handlers ](/docs/v2.12/consuming-messages/queueable-handlers) - Advanced usage -------------- - [ Replacing the default serializer/deserializer ](/docs/v2.12/advanced-usage/replacing-default-serializer) - [ Graceful shutdown ](/docs/v2.12/advanced-usage/graceful-shutdown) - [ SASL Authentication ](/docs/v2.12/advanced-usage/sasl-authentication) - [ Custom Committers ](/docs/v2.12/advanced-usage/custom-committers) - [ Manual Commit ](/docs/v2.12/advanced-usage/manual-commit) - [ Middlewares ](/docs/v2.12/advanced-usage/middlewares) - [ Stop consumer after last messages ](/docs/v2.12/advanced-usage/stop-consumer-after-last-message) - [ Stop consumer on demand ](/docs/v2.12/advanced-usage/stopping-a-consumer) - [ Writing custom loggers ](/docs/v2.12/advanced-usage/custom-loggers) - [ Before and after callbacks ](/docs/v2.12/advanced-usage/before-callbacks) - [ Setting global configurations ](/docs/v2.12/advanced-usage/setting-global-configuration) - [ Sending multiple messages with the same producer ](/docs/v2.12/advanced-usage/sending-multiple-messages-with-the-same-producer) - Testing ------- - [ Kafka fake ](/docs/v2.12/testing/fake) - [ Assert Published ](/docs/v2.12/testing/assert-published) - [ Assert published On ](/docs/v2.12/testing/assert-published-on) - [ Assert nothing published ](/docs/v2.12/testing/assert-nothing-published) - [ Assert published times ](/docs/v2.12/testing/assert-published-times) - [ Assert published on times ](/docs/v2.12/testing/assert-published-on-times) - [ Mocking your kafka consumer ](/docs/v2.12/testing/mocking-your-kafka-consumer) Handling failed messages ========================== When a message handler throws an exception, the consumer first calls the `failed` method of the consumer class. By default, it rethrows the exception, which is then logged and reported through the Laravel exception handler. What happens to the message next depends on how the consumer is configured. 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! [](#content-default-behavior "Permalink")Default behavior --------------------------------------------------------- Without a dead letter queue, the consumer moves on to the next message and the offset of the failed message is committed. **The failed message is not consumed again**, so the default behavior is at most once delivery for messages whose handler fails. This is not limited to auto commit mode. Kafka does not acknowledge messages individually: a committed offset is the position a consumer group resumes from in a partition. Skipping the commit of a failed message is not enough to consume it again, because committing any later message of the same partition moves that position past it. With auto commit enabled, librdkafka also commits the offset of every fetched message in the background, whether the handler succeeded or not. Failed messages can be retried before they are handled as failed. After that, there are two ways to keep them from being lost: sending them to a dead letter queue, or stopping the consumer. [](#content-retrying-failed-messages "Permalink")Retrying failed messages ------------------------------------------------------------------------- Failures caused by a temporary problem, such as a dependency that is briefly unavailable, can be retried with the `retryFailedMessages` method. It receives the number of retries and, optionally, the time to wait before each retry in milliseconds: ``` $consumer = \Junges\Kafka\Facades\Kafka::consumer(['orders']) ->withConsumerGroupId('orders-group') ->retryFailedMessages(3, backoffInMs: 1000) ->withDlq() ->withHandler(new OrderHandler) ->build(); ``` When the handler throws an exception, it is called again with the same message, up to the given number of times. Middlewares run again on every attempt. Once all retries are used, the message is handled as failed: the `failed` method of the consumer class is called, and the message is sent to the dead letter queue, stops the consumer, or is skipped, depending on the configuration. Retries also end early when the consumer is asked to stop, for instance by a termination signal. The consumer waits during the backoff, so no other message is consumed while a message is being retried. Keep the total time spent retrying a message (the number of retries multiplied by the backoff, plus the time the handler takes) well below the `max.poll.interval.ms` consumer option, 5 minutes by default. A consumer that does not poll Kafka within that interval is removed from the consumer group. Longer outages are better handled by a dead letter queue or by stopping the consumer. The `SeekToCurrentErrorCommitter` committer is deprecated in favor of this method, as it does not make failed messages be consumed again. [](#content-sending-failed-messages-to-a-dead-letter-queue "Permalink")Sending failed messages to a dead letter queue --------------------------------------------------------------------------------------------------------------------- When a dead letter queue is configured with `withDlq`, failed messages are published to the dead letter queue topic before their offsets are committed, and the consumer moves on to the next message. See [ configuring a dead letter queue ](configuring-consumer-options) for details. [](#content-stopping-the-consumer-on-failure "Permalink")Stopping the consumer on failure ----------------------------------------------------------------------------------------- If a failed message must be processed before any later message of the same partition, use the `stopOnFailure` method: ``` $consumer = \Junges\Kafka\Facades\Kafka::consumer(['orders']) ->withConsumerGroupId('orders-group') ->stopOnFailure() ->withHandler(new OrderHandler) ->build(); $consumer->consume(); ``` When a message fails and there is no dead letter queue, the consumer is closed, leaving the consumer group, and throws a `Junges\Kafka\Exceptions\ConsumerException`. The original exception is available through the `getPrevious` method. The offset of the failed message is not committed, so the next consumer of its partition starts from the failed message. This works both in auto commit mode, where the offsets of the messages processed before the failure are committed when the consumer is closed, and in [ manual commit ](../advanced-usage/manual-commit) mode. The consumer process is expected to exit, and to be restarted by a process monitor such as Supervisor. Keep in mind that: - Messages are delivered at least once. A message can be processed again after a restart, so handlers should be idempotent. - A message that always fails stops the consumer every time it is consumed, blocking its partition until the cause is fixed. - Make sure your process monitor keeps restarting the consumer. Supervisor, for instance, considers a process that exits within `startsecs` seconds of starting as a failed start, and gives up after `startretries` failed starts. - When a dead letter queue is also configured, failed messages are sent to it and the consumer does not stop. Stopping only happens for failures that can not be sent anywhere else. With auto commit enabled, both `stopOnFailure` and `retryFailedMessages` set the `enable.auto.offset.store` option to `false`, and the consumer stores the offset of each message only after it is processed. This is what keeps librdkafka from committing the offset of a failed message in the background. Messages consumed by [ queueable handlers ](queueable-handlers) are processed by the queue worker, so their failures are handled by the queue instead. Previous [ Configuring consumer options ](https://laravelkafka.com/docs/v2.12/consuming-messages/configuring-consumer-options) Next [ Custom deserializers ](https://laravelkafka.com/docs/v2.12/consuming-messages/custom-deserializers) Sponsors [ version="1.0" encoding="UTF-8"? EasyCal ](https://easycal.app/) [ Search ⌘ K ](https://typesense.org/)