morningman opened a new pull request, #68338:
URL: https://github.com/apache/doris/pull/68338
### What problem does this PR solve?
Related PR: #68101 (the one connection pool, which made the leak visible),
#68266 (the bearer token as the session's credential, which removed its last
cap), #67503 / #62259 (the Arrow Flight deferral gate this generalizes)
Problem Summary:
**Context.** A Remote Doris catalog with `use_arrow_flight = true` reads
another Doris cluster over Arrow Flight SQL: for every scan,
`RemoteDorisScanNode` on the local FE performs a Flight SQL handshake against a
remote FE (`authenticateBasicToken`), runs the query there (`GetFlightInfo`),
and hands the endpoints - a ticket per remote BE - to the local BE, which reads
the rows with `DoGet` straight from the remote BEs.
The handshake is not free on the remote side: since #68101 a Flight SQL
session is a connection in the remote FE's one connection pool, counted against
`qe_max_connection`, the Arrow Flight SQL sub-quota and the catalog user's
`max_user_connections`; since #68266 the bearer token is that session's name in
the pool and nothing else, so the session ends only on `CloseSession`, `KILL
CONNECTION`, `wait_timeout` (8h by default) or an FE restart.
The regression framework has a leak of its own of the same shape: a suite's
Doris connections are `ThreadLocal` to the thread that opened them, and only
the suite thread and `Suite.thread()` close theirs; a `sql` on any other thread
opens a connection nobody closes.
**1. The problem, and what it cost**
- `RemoteDorisScanNode.executeFlightSqlQuery` closed the gRPC channel and
the allocator in a try-with-resources but never sent `CloseSession`. Every scan
of a Remote Doris table therefore left one Flight SQL session behind on the
remote FE, under the catalog user, until `wait_timeout`. Before #68101 this was
invisible: Flight sessions had a pool of their own, were not counted per user,
and a per-user LRU of `max_user_connections / 2` tokens evicted the oldest.
After #68101 the leaked sessions eat the catalog user's quota on the remote FE;
after #68266 nothing caps them at all. A hundred scans within 8h and the user -
MySQL clients included - is refused there with `Reach limit of connections`.
- This is what broke the external regression pipeline on 2026-09-21
(TeamCity 1053416 on #68308): the `remote_doris` suites point the catalog at
the FE under test with user `root`; 49 scans left 48 Flight sessions (`Arrow
Flight SQL: 512 (current: 48)` in the refusal), which took half of root's 100;
the other half was taken by the framework leak below, and 14 MySQL connections
were refused in a six-second window.
- The framework leak: `Awaitility.await()...until { sql ... }` evaluates the
condition on Awaitility's own thread, which dies with the `await()`. Each call
leaked one root connection until the client JVM garbage-collected it (the FE
logs those as `No more data to be read. Close connection`). 230 suites call
`Awaitility.await()` directly; in the failing run one suite opened 22 such
connections in 46 seconds, and 29 of them were collected in one GC at the
moment the refusals stopped. Suites that call `sql` from threads of their own
(`Thread.start { streamLoad }`, an `Executors` pool) leak the same way - a P0
run of the same day shows 76 Awaitility connections and 89 own-thread
connections opened by root within one minute, all left to the garbage collector
- and the two docker helpers dropped the connection their action opened without
closing it.
**2. What this PR does, and why it helps**
FE:
- `RemoteDorisFlightSession` (new): the session as an object - handshake,
`execute`, and `close()` = `CloseSession` (bounded to 5s so a remote FE that
stopped answering cannot hang the local query's teardown; the session is then
left to its `wait_timeout` as before) followed by the channel and the
allocator. `open()` leaves nothing behind when the handshake is refused; a
query that fails closes its session at once, so the retry on the next node
leaves nothing behind either. Idempotent.
- `RemoteDorisScanNode` keeps the session from `getSplits` until `stop()`,
which the coordinator calls when the local query closes or is cancelled - i.e.
when the local BE is done with the remote query's endpoints. It cannot be
closed right after `GetFlightInfo`: the remote FE cancels whatever a closed
session was still running, and when the remote table is itself an external
table scanned in batch mode the remote query is deferred there and still
running while the local BE reads.
- For the same reason the local coordinator has to outlive dispatch when the
local query is itself an Arrow Flight SQL query (otherwise #67503 closes it
right after `exec()`, while the local BE may still be reading). The deferral
gate's predicate is generalized from "has a batch split source" to "the BE
still depends on this scan after dispatch": `ScanNode.hasBatchSplitSource()` ->
`coordinatorMustOutliveDispatch()`, `Coordinator.hasBatchSplitSource()` ->
`mustOutliveDispatch()`; `RemoteDorisScanNode` adds its open session as the
second reason. Batch-mode external scans behave exactly as before.
Regression framework:
- `Awaitility.pollInSameThread()` at framework start-up: every `until { }`
now runs on the suite thread and reuses the suite's connection, as
`Suite.awaitUntil` already did. The trade-off (an `atMost()` cannot interrupt a
condition that blocks) is bounded by the timeouts of the statements a condition
runs.
- `SuiteContext` records every connection its thread-local accessors open,
with the thread that opened it. On every statement it closes the connections of
threads that have finished (a suite that starts a thread per step -
`Thread.start { streamLoad }; join`, as the mow flexible suites do 72 times -
now holds at most the connections of the threads still running,
deterministically, where before the population depended on the JVM's next GC),
and when the suite ends it closes whatever is left, with a warning naming the
suite.
- The two docker helpers (`docker`, `dockers`) close the connection their
action opened before restoring the original one (and the multi-cluster one now
types that original as the `ConnectionInfo` it is).
Tests:
- `RemoteDorisScanNodeTest`: an in-process Flight SQL server counts the
sessions it is asked to close. The session lives from the query until `stop()`;
`stop()` twice closes once; a failed query closes at once; a refused handshake
opens nothing; a session handed over after `stop()` is closed at once; and the
coordinator of a query with such a scan `mustOutliveDispatch()`.
- `ArrowFlightDeferralGateTest` follows the rename.
- Regression
`external_table_p0/remote_doris/test_remote_doris_flight_session`: a catalog
logging in as a user of its own scans a table five times and asserts after each
scan that `information_schema.processlist` holds no `ArrowFlightSQL` session of
that user - the assertion that fails on master.
What it buys: a Remote Doris scan costs the remote FE one session for
exactly the duration of the local query, whatever the protocol of the local
client; the catalog user's quota on the remote FE is no longer consumed by
history; and the regression framework no longer manufactures the MySQL half of
the pressure.
**3. The classes, and how they call each other**
- `RemoteDorisScanNode` (existing): `getSplits` -> `executeQuery` ->
`executeFlightSqlQuery(host, user, password, sql, timeout)`:
`RemoteDorisFlightSession.open` + `execute`, then `keepFlightSession`. `stop()`
(from `Coordinator.close()` / `cancel()`) closes the session;
`coordinatorMustOutliveDispatch()` is true while one is held.
- `RemoteDorisFlightSession` (new): `open` (FlightClient +
`authenticateBasicToken`), `execute` (`FlightSqlClient.execute`), `close`
(`closeSession` with a 5s deadline, then client and allocator).
- `ScanNode.coordinatorMustOutliveDispatch()` (renamed from
`hasBatchSplitSource`): `splitAssignment != null`, overridable.
- `Coordinator.mustOutliveDispatch()` (renamed): any scan node's
`coordinatorMustOutliveDispatch()`.
- `StmtExecutor.executeAndSendResult`: the deferral gate now reads
`coord.mustOutliveDispatch()`.
- Remote FE, untouched: `DorisFlightSqlProducer.closeSession` ->
`FlightSessionsInConnectPool.closeConnectContext` -> `ConnectContext.cleanup()`
+ `cancelQuery`.
- `RegressionTest.initGroovyEnv`: `Awaitility.pollInSameThread()`.
- `SuiteContext`: `openedDorisConnections` (connection -> opening thread),
`trackDorisConnection`, `closeConnectionsOfFinishedThreads` (from
`getConnection()`, i.e. every statement), `closeDorisConnection`,
`closeLeftoverDorisConnections` (from `close()`); `Suite.dockerImpl` /
`dockers` call `closeDorisConnection`.
```
local FE remote FE
remote BE
RemoteDorisScanNode.getSplits
'- executeFlightSqlQuery
|- RemoteDorisFlightSession.open ---- handshake --> openSession
(pool: +1 for the catalog user)
|- session.execute ----------------- GetFlightInfo --> runs the query
---------------> result buffered
'- keepFlightSession
(ticket per BE)
Coordinator.exec -> local BE ------------------------------ DoGet(ticket)
--------------------> rows
(Arrow Flight local client: coordinator kept alive, mustOutliveDispatch()
== true)
Coordinator.close / cancel
'- scanNode.stop
'- session.close ------------------- CloseSession -->
closeConnectContext (pool: -1)
```
### Release note
None
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- [x] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [x] Yes. <!-- Explain the behavior change -->
- A Remote Doris scan ends its Flight SQL session on the remote FE
when the local query ends; the remote FE's `SHOW PROCESSLIST` no longer
accumulates `ArrowFlightSQL` sessions of the catalog user.
- An Arrow Flight SQL query on the local FE that scans a Remote
Doris table keeps its coordinator until the session's next command, its
teardown or the deferred-query idle reaper, like a batch-mode external scan
does (#62259).
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should
merge into -->
--
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]
