puller: close kvclient correctly when stopping a processor (#11957)

cdc/processor/processor.go

@@ -1089,4 +1089,8 @@ func (d *ddlHandler) Run(ctx context.Context, _ ...chan<- error) error {
 
 func (d *ddlHandler) WaitForReady(_ context.Context) {}
 
-func (d *ddlHandler) Close() {}
+func (d *ddlHandler) Close() {
+	if d.puller != nil {
+		d.puller.Close()
+	}
+}

cdc/processor/sourcemanager/manager.go

@@ -280,11 +280,15 @@ func (m *SourceManager) Close() {
 		zap.String("changefeed", m.changefeedID.ID))
 
 	start := time.Now()
-	m.tablePullers.Range(func(span tablepb.Span, value interface{}) bool {
-		value.(pullerwrapper.Wrapper).Close()
-		return true
-	})
-	log.Info("All pullers have been closed",
+	if m.multiplexing {
+		m.multiplexingPuller.puller.Close()
+	} else {
+		m.tablePullers.Range(func(span tablepb.Span, value interface{}) bool {
+			value.(pullerwrapper.Wrapper).Close()
+			return true
+		})
+	}
+	log.Info("SourceManager puller have been closed",
 		zap.String("namespace", m.changefeedID.Namespace),
 		zap.String("changefeed", m.changefeedID.ID),
 		zap.Duration("cost", time.Since(start)))

cdc/puller/ddl_puller.go

@@ -97,6 +97,16 @@ func (p *ddlJobPullerImpl) Run(ctx context.Context, _ ...chan<- error) error {
 	return p.run(ctx)
 }
 
+// WaitForReady implements util.Runnable.
+func (p *ddlJobPullerImpl) WaitForReady(_ context.Context) {}
+
+// Close implements util.Runnable.
+func (p *ddlJobPullerImpl) Close() {
+	if p.multiplexing {
+		p.multiplexingPuller.Close()
+	}
+}
+
 func (p *ddlJobPullerImpl) handleRawKVEntry(ctx context.Context, ddlRawKV *model.RawKVEntry) error {
 	if ddlRawKV == nil {
 		return nil
@@ -185,12 +195,6 @@ func (p *ddlJobPullerImpl) runMultiplexing(ctx context.Context) error {
 	return eg.Wait()
 }
 
-// WaitForReady implements util.Runnable.
-func (p *ddlJobPullerImpl) WaitForReady(_ context.Context) {}
-
-// Close implements util.Runnable.
-func (p *ddlJobPullerImpl) Close() {}
-
 // Output the DDL job entry, it contains the DDL job and the error.
 func (p *ddlJobPullerImpl) Output() <-chan *model.DDLJobEntry {
 	return p.outputCh
@@ -794,6 +798,12 @@ func (h *ddlPullerImpl) Run(ctx context.Context) error {
 		zap.String("changefeed", h.changefeedID.ID),
 		zap.Uint64("resolvedTS", atomic.LoadUint64(&h.resolvedTS)))
 
+	defer func() {
+		log.Info("DDL puller stopped",
+			zap.String("namespace", h.changefeedID.Namespace),
+			zap.String("changefeed", h.changefeedID.ID))
+	}()
+
 	return g.Wait()
 }
 
@@ -811,10 +821,13 @@ func (h *ddlPullerImpl) PopFrontDDL() (uint64, *timodel.Job) {
 
 // Close the ddl puller, release all resources.
 func (h *ddlPullerImpl) Close() {
-	log.Info("close the ddl puller",
+	h.cancel()
+	if h.ddlJobPuller != nil {
+		h.ddlJobPuller.Close()
+	}
+	log.Info("DDL puller closed",
 		zap.String("namespace", h.changefeedID.Namespace),
 		zap.String("changefeed", h.changefeedID.ID))
-	h.cancel()
 }
 
 func (h *ddlPullerImpl) ResolvedTs() uint64 {

cdc/puller/multiplexing_puller.go

@@ -277,6 +277,16 @@ func (p *MultiplexingPuller) run(ctx context.Context, includeClient bool) error
 	return g.Wait()
 }
 
+// Close closes the puller.
+func (p *MultiplexingPuller) Close() {
+	if p.client != nil {
+		p.client.Close()
+	}
+	log.Info("MultiplexingPuller is closed",
+		zap.String("namespace", p.changefeed.Namespace),
+		zap.String("changefeed", p.changefeed.ID))
+}
+
 func (p *MultiplexingPuller) handleInputCh(ctx context.Context, inputCh <-chan kv.MultiplexingEvent) error {
 	for {
 		var e kv.MultiplexingEvent