fix cdc watermark add failure handling (#25925)

- Updated pkg/cdc/watermark_updater.go: execAddWM now returns immediately when the
catalog INSERT fails, so it no longer writes the watermark into cacheCommitted. The
existing onJobs cleanup path then propagates the error to the waiting job.

- Updated pkg/cdc/watermark_updater_test.go: added
TestAuditAddWatermarkFailureIsReturnedAndNotCached, covering the case where SELECT
returns no watermark row, INSERT fails with an injected error, the job receives that
error, and the failed watermark is not cached.

Approved by: @gouhongshen

Author: jiangxinmeng1

pkg/cdc/watermark_updater.go

@@ -813,6 +813,7 @@ func (u *CDCWatermarkUpdater) execAddWM() (errMsg string, err error) {
 	err = u.ie.Exec(ctx, addSql, ie.SessionOverrideOptions{})
 	if err != nil {
 		errMsg = fmt.Sprintf("add sql \"%s\" failed", addSql)
+		return
 	}
 	u.Lock()
 	defer u.Unlock()

pkg/cdc/watermark_updater_test.go

@@ -48,6 +48,10 @@ type retryableMockExecutor struct {
 	lastSQL       string
 }
 
+type failAddWatermarkExecutor struct {
+	insertErr error
+}
+
 func (m *retryableMockExecutor) Exec(_ context.Context, sql string, _ ie.SessionOverrideOptions) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -66,6 +70,30 @@ func (m *retryableMockExecutor) Query(_ context.Context, _ string, _ ie.SessionO
 
 func (m *retryableMockExecutor) ApplySessionOverride(_ ie.SessionOverrideOptions) {}
 
+func (m *failAddWatermarkExecutor) Exec(_ context.Context, sql string, _ ie.SessionOverrideOptions) error {
+	if strings.HasPrefix(sql, "INSERT INTO `mo_catalog`.`mo_cdc_watermark`") {
+		return m.insertErr
+	}
+	return nil
+}
+
+func (m *failAddWatermarkExecutor) Query(_ context.Context, sql string, _ ie.SessionOverrideOptions) ie.InternalExecResult {
+	if strings.HasPrefix(sql, "SELECT") {
+		return &InternalExecResultForTest{
+			resultSet: &MysqlResultSetForTest{
+				Data: [][]interface{}{},
+			},
+		}
+	}
+	return &InternalExecResultForTest{
+		resultSet: &MysqlResultSetForTest{
+			Data: [][]interface{}{},
+		},
+	}
+}
+
+func (m *failAddWatermarkExecutor) ApplySessionOverride(_ ie.SessionOverrideOptions) {}
+
 func newWmMockSQLExecutor() *wmMockSQLExecutor {
 	return &wmMockSQLExecutor{
 		mp: make(map[string]string),
@@ -116,6 +144,31 @@ func (m *wmMockSQLExecutor) Query(ctx context.Context, sql string, pts ie.Sessio
 
 func (m *wmMockSQLExecutor) ApplySessionOverride(opts ie.SessionOverrideOptions) {}
 
+func TestAuditAddWatermarkFailureIsReturnedAndNotCached(t *testing.T) {
+	insertErr := moerr.NewInternalErrorNoCtx("injected watermark insert failure")
+	updater := NewCDCWatermarkUpdater("add-failure", &failAddWatermarkExecutor{
+		insertErr: insertErr,
+	})
+	key := WatermarkKey{
+		AccountId: 1,
+		TaskId:    "task",
+		DBName:    "db",
+		TableName: "tbl",
+	}
+	watermark := types.BuildTS(10, 1)
+	job := NewGetOrAddCommittedWMJob(context.Background(), &key, &watermark)
+
+	updater.onJobs(job)
+
+	result := job.GetResult()
+	require.ErrorIs(t, result.Err, insertErr)
+
+	updater.RLock()
+	_, ok := updater.cacheCommitted[key]
+	updater.RUnlock()
+	require.False(t, ok)
+}
+
 func TestWatermarkUpdater_CommitRetrySuccess(t *testing.T) {
 	exec := &retryableMockExecutor{failRemaining: 1}
 	updater := NewCDCWatermarkUpdater("retry-success", exec)