fresh-borzoni commented on code in PR #404:
URL: https://github.com/apache/fluss-rust/pull/404#discussion_r2885978911
##########
crates/fluss/src/client/write/writer_client.rs:
##########
@@ -146,21 +163,48 @@ impl WriterClient {
Ok((bucket_assigner, bucket_id))
}
- pub async fn close(self) -> Result<()> {
- self.shutdown_tx
- .send(())
- .await
- .map_err(|e| Error::UnexpectedError {
- message: format!("Failed to close write client: {e:?}"),
- source: None,
- })?;
-
- self.sender_join_handle
- .await
- .map_err(|e| Error::UnexpectedError {
- message: format!("Failed to close write client: {e:?}"),
- source: None,
- })?;
+ /// Close the writer with a timeout. Matches Java's two-phase shutdown:
+ ///
+ /// 1. **Graceful**: Signal the sender to drain all remaining batches.
+ /// `accumulator.close()` makes all batches immediately ready (no need
+ /// to wait for `batch_timeout_ms`).
+ /// 2. **Force** (if timeout exceeded): Abort the sender task and fail
+ /// all remaining batches with an error.
+ ///
+ /// Idempotent: calling `close` a second time returns `Ok(())` immediately.
+ pub async fn close(&self, timeout: Duration) -> Result<()> {
+ // Take shutdown_tx and join_handle out of their Mutexes.
Review Comment:
this shouldn't be part of public API, writer_client is used when you create
AppendWriter, UpsertWriter. I will assign crate visibility, it looks more
sensible for me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
