Call worker_fn for each batch, instead of waiting for max_messages.

If max_messages is 0 (the default), messages accumulate in memory and are never processed.

---

Migrated from D2205 (view on Phabricator)