fix: allow aborting a failed CDC 2PC import to release the peer (#51083)

## Summary In CDC replication mode, a 2PC bulk import (`auto_commit=false`) runs **independently** on the primary (source) and standby clusters against the same files. Only the source can broadcast `Commit`/`Rollback`; the standby replays them via CDC. When the import fails on **one** cluster but the other holds a replicated `Uncommitted` job, that job had no recovery path: - **Mode 1 (source fails, standby `Uncommitted`):** the failure never broadcast a `RollbackImport`, and `AbortImport` on a terminal-`Failed` source was rejected — so the standby stayed `Uncommitted` indefinitely, holding invisible imported segments. (Observed in production: standby stuck 2+ days.) - **Mode 2 (standby fails, source `Uncommitted`):** if the platform commits the source before confirming the standby is prepared, the source commits while the standby's commit ack no-ops on its `Failed` job → silent divergence. issue: #51070 ## What this changes (datacoord) - **Abort a failed source (`AbortImport`):** a `Failed` source (its own import failed) is now abortable — it broadcasts `RollbackImport` to release the peer's `Uncommitted` job. `Committing`/`Completed` stay rejected. Lets an operator recover **immediately**, instead of waiting for GC. - **GC self-heal (`importChecker.checkGC`):** before GC removes a `Failed` replicate-source job, it broadcasts `RollbackImport` so the peer is released **automatically at retention expiry**. Transient broadcast failure keeps the job and retries on the next tick; a standby's `ErrNotPrimary` proceeds; **job removal itself is the idempotency guard** (once gone, never re-broadcast). This closes the recovery-window gap without a persisted flag or proto change. - **Divergence signal (`CommitImport` ack):** a commit ack landing on a `Failed` replica now logs at WARN (Mode 2). - `isReplicatingClusterNow`: the GC decision now **propagates** replication-check errors instead of tolerating them — an indeterminate status (transient balancer error, `OnShutdownError`) retains the job and retries next tick, since a false "not replicating" would irreversibly strand the replicating peer; only permanent broadcast errors (`ErrNotPrimary` from a standby, `ErrCollectionNotFound` after a replicated drop) proceed to GC. The balancer access is bounded by a 10s timeout so an unregistered balancer cannot park the checker loop. - Minor: the coordinator callbacks the checker needs are bundled into an `importCheckerHooks` struct instead of a growing positional-arg list. ## Recovery guarantee The peer's `Uncommitted` job is released either **immediately** via `AbortImport`, or **automatically** when the failed source reaches GC retention — no permanent stranding, and no dependence on a manual abort within the 3h retention window. ## Not in scope (follow-ups) - **Idempotent re-broadcast dedup:** repeated `AbortImport` on a real-failure source re-broadcasts each time (harmless — flusher/standby-ack no-op). A persisted `rollback_broadcasted` flag would dedup this. - **Eager self-heal:** broadcasting the rollback at *failure* time (in `checkFailedJob`) rather than at GC time, to shorten peer-stranding to seconds. - **Mode 2 prevention:** a standby→coordinator prepared-vote so the source refuses to commit unless all replicas are prepared. ## Test plan - `checkGC` self-heal: failed replicate-source broadcasts-then-removes with transient-retry; `ErrNotPrimary` (standby) proceeds; non-replicating skips the broadcast. - `AbortImport`: failed-source broadcasts rollback; `Committing` and `Completed` rejected; previously-user-aborted is idempotent. - CI `ci-v2/ut-go` compiles and runs these tests. (Local datacoord UT was not run for this revision due to cgo core drift in the worktree.) 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: bigsheeper <yihao.dai@zilliz.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-21 02:05:41 +00:00 · 2026-07-20 10:48:41 +08:00
parent 29891a36b7
commit 10cd30c66e
6 changed files with 543 additions and 46 deletions
@@ -145,6 +145,27 @@ func isReplicatingCluster(cfg *commonpb.ReplicateConfiguration) bool {
 	return cfg != nil && (len(cfg.GetCrossClusterTopology()) > 0 || len(cfg.GetClusters()) > 1)
 }

+// isReplicatingClusterNow reports whether this cluster is currently part of a CDC
+// replication topology. A non-nil error means the status could not be determined (e.g. a
+// transient balancer error, or OnShutdownError while streamingcoord is stopping before
+// datacoord); the caller must treat that as indeterminate rather than "not replicating",
+// because at GC time a false "not replicating" would irreversibly drop a replicating job
+// without releasing the peer. A nil assignment is an unambiguous "not replicating".
+func (s *Server) isReplicatingClusterNow(ctx context.Context) (bool, error) {
+	balancer, err := balance.GetWithContext(ctx)
+	if err != nil {
+		return false, err
+	}
+	assignment, err := balancer.GetLatestChannelAssignment()
+	if err != nil {
+		return false, err
+	}
+	if assignment == nil {
+		return false, nil
+	}
+	return isReplicatingCluster(assignment.ReplicateConfiguration), nil
+}
+
 // broadcastImport broadcasts the import message to all vchannels.
 // This method is called from the new ImportV2 flow where proxy calls DataCoord directly.
 func (s *Server) broadcastImport(ctx context.Context,
@@ -235,8 +256,11 @@ func (c *DDLCallbacks) commitImportV2AckCallback(ctx context.Context, result mes
 			mlog.FieldJobID(jobID), mlog.String("state", job.GetState().String()))
 		return nil
 	case internalpb.ImportJobState_Failed:
-		mlog.Info(ctx, "CommitImport: job already failed, no-op",
-			mlog.FieldJobID(jobID))
+		// Divergence signal: the source committed but this replica already failed, so
+		// this replica will NOT make the data visible. Left as a no-op here; surfaced
+		// at WARN for alerting.
+		mlog.Warn(ctx, "CommitImport ack landed on a Failed import job; this replica will NOT commit while the source commits — potential primary/standby divergence",
+			mlog.FieldJobID(jobID), mlog.String("reason", job.GetReason()))
 		return nil
 	default:
 		// CommitImport may be replicated before the local import task reaches
@@ -22,16 +22,19 @@ import (
 	"sync"
 	"time"

+	"github.com/cockroachdb/errors"
 	"github.com/samber/lo"

 	"github.com/milvus-io/milvus/internal/datacoord/allocator"
 	"github.com/milvus-io/milvus/internal/datacoord/broker"
+	"github.com/milvus-io/milvus/internal/streamingcoord/server/broadcaster"
 	"github.com/milvus-io/milvus/internal/util/importutilv2"
 	"github.com/milvus-io/milvus/pkg/v3/metrics"
 	"github.com/milvus-io/milvus/pkg/v3/mlog"
 	"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v3/proto/internalpb"
 	"github.com/milvus-io/milvus/pkg/v3/util/funcutil"
+	"github.com/milvus-io/milvus/pkg/v3/util/merr"
 	"github.com/milvus-io/milvus/pkg/v3/util/tsoutil"
 )

@@ -40,6 +43,23 @@ type ImportChecker interface {
 	Close()
 }

+// importCheckerHooks bundles the coordinator callbacks the import checker invokes,
+// injected as one named unit (instead of a growing positional-arg list) so the checker
+// does not depend on *Server. A nil callback disables the corresponding behavior; tests
+// inject only the hooks they exercise.
+type importCheckerHooks struct {
+	// commitImport broadcasts a CommitImport WAL message. Required in production; a nil
+	// value is a programming error only when reached on the auto_commit=true path.
+	commitImport func(ctx context.Context, job ImportJob) error
+	// rollbackImport broadcasts a RollbackImport WAL message. nil disables GC self-heal.
+	rollbackImport func(ctx context.Context, job ImportJob) error
+	// isReplicatingCluster reports whether this cluster is currently replicating. A
+	// non-nil error means the status is indeterminate (e.g. a transient balancer error
+	// during shutdown) and the caller must not make an irreversible GC decision. nil hook
+	// is treated as "not replicating" (GC self-heal disabled).
+	isReplicatingCluster func(ctx context.Context) (bool, error)
+}
+
 type importChecker struct {
 	ctx        context.Context
 	meta       *meta
@@ -49,9 +69,7 @@ type importChecker struct {
 	ci         CompactionInspector
 	handler    Handler

-	// commitImportFn broadcasts a CommitImport WAL message.
-	// Injected at construction time so the checker does not depend on *Server.
-	commitImportFn func(ctx context.Context, job ImportJob) error
+	hooks importCheckerHooks

 	closeOnce sync.Once
 	closeChan chan struct{}
@@ -64,35 +82,44 @@ func NewImportChecker(ctx context.Context,
 	importMeta ImportMeta,
 	ci CompactionInspector,
 	handler Handler,
-	commitImportFn func(ctx context.Context, job ImportJob) error, // required; nil OK for tests
+	hooks importCheckerHooks,
 ) ImportChecker {
 	return &importChecker{
-		ctx:            ctx,
-		meta:           meta,
-		broker:         broker,
-		alloc:          alloc,
-		importMeta:     importMeta,
-		ci:             ci,
-		handler:        handler,
-		commitImportFn: commitImportFn,
-		closeChan:      make(chan struct{}),
+		ctx:        ctx,
+		meta:       meta,
+		broker:     broker,
+		alloc:      alloc,
+		importMeta: importMeta,
+		ci:         ci,
+		handler:    handler,
+		hooks:      hooks,
+		closeChan:  make(chan struct{}),
 	}
 }

+// Start runs the checker loops until Close. The state-machine loop and the
+// timeout/GC loop deliberately run on separate goroutines: checkGC's rollback
+// broadcast can park on the ctx-insensitive resource-key lock (see checkGC), and
+// isolating it guarantees the state machine keeps making progress no matter how
+// long GC blocks. All state shared by the two loops lives behind importMeta's
+// mutex (which already serves concurrent RPC and ack-callback goroutines), and
+// UpdateJob refuses transitions out of Completed/Failed, so the loops cannot
+// resurrect or regress each other's terminal states.
 func (c *importChecker) Start() {
 	mlog.Info(c.ctx, "start import checker")
-	var (
-		ticker1 = time.NewTicker(Params.DataCoordCfg.ImportCheckIntervalHigh.GetAsDuration(time.Second)) // 2s
-		ticker2 = time.NewTicker(Params.DataCoordCfg.ImportCheckIntervalLow.GetAsDuration(time.Second))  // 2min
-	)
-	defer ticker1.Stop()
-	defer ticker2.Stop()
+	go c.runGCLoop()
+	c.runStateMachineLoop()
+}
+
+func (c *importChecker) runStateMachineLoop() {
+	ticker := time.NewTicker(Params.DataCoordCfg.ImportCheckIntervalHigh.GetAsDuration(time.Second)) // 2s
+	defer ticker.Stop()
 	for {
 		select {
 		case <-c.closeChan:
-			mlog.Info(c.ctx, "import checker exited")
+			mlog.Info(c.ctx, "import checker state-machine loop exited")
 			return
-		case <-ticker1.C:
+		case <-ticker.C:
 			jobs := c.importMeta.GetJobBy(c.ctx)
 			for _, job := range jobs {
 				if !funcutil.SliceSetEqual[string](job.GetVchannels(), job.GetReadyVchannels()) {
@@ -122,7 +149,19 @@ func (c *importChecker) Start() {
 					c.checkFailedJob(job)
 				}
 			}
-		case <-ticker2.C:
+		}
+	}
+}
+
+func (c *importChecker) runGCLoop() {
+	ticker := time.NewTicker(Params.DataCoordCfg.ImportCheckIntervalLow.GetAsDuration(time.Second)) // 2min
+	defer ticker.Stop()
+	for {
+		select {
+		case <-c.closeChan:
+			mlog.Info(c.ctx, "import checker gc loop exited")
+			return
+		case <-ticker.C:
 			jobs := c.importMeta.GetJobBy(c.ctx)
 			for _, job := range jobs {
 				c.tryTimeoutJob(job)
@@ -469,11 +508,11 @@ func (c *importChecker) checkUncommittedJob(job ImportJob) {
 	// collection-level resource-key lock serializes overlapping broadcasts, the
 	// ack callback only transitions when the job is still Uncommitted, and
 	// HandleCommitVchannel is idempotent on committed_vchannels.
-	if c.commitImportFn == nil {
-		log.Error(c.ctx, "commitImportFn is nil but auto_commit=true; this is a programming error")
+	if c.hooks.commitImport == nil {
+		log.Error(c.ctx, "commit hook is nil but auto_commit=true; this is a programming error")
 		return
 	}
-	if err := c.commitImportFn(c.ctx, job); err != nil {
+	if err := c.hooks.commitImport(c.ctx, job); err != nil {
 		log.Warn(c.ctx, "auto-commit broadcast failed, will retry on next tick", mlog.Err(err))
 	}
 }
@@ -601,6 +640,57 @@ func (c *importChecker) checkGC(job ImportJob) {
 		if !shouldRemoveJob {
 			return
 		}
+		// In a CDC replicating cluster, a failed 2PC source import must release the
+		// peer cluster's replicated Uncommitted job before we drop it — otherwise the
+		// peer is stranded with invisible imported segments and no recovery path, since
+		// source GC never touches the peer. Removal of the job is itself the idempotency
+		// guard: once gone we never re-broadcast. Auto-commit jobs have no 2PC peer to
+		// release, so they skip the gate entirely.
+		if c.hooks.rollbackImport != nil && c.hooks.isReplicatingCluster != nil &&
+			job.GetState() == internalpb.ImportJobState_Failed && !job.GetAutoCommit() {
+			// The check reaches the streaming balancer future, which blocks until the
+			// balancer is registered — under the server-lifetime c.ctx that would park
+			// the GC loop during the window before streamingcoord registers
+			// it (e.g. a restart recovering a job already past retention). Bound it like
+			// checkCollection does; a timeout is just another indeterminate status.
+			replicateCheckCtx, cancel := context.WithTimeout(c.ctx, 10*time.Second)
+			replicating, err := c.hooks.isReplicatingCluster(replicateCheckCtx)
+			cancel()
+			switch {
+			case err != nil:
+				// Indeterminate replication status (e.g. a transient balancer error during
+				// shutdown, when streamingcoord stops before datacoord). Removing the job now
+				// could strand a replicating peer's Uncommitted job with no recovery path,
+				// which is irreversible — a false "not replicating" costs nothing but a retry,
+				// so keep the job and re-evaluate on the next GC tick.
+				log.Warn(c.ctx, "cannot determine replication status before GC of failed import job, will retry", mlog.Err(err))
+				return
+			case replicating:
+				// Broadcast the RollbackImport to release the peer. A transient error keeps
+				// the job to retry next tick; a permanent error (standby ErrNotPrimary, or the
+				// collection was dropped — itself a replicated DDL, so the peer fails its own
+				// job independently) falls through to GC, since retrying it forever would leak
+				// the job's metadata.
+				//
+				// Bound the broadcast like the replication check above: it blocks in
+				// BlockUntilDone until every vchannel append succeeds, and under the
+				// server-lifetime c.ctx an unavailable streamingnode would park this
+				// loop until shutdown. A timeout is just another transient status —
+				// keep the job and retry on the next GC tick. The resource-key lock on
+				// the broadcast path is still ctx-insensitive (making it fail-fast is a
+				// follow-up), which is one reason this GC loop runs on its own
+				// goroutine (see Start): even an unbounded park here can only delay
+				// GC, never the import state machine.
+				rollbackCtx, rollbackCancel := context.WithTimeout(c.ctx, 10*time.Second)
+				err := c.hooks.rollbackImport(rollbackCtx, job)
+				rollbackCancel()
+				if err != nil && !isPermanentRollbackErr(err) {
+					log.Warn(c.ctx, "failed to broadcast rollback before GC of failed replicate import job, will retry", mlog.Err(err))
+					return
+				}
+				log.Info(c.ctx, "proceeding with GC of failed replicate import job after rollback attempt")
+			}
+		}
 		err := c.importMeta.RemoveJob(c.ctx, job.GetJobID())
 		if err != nil {
 			log.Warn(c.ctx, "remove import job failed", mlog.Err(err))
@@ -609,3 +699,20 @@ func (c *importChecker) checkGC(job ImportJob) {
 		log.Info(c.ctx, "import job removed")
 	}
 }
+
+// isPermanentRollbackErr reports whether a RollbackImport broadcast error is permanent,
+// i.e. retrying it can never succeed, so the failed job should still be GC'd rather than
+// retried forever (which would leak its metadata). Everything else is treated as transient
+// and retried on the next GC tick — misclassifying a transient error as permanent would
+// drop a replicating job without releasing the peer, which is irreversible.
+func isPermanentRollbackErr(err error) bool {
+	// ErrNotPrimary: this cluster is a replication standby, not the primary that owns the
+	// broadcast; its own failed job is independent and safe to drop.
+	// ErrCollectionNotFound: the collection was dropped. DropCollection is itself a
+	// replicated DDL, so the peer marks its own import job Failed independently — there is
+	// no peer left to release, and the broadcast can never succeed.
+	// errRollbackImportNoVchannels: the job carries no vchannels (fixed at creation), so
+	// the broadcast has no peer to address and can never succeed.
+	return errors.Is(err, broadcaster.ErrNotPrimary) || errors.Is(err, merr.ErrCollectionNotFound) ||
+		errors.Is(err, errRollbackImportNoVchannels)
+}
@@ -33,10 +33,12 @@ import (
 	"github.com/milvus-io/milvus/internal/datacoord/allocator"
 	broker2 "github.com/milvus-io/milvus/internal/datacoord/broker"
 	"github.com/milvus-io/milvus/internal/metastore/mocks"
+	"github.com/milvus-io/milvus/internal/streamingcoord/server/broadcaster"
 	"github.com/milvus-io/milvus/pkg/v3/mlog"
 	"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v3/proto/internalpb"
 	"github.com/milvus-io/milvus/pkg/v3/proto/rootcoordpb"
+	"github.com/milvus-io/milvus/pkg/v3/util/merr"
 	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/v3/util/timerecord"
 	"github.com/milvus-io/milvus/pkg/v3/util/tsoutil"
@@ -88,7 +90,7 @@ func (s *ImportCheckerSuite) SetupTest() {
 		}, nil
 	}).Maybe()

-	checker := NewImportChecker(context.TODO(), meta, broker, s.alloc, importMeta, ci, handler, nil).(*importChecker)
+	checker := NewImportChecker(context.TODO(), meta, broker, s.alloc, importMeta, ci, handler, importCheckerHooks{}).(*importChecker)
 	s.checker = checker

 	job := &importJob{
@@ -508,6 +510,277 @@ func (s *ImportCheckerSuite) TestCheckGC() {
 	s.Equal(0, len(s.importMeta.GetJobBy(context.TODO())))
 }

+// setupGCReadyFailedJob puts the suite's job into a Failed, past-cleanup-ts state with
+// a single import task that has no live segments and is unassigned, so checkGC is one
+// step away from removing the job (the only remaining gate is the replicate rollback).
+func (s *ImportCheckerSuite) setupGCReadyFailedJob() {
+	catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog)
+	catalog.EXPECT().SaveImportTask(mock.Anything, mock.Anything).Return(nil)
+	catalog.EXPECT().SaveImportJob(mock.Anything, mock.Anything).Return(nil)
+
+	taskProto := &datapb.ImportTaskV2{
+		JobID:  s.jobID,
+		TaskID: 1,
+		State:  datapb.ImportTaskStateV2_Failed,
+		NodeID: NullNodeID,
+	}
+	task := &importTask{tr: timerecord.NewTimeRecorder("import task")}
+	task.task.Store(taskProto)
+	s.NoError(s.importMeta.AddTask(context.TODO(), task))
+
+	s.NoError(s.importMeta.UpdateJob(context.TODO(), s.jobID, UpdateJobState(internalpb.ImportJobState_Failed)))
+	GCRetention := Params.DataCoordCfg.ImportTaskRetention.GetAsDuration(time.Second)
+	job := s.importMeta.GetJob(context.TODO(), s.jobID)
+	job.(*importJob).CleanupTs = tsoutil.AddPhysicalDurationOnTs(job.GetCleanupTs(), GCRetention*-2)
+	s.NoError(s.importMeta.AddJob(context.TODO(), job))
+}
+
+// A failed source in a replicating cluster must broadcast RollbackImport before its job
+// is GC'd. A transient broadcast error keeps the job alive to retry; once it succeeds the
+// job is removed.
+func (s *ImportCheckerSuite) TestCheckGCReplicateSourceBroadcastsRollback() {
+	s.setupGCReadyFailedJob()
+	catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog)
+	catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil)
+	catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil)
+
+	rollbackCalls := 0
+	rollbackErr := errors.New("broadcast failed")
+	s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error {
+		rollbackCalls++
+		return rollbackErr
+	}
+	s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil }
+
+	// First tick: broadcast fails → job retained (tasks already removed), rollback attempted.
+	s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID))
+	s.Equal(1, rollbackCalls)
+	s.Equal(1, len(s.importMeta.GetJobBy(context.TODO())))
+
+	// Next tick: broadcast succeeds → job removed.
+	rollbackErr = nil
+	s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID))
+	s.Equal(2, rollbackCalls)
+	s.Equal(0, len(s.importMeta.GetJobBy(context.TODO())))
+}
+
+// A standby (not the replication primary) gets ErrNotPrimary from the broadcast; that is
+// treated as success so its own failed job is still GC'd.
+func (s *ImportCheckerSuite) TestCheckGCReplicateNotPrimaryProceeds() {
+	s.setupGCReadyFailedJob()
+	catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog)
+	catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil)
+	catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil)
+
+	s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error { return broadcaster.ErrNotPrimary }
+	s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil }
+
+	s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID))
+	s.Equal(0, len(s.importMeta.GetJobBy(context.TODO())))
+}
+
+// A non-replicating cluster must not broadcast any rollback; the failed job is GC'd as before.
+func (s *ImportCheckerSuite) TestCheckGCNonReplicatingSkipsRollback() {
+	s.setupGCReadyFailedJob()
+	catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog)
+	catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil)
+	catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil)
+
+	rollbackCalls := 0
+	s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error {
+		rollbackCalls++
+		return nil
+	}
+	s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return false, nil }
+
+	s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID))
+	s.Equal(0, rollbackCalls)
+	s.Equal(0, len(s.importMeta.GetJobBy(context.TODO())))
+}
+
+// The replication check reaches the streaming balancer future, which blocks until the
+// balancer is registered; checkGC must pass a deadline-bounded ctx so an unregistered
+// balancer (datacoord ready before streamingcoord) cannot park the whole checker loop.
+func (s *ImportCheckerSuite) TestCheckGCReplicateCheckCtxIsBounded() {
+	s.setupGCReadyFailedJob()
+	catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog)
+	catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil)
+	catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil)
+
+	s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error { return nil }
+	hasDeadline := false
+	s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) {
+		_, hasDeadline = ctx.Deadline()
+		return false, nil
+	}
+
+	s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID))
+	s.True(hasDeadline)
+}
+
+// When the replication status cannot be determined (e.g. a transient balancer error during
+// shutdown), GC must NOT drop the job: a false "not replicating" would strand a replicating
+// peer with no recovery path. The job is retained and no rollback is broadcast.
+func (s *ImportCheckerSuite) TestCheckGCReplicateIndeterminateRetainsJob() {
+	s.setupGCReadyFailedJob()
+	// The task-cleanup loop runs before the replication gate, so the task is removed
+	// even though the job itself is retained; DropImportJob must NOT be called.
+	catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog)
+	catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil)
+
+	rollbackCalls := 0
+	s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error {
+		rollbackCalls++
+		return nil
+	}
+	s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) {
+		return false, errors.New("balancer not ready")
+	}
+
+	s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID))
+	s.Equal(0, rollbackCalls)
+	s.Equal(1, len(s.importMeta.GetJobBy(context.TODO())))
+}
+
+// A permanent rollback error (e.g. the collection was dropped → ErrCollectionNotFound) must
+// NOT be retried forever, which would leak the job's metadata. The job is GC'd instead.
+func (s *ImportCheckerSuite) TestCheckGCReplicatePermanentRollbackErrRemovesJob() {
+	s.setupGCReadyFailedJob()
+	catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog)
+	catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil)
+	catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil)
+
+	rollbackCalls := 0
+	s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error {
+		rollbackCalls++
+		return merr.WrapErrCollectionNotFound(job.GetCollectionID())
+	}
+	s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil }
+
+	s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID))
+	s.Equal(1, rollbackCalls)
+	s.Equal(0, len(s.importMeta.GetJobBy(context.TODO())))
+}
+
+// The rollback broadcast can block until every vchannel append succeeds (or forever on
+// an unavailable streamingnode) under the server-lifetime c.ctx; checkGC must pass a
+// deadline-bounded ctx so a stuck broadcast cannot park the whole checker loop.
+func (s *ImportCheckerSuite) TestCheckGCReplicateRollbackCtxIsBounded() {
+	s.setupGCReadyFailedJob()
+	catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog)
+	catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil)
+	catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil)
+
+	hasDeadline := false
+	s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error {
+		_, hasDeadline = ctx.Deadline()
+		return nil
+	}
+	s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil }
+
+	s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID))
+	s.True(hasDeadline)
+	s.Equal(0, len(s.importMeta.GetJobBy(context.TODO())))
+}
+
+// A job without vchannels can never deliver its rollback (Vchannels are fixed at
+// creation), so the error must be classified permanent and the job GC'd — while a
+// generic ImportSysFailed error must stay transient.
+func (s *ImportCheckerSuite) TestCheckGCReplicateNoVchannelsRollbackErrRemovesJob() {
+	server := &Server{}
+	err := server.broadcastRollbackImportMessage(context.TODO(), &importJob{ImportJob: &datapb.ImportJob{JobID: 1}})
+	s.Error(err)
+	s.True(isPermanentRollbackErr(err))
+	s.False(isPermanentRollbackErr(merr.WrapErrImportSysFailedMsg("some transient failure")))
+
+	s.setupGCReadyFailedJob()
+	catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog)
+	catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil)
+	catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil)
+
+	s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error {
+		return server.broadcastRollbackImportMessage(ctx, &importJob{ImportJob: &datapb.ImportJob{JobID: job.GetJobID()}})
+	}
+	s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil }
+
+	s.checker.checkGC(s.importMeta.GetJob(context.TODO(), s.jobID))
+	s.Equal(0, len(s.importMeta.GetJobBy(context.TODO())))
+}
+
+// The GC loop and the state-machine loop must run on separate goroutines: a
+// rollback broadcast parked on the ctx-insensitive resource-key lock (or any
+// other stall inside checkGC) must delay only GC, never the import state
+// machine. This test parks checkGC's rollback forever and asserts the state
+// machine still processes another job meanwhile.
+func (s *ImportCheckerSuite) TestStateMachineProgressesWhileGCRollbackParked() {
+	params := paramtable.Get()
+	params.Save(params.DataCoordCfg.ImportCheckIntervalHigh.Key, "0.05")
+	params.Save(params.DataCoordCfg.ImportCheckIntervalLow.Key, "0.05")
+	defer params.Reset(params.DataCoordCfg.ImportCheckIntervalHigh.Key)
+	defer params.Reset(params.DataCoordCfg.ImportCheckIntervalLow.Key)
+
+	s.setupGCReadyFailedJob()
+	catalog := s.importMeta.(*importMeta).catalog.(*mocks.DataCoordCatalog)
+	catalog.EXPECT().DropImportTask(mock.Anything, mock.Anything).Return(nil).Maybe()
+	// Teardown releases the parked rollback, letting the in-flight checkGC run to
+	// completion (RemoveJob, then checkCollection) while the loops shut down.
+	catalog.EXPECT().DropImportJob(mock.Anything, mock.Anything).Return(nil).Maybe()
+	s.checker.broker.(*broker2.MockBroker).EXPECT().HasCollection(mock.Anything, mock.Anything).Return(true, nil).Maybe()
+
+	rollbackEntered := make(chan struct{}, 1)
+	rollbackRelease := make(chan struct{})
+	s.checker.hooks.rollbackImport = func(ctx context.Context, job ImportJob) error {
+		select {
+		case rollbackEntered <- struct{}{}:
+		default:
+		}
+		// Park forever, ignoring ctx — simulates the ctx-insensitive lock.
+		<-rollbackRelease
+		return nil
+	}
+	s.checker.hooks.isReplicatingCluster = func(ctx context.Context) (bool, error) { return true, nil }
+
+	go s.checker.Start()
+	defer s.checker.Close()
+	defer close(rollbackRelease)
+
+	// Wait until the GC loop is parked inside the rollback broadcast.
+	select {
+	case <-rollbackEntered:
+	case <-time.After(10 * time.Second):
+		s.FailNow("GC loop never reached the rollback broadcast")
+	}
+
+	// Feed the state machine a Failed job with a live task; only the
+	// state-machine loop (checkFailedJob → tryFailingTasks) can fail the task.
+	jobB := &importJob{
+		ImportJob: &datapb.ImportJob{
+			JobID:          s.jobID + 100,
+			CollectionID:   1,
+			Vchannels:      []string{"ch1"},
+			ReadyVchannels: []string{"ch1"},
+			State:          internalpb.ImportJobState_Failed,
+			CleanupTs:      tsoutil.ComposeTSByTime(time.Now().Add(24 * time.Hour)), // never GC-ready
+		},
+		tr: timerecord.NewTimeRecorder("import job"),
+	}
+	s.NoError(s.importMeta.AddJob(context.TODO(), jobB))
+	taskProto := &datapb.ImportTaskV2{
+		JobID:  jobB.GetJobID(),
+		TaskID: 999,
+		State:  datapb.ImportTaskStateV2_Pending,
+		NodeID: NullNodeID,
+	}
+	task := &importTask{tr: timerecord.NewTimeRecorder("import task")}
+	task.task.Store(taskProto)
+	s.NoError(s.importMeta.AddTask(context.TODO(), task))
+
+	s.Eventually(func() bool {
+		tasks := s.importMeta.GetTaskBy(context.TODO(), WithJob(jobB.GetJobID()))
+		return len(tasks) == 1 && tasks[0].GetState() == datapb.ImportTaskStateV2_Failed
+	}, 10*time.Second, 20*time.Millisecond)
+}
+
 func (s *ImportCheckerSuite) TestCheckCollection() {
 	mockErr := errors.New("mock err")

@@ -603,7 +876,7 @@ func TestImportCheckerCompaction(t *testing.T) {
 	cim := NewMockCompactionInspector(t)
 	handler := NewNMockHandler(t)

-	checker := NewImportChecker(context.TODO(), meta, broker, alloc, importMeta, cim, handler, nil).(*importChecker)
+	checker := NewImportChecker(context.TODO(), meta, broker, alloc, importMeta, cim, handler, importCheckerHooks{}).(*importChecker)

 	job := &importJob{
 		ImportJob: &datapb.ImportJob{
@@ -814,13 +1087,13 @@ func (s *ImportCheckerSuite) TestCheckUncommittedJob_AutoCommitTrue() {
 	})

 	commitCalled := false
-	s.checker.commitImportFn = func(ctx context.Context, job ImportJob) error {
+	s.checker.hooks.commitImport = func(ctx context.Context, job ImportJob) error {
 		commitCalled = true
 		return nil
 	}

 	s.checker.checkUncommittedJob(s.importMeta.GetJob(context.TODO(), s.jobID))
-	s.True(commitCalled, "commitImportFn should be called when auto_commit=true")
+	s.True(commitCalled, "commit hook should be called when auto_commit=true")
 }

 func (s *ImportCheckerSuite) TestCheckUncommittedJob_AutoCommitFalse() {
@@ -831,25 +1104,25 @@ func (s *ImportCheckerSuite) TestCheckUncommittedJob_AutoCommitFalse() {
 	})

 	commitCalled := false
-	s.checker.commitImportFn = func(ctx context.Context, job ImportJob) error {
+	s.checker.hooks.commitImport = func(ctx context.Context, job ImportJob) error {
 		commitCalled = true
 		return nil
 	}

 	s.checker.checkUncommittedJob(s.importMeta.GetJob(context.TODO(), s.jobID))
-	s.False(commitCalled, "commitImportFn must NOT be called when auto_commit=false")
+	s.False(commitCalled, "commit hook must NOT be called when auto_commit=false")
 	// Job state must remain Uncommitted.
 	s.Equal(internalpb.ImportJobState_Uncommitted, s.importMeta.GetJob(context.TODO(), s.jobID).GetState())
 }

 func (s *ImportCheckerSuite) TestCheckUncommittedJob_NilFn_AutoCommitTrue() {
-	// commitImportFn=nil with auto_commit=true is a programming error; the checker
+	// commit hook=nil with auto_commit=true is a programming error; the checker
 	// must log an error and return without crashing (no panic in the ticker goroutine).
 	s.manuallyUpdateJob(s.jobID, func(job ImportJob) {
 		job.(*importJob).State = internalpb.ImportJobState_Uncommitted
 		job.(*importJob).AutoCommit = true
 	})
-	s.checker.commitImportFn = nil
+	s.checker.hooks.commitImport = nil

 	s.NotPanics(func() {
 		s.checker.checkUncommittedJob(s.importMeta.GetJob(context.TODO(), s.jobID))
@@ -859,7 +1132,7 @@ func (s *ImportCheckerSuite) TestCheckUncommittedJob_NilFn_AutoCommitTrue() {

 // TestCheckUncommittedJob_RepeatedTicks_Safe verifies that ticker re-entry into
 // checkUncommittedJob before the ack callback transitions the job state is safe.
-// commitImportFn is invoked once per tick; correctness against the resulting
+// the commit hook is invoked once per tick; correctness against the resulting
 // duplicate broadcasts is guaranteed by the broadcaster's resource-key lock,
 // the ack callback's state guard, and HandleCommitVchannel's idempotency.
 func (s *ImportCheckerSuite) TestCheckUncommittedJob_RepeatedTicks_Safe() {
@@ -869,7 +1142,7 @@ func (s *ImportCheckerSuite) TestCheckUncommittedJob_RepeatedTicks_Safe() {
 	})

 	callCount := 0
-	s.checker.commitImportFn = func(ctx context.Context, job ImportJob) error {
+	s.checker.hooks.commitImport = func(ctx context.Context, job ImportJob) error {
 		callCount++
 		return nil
 	}
@@ -877,7 +1150,7 @@ func (s *ImportCheckerSuite) TestCheckUncommittedJob_RepeatedTicks_Safe() {
 	for i := 0; i < 3; i++ {
 		s.checker.checkUncommittedJob(s.importMeta.GetJob(context.TODO(), s.jobID))
 	}
-	s.Equal(3, callCount, "each tick must call commitImportFn; broadcaster handles dedup")
+	s.Equal(3, callCount, "each tick must call the commit hook; broadcaster handles dedup")
 	s.Equal(internalpb.ImportJobState_Uncommitted, s.importMeta.GetJob(context.TODO(), s.jobID).GetState(),
 		"state must remain Uncommitted until the ack callback fires")
 }
@@ -343,7 +343,11 @@ func (s *Server) initDataCoord() error {

 	s.importInspector = NewImportInspector(s.ctx, s.meta, s.importMeta, s.globalScheduler)

-	s.importChecker = NewImportChecker(s.ctx, s.meta, s.broker, s.allocator, s.importMeta, s.compactionInspector, s.handler, s.broadcastCommitImportMessage)
+	s.importChecker = NewImportChecker(s.ctx, s.meta, s.broker, s.allocator, s.importMeta, s.compactionInspector, s.handler, importCheckerHooks{
+		commitImport:         s.broadcastCommitImportMessage,
+		rollbackImport:       s.broadcastRollbackImportMessage,
+		isReplicatingCluster: s.isReplicatingClusterNow,
+	})

 	// init file resource observer
 	if s.fileResourceObserver != nil {
@@ -2864,12 +2864,20 @@ func (s *Server) broadcastCommitImportMessage(ctx context.Context, job ImportJob
 	return err
 }

+// errRollbackImportNoVchannels marks a rollback that can never be delivered: the job
+// carries no vchannels, so there is no peer to address. A job's Vchannels are fixed at
+// creation, so retrying can never succeed — isPermanentRollbackErr classifies this as
+// permanent so GC proceeds instead of retaining the job forever. A plain sentinel
+// attached via errors.Mark, NOT a merr error: merr's errors.Is matches by error code,
+// which would make every ImportSysFailed error (mostly transient) match it.
+var errRollbackImportNoVchannels = errors.New("import job has no vchannels")
+
 // broadcastRollbackImportMessage broadcasts a RollbackImport WAL message for the given import job.
 // Targets the job's data vchannels, matching the CommitImport routing.
 func (s *Server) broadcastRollbackImportMessage(ctx context.Context, job ImportJob) error {
 	vchannels := job.GetVchannels()
 	if len(vchannels) == 0 {
-		return merr.WrapErrImportSysFailedMsg("job %d has no vchannels", job.GetJobID())
+		return errors.Mark(merr.WrapErrImportSysFailedMsg("job %d has no vchannels", job.GetJobID()), errRollbackImportNoVchannels)
 	}

 	broadcaster, err := s.startBroadcastWithCollectionID(ctx, job.GetCollectionID())
@@ -2954,23 +2962,38 @@ func (s *Server) CommitImport(ctx context.Context, req *datapb.CommitImportReque
 	)
 }

-// AbortImport aborts a 2PC import job that has not yet been committed.
-// It broadcasts a RollbackImport WAL message to cancel the job.
-// Returns an error if the job is already committed or committing.
+// AbortImport rolls back a 2PC import job that has not been committed by
+// broadcasting a RollbackImport WAL message. A job that has already Failed (for
+// example because its own import failed) is still abortable, so the control plane
+// can proactively release the peer cluster's replicated Uncommitted job instead of
+// waiting for the failed source's GC self-heal (see importChecker.checkGC).
+// Committing/Completed jobs are terminal and rejected.
+//
+// Behavior change for non-CDC clusters: aborting an already-Failed 2PC job used to
+// return an error; it now succeeds and (re)broadcasts a RollbackImport, which the
+// flusher no-ops. Repeated aborts on a real-failure source therefore re-broadcast
+// each time — harmless, but not deduplicated (there is no persisted "rolled back"
+// flag in this change; that idempotency is a follow-up).
 func (s *Server) AbortImport(ctx context.Context, req *datapb.AbortImportRequest) (*commonpb.Status, error) {
 	return s.validateAndExecuteImportAction(ctx, req.GetJobId(),
 		func(job ImportJob) *commonpb.Status {
 			state := job.GetState()
+			// Idempotent only for a job that was previously Uncommitted and then
+			// user-aborted (its reason is rewritten to importJobReasonAbortedByUser). A
+			// source that failed on its own keeps its real failure reason, so this does
+			// NOT fire for it and each abort re-broadcasts (see the note above).
 			if state == internalpb.ImportJobState_Failed && job.GetReason() == importJobReasonAbortedByUser {
 				return merr.Success()
 			}
-			if state == internalpb.ImportJobState_Failed ||
-				state == internalpb.ImportJobState_Committing ||
+			// Committed states are truly terminal and cannot be rolled back.
+			if state == internalpb.ImportJobState_Committing ||
 				state == internalpb.ImportJobState_Completed {
 				return merr.Status(merr.WrapErrImportFailed(
 					fmt.Sprintf("job %d is in terminal/committed state %s, abort not allowed", req.GetJobId(), state)))
 			}
-			return nil // proceed
+			// Uncommitted, or a Failed source (its own import failed) → broadcast the
+			// rollback so the peer cluster's replicated Uncommitted job is released.
+			return nil
 		},
 		func(ctx context.Context, job ImportJob) error {
 			mlog.Info(context.TODO(), "aborting import job via WAL broadcast")
@@ -5811,3 +5811,69 @@ type (
 	embeddedBroadcastAPI struct{ broadcaster.BroadcastAPI }
 	embeddedBroker       struct{ broker.Broker }
 )
+
+func TestAbortImport_FailedSourceBroadcastsRollback(t *testing.T) {
+	ctx := context.Background()
+	// A source whose own import failed (real reason, not user-aborted) must still be
+	// abortable, so the control plane can release the peer cluster's Uncommitted job.
+	job := &importJob{ImportJob: &datapb.ImportJob{
+		JobID: 2101, CollectionID: 300,
+		State: internalpb.ImportJobState_Failed, Reason: "disk quota exceeded",
+	}}
+	im := NewMockImportMeta(t)
+	im.EXPECT().GetJob(mock.Anything, int64(2101)).Return(job).Times(2)
+
+	server := &Server{importMeta: im, importJobLock: lock.NewKeyLock[int64]()}
+	server.stateCode.Store(commonpb.StateCode_Healthy)
+	// The broadcast is the assertion of record: a regression that short-circuits
+	// AbortImport on a Failed job (returning success without running the action)
+	// would skip the rollback and strand the peer, so success alone is not enough.
+	rollbackCalls := 0
+	var rollbackJobID int64
+	bm := mockey.Mock((*Server).broadcastRollbackImportMessage).To(
+		func(s *Server, ctx context.Context, job ImportJob) error {
+			rollbackCalls++
+			rollbackJobID = job.GetJobID()
+			return nil
+		}).Build()
+	defer bm.UnPatch()
+
+	resp, err := server.AbortImport(ctx, &datapb.AbortImportRequest{JobId: 2101})
+	assert.NoError(t, err)
+	assert.True(t, merr.Ok(resp))
+	assert.Equal(t, 1, rollbackCalls)
+	assert.Equal(t, int64(2101), rollbackJobID)
+}
+
+func TestAbortImport_CompletedRejected(t *testing.T) {
+	ctx := context.Background()
+	job := &importJob{ImportJob: &datapb.ImportJob{
+		JobID: 2103, State: internalpb.ImportJobState_Completed,
+	}}
+	im := NewMockImportMeta(t)
+	im.EXPECT().GetJob(mock.Anything, int64(2103)).Return(job).Once()
+
+	server := &Server{importMeta: im, importJobLock: lock.NewKeyLock[int64]()}
+	server.stateCode.Store(commonpb.StateCode_Healthy)
+
+	resp, err := server.AbortImport(ctx, &datapb.AbortImportRequest{JobId: 2103})
+	assert.NoError(t, err)
+	assert.False(t, merr.Ok(resp))
+}
+
+func TestAbortImport_CommittingRejected(t *testing.T) {
+	ctx := context.Background()
+	// Committing is mid-commit and cannot be rolled back.
+	job := &importJob{ImportJob: &datapb.ImportJob{
+		JobID: 2104, State: internalpb.ImportJobState_Committing,
+	}}
+	im := NewMockImportMeta(t)
+	im.EXPECT().GetJob(mock.Anything, int64(2104)).Return(job).Once()
+
+	server := &Server{importMeta: im, importJobLock: lock.NewKeyLock[int64]()}
+	server.stateCode.Store(commonpb.StateCode_Healthy)
+
+	resp, err := server.AbortImport(ctx, &datapb.AbortImportRequest{JobId: 2104})
+	assert.NoError(t, err)
+	assert.False(t, merr.Ok(resp))
+}