github-actions[bot] commented on code in PR #69421:
URL: https://github.com/apache/doris/pull/69421#discussion_r4345042310
##########
fe/fe-core/src/main/java/org/apache/doris/arrowflight/protocol/FlightProtocolAdapter.java:
##########
@@ -588,9 +589,23 @@ public <E extends Exception> void
runCommand(ConnectContext ctx, SessionAction<E
});
}
+ public boolean hasPendingResults() {
+ // Expiration closes Arrow buffers. Serialize cleanup with DoGet,
which may
+ // still be sending a result obtained before its cache entry expired.
+ if (!commandLock.tryLock()) {
+ return true;
+ }
+ try {
+ return channel.hasPendingResults();
+ } finally {
+ commandLock.unlock();
+ }
+ }
+
/** {@link #runCommand} for a command that returns a value. */
public <T, E extends Exception> T callCommand(ConnectContext ctx,
SessionCommand<T, E> command) throws E {
acquireCommandLock(ctx);
+ QueryAdmissionController.INSTANCE.acquireContinuation();
Review Comment:
[P2] Include CloseSession in continuation accounting
Only commands using `callCommand()` receive this reservation;
`DorisFlightSqlProducer.closeSession()` calls teardown directly. For a Flight
`BEGIN; INSERT INTO ... VALUES ...` session, CloseSession clears its cached
results/deferred coordinators before `closeTxn()` waits for the BE rollback
response (up to five seconds). During that wait the command lock is free, the
MySQL idle scanner skips this session, and all four drain predicates can report
no work. Shutdown then closes Flight while the accepted CloseSession is
unfinished: [Arrow 19.0.0
FlightServer.close](https://github.com/apache/arrow-java/blob/v19.0.0/flight/flight-core/src/main/java/org/apache/arrow/flight/FlightServer.java#L145-L168)
force-closes RPCs after three seconds, so a slow rollback can lose its
response despite the default 300-second grace budget. Reserve a continuation
through CloseSession's teardown and response without taking the command lock,
since teardown must be able to cancel an active command.
##########
fe/fe-core/src/main/java/org/apache/doris/DorisFE.java:
##########
@@ -726,19 +733,72 @@ public static boolean isServerReady() {
}
private static void gracefulShutdown() {
- // wait for all queries to finish
+ // Health is already 503, but keep serving during LB / DNS convergence.
try {
- long now = System.currentTimeMillis();
- List<Coordinator> allCoordinators =
QeProcessorImpl.INSTANCE.getAllCoordinators();
- while (!allCoordinators.isEmpty() && System.currentTimeMillis() -
now < 300 * 1000L) {
- Thread.sleep(1000);
- allCoordinators =
QeProcessorImpl.INSTANCE.getAllCoordinators();
- LOG.info("waiting {} queries to finish before shutdown",
allCoordinators.size());
- }
- } catch (Throwable t) {
- LOG.error("", t);
+ LOG.info("graceful shutdown: serving during drain for {} seconds",
Config.graceful_shutdown_drain_seconds);
+ TimeUnit.SECONDS.sleep(Math.max(0,
Config.graceful_shutdown_drain_seconds));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn("graceful shutdown drain interrupted", e);
}
+ // Close admission before the listener: an accepted connection can
race its closure.
+ QueryAdmissionController.INSTANCE.stopAccepting();
+ if (qeService != null) {
+ qeService.stopAcceptNewConnections();
+ }
+ try {
+ long deadline = System.nanoTime()
+ + TimeUnit.SECONDS.toNanos(Math.max(0,
Config.grace_shutdown_wait_seconds));
+ waitForGracefulShutdown(ExecuteEnv.getInstance().getScheduler(),
deadline);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn("graceful shutdown wait interrupted", e);
+ } catch (Throwable t) {
+ LOG.error("graceful shutdown failed while draining requests", t);
+ }
LOG.info("graceful shutdown finished");
}
+
+ static void waitForGracefulShutdown(ConnectScheduler scheduler, long
deadline) throws InterruptedException {
+ // XNIO invokes close listeners synchronously, including transaction
rollback RPCs.
+ // Keep those waits on a daemon worker so even an idle session cannot
stall the deadline.
+ ExecutorService idleCloser =
ThreadPoolManager.newDaemonFixedThreadPool(
+ 1, 1, "graceful-idle-mysql-closer", false);
+ try {
+ Future<?> idleCleanup = null;
+ while (true) {
+ if (idleCleanup == null) {
+ idleCleanup = idleCloser.submit(() ->
scheduler.closeIdleMysqlConnections(deadline));
+ }
+ int activeRequests =
QueryAdmissionController.INSTANCE.getActiveRequests();
+ List<Coordinator> coordinators =
QeProcessorImpl.INSTANCE.getAllCoordinators();
+ boolean pendingFlightResults =
scheduler.hasPendingFlightResults();
+ boolean pendingIdleCleanup = !idleCleanup.isDone();
+ // A session's transaction may still be rolling back after its
last request ended.
+ if (activeRequests == 0 && coordinators.isEmpty() &&
!pendingFlightResults && !pendingIdleCleanup) {
Review Comment:
[P2] Wait for session teardown after the final request finishes
A completed scan may have skipped a session that was still executing an
admitted transactional INSERT. If that request publishes COM_SLEEP and releases
its count before `getActiveRequests()` is sampled, this condition sees zero
work and exits before the rescan at lines 792-794. The now-idle transaction is
never closed or rolled back, even with most of the grace period remaining; the
later `MysqlServer.stop()` only closes the accepting listener. An admitted
COM_QUIT exposes the same cleanup-ownership gap: its count is released before
`ReadListener` calls `cleanup()`, and idle scans skip its killed context while
rollback runs. Require a cleanup pass after request quiescence and track
teardown already owned by other workers through completion or the deadline. The
existing deadline thread covers blocking cleanup; this path incorrectly
declares cleanup complete.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/ConnectScheduler.java:
##########
@@ -124,6 +125,27 @@ public Map<Integer, ConnectContext> getConnectionMap() {
return connectPoolMgr.getConnectionMap();
}
+ public void closeIdleMysqlConnections(long deadline) {
+ for (ConnectContext context : getConnectionMap().values()) {
+ // A close listener can block on transaction rollback. Do not
start another
+ // teardown once that wait consumed the shutdown budget or was
interrupted.
+ if (System.nanoTime() >= deadline ||
Thread.currentThread().isInterrupted()) {
+ break;
+ }
+ try {
+ context.closeIdleMysqlConnection();
Review Comment:
[P2] Bind the session context on the idle cleanup worker
The new daemon worker closes each session without installing its
thread-local context. On a non-cloud follower, `BEGIN; INSERT INTO t SELECT
...` leaves `isTransactionBegan=true` in the session after the master's
transaction state is copied back. This close invokes `unregisterConnection()`
-> `closeTxn()` -> `TransactionEntry.abortTransaction()`, which constructs
`MasterOpExecutor` with `ConnectContext.get()`. That is null on this worker,
and the constructor immediately dereferences `ctx.getEnv()`. `closeTxn()`
catches the failure and clears the transaction entry, so the session disappears
while its master-side transaction stays PREPARE until timeout. Bind and
restore/clear each session's context around the close, as the normal MySQL
worker does. This is distinct from the existing rollback-deadline comment: the
rollback RPC is never dispatched here.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
