fix: IntermediateWriter closes underlying writer twice

[waynexia]

I hereby agree to the terms of the GreptimeDB CLA

What's changed and what's your intention?

futures_util::forward() will flush and close the writer when stream is finished. From its document https://docs.rs/futures-util/0.3.30/futures_util/stream/trait.StreamExt.html#method.forward :

It will complete once the stream is exhausted, the sink has received and flushed all items, and the sink is closed.

Test can pass before because Vec won't report errors when it gets closed twice.

Checklist

• I have written the necessary rustdoc comments.
• I have added the necessary unit tests and integration tests.
• This PR does not require documentation updates.

Refer to a related PR or issue link (optional)

[codecov]

Codecov Report

Attention: 3 lines in your changes are missing coverage. Please review.

Comparison is base (b0c3be3) 85.62% compared to head (95d484b) 85.17%.
Report is 2 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3248      +/-   ##
==========================================
- Coverage   85.62%   85.17%   -0.46%     
==========================================
  Files         856      856              
  Lines      140092   140107      +15     
==========================================
- Hits       119957   119334     -623     
- Misses      20135    20773     +638