feat(writer): the RollingFileWriter closes files in background when concurrency is set

[fvaleye]

Which issue does this PR close?

• Closes Discussion: Optimize RollingFileWriter by closing asynchronously #1551

What changes are included in this PR?

My general approach:

• Non-blocking rollover: new files start immediately while old ones close concurrently, if we reach the limit
• Controlled concurrency: uses max_concurrent_closes to prevent memory exhaustion. By default, the default behavior is kept by having only one background task running.
• Resource management: Automatic cleanup and proper error propagation when calling close()

The main idea suggested by the issue is to improve the RollingFileWriter to close files in the background, improving performance by allowing writes to continue without waiting for the previous file to be fully closed.

Key changes:

• A JoinSet of all closing futures is now used to manage background closing tasks.
• When rolling over, the writer spawns a task to close the previous file and immediately starts writing to a new one.
• The close method now waits for all background tasks to complete before returning.
• Concurrency can be configured via a parameter to control how many files are closed in parallel.

Are these changes tested?

Yes, with unit-testing and local benchmark tests

[liurenjie1024]

Thanks @fvaleye for this pr, but I don't think we should do this now. As I said in previous discussion, such kind of optimization typically consumes more memory since an unclosed writer will hold memory in it. This optimization make things difficult to predict and manage when integrated with compute engine.

[fvaleye]

Thanks @fvaleye for this pr, but I don't think we should do this now. As I said in previous discussion, such kind of optimization typically consumes more memory since an unclosed writer will hold memory in it. This optimization make things difficult to predict and manage when integrated with compute engine.

Thanks for your feedback @liurenjie1024 !
I understand, that's exactly why I took a conservative approach, using default settings and flexible concurrency parameters with safeguards.
I will close this PR for now.