maintainer: fix remove changefeed and some data race (#743)

Signed-off-by: dongmen <414110582@qq.com>
pingcap · Jan 2, 2025 · 11ba4e3 · 11ba4e3
1 parent 05c03b4
commit 11ba4e3
Show file tree

Hide file tree

Showing 13 changed files with 281 additions and 265 deletions.
diff --git a/.github/workflows/integration_test_mysql.yaml b/.github/workflows/integration_test_mysql.yaml
@@ -75,8 +75,8 @@ jobs:
         if: ${{ always() }}
         run: |
           DIR=$(sudo find /tmp/tidb_cdc_test/ -type d -name 'cdc_data' -exec dirname {} \;)
-          CASE=$(basename $DIR)
           [ -z "$DIR" ] && exit 0
+          CASE=$(basename $DIR)
           mkdir -p ./logs/$CASE
           cat $DIR/stdout.log
           tail -n 10 $DIR/cdc.log

diff --git a/cmd/main.go b/cmd/main.go
@@ -15,6 +15,7 @@ package main
 
 import (
 	"os"
+	"slices"
 	"strings"
 
 	"github.com/pingcap/log"
@@ -48,6 +49,7 @@ func addNewArchCommandTo(cmd *cobra.Command) {
 }
 
 func isNewArchEnabledByConfig(serverConfigFilePath string) bool {
+
 	cfg := config.GetDefaultServerConfig()
 	if len(serverConfigFilePath) > 0 {
 		// strict decode config file, but ignore debug item
@@ -80,6 +82,12 @@ func parseConfigFlagFromOSArgs() string {
 			serverConfigFilePath = os.Args[i+2]
 		}
 	}
+
+	// If the command is `cdc cli changefeed`, means it's not a server config file.
+	if slices.Contains(os.Args, "cli") && slices.Contains(os.Args, "changefeed") {
+		serverConfigFilePath = ""
+	}
+
 	return serverConfigFilePath
 }
 

diff --git a/docs/design/2024-12-20-ticdc-flow-control.md b/docs/design/2024-12-20-ticdc-flow-control.md
@@ -12,7 +12,7 @@ TiCDC processes data in two main parts:
 
 The following diagram illustrates the relationship between the **data puller** and **data sinker**:
 
-![Data Flow](./medias/flow-control-1.png)
+![Data Flow](../media/flow-control-1.png)
 <!-- The source file for this diagram: docs/design/medias/flow-control-1.puml -->
 
 In this architecture, **EventService** and **EventCollector** facilitate communication between the two parts:  

diff --git a/downstreamadapter/eventcollector/event_collector.go b/downstreamadapter/eventcollector/event_collector.go
@@ -334,7 +334,10 @@ func (c *EventCollector) mustSendDispatcherRequest(target node.ID, topic string,
 
 	if err != nil {
 		log.Info("failed to send dispatcher request message to event service, try again later",
-			zap.Stringer("target", target),
+			zap.String("changefeedID", req.Dispatcher.GetChangefeedID().ID().String()),
+			zap.Stringer("dispatcher", req.Dispatcher.GetId()),
+			zap.Any("target", target.String()),
+			zap.Any("request", req),
 			zap.Error(err))
 		// Put the request back to the channel for later retry.
 		c.dispatcherRequestChan.In() <- DispatcherRequestWithTarget{
@@ -570,6 +573,7 @@ func (d *dispatcherStat) handleHandshakeEvent(event dispatcher.DispatcherEvent,
 func (d *dispatcherStat) handleReadyEvent(event dispatcher.DispatcherEvent, eventCollector *EventCollector) {
 	d.eventServiceInfo.Lock()
 	defer d.eventServiceInfo.Unlock()
+
 	if event.GetType() != commonEvent.TypeReadyEvent {
 		log.Panic("should not happen")
 	}
@@ -658,7 +662,7 @@ func (d *dispatcherStat) unregisterDispatcher(eventCollector *EventCollector) {
 		ActionType: eventpb.ActionType_ACTION_TYPE_REMOVE,
 	})
 	// unregister from remote event service if have
-	if d.eventServiceInfo.serverID != eventCollector.serverId {
+	if d.eventServiceInfo.serverID != "" && d.eventServiceInfo.serverID != eventCollector.serverId {
 		eventCollector.mustSendDispatcherRequest(d.eventServiceInfo.serverID, eventServiceTopic, DispatcherRequest{
 			Dispatcher: d.target,
 			ActionType: eventpb.ActionType_ACTION_TYPE_REMOVE,