zhangshenghang opened a new issue, #12715: URL: https://github.com/apache/seatunnel/issues/12715
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues. ### What happened `DorisSourceReader.pollNext()` replaces its single `valueReader` field for every split **without closing the previous one**: ```java // DorisSourceReader.java public void pollNext(Collector<SeaTunnelRow> output) throws Exception { synchronized (output.getCheckpointLock()) { DorisSourceSplit nextSplit = splitsQueue.poll(); if (nextSplit != null) { ... valueReader = new DorisValueReader(partition, dorisSourceConfig, dorisSourceTable); // previous reader silently dropped while (valueReader.hasNext()) { ... } } ... } } ``` Resource-release chain (verified in the connector source): - `DorisValueReader.close()` is the **only** place in the connector that sends the `closeScanner` Thrift RPC, which frees the scan context on the Doris BE. - `BackendClient.close()`, the only place that closes the underlying Thrift `TSocket`, is `private` and is only invoked at the end of `BackendClient.closeScanner()`. - Therefore a `DorisValueReader` that is never `close()`d leaks **both** one TCP connection to the Doris BE and one BE-side scanner context (which holds memory and tablet readers until the BE times it out). `DorisSourceReader.close()` runs only once at reader shutdown, so it can only release the **last** split's reader. Any job whose table is split into N `PartitionDefinition` splits leaks (N-1) scanners and sockets **per reader task**. Multiple splits are the normal case: partitioned/sharded tables, larger tablets counts, and multi-table sync all produce several splits, and `DorisSourceSplitEnumerator.assignSplit()` hands the whole split list to each reader at once. Nothing auto-closes the reader on end-of-split either: `DorisValueReader.hasNext()` simply returns `false` at EOS without releasing anything. Note: #12325 (Oct 2026) hardened `DorisValueReader.close()` itself (stop the async fetch thread before taking the lock), but the overwrite path in `pollNext()` still bypasses `close()` entirely. Expected behavior: `pollNext()` closes the previous `DorisValueReader` before creating the next one, so each finished split releases its scanner context and socket. ### SeaTunnel Version dev (commit 0278a6a74) ### SeaTunnel Config ```conf env { job.mode = "BATCH" } source { Doris { source_table_names = ["db.tbl"] # a partitioned / multi-tablet table => multiple splits fenodes = "fe:8030" username = "user" password = "xxx" database = "db" table-list = [ {database = "db", table = "tbl"} ] } } sink { Console {} } ``` ### Running Command ```shell $SEATUNNEL_HOME/bin/seatunnel.sh --config ./doris.conf -e run ``` ### Error Exception No error surfaces to the user — the leak is silent. Observable effects: - `netstat`/`ss` on the job node shows an accumulating number of open TCP connections to Doris BE ports per reader task. - The Doris BE holds one scan context per finished split until its internal timeout; with large or frequent jobs this shows up as BE memory pressure (`be.INFO` scan contexts). ### Zeta or Flink or Spark Version Zeta ### Java or Scala Version Java 17 ### Are you willing to submit PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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]
