[flink] Close writer immediately on sink failover

[loserwang1024]

Search before asking

• I searched in the 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).

@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?

• I'm willing to submit a PR!

[firasbouzazi]

can i take the issue ?