CDC backoff cancel

internal/impl/mssqlserver/replication/sleep_test.go

@@ -0,0 +1,35 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package replication
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSleepInterruptible(t *testing.T) {
+	t.Run("cancellation cuts the sleep short", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(t.Context())
+		go func() {
+			time.Sleep(10 * time.Millisecond)
+			cancel()
+		}()
+		start := time.Now()
+		err := sleepInterruptible(ctx, time.Hour)
+		require.ErrorIs(t, err, context.Canceled)
+		require.Less(t, time.Since(start), 10*time.Second)
+	})
+
+	t.Run("uncancelled sleep elapses and returns nil", func(t *testing.T) {
+		require.NoError(t, sleepInterruptible(t.Context(), time.Millisecond))
+	})
+}

internal/impl/mssqlserver/replication/stream.go

@@ -496,12 +496,28 @@ func (r *ChangeTableStream) ReadChangeTables(ctx context.Context, db *sql.DB, st
 				startLSN = lastLSN
 			} else {
 				r.log.Debug("No more changes across all change tables, backing off...")
-				time.Sleep(r.backoffInterval)
+				if err := sleepInterruptible(ctx, r.backoffInterval); err != nil {
+					return err
+				}
 			}
 		}
 	}
 }
 
+// sleepInterruptible pauses for d unless ctx is cancelled first, so a
+// shutdown or a poisoned-publisher teardown never has to wait out the
+// change-table backoff.
+func sleepInterruptible(ctx context.Context, d time.Duration) error {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case <-timer.C:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
 // UserDefinedTable represents a found user's SQL Server table (called a user-defined table) in SQL.
 type UserDefinedTable struct {
 	Schema   string

internal/impl/oracledb/logminer/logminer.go

@@ -176,15 +176,33 @@ func (lm *LogMiner) ReadChanges(ctx context.Context, startPos replication.SCN) (
 					lm.log.Debugf("Caught up with redo logs, backing off for %s...", lm.cfg.MiningBackoffInterval)
 					lm.caughtUpLogged = true
 				}
-				time.Sleep(lm.cfg.MiningBackoffInterval)
+				if err := sleepInterruptible(ctx, lm.cfg.MiningBackoffInterval); err != nil {
+					return err
+				}
 			} else {
 				lm.caughtUpLogged = false
-				time.Sleep(lm.cfg.MiningInterval)
+				if err := sleepInterruptible(ctx, lm.cfg.MiningInterval); err != nil {
+					return err
+				}
 			}
 		}
 	}
 }
 
+// sleepInterruptible pauses for d unless ctx is cancelled first, so a
+// shutdown or a poisoned-publisher teardown never has to wait out a
+// mining backoff.
+func sleepInterruptible(ctx context.Context, d time.Duration) error {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case <-timer.C:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
 // FindStartPos returns the database's current SCN so that streaming begins from
 // the present moment rather than replaying historical redo logs.
 func (lm *LogMiner) FindStartPos(ctx context.Context) (replication.SCN, error) {

internal/impl/oracledb/logminer/sleep_test.go

@@ -0,0 +1,35 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package logminer
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSleepInterruptible(t *testing.T) {
+	t.Run("cancellation cuts the sleep short", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(t.Context())
+		go func() {
+			time.Sleep(10 * time.Millisecond)
+			cancel()
+		}()
+		start := time.Now()
+		err := sleepInterruptible(ctx, time.Hour)
+		require.ErrorIs(t, err, context.Canceled)
+		require.Less(t, time.Since(start), 10*time.Second)
+	})
+
+	t.Run("uncancelled sleep elapses and returns nil", func(t *testing.T) {
+		require.NoError(t, sleepInterruptible(t.Context(), time.Millisecond))
+	})
+}