You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "loserwang1024 (via GitHub)" <gi...@apache.org> on 2026/08/03 06:05:15 UTC
Re: [PR] [flink][client] Close writer immediately on sink failover [fluss]
loserwang1024 commented on code in PR #3817:
URL: https://github.com/apache/fluss/pull/3817#discussion_r3701663307
##########
fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java:
##########
@@ -319,26 +326,46 @@ private Sender newSender(short acks, int retries) {
}
public void close(Duration timeout) {
- LOG.info("Closing writer.");
+ long timeoutMs = timeout.toMillis();
+ LOG.info("Closing writer with timeout {} ms.", timeoutMs);
writerMetricGroup.close();
if (sender != null) {
- sender.initiateClose();
+ if (timeoutMs > 0) {
+ // graceful close: stop accepting new records, but keep the sender running to
+ // send out the pending records.
+ sender.initiateClose();
+ } else {
+ // immediate close: abandon all unsent and in-flight requests right away so
+ // that close never blocks (e.g. during Flink task failover/cancellation).
+ sender.forceClose();
+ }
}
if (ioThreadPool != null) {
ioThreadPool.shutdown();
try {
- if (!ioThreadPool.awaitTermination(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
+ if (!ioThreadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
+ // the graceful close didn't finish within the timeout (or an immediate
+ // close was requested): abandon the pending requests so that the sender
+ // thread can exit its drain loop, and interrupt it to break any blocking
+ // waits.
+ if (sender != null) {
Review Comment:
Please implement it referring kafka.
<img width="1964" height="1490" alt="Image" src="https://github.com/user-attachments/assets/6c6e67d4-298b-4a4f-8ea2-8e58996fbcee" />
##########
fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java:
##########
@@ -319,26 +326,46 @@ private Sender newSender(short acks, int retries) {
}
public void close(Duration timeout) {
- LOG.info("Closing writer.");
+ long timeoutMs = timeout.toMillis();
+ LOG.info("Closing writer with timeout {} ms.", timeoutMs);
writerMetricGroup.close();
if (sender != null) {
- sender.initiateClose();
+ if (timeoutMs > 0) {
Review Comment:
Is it needed?if timeout <= 0, oThreadPool.awaitTermination will return false.
<img width="1314" height="1206" alt="Image" src="https://github.com/user-attachments/assets/4a323e6a-41d9-4f2a-bfdc-1acb27642f85" />
--
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