(doris) branch master updated: [fix](streaming-job) Fix PG replication slot leak when streaming task is cancelled during pause/resume (#62010)

[di...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new c6dea59bac6 [fix](streaming-job) Fix PG replication slot leak when streaming task is cancelled during pause/resume  (#62010)
c6dea59bac6 is described below

commit c6dea59bac6f077a821d5130c390ebb8bcba7d3a
Author: wudi <wu...@selectdb.com>
AuthorDate: Thu Apr 2 12:12:57 2026 +0800

    [fix](streaming-job) Fix PG replication slot leak when streaming task is cancelled during pause/resume  (#62010)
    
    ### What problem does this PR solve?
    
      Problem Summary:
    
    When a PostgreSQL CDC streaming job is paused and resumed, the PG
    replication slot
      can be permanently leaked, causing all subsequent tasks to fail with:
      `replication slot "doris_cdc_xxx" is active for PID xxx`
    
      **Root cause:**
    
    The CDC client reuses a single `SourceReader` instance per jobId
    (`Env.getOrCreateReader`).
    When FE cancels a task (PAUSE), the BE HTTP connection is closed, but
    the CDC client's
    `buildStreamRecords` thread may still be blocked in `pollRecords` (up to
    15s timeout).
    Before the old task finishes, the new task (after RESUME) arrives at the
    same CDC client
    and calls `prepareStreamSplit`, which overwrites `this.streamReader`
    with a new Fetcher
    without closing the old one. The old Debezium reader (holding the PG
    replication connection)
    is leaked — its reference is lost, so `finishSplitRecords` in the old
    task's finally block
      closes the new Fetcher instead, and the PG slot is never released.
    
    From the logs, the slot remained occupied for 25+ minutes until the test
    timed out:
    Failed to start replication stream at LSN{0/318EBC8}; when setting up
    multiple connectors
    for the same database host, please make sure to use a distinct
    replication slot name for each.
    
      **Fix:**
    
      Close the previous stream/binlog reader before creating a new one in
    `prepareStreamSplit` (PG) and `prepareBinlogSplit` (MySQL). This ensures
    the old
    Debezium connection is properly released when a new task reuses the same
    SourceReader.
---
 .../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