Running consumers in production | 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) Running consumers in production ================================= Consumers are long running processes. In production, they need a process monitor that keeps them running, restarts them when they exit, and stops them gracefully during deployments. 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-supervisor "Permalink")Supervisor --------------------------------------------- Supervisor is a process monitor commonly used in Linux environments. To install it on Ubuntu, you may use the following command: ``` sudo apt-get install supervisor ``` On macOS, you can use Homebrew: ``` brew install supervisor ``` Supervisor configuration files are typically stored in the `/etc/supervisor/conf.d` directory. Within this directory, you may create any number of configuration files that instruct Supervisor how your processes should be monitored. For example, let's create an `orders-consumer.conf` file that starts and monitors a [ consumer class ](../consuming-messages/class-structure): ``` [program:orders-consumer] directory=/var/www/html process_name=%(program_name)s_%(process_num)02d command=php artisan kafka:consume OrdersConsumer numprocs=3 autostart=true autorestart=true startsecs=1 startretries=10 stopwaitsecs=60 redirect_stderr=true stdout_logfile=/var/log/supervisor/orders-consumer.log ``` Once the configuration file has been created, update the Supervisor configuration and start the processes: ``` sudo supervisorctl reread sudo supervisorctl update sudo supervisorctl start orders-consumer:* ``` ### [](#content-how-many-processes-to-run "Permalink")How many processes to run Consumers of the same group share the partitions of the topics they consume, and each partition is consumed by a single consumer of the group. Running more consumer processes than partitions leaves the extra processes idle, so `numprocs` should not exceed the number of partitions. ### [](#content-stopping-gracefully "Permalink")Stopping gracefully When Supervisor stops a process, it sends a `SIGTERM` signal, and the consumer finishes the message it is handling before exiting, see [ graceful shutdown ](graceful-shutdown). Supervisor waits `stopwaitsecs` seconds before killing the process, so set it higher than the longest time your handler may take to process a message, including the [ retries ](../consuming-messages/handling-failed-messages#retrying-failed-messages) of a failed message and the wait between them. ### [](#content-restarting-after-failures "Permalink")Restarting after failures When a message fails and there is no dead letter queue, the consumer stops and exits with an error, so the message is consumed again once the process is restarted. Supervisor considers a process that exits within `startsecs` seconds of starting as a failed start, and gives up on it after `startretries` failed starts. Keep `startsecs` low, so a consumer that stops on a failed message is restarted, and monitor the logs, since a message that always fails stops the consumer every time it is consumed. [](#content-restarting-consumers-after-deployments "Permalink")Restarting consumers after deployments ----------------------------------------------------------------------------------------------------- Consumers load your application code once, when they start, so they must be restarted after deploying new code. The `kafka:restart-consumers` command asks every running consumer to stop gracefully once it finishes the message it is handling, and your process monitor starts them again: ``` php artisan kafka:restart-consumers ``` To restart only some consumers, pass their names. [ Consumer classes ](../consuming-messages/class-structure#naming-consumers) are named after their class by default, and can be given by their name in the `App\Kafka\Consumers` namespace, like with the `kafka:consume` command: ``` php artisan kafka:restart-consumers OrdersConsumer PaymentsConsumer ``` The command stores the restart time in the cache, and consumers check it every second. All consumers must share the cache store defined by the `cache_driver` key of the `config/kafka.php` file, which defaults to your application's cache store, so it can't be a store local to each server, such as `file` or `array`, when consumers run on several servers. [](#content-limiting-memory-usage "Permalink")Limiting memory usage ------------------------------------------------------------------- Long running PHP processes may slowly accumulate memory. To keep it in check, make consumers stop after a number of messages or an amount of time, and let the process monitor start them again: ``` php artisan kafka:consume OrdersConsumer --max-messages=10000 --max-time=3600 ``` [](#content-producers-in-long-running-processes "Permalink")Producers in long running processes ----------------------------------------------------------------------------------------------- Messages published with `Kafka::publish()` are queued and delivered in the background. They are flushed when the application terminates, but long running processes don't terminate after each unit of work: - **Consumers** flush the messages published while handling a message before its offset is stored or committed. - **Queue workers** flush them after each job. - **Laravel Octane** flushes them when each request terminates. The Kafka manager is resolved again on every request, which creates a new producer, and its broker connections, for each request. To reuse the same producer for every request handled by a worker, add `Junges\Kafka\Factory::class` to the `warm` array of your `config/octane.php` file. - **Other long running commands** should call `Kafka::flush()` whenever they need their messages to be delivered, as they are only flushed when the command exits. Previous [ Graceful shutdown ](https://laravelkafka.com/docs/v3.0/advanced-usage/graceful-shutdown) Next [ SASL Authentication ](https://laravelkafka.com/docs/v3.0/advanced-usage/sasl-authentication) Sponsors [ version="1.0" encoding="UTF-8"? EasyCal ](https://easycal.app/) [ Search ⌘ K ](https://typesense.org/)