owner(cdc): drain warnings when closing changefeed (#11280)

close #11277
pingcap · Jun 11, 2024 · cded157 · cded157
1 parent d662c77
commit cded157
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 0 deletions.
diff --git a/cdc/owner/changefeed.go b/cdc/owner/changefeed.go
@@ -811,6 +811,20 @@ func (c *changefeed) releaseResources(ctx context.Context) {
 	c.initialized.Store(false)
 	c.isReleased = true
 
+	// when closing a changefeed, we must clean the warningCh.
+	// otherwise, the old warning errors will be handled when the reused changefeed instance is ticked again
+OUT:
+	for {
+		select {
+		case err := <-c.warningCh:
+			log.Warn("drain owner warnings",
+				zap.String("namespace", c.id.Namespace),
+				zap.String("changefeed", c.id.ID),
+				zap.Error(err))
+		default:
+			break OUT
+		}
+	}
 	log.Info("changefeed closed",
 		zap.String("namespace", c.id.Namespace),
 		zap.String("changefeed", c.id.ID),

diff --git a/cdc/owner/feed_state_manager.go b/cdc/owner/feed_state_manager.go
@@ -468,6 +468,7 @@ func (m *feedStateManager) HandleWarning(errs ...*model.RunningError) {
 				"it will be failed",
 				zap.String("namespace", m.state.GetID().Namespace),
 				zap.String("changefeed", m.state.GetID().ID),
+				zap.Time("checkpointTsAdvanced", m.checkpointTsAdvanced),
 				zap.Uint64("checkpointTs", m.state.GetChangefeedStatus().CheckpointTs),
 				zap.Duration("checkpointTime", currTime.Sub(ckptTime)),
 			)