diff --git a/internal/impl/mssqlserver/replication/sleep_test.go b/internal/impl/mssqlserver/replication/sleep_test.go new file mode 100644 index 0000000000..d4b97f9b7e --- /dev/null +++ b/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)) + }) +} diff --git a/internal/impl/mssqlserver/replication/stream.go b/internal/impl/mssqlserver/replication/stream.go index e53fa734a6..5f1772119f 100644 --- a/internal/impl/mssqlserver/replication/stream.go +++ b/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 diff --git a/internal/impl/oracledb/logminer/logminer.go b/internal/impl/oracledb/logminer/logminer.go index 5dd56b77aa..4771b72f8d 100644 --- a/internal/impl/oracledb/logminer/logminer.go +++ b/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) { diff --git a/internal/impl/oracledb/logminer/sleep_test.go b/internal/impl/oracledb/logminer/sleep_test.go new file mode 100644 index 0000000000..6cc103c6af --- /dev/null +++ b/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)) + }) +}