ddl notifier: fix shared session by List and DeleteAndCommit (#58833)

close #58824

Author: lance6716

pkg/ddl/notifier/BUILD.bazel

@@ -36,7 +36,7 @@ go_test(
     ],
     embed = [":notifier"],
     flaky = True,
-    shard_count = 9,
+    shard_count = 10,
     deps = [
         "//pkg/ddl",
         "//pkg/ddl/session",

pkg/ddl/notifier/subscribe.go

@@ -231,12 +231,19 @@ func (n *DDLNotifier) processEvents(ctx context.Context) error {
 			}
 
 			if change.processedByFlag == n.handlersBitMap {
-				if err3 := n.store.DeleteAndCommit(
+				s3, err3 := n.sysSessionPool.Get()
+				if err3 != nil {
+					return errors.Trace(err3)
+				}
+				sess4Del := sess.NewSession(s3.(sessionctx.Context))
+				err3 = n.store.DeleteAndCommit(
 					ctx,
-					sess4List,
+					sess4Del,
 					change.ddlJobID,
 					int(change.subJobID),
-				); err3 != nil {
+				)
+				n.sysSessionPool.Put(s3)
+				if err3 != nil {
 					logutil.Logger(ctx).Error("Error deleting change",
 						zap.Int64("ddlJobID", change.ddlJobID),
 						zap.Int64("subJobID", change.subJobID),

pkg/ddl/notifier/testkit_test.go

@@ -108,12 +108,7 @@ func TestBasicPubSub(t *testing.T) {
 		return nil
 	}
 	n.RegisterHandler(notifier.TestHandlerID, testHandler)
-
-	done := make(chan struct{})
-	go func() {
-		n.OnBecomeOwner()
-		close(done)
-	}()
+	n.OnBecomeOwner()
 
 	tk2 := testkit.NewTestKit(t, store)
 	se := sess.NewSession(tk2.Session())
@@ -138,7 +133,6 @@ func TestBasicPubSub(t *testing.T) {
 	require.Equal(t, event2, seenChanges[1])
 	require.Equal(t, event3, seenChanges[2])
 	n.OnRetireOwner()
-	<-done
 }
 
 func TestDeliverOrderAndCleanup(t *testing.T) {
@@ -186,12 +180,7 @@ func TestDeliverOrderAndCleanup(t *testing.T) {
 	n.RegisterHandler(3, h1)
 	n.RegisterHandler(4, h2)
 	n.RegisterHandler(9, h3)
-
-	done := make(chan struct{})
-	go func() {
-		n.OnBecomeOwner()
-		close(done)
-	}()
+	n.OnBecomeOwner()
 
 	tk2 := testkit.NewTestKit(t, store)
 	se := sess.NewSession(tk2.Session())
@@ -220,7 +209,6 @@ func TestDeliverOrderAndCleanup(t *testing.T) {
 	require.Equal(t, []int64{1000, 1001, 1002}, *id3)
 
 	n.OnRetireOwner()
-	<-done
 }
 
 func TestPubSub(t *testing.T) {
@@ -352,12 +340,7 @@ func Test2OwnerForAShortTime(t *testing.T) {
 		return nil
 	}
 	n.RegisterHandler(notifier.TestHandlerID, testHandler)
-
-	done := make(chan struct{})
-	go func() {
-		n.OnBecomeOwner()
-		close(done)
-	}()
+	n.OnBecomeOwner()
 
 	tk2 := testkit.NewTestKit(t, store)
 	se := sess.NewSession(tk2.Session())
@@ -383,7 +366,6 @@ func Test2OwnerForAShortTime(t *testing.T) {
 	tk2.MustQuery("SELECT * FROM test.result").Check(testkit.Rows())
 
 	n.OnRetireOwner()
-	<-done
 }
 
 func TestPaginatedList(t *testing.T) {
@@ -452,3 +434,58 @@ func TestPaginatedList(t *testing.T) {
 		return count.Load() == 8
 	}, 5*time.Second, 500*time.Millisecond)
 }
+
+func TestBeginTwice(t *testing.T) {
+	conf := new(log.Config)
+	logFilename := path.Join(t.TempDir(), "/testBeginTwice.log")
+	conf.File.Filename = logFilename
+	lg, p, e := log.InitLogger(conf)
+	require.NoError(t, e)
+	rs := log.ReplaceGlobals(lg, p)
+	defer rs()
+
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("USE test")
+	tk.MustExec("DROP TABLE IF EXISTS " + ddl.NotifierTableName)
+	tk.MustExec(ddl.NotifierTableSQL)
+
+	s := notifier.OpenTableStore("test", ddl.NotifierTableName)
+	sessionPool := util.NewSessionPool(
+		5,
+		func() (pools.Resource, error) {
+			return testkit.NewTestKit(t, store).Session(), nil
+		},
+		nil,
+		nil,
+	)
+
+	n := notifier.NewDDLNotifier(sessionPool, s, 50*time.Millisecond)
+
+	testHandler := func(context.Context, sessionctx.Context, *notifier.SchemaChangeEvent) error {
+		return nil
+	}
+	n.RegisterHandler(notifier.TestHandlerID, testHandler)
+	n.OnBecomeOwner()
+
+	tk2 := testkit.NewTestKit(t, store)
+	se := sess.NewSession(tk2.Session())
+	ctx := context.Background()
+	event1 := notifier.NewCreateTableEvent(&model.TableInfo{ID: 1000, Name: ast.NewCIStr("t1")})
+	err := notifier.PubSchemeChangeToStore(ctx, se, 1, -1, event1, s)
+	require.NoError(t, err)
+
+	// after handler processed the event, wait to ensure the record is deleted by DDL notifier
+	require.Eventually(t, func() bool {
+		changes := make([]*notifier.SchemaChange, 8)
+		result, closeFn := s.List(ctx, se)
+		count, err2 := result.Read(changes)
+		require.NoError(t, err2)
+		closeFn()
+		return count == 0
+	}, time.Second, 50*time.Millisecond)
+
+	content, err := os.ReadFile(logFilename)
+	require.NoError(t, err)
+	require.NotContains(t, string(content), "context provider not set")
+}