(flink) branch master updated: [FLINK-34644][runtime] Fix race condition in RestServerEndpoint shutdown (#27708)

[martijnvisser]

This is an automated email from the ASF dual-hosted git repository.

martijnvisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 5cb890d408c [FLINK-34644][runtime] Fix race condition in 
RestServerEndpoint shutdown (#27708)
5cb890d408c is described below

commit 5cb890d408ceb4dd7cbd0960528892ca0d82642b
Author: Martijn Visser <2989614+martijnvis...@users.noreply.github.com>
AuthorDate: Sat Feb 28 05:33:06 2026 -0800

    [FLINK-34644][runtime] Fix race condition in RestServerEndpoint shutdown 
(#27708)
    
    * [FLINK-34644][runtime] Fix race condition in RestServerEndpoint shutdown 
causing flaky test
    
    Change thenAccept to thenCompose in AbstractRestHandler.respondToRequest()
    so that the returned future only completes after the HTTP response is fully
    flushed to the network. Previously, the future returned by
    HandlerUtils.sendResponse() was silently discarded, allowing
    InFlightRequestTracker.deregisterRequest() to fire before the response
    write completed. Under load this let shutDownInternal() tear down the
    Netty event loops while response bytes were still in flight, causing
    ConnectionClosedException on the client side.
    
    * [FLINK-34644][sql-gateway] Apply same thenAccept to thenCompose fix in 
AbstractSqlGatewayRestHandler
    
    Apply the same fix as AbstractRestHandler: change thenAccept to
    thenCompose in respondToRequest() so the returned future only completes
    after the HTTP response is fully flushed. This prevents the same race
    condition where in-flight request deregistration could fire before the
    response write completes.
---
 .../java/org/apache/flink/runtime/rest/handler/AbstractRestHandler.java | 2 +-
 .../flink/table/gateway/rest/handler/AbstractSqlGatewayRestHandler.java | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractRestHandler.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractRestHandler.java
index 2ecbe743cda..1c104bb4dd4 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractRestHandler.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/AbstractRestHandler.java
@@ -85,7 +85,7 @@ public abstract class AbstractRestHandler<
             response = FutureUtils.completedExceptionally(e);
         }
 
-        return response.thenAccept(
+        return response.thenCompose(
                 resp ->
                         HandlerUtils.sendResponse(
                                 ctx,
diff --git 
a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/AbstractSqlGatewayRestHandler.java
 
b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/AbstractSqlGatewayRestHandler.java
index a3613e38a06..2fc4a08e69f 100644
--- 
a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/AbstractSqlGatewayRestHandler.java
+++ 
b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/AbstractSqlGatewayRestHandler.java
@@ -88,7 +88,7 @@ public abstract class AbstractSqlGatewayRestHandler<
             response = FutureUtils.completedExceptionally(e);
         }
 
-        return response.thenAccept(
+        return response.thenCompose(
                 resp ->
                         HandlerUtils.sendResponse(
                                 ctx,