You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/04/03 15:10:30 UTC
(doris) branch branch-4.1 updated: branch-4.1: [fix](streaming-job) Fix PG replication slot leak when streaming task is cancelled during pause/resume #62010 (#62033)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new b8e113b5f2a branch-4.1: [fix](streaming-job) Fix PG replication slot leak when streaming task is cancelled during pause/resume #62010 (#62033)
b8e113b5f2a is described below
commit b8e113b5f2ad88702ffc440b622f6dbca201a63e
Author: github-actions[bot] <41...@users.noreply.github.com>
AuthorDate: Fri Apr 3 23:10:22 2026 +0800
branch-4.1: [fix](streaming-job) Fix PG replication slot leak when streaming task is cancelled during pause/resume #62010 (#62033)
Cherry-picked from #62010
Co-authored-by: wudi <wu...@selectdb.com>
---
.../cdcclient/source/reader/JdbcIncrementalSourceReader.java | 12 ++++++++++++
.../cdcclient/source/reader/mysql/MySqlSourceReader.java | 12 ++++++++++++
2 files changed, 24 insertions(+)
diff --git a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
index e7245334ffb..3d9167d6124 100644
--- a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
+++ b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
@@ -351,6 +351,18 @@ public abstract class JdbcIncrementalSourceReader extends AbstractCdcSourceReade
}
Tuple2<SourceSplitBase, Boolean> splitFlag = createStreamSplit(offsetMeta, baseReq);
this.streamSplit = splitFlag.f0.asStreamSplit();
+
+ // Close previous stream reader to release resources (e.g. PG replication slot)
+ // before creating a new one. This prevents connection leaks when a cancelled
+ // task's reader is still active while a new task arrives.
+ if (this.streamReader != null) {
+ LOG.info(
+ "Closing previous stream reader before creating new one for job {}",
+ baseReq.getJobId());
+ closeReaderInternal(this.streamReader);
+ this.streamReader = null;
+ }
+
this.streamReader = getBinlogSplitReader(baseReq);
LOG.info("Prepare stream split: {}", this.streamSplit.toString());
diff --git a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
index 9d77f3912eb..14f6a6de1d4 100644
--- a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
+++ b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
@@ -338,6 +338,18 @@ public class MySqlSourceReader extends AbstractCdcSourceReader {
tryLoadTableSchemasFromRequest(baseReq);
Tuple2<MySqlSplit, Boolean> splitFlag = createBinlogSplit(offsetMeta, baseReq);
this.binlogSplit = (MySqlBinlogSplit) splitFlag.f0;
+
+ // Close previous binlog reader to release resources before creating a new one.
+ // This prevents connection leaks when a cancelled task's reader is still active
+ // while a new task arrives.
+ if (this.binlogReader != null) {
+ LOG.info(
+ "Closing previous binlog reader before creating new one for job {}",
+ baseReq.getJobId());
+ this.binlogReader.close();
+ this.binlogReader = null;
+ }
+
this.binlogReader = getBinlogSplitReader(baseReq);
LOG.info("Prepare binlog split: {}", this.binlogSplit.toString());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org