Zeebe worker stops processing after instances processed = max_task_count

[Andy-JB]

Describe the bug
When the number of instances waiting in Zeebe is greater than max_task_count, starting a Zeebe worker will process max_task_count instances before ceasing to process.

To Reproduce
Steps to reproduce the behavior:

1. Start Zeebe (with no worker running)
2. Create a number of instances > max_task_count
3. Start Zeebe worker
4. max_task_count number of instances will be processed by Zeebe worker.
5. Zeebe worker does not process any more instances. It will need to be restarted to continue processing

Expected behavior
Zeebe worker continues processing all instances in Zeebe.

Screenshots

Before starting Zeebe worker with max_task_count=32:

After starting Zeebe worker:

Desktop (please complete the following information):

• OS: iOS 11.4
• Browser: Chrome
• Version: Zeebe 1.1.0, pyzeebe 3.0.0rc2

Extra info
Tested using the worker example from the docs and zeebe-docker-compose

This bug also occurs on Kubernetes v1.11.0

[Andy-JB]

I've tested a fix for this issue in job_poller.py.
If calculate_max_jobs_to_activate() > 0 returns false, the thread will wait 5 seconds before checking again.

The wait time could also be an argument of ZeebeWorker.

Hi @JonatanMartens, what do you think? Would you like me to submit a pull request?

    async def poll(self):
        while self.should_poll():
            while self.calculate_max_jobs_to_activate() > 0:
                await self.poll_once()
            await asyncio.sleep(5)
...

    def should_poll(self) -> bool:
        return not self.stop_event.is_set() \
            and (self.zeebe_adapter.connected or self.zeebe_adapter.retrying_connection) \
            and self.calculate_max_jobs_to_activate() > 0

[JonatanMartens]

Hey @Andy-JB,

I can't seem to replicate the issue on my machine, can you provide how you use the worker?

It would be really helpful if you could submit a pr, if you can also add a test so that we know it was fixed.

Thank you for opening this issue!

[tomas-zemres]

We are testing version 3.0.0rc3 and we also found this bug.

It can be simply reproduced by following scenario:

• Implement worker with simple task with asyncio.sleep(1)
• stop the worker
• run 100 new processes in Zeebe
  (in Zeebe is 100 pending workflows now)
• Start the worker
  • the worker takes 32 tasks and runs 32 x the sleep(1) in parallel, but after that the "self.should_poll()" returns False, because
    self.calculate_max_jobs_to_activate() returns 0. When the self.should_poll() returns False the polling jobs is stopped and following jobs are never processed (until worker is manually restarted)

[maio]

Hello,

basically whenever calculate_max_jobs_to_activate() returns 0 (e.g. when worker is currently processing >32 jobs) JobPoller will stop polling for new jobs because it will return from poll(self) method.

[lesnek]

We encountered the same issue in our project when traffic became 100 processes/second (incident), when we have max task count 32

Also I like more the "tmp fix" mentioned here, then just waiting

[maio]

This issue is fixed for us in pre-release/3.0.0 thanks to merge above. It's maybe OK now to close this? cc: @lesnek