fix: bug: topic reader close method freezes indefinitely

[Roshan931]

Fixes #679

Root cause

Topic reader close(flush=True) waits forever for commit acks on an already-broken stream

Changes

• ReaderStream.flush() now stops waiting for commit acks as soon as the stream's first error is set, instead of awaiting ack futures that can never be resolved; close(flush=True) then proceeds to tear down the stream and fail pending waiters with PublicTopicReaderPartitionExpiredError. Added two regression tests.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 82.66%. Comparing base (d6391fb) to head (3bbfe47).

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #920      +/-   ##
==========================================
- Coverage   82.67%   82.66%   -0.02%     
==========================================
  Files          99       99              
  Lines       12961    12965       +4     
  Branches     1269     1269              
==========================================
+ Hits        10716    10718       +2     
- Misses       1794     1795       +1     
- Partials      451      452       +1     

Flag	Coverage Δ
integration	80.50% <100.00%> (+0.05%)	⬆️
unit	49.24% <66.66%> (+0.02%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
ydb/_topic_reader/topic_reader_asyncio.py	86.65% <100.00%> (-0.24%)	⬇️

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.