Stop fetching when flow stops

[ekerstens]

Description

Changes

1. During recovery, resume the consumer after the app. When the app flow_control resumes it clears all queues, and resuming the consumer after the flow control makes sure that we don't write data to the app which is then discarded.
2. Added logic to the consumer to cancel the fetching coroutine during a rebalance.
3. Moved the wait for flow to resume from consumer._wait_next_records to consumer.getmany to avoid a lock during rebalance
4. During a rebalance, the app waits for the consumer to cancel fetching before continuing, to avoid potential issues of discarding a fetch which occurs after assignment.

[patkivikram]

Should this be continue?

[ekerstens]

Reason for break is that even if the fetcher has completed (I have seen cases where both happen around the same time) to ignore the result and return nothing

[codecov-commenter]

Codecov Report

Merging #253 (032436b) into master (a0e9a31) will decrease coverage by 0.04%.
The diff coverage is 79.31%.

@@            Coverage Diff             @@
##           master     #253      +/-   ##
==========================================
- Coverage   94.26%   94.22%   -0.05%     
==========================================
  Files         100      100              
  Lines       10814    10834      +20     
  Branches     1477     1482       +5     
==========================================
+ Hits        10194    10208      +14     
- Misses        549      553       +4     
- Partials       71       73       +2     

Impacted Files	Coverage Δ
faust/transport/consumer.py	94.24% <77.77%> (-0.90%)	⬇️
faust/app/base.py	99.51% <100.00%> (+<0.01%)	⬆️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update a0e9a31...032436b. Read the comment docs.

[patkivikram]

Don't we want to process the exception?