Loss messages from Priority Queue

[Buravenkov]

Hello, we have encountered the following issue. When redeploying the application, the old k8s pod stops working, and a new one is deployed. During this process, we lose messages from the priority queue. To process the queues, we use the jobs plugin with the Kafka driver configured with the following settings.

kafka:
  brokers: "${KAFKA_BROKER}"

jobs:
  num_pollers: 6
  timeout: 10
  pipeline_size: 3000
  pool:
    num_workers: 5
    allocate_timeout: 30s
    supervisor:
      max_worker_memory: 128
      watch_tick: 30s
  pipelines:
    topic1:
      driver: kafka
      config:
        priority: 1
        prefetch: 100
        consumer_options:
          topics: ["topic1"]
          consume_regexp: false
        group_options:
          group_id: php_kafka_consumer
    topic2:
      driver: kafka
      config:
        priority: 1
        prefetch: 500
        consumer_options:
          topics: [ "topic2" ]
          consume_regexp: false
        group_options:
          group_id: php_kafka_consumer
    topic3:
      driver: kafka
      config:
        priority: 1
        prefetch: 500
        consumer_options:
          topics: [ "topic3" ]
          consume_regexp: false
        group_options:
          group_id: php_kafka_consumer
    topic4:
      driver: kafka
      config:
        priority: 1
        prefetch: 500
        consumer_options:
          topics: [ "topic4" ]
          consume_regexp: false
        group_options:
          group_id: php_kafka_consumer
    topic5:
      driver: kafka
      config:
        priority: 1
        prefetch: 100
        consumer_options:
          topics: [ "topic5" ]
          consume_regexp: false
        group_options:
          group_id: php_kafka_consumer
  consume:
    - topic1
    - topic2
    - topic3
    - topic4
    - topic5

As far as I understand, when messages enter the PQ, the offsets in the topics are immediately shifted. Could you please tell me if it’s possible to set enable.auto.commit = false and commit the offset only after the message has been processed by the worker? Or at least after the message has been loaded into the worker's memory?

Php worker execute following code:

        try {
            while ($task = $this->consumer->waitTask()) {
                try {
                    $handler = $this->getHandler($task->getQueue());
                    $handler->handle($task);
                } catch (Throwable $e) {
                    $this->logger->error(
                        "Task processing error: {$e->getMessage()}",
                        ['payload' => $task->getPayload()]
                    );
                    $task->fail($e->getMessage());
                    continue;
                }
                $task->complete();
            }
        } catch (Throwable $e) {
            $this->logger->error('Consumer error:', ['message' => $e->getMessage()]);
        }

And we use a 2023.1.1 version of roadrunner app and this "spiral/roadrunner-jobs": "^3.0" composer-package. Please let me know if you need any additional information.

[rustatian]

Hey @Buravenkov 👋🏻
RR commits records only after the message is processed: code, code
Kafka lacks a per-message commit feature, like RMQ, for example. Thus, a message with offset 3 (for example) might be committed while 1 and 2 are still in process. But, also, try to update RR to the latest version. v2023 is too old.

[rustatian]

You may also set jobs.pipeline_size (size of the PQ) to a minimal value.

[Buravenkov]

@rustatian Can I ask one more question – for upgrade to version 2024.3.2 of rr, do we also need to update the version of the package "spiral/roadrunner-jobs": "^3.0"?

[rustatian]

You may ask as many questions as you have 😄
I think this is not required, but highly advised because of bug fixes. There're no breaking changes in RR and PHP client packages.

[Buravenkov]

Thank you for your responses ❤️
We will try the solutions, and I will return either to close the discussion or for clarifying information.

[rustatian]

Sure, good luck ❤️

[Buravenkov]

@rustatian Hello!

We have upgraded RoadRunner to version v2024.3.2, but we still noticed that some messages were lost. At the same time, we reduced the size of the priority queue to 100 messages, and during a redeployment, we saw that messages from this queue were read within 3 seconds. However, 67 messages were still lost.

Because of this, we have a question: do we need to explicitly handle graceful shutdown signals in the PHP code of the workers and implement some logic that prevents the worker from terminating until it has completed all the tasks that the pollers loaded into its memory?

I couldn't find any information in the documentation that confirms or refutes the claim that a jobs worker will complete all tasks loaded into its memory when RoadRunner receives a shutdown signal.

P.S. We are sending a SIGTERM signal to stop it.

[Buravenkov]

Do you know a way to guarantee that all tasks in the worker's memory will be completed before the worker is terminated?

[rustatian]

Then it turns out that whatever was loaded into the worker's memory will be lost upon redeployment for the consumer group since the offset will be moved at the moment RoadRunner reads messages from the topics?

No, I mentioned this initially. RR does not use auto commit, but auto commit marks. Every successfully (ack-ed) processed message from the PHP worker then market as processed.

the offset will be moved at the moment RoadRunner reads messages from the topics?

RR does not use offset commits as I mentioned before with links to the actual code. RR marks records after this call from PHP $task->complete();.

Do you know a way to guarantee that all tasks in the worker's memory will be completed before the worker is terminated?

There are no tasks in the worker's memory. All tasks are in the RR's priority queue. You may reduce number of pre-fetched messages, I see that you use 100 and 500 messages to prefetch.
It is hard to say what is happening in your case, since there is no info about the broker configuration.

---

But, if you still have a problem, please, create a ticket with a broker's configuration, simple PHP worker (w/o you code), RR configuration (minimal), and steps to reproduce the problem.

[Buravenkov]

Hi @rustatian, we found the reason why the offset was shifting during redeployment. Since RoadRunner assigns tasks to workers in a random order, in a situation where there are 10 messages in the topic, it may assign a message with an offset of 10 to a worker first. The worker processes it, and after calling complete, RoadRunner commits this offset. Then, redeployment occurs, and if the worker doesn't manage to process everything it had in memory, it will be lost. This means that messages with offsets before 10 may remain unprocessed. Moreover, during the next service launch, these messages will no longer be included in the processing.
Switching to version rr 2024.3.2 helped us, and specifying a smaller value in pipeline_size than the total number of workers in prefetch resolved the issue. This way, messages are processed before the microservice redeployment.
However, could you please advise if there is any way to make rr-jobs with the Kafka driver receive messages in order?

[rustatian]

Hey @Buravenkov 👋🏻

we found the reason why the offset was shifting during redeployment. Since RoadRunner assigns tasks to workers in a random order,

I was told about that literally in the first message 😃

Kafka lacks a per-message commit feature, like RMQ, for example. Thus, a message with offset 3 (for example) might be committed while 1 and 2 are still in process. But, also, try to update RR to the latest version. v2023 is too old.

---

However, could you please advise if there is any way to make rr-jobs with the Kafka driver receive messages in order?

Unfortunately, you can't achieve that with RR. RR sends messages to the first free workers in the stack. So only having 1 worker you may receive them in order. However, there is no limitation using rdkafka PHP extension for example.

[Buravenkov]

@rustatian Yes, I somehow missed that, thank you for highlighting it again.
We are planning to use the service plugin with rdkafka instead of jobs. I think the thread can be closed. Thanks for the help!

[Buravenkov]

So only having 1 worker you may receive them in order

One more question. You say that with one worker, messages will be processed sequentially, but I noticed that in the worker, messages still arrive in random order. I created a topic, pushed 10 messages into it {"a": 1} ... {"a": 10}, and noticed that the output in the worker showed that the first message taken was {"a": 10}. I don’t understand why this happens. Is it better to create an issue about this, or can you answer why this happens without additional information?
The settings are the same: only 1 pipeline, 1 worker, and pipeline_size = 10.