fix(run_task_in_threads): Commit offsets even when consumer is idle (#…

…276)

arroyo/processing/strategies/run_task_in_threads.py

@@ -85,6 +85,15 @@ def submit(
 
     def poll(self) -> None:
         self.__forward_invalid_offsets()
+
+        if not self.__queue:
+            # specifically for the case where we are idling, poll the next step
+            # so that committing can occur. if there is stuff in the queue and
+            # we are waiting for a future to be finished, we do not really need
+            # to forward polls.
+            self.__next_step.poll()
+            return
+
         while self.__queue:
             message, future = self.__queue[0]
             next_message: Message[TResult]

tests/processing/strategies/test_all.py

@@ -289,3 +289,22 @@ def test_join(strategy_factory: StrategyFactory) -> None:
     assert next_step.close.call_args_list == [call()]
 
     assert next_step.join.call_args_list == [call(timeout=None)]
+
+
+@pytest.mark.parametrize("strategy_factory", FACTORIES)
+def test_poll_next_step(
+    request: pytest.FixtureRequest, strategy_factory: StrategyFactory
+) -> None:
+    next_step = Mock()
+
+    step = strategy_factory(next_step)
+    request.addfinalizer(step.terminate)
+
+    # Ensure that polling a strategy forwards the poll unconditionally even if
+    # there are no messages to process, or no progress at all. Otherwise there
+    # are weird effects where all messages on a test topic (such as in
+    # benchmarking/QE) have been processed but the last batch of offsets never
+    # gets committed.
+    step.poll()
+
+    assert next_step.poll.call_args_list == [call()]