Fix a PANIC in coordinator.

api/v2/coordinator.go

@@ -29,7 +29,7 @@ import (
 func (h *OpenAPIV2) ResignOwner(c *gin.Context) {
 	o, _ := h.server.GetCoordinator()
 	if o != nil {
-		o.AsyncStop()
+		o.Stop()
 	}
 
 	c.JSON(getStatus(c), &EmptyResponse{})

cmd/cdc/server/server.go

@@ -112,6 +112,16 @@ func (o *options) run(cmd *cobra.Command) error {
 	log.Info("TiCDC(new arch) server created",
 		zap.Strings("pd", o.pdEndpoints), zap.Stringer("config", o.serverConfig))
 
+	// shutdown is used to notify the server to shutdown (by closing the context) when receiving the signal, and exit the process.
+	shutdown := func() <-chan struct{} {
+		done := make(chan struct{})
+		cancel()
+		close(done)
+		return done
+	}
+	// Gracefully shutdown the server when receiving the signal, and exit the process.
+	util.InitSignalHandling(shutdown, cancel)
+
 	err = svr.Run(ctx)
 	if err != nil && !errors.Is(errors.Cause(err), context.Canceled) {
 		log.Warn("cdc server exits with error", zap.Error(err))

coordinator/coordinator.go

@@ -123,7 +123,6 @@ func New(node *node.Info,
 	mc.RegisterHandler(messaging.CoordinatorTopic, c.recvMessages)
 
 	c.taskScheduler = threadpool.NewThreadPoolDefault()
-	c.closed.Store(false)
 
 	controller := NewController(
 		c.version,
@@ -147,18 +146,41 @@ func New(node *node.Info,
 				log.Info("Coordinator changed, and I am not the coordinator, stop myself",
 					zap.String("selfID", string(c.nodeInfo.ID)),
 					zap.String("newCoordinatorID", newCoordinatorID))
-				c.AsyncStop()
+				c.Stop()
 			}
 		})
 
 	return c
 }
 
-func (c *coordinator) recvMessages(_ context.Context, msg *messaging.TargetMessage) error {
+func (c *coordinator) recvMessages(ctx context.Context, msg *messaging.TargetMessage) error {
 	if c.closed.Load() {
 		return nil
 	}
-	c.eventCh.In() <- &Event{message: msg}
+
+	defer func() {
+		if r := recover(); r != nil {
+			// There is chance that:
+			// 1. Just before the c.eventCh is closed, the recvMessages is called
+			// 2. Then the goroutine(call it g1) that calls recvMessages is scheduled out by runtime, and the msg is in flight
+			// 3. The c.eventCh is closed by another goroutine(call it g2)
+			// 4. g1 is scheduled back by runtime, and the msg is sent to the closed channel
+			// 5. g1 panics
+			// To avoid the panic, we have two choices:
+			// 1. Use a mutex to protect this function, but it will reduce the throughput
+			// 2. Recover the panic, and log the error
+			// We choose the second option here.
+			log.Error("panic in recvMessages", zap.Any("msg", msg), zap.Any("panic", r))
+		}
+	}()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+		c.eventCh.In() <- &Event{message: msg}
+	}
+
 	return nil
 }
 
@@ -385,13 +407,14 @@ func (c *coordinator) GetChangefeed(ctx context.Context, changefeedDisplayName c
 	return c.controller.GetChangefeed(ctx, changefeedDisplayName)
 }
 
-func (c *coordinator) AsyncStop() {
+func (c *coordinator) Stop() {
 	if c.closed.CompareAndSwap(false, true) {
 		c.mc.DeRegisterHandler(messaging.CoordinatorTopic)
 		c.controller.Stop()
 		c.taskScheduler.Stop()
-		c.eventCh.CloseAndDrain()
 		c.cancel()
+		// close eventCh after cancel, to avoid send or get event from the channel
+		c.eventCh.CloseAndDrain()
 	}
 }
 

coordinator/coordinator_test.go

@@ -22,6 +22,7 @@ import (
 	"net/http"
 	"net/http/pprof"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -477,6 +478,132 @@ func TestBootstrapWithUnStoppedChangefeed(t *testing.T) {
 	}, waitTime, time.Millisecond*5)
 }
 
+func TestConcurrentStopAndSendEvents(t *testing.T) {
+	// Setup context
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Initialize node info
+	info := node.NewInfo("127.0.0.1:28600", "")
+	etcdClient := newMockEtcdClient(string(info.ID))
+	nodeManager := watcher.NewNodeManager(nil, etcdClient)
+	appcontext.SetService(watcher.NodeManagerName, nodeManager)
+	nodeManager.GetAliveNodes()[info.ID] = info
+
+	// Initialize message center
+	mc := messaging.NewMessageCenter(ctx, info.ID, 0, config.NewDefaultMessageCenterConfig(), nil)
+	mc.Run(ctx)
+	defer mc.Close()
+	appcontext.SetService(appcontext.MessageCenter, mc)
+
+	// Initialize backend
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	backend := mock_changefeed.NewMockBackend(ctrl)
+	backend.EXPECT().GetAllChangefeeds(gomock.Any()).Return(map[common.ChangeFeedID]*changefeed.ChangefeedMetaWrapper{}, nil).AnyTimes()
+
+	// Create coordinator
+	cr := New(info, &mockPdClient{}, pdutil.NewClock4Test(), backend, "test-gc-service", 100, 10000, time.Millisecond*10)
+	co := cr.(*coordinator)
+
+	// Number of goroutines for each operation
+	const (
+		sendEventGoroutines = 10
+		stopGoroutines      = 5
+		eventsPerGoroutine  = 100
+	)
+
+	var wg sync.WaitGroup
+	wg.Add(sendEventGoroutines + stopGoroutines)
+
+	// Start the coordinator
+	ctxRun, cancelRun := context.WithCancel(ctx)
+	go func() {
+		err := cr.Run(ctxRun)
+		if err != nil && err != context.Canceled {
+			t.Errorf("Coordinator Run returned unexpected error: %v", err)
+		}
+	}()
+
+	// Give coordinator some time to initialize
+	time.Sleep(100 * time.Millisecond)
+
+	// Start goroutines to send events
+	for i := 0; i < sendEventGoroutines; i++ {
+		go func(id int) {
+			defer wg.Done()
+			defer func() {
+				// Recover from potential panics
+				if r := recover(); r != nil {
+					t.Errorf("Panic in send event goroutine %d: %v", id, r)
+				}
+			}()
+
+			for j := 0; j < eventsPerGoroutine; j++ {
+				// Try to send an event
+				if co.closed.Load() {
+					// Coordinator is already closed, stop sending
+					return
+				}
+
+				msg := &messaging.TargetMessage{
+					Topic: messaging.CoordinatorTopic,
+					Type:  messaging.TypeMaintainerHeartbeatRequest,
+				}
+
+				// Use recvMessages to send event to channel
+				err := co.recvMessages(ctx, msg)
+				if err != nil && err != context.Canceled {
+					t.Logf("Failed to send event in goroutine %d: %v", id, err)
+				}
+
+				// Small sleep to increase chance of race conditions
+				time.Sleep(time.Millisecond)
+			}
+		}(i)
+	}
+
+	// Start goroutines to stop the coordinator
+	for i := 0; i < stopGoroutines; i++ {
+		go func(id int) {
+			defer wg.Done()
+			// Small delay to ensure some events are sent first
+			time.Sleep(time.Duration(10+id*5) * time.Millisecond)
+			co.Stop()
+		}(i)
+	}
+
+	// Wait for all goroutines to complete
+	wg.Wait()
+
+	// Cancel the context to ensure the coordinator stops
+	cancelRun()
+
+	// Give some time for the coordinator to fully stop
+	time.Sleep(100 * time.Millisecond)
+
+	// Verify that the coordinator is closed
+	require.True(t, co.closed.Load())
+
+	// Verify that event channel is closed
+	select {
+	case _, ok := <-co.eventCh.Out():
+		require.False(t, ok, "Event channel should be closed")
+	default:
+		// Channel might be already drained, which is fine
+	}
+
+	// Try sending another event - should not panic but may return error
+	msg := &messaging.TargetMessage{
+		Topic: messaging.CoordinatorTopic,
+		Type:  messaging.TypeMaintainerHeartbeatRequest,
+	}
+
+	err := co.recvMessages(ctx, msg)
+	require.NoError(t, err)
+	require.True(t, co.closed.Load())
+}
+
 type maintainNode struct {
 	cancel  context.CancelFunc
 	mc      messaging.MessageCenter

downstreamadapter/dispatchermanager/heartbeat_collector.go

@@ -212,6 +212,7 @@ func (c *HeartBeatCollector) RecvMessages(_ context.Context, msg *messaging.Targ
 }
 
 func (c *HeartBeatCollector) Close() {
+	log.Info("heartbeat collector is closing")
 	c.mc.DeRegisterHandler(messaging.HeartbeatCollectorTopic)
 	c.cancel()
 	c.wg.Wait()

downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go

@@ -273,7 +273,9 @@ func (m *DispatcherOrchestrator) sendResponse(to node.ID, topic string, msg mess
 }
 
 func (m *DispatcherOrchestrator) Close() {
+	log.Info("dispatcher orchestrator is closing")
 	m.mc.DeRegisterHandler(messaging.DispatcherManagerManagerTopic)
+	log.Info("dispatcher orchestrator closed")
 }
 
 // handleDispatcherError creates and sends an error response for create dispatcher-related failures

downstreamadapter/eventcollector/event_collector.go

@@ -161,9 +161,9 @@ func (c *EventCollector) Run(ctx context.Context) {
 }
 
 func (c *EventCollector) Close() {
+	log.Info("event collector is closing")
 	c.cancel()
 	c.ds.Close()
-
 	c.changefeedIDMap.Range(func(key, value any) bool {
 		cfID := value.(common.ChangeFeedID)
 		// Remove metrics for the changefeed.

pkg/messaging/message_center.go

@@ -278,6 +278,7 @@ func (mc *messageCenter) ReceiveCmd() (*TargetMessage, error) {
 
 // Close stops the grpc server and stops all the connections to the remote targets.
 func (mc *messageCenter) Close() {
+	log.Info("message center is closing", zap.Stringer("id", mc.id))
 	mc.remoteTargets.RLock()
 	defer mc.remoteTargets.RUnlock()
 

pkg/messaging/proto/message.proto

@@ -14,7 +14,6 @@ message Message {
     int32 type = 5;
     // topic is the destination of the message, it is used to route the message to the correct handler.
     string topic = 6;
-    // TODO, change to real types
     repeated bytes payload = 7;
 }
 

pkg/messaging/target.go

@@ -116,36 +116,36 @@ func (s *remoteMessageTarget) Epoch() uint64 {
 func (s *remoteMessageTarget) sendEvent(msg ...*TargetMessage) error {
 	if !s.eventSender.ready.Load() {
 		s.connectionNotfoundErrorCounter.Inc()
-		return AppError{Type: ErrorTypeConnectionNotFound, Reason: "Stream has not been initialized"}
+		return AppError{Type: ErrorTypeConnectionNotFound, Reason: fmt.Sprintf("Stream has not been initialized, target: %s", s.targetId)}
 	}
 	select {
 	case <-s.ctx.Done():
 		s.connectionNotfoundErrorCounter.Inc()
-		return AppError{Type: ErrorTypeConnectionNotFound, Reason: "Stream has been closed"}
+		return AppError{Type: ErrorTypeConnectionNotFound, Reason: fmt.Sprintf("Stream has been closed, target: %s", s.targetId)}
 	case s.sendEventCh <- s.newMessage(msg...):
 		s.sendEventCounter.Add(float64(len(msg)))
 		return nil
 	default:
 		s.congestedEventErrorCounter.Inc()
-		return AppError{Type: ErrorTypeMessageCongested, Reason: "Send event message is congested"}
+		return AppError{Type: ErrorTypeMessageCongested, Reason: fmt.Sprintf("Send event message is congested, target: %s", s.targetId)}
 	}
 }
 
 func (s *remoteMessageTarget) sendCommand(msg ...*TargetMessage) error {
 	if !s.commandSender.ready.Load() {
 		s.connectionNotfoundErrorCounter.Inc()
-		return AppError{Type: ErrorTypeConnectionNotFound, Reason: "Stream has not been initialized"}
+		return AppError{Type: ErrorTypeConnectionNotFound, Reason: fmt.Sprintf("Stream has not been initialized, target: %s", s.targetId)}
 	}
 	select {
 	case <-s.ctx.Done():
 		s.connectionNotfoundErrorCounter.Inc()
-		return AppError{Type: ErrorTypeConnectionNotFound, Reason: "Stream has been closed"}
+		return AppError{Type: ErrorTypeConnectionNotFound, Reason: fmt.Sprintf("Stream has been closed, target: %s", s.targetId)}
 	case s.sendCmdCh <- s.newMessage(msg...):
 		s.sendCmdCounter.Add(float64(len(msg)))
 		return nil
 	default:
 		s.congestedCmdErrorCounter.Inc()
-		return AppError{Type: ErrorTypeMessageCongested, Reason: "Send command message is congested"}
+		return AppError{Type: ErrorTypeMessageCongested, Reason: fmt.Sprintf("Send command message is congested, target: %s", s.targetId)}
 	}
 }
 

pkg/server/coordinator.go

@@ -27,7 +27,7 @@ import (
 // 4. manager gc safe point
 // 5. response for open API call
 type Coordinator interface {
-	AsyncStop()
+	Stop()
 	// Run handles messages
 	Run(ctx context.Context) error
 	// ListChangefeeds returns all changefeeds

server/module_election.go

@@ -144,7 +144,7 @@ func (e *elector) campaignCoordinator(ctx context.Context) error {
 		e.svr.setCoordinator(co)
 		err = co.Run(ctx)
 		// When coordinator exits, we need to stop it.
-		e.svr.coordinator.AsyncStop()
+		e.svr.coordinator.Stop()
 		e.svr.setCoordinator(nil)
 		log.Info("coordinator stop", zap.String("captureID", string(e.svr.info.ID)),
 			zap.Int64("coordinatorVersion", coordinatorVersion), zap.Error(err))