[I] [flink] Support timeout-based writer close and close Fluss sink immediately on failover [fluss]

["loserwang1024 (via GitHub)" <gi...@apache.org>]

loserwang1024 opened a new issue, #3809:
URL: https://github.com/apache/fluss/issues/3809

   ### Search before asking
   
   - [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.
   
   
   ### Fluss version
   
   0.9.0 (latest release)
   
   ### Please describe the bug 🐞
   
   Currently, the Fluss Flink sink closes the client through `Connection.close()`, which closes the underlying `WriterClient` with `Duration.ofMillis(Long.MAX_VALUE)`:
   
   - `fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java`
   - `writerClient.close(Duration.ofMillis(Long.MAX_VALUE))`
   
   This means sink close may wait indefinitely for pending writer requests to finish. During Flink task failover or cancellation, if the sender still has records that cannot be sent downstream, the task may be blocked in close and failover cannot proceed promptly.
   
   This behavior is unreasonable for Flink failover/cancel paths. The Kafka sink handles this differently: `FlinkKafkaInternalProducer#close()` closes immediately by calling `super.close(Duration.ZERO)`.
   ```java
   @Override
   public void close() {
       if (!closed) {
           LOG.debug("Closing immediately {}", this);
           super.close(Duration.ZERO);
           closed = true;
       }
   }
   ```
   
   
   ### Solution
   
   Fluss should provide a timeout-based close API that allows callers to choose between graceful close and immediate/fast close.For the Flink sink, close during task failover/cancel should use immediate close semantics, similar to Kafka's FlinkKafkaInternalProducer#close().
   
   ### Are you willing to submit a PR?
   
   - [x] I'm willing to submit a PR!


-- 
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: issues-unsubscribe@fluss.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] [flink] Close writer immediately on sink failover [fluss]

["firasbouzazi (via GitHub)" <gi...@apache.org>]

firasbouzazi commented on issue #3809:
URL: https://github.com/apache/fluss/issues/3809#issuecomment-5128874894

   can i take the issue ?


-- 
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: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] [flink] Close writer immediately on sink failover [fluss]

["loserwang1024 (via GitHub)" <gi...@apache.org>]

loserwang1024 commented on issue #3809:
URL: https://github.com/apache/fluss/issues/3809#issuecomment-5138548092

   @firasbouzazi just do it 


-- 
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: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] [flink] Close writer immediately on sink failover [fluss]

["loserwang1024 (via GitHub)" <gi...@apache.org>]

loserwang1024 closed issue #3809: [flink] Close writer immediately on sink failover
URL: https://github.com/apache/fluss/issues/3809


-- 
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: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org