Fixed filter not acking filtered out messages. (#208)

* Fixed filter not acking filtered out messages.

* Removed debug print from test.

* Added Cython implementation for the filter fix.

Co-authored-by: Matthew Drago <matthew.drago@gig.com>
Co-authored-by: Vikram Patki <54442035+patkivikram@users.noreply.github.com>
Co-authored-by: Taybin Rutkin <taybin@users.noreply.github.com>

Author: 4 people

faust/_cython/streams.pyx

@@ -109,7 +109,8 @@ cdef class StreamIterator:
             object consumer
         consumer = self.consumer
         last_stream_to_ack = False
-        if do_ack and event is not None:
+        # if do_ack and event is not None:
+        if event is not None and (do_ack or event.value is self._skipped_value):
             message = event.message
             if not message.acked:
                 refcount = message.refcount

faust/streams.py

@@ -1113,6 +1113,8 @@ async def _py_aiter(self) -> AsyncIterator[T_co]:
                                 value = await _maybe_async(processor(value))
                         value = await on_merge(value)
                     except Skip:
+                        # We want to ack the filtered message
+                        # otherwise the lag would increase
                         value = skipped_value
 
                 try:
@@ -1121,7 +1123,9 @@ async def _py_aiter(self) -> AsyncIterator[T_co]:
                         yield value
                 finally:
                     self.current_event = None
-                    if do_ack and event is not None:
+                    # We want to ack the filtered out message
+                    # otherwise the lag would increase
+                    if event is not None and (do_ack or value is skipped_value):
                         # This inlines self.ack
                         last_stream_to_ack = event.ack()
                         message = event.message
@@ -1130,6 +1134,7 @@ async def _py_aiter(self) -> AsyncIterator[T_co]:
                         on_stream_event_out(tp, offset, self, event, sensor_state)
                         if last_stream_to_ack:
                             on_message_out(tp, offset, message)
+
         except StopAsyncIteration:
             # We are not allowed to propagate StopAsyncIteration in __aiter__
             # (if we do, it'll be converted to RuntimeError by CPython).

tests/functional/test_streams.py

@@ -222,6 +222,19 @@ async def myfilter(value):
     assert i == 3
 
 
+@pytest.mark.asyncio
+async def test_stream_filter_acks_filtered_out_messages(app, event_loop):
+    """
+    Test the filter function acknowledges the filtered out
+     messages regardless of the ack setting.
+    """
+    values = [1000, 3000, 99, 5000, 3, 9999]
+    async with app.stream(values).filter(lambda x: x > 1000) as stream:
+        async for event in stream.events():
+            assert event.value > 1000
+    assert len(app.consumer.unacked) == 0
+
+
 @pytest.mark.asyncio
 async def test_events(app):
     async with new_stream(app) as stream: