Recovery hang due to backpressure changes

[patkivikram]

Checklist

• I have included information about relevant versions
• I have verified that the issue persists when using the master branch of Faust.

Steps to reproduce

Tell us what you did to cause something to happen.

Expected behavior

Tell us what you expected to happen.

Actual behavior

If the stream buffer gets full and the worker is still in recovery, there is a chance of it stuck in the recovery loop forever

Full traceback

This keeps going on forever
+Remaining for active recovery--------------+-----------+-------------+-------------+-----------+
| topic | partition | need offset | have offset | remaining |
+-------------------------------------------+-----------+-------------+-------------+-----------+
| ifos0.5-correletable_keys_table-changelog | 0 | 4574 | 4256 | 318 |
| ifos0.5-kv_table-changelog | 2 | 26857 | 26617 | 240 |
| ifos0.5-triggers_table-changelog | 0 | 3691 | 1347 | 2344 |
+-------------------------------------------+-----------+-------------+-------------+-----------+

Paste the full traceback (if there is any)

Versions

• Python version
• Faust version
• Operating system
• Kafka version
• RocksDB version (if applicable)