fix CI

Signed-off-by: lance6716 <[email protected]>

Author: lance6716

pkg/ddl/notifier/store.go

@@ -35,7 +35,8 @@ type Store interface {
 		se *sess.Session,
 		ddlJobID int64,
 		multiSchemaChangeID int64,
-		processedBy uint64,
+		oldProcessedBy uint64,
+		newProcessedBy uint64,
 	) error
 	DeleteAndCommit(ctx context.Context, se *sess.Session, ddlJobID int64, multiSchemaChangeID int) error
 	// List will start a transaction of given session and read all schema changes
@@ -87,17 +88,28 @@ func (t *tableStore) UpdateProcessed(
 	se *sess.Session,
 	ddlJobID int64,
 	multiSchemaChangeID int64,
-	processedBy uint64,
+	oldProcessedBy uint64,
+	newProcessedBy uint64,
 ) error {
 	sql := fmt.Sprintf(`
 		UPDATE %s.%s
 		SET processed_by_flag = %d
-		WHERE ddl_job_id = %d AND sub_job_id = %d`,
+		WHERE ddl_job_id = %d AND sub_job_id = %d AND processed_by_flag = %d`,
 		t.db, t.table,
-		processedBy,
-		ddlJobID, multiSchemaChangeID)
+		newProcessedBy,
+		ddlJobID, multiSchemaChangeID, oldProcessedBy,
+	)
 	_, err := se.Execute(ctx, sql, "ddl_notifier")
-	return err
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if se.GetSessionVars().StmtCtx.AffectedRows() == 0 {
+		return errors.Errorf(
+			"failed to update processed_by_flag, maybe the row has been updated by other owner. ddl_job_id: %d, sub_job_id: %d",
+			ddlJobID, multiSchemaChangeID,
+		)
+	}
+	return nil
 }
 
 // DeleteAndCommit implements Store interface.

pkg/ddl/notifier/subscribe.go

@@ -298,17 +298,14 @@ func (n *DDLNotifier) processEventForHandler(
 			zap.Duration("duration", time.Since(now)))
 	}
 
-	if err = n.store.UpdateProcessed(
+	return errors.Trace(n.store.UpdateProcessed(
 		ctx,
 		session,
 		change.ddlJobID,
 		change.subJobID,
+		change.processedByFlag,
 		newFlag,
-	); err != nil {
-		return errors.Trace(err)
-	}
-
-	return nil
+	))
 }
 
 // Stop stops the background loop.

pkg/ddl/notifier/testkit_test.go

@@ -318,9 +318,9 @@ func Test2OwnerForAShortTime(t *testing.T) {
 
 	s := notifier.OpenTableStore("test", ddl.NotifierTableName)
 	sessionPool := util.NewSessionPool(
-		1,
+		4,
 		func() (pools.Resource, error) {
-			return tk.Session(), nil
+			return testkit.NewTestKit(t, store).Session(), nil
 		},
 		nil,
 		nil,
@@ -360,7 +360,7 @@ func Test2OwnerForAShortTime(t *testing.T) {
 		if !bytes.Contains(content, []byte("Error processing change")) {
 			return false
 		}
-		return bytes.Contains(content, []byte("Write conflict"))
+		return bytes.Contains(content, []byte("maybe the row has been updated by other owner"))
 	}, time.Second, 25*time.Millisecond)
 	// the handler should not commit
 	tk2.MustQuery("SELECT * FROM test.result").Check(testkit.Rows())