Problems with offset commits with filter/group_by pattern

[faust-streaming]

Problems with offset commits with filter/group_by pattern

[krzysieksulejczak]

Hi,

I work with @remy-sl on the same faust service. Following the problem I have a few more observations.

Events which should be filtered out are sent to repartitioned topic

It looks like stream.filter(...).group_by(...) pipeline attaches group_by processor to original stream and filter to repartitioned stream. Original stream .info() looks like this:
{'app': <App(builds-aggregator1): [URL('...')] running agents(<AgentManager: running >) 0x7f37a7ec9220>, 'channel': <(*)Topic: ...@0x7f37a58953d0>, 'processors': [<function Stream.group_by.<locals>.repartition at 0x7f37a5820550>], 'on_start': None, 'loop': <_UnixSelectorEventLoop running=True closed=False debug=False>, 'combined': [], 'beacon': Node: Worker/App/AgentManager/Agent: .../Stream: Topic: ..., 'concurrency_index': None, 'prev': None, 'active_partitions': None}
Repartitioned stream info() looks like this:
{'app': <App(builds-aggregator1): [URL('...')] running agents(<AgentManager: running >) 0x7f37a7ec9220>, 'channel': <(*)Topic: ...@0x7f37a5828070>, 'processors': [<function Stream.filter.<locals>.on_value at 0x7f37a58200d0>], 'on_start': <bound method Service.maybe_start of <Stream: init <(*)Topic: vd...x7f37a58953d0>>>, 'loop': <_UnixSelectorEventLoop running=True closed=False debug=False>, 'combined': [], 'beacon': Node: Worker/App/AgentManager/Agent: .../Stream: Topic: .../Stream: Topic: ..., 'concurrency_index': None, 'prev': <Stream: init <(*)Topic: vd...x7f37a58953d0>>, 'active_partitions': None}

Commit offset not advancing

It looks like commit offset stops at first occurrence of Skip() exception from filter(). .filter(lambda v: True) works without problems, while .filter(lambda v: False) makes commit offset freeze. After briefly looking at code I think the problem might lay here: https://github.com/faust-streaming/faust/blob/master/faust/streams.py#L909
When skipped_value occurs finally block is not executed in try/finally below, where ack() is executed.

Hope that helps.

[mdrago98]

Hi,

We're facing a related problem when using stream.noack() in conjunction with filter function where the offsets aren't committed properly and the filtered out messages would increase the lag on the consumer group.
Was there any progress on this issue?