RepartitionExec should not error if output has hung up

[alamb]

Which issue does this PR close?

Closes #575

Rationale for this change

See #575 for gory details, but the idea is that an output hanging up is not an error (it is a LIMIT). However, we also should stop repartition exec from reading input that will never be read.

What changes are included in this PR?

1. Ignore errors when sending to output that has hung up
2. Stop pulling from input if all outputs have hung up.

Are there any user-facing changes?

Avoid intermittent errors (that probably started appearing after #521 as previously runtime errors were ignored)

[alamb]

I did not write a test that early shutdown stop consuming input (because repartition exec uses unbounded channels, it buffers indefinitely and thus the timing control has to be very precise)

🤔 mostly that sounds like an excuse

[codecov-commenter]

Codecov Report

Merging #576 (33f07f8) into master (51e5445) will increase coverage by 0.02%.
The diff coverage is 90.00%.

❗ Current head 33f07f8 differs from pull request most recent head e45efcf. Consider uploading reports for the commit e45efcf to get more accurate results

@@            Coverage Diff             @@
##           master     #576      +/-   ##
==========================================
+ Coverage   76.02%   76.05%   +0.02%     
==========================================
  Files         156      156              
  Lines       27063    27126      +63     
==========================================
+ Hits        20575    20631      +56     
- Misses       6488     6495       +7     

Impacted Files	Coverage Δ
datafusion/src/test/exec.rs	74.73% <82.75%> (+3.52%)	⬆️
datafusion/src/physical_plan/repartition.rs	87.02% <95.12%> (+0.85%)	⬆️

---

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 51e5445...e45efcf. Read the comment docs.

[alamb]

@Dandandan / @andygrove would you be ok if I merged this bug fix (it is causing us trouble in IOx)?

[Dandandan]

Makes sense 👍