data race in coordinator

[CharlesCheung96]

=== Command to ticdc(new arch).
==================
WARNING: DATA RACE
Write at 0x00c0033bb8f0 by goroutine 1898:
  github.com/pingcap/ticdc/pkg/bootstrap.(*Bootstrapper[go.shape.struct { Statuses []*github.com/pingcap/ticdc/heartbeatpb.MaintainerStatus "protobuf:\"bytes,1,rep,name=statuses,proto3\" json:\"statuses,omitempty\"" }]).HandleBootstrapResponse()
      /root/ticdc/pkg/bootstrap/bootstrap.go:68 +0x1a5
  github.com/pingcap/ticdc/coordinator.(*Controller).onMaintainerBootstrapResponse()
      /root/ticdc/coordinator/controller.go:234 +0x277
  github.com/pingcap/ticdc/coordinator.(*Controller).onMessage()
      /root/ticdc/coordinator/controller.go:185 +0x2fb
  github.com/pingcap/ticdc/coordinator.(*Controller).HandleEvent()
      /root/ticdc/coordinator/controller.go:145 +0x25d
  github.com/pingcap/ticdc/coordinator.(*StreamHandler).Handle()
      /root/ticdc/coordinator/coodinator_event.go:51 +0xb6
  github.com/pingcap/ticdc/utils/dynstream.(*stream[go.shape.int,go.shape.string,go.shape.*uint8,go.shape.*uint8,go.shape.*uint8]).handleLoop()
      /root/ticdc/utils/dynstream/stream.go:319 +0x947
  github.com/pingcap/ticdc/utils/dynstream.(*stream[go.shape.int,go.shape.string,go.shape.*uint8,go.shape.*uint8,go.shape.*uint8]).start.gowrap1()
      /root/ticdc/utils/dynstream/stream.go:213 +0xaa

Previous read at 0x00c0033bb8f0 by goroutine 1781:
  github.com/pingcap/ticdc/pkg/bootstrap.(*Bootstrapper[go.shape.struct { Statuses []*github.com/pingcap/ticdc/heartbeatpb.MaintainerStatus "protobuf:\"bytes,1,rep,name=statuses,proto3\" json:\"statuses,omitempty\"" }]).checkAllCaptureInitialized()
      /root/ticdc/pkg/bootstrap/bootstrap.go:144 +0x1b0
  github.com/pingcap/ticdc/pkg/bootstrap.(*Bootstrapper[go.shape.struct { Statuses []*github.com/pingcap/ticdc/heartbeatpb.MaintainerStatus "protobuf:\"bytes,1,rep,name=statuses,proto3\" json:\"statuses,omitempty\"" }]).fistBootstrap()
      /root/ticdc/pkg/bootstrap/bootstrap.go:157 +0xe6
  github.com/pingcap/ticdc/pkg/bootstrap.(*Bootstrapper[go.shape.struct { Statuses []*github.com/pingcap/ticdc/heartbeatpb.MaintainerStatus "protobuf:\"bytes,1,rep,name=statuses,proto3\" json:\"statuses,omitempty\"" }]).HandleRemoveNodes()
      /root/ticdc/pkg/bootstrap/bootstrap.go:108 +0x7c9
  github.com/pingcap/ticdc/coordinator.(*Controller).onNodeChanged()
      /root/ticdc/coordinator/controller.go:218 +0xa04
  github.com/pingcap/ticdc/coordinator.NewController.func1()
      /root/ticdc/coordinator/controller.go:110 +0x44
  github.com/pingcap/ticdc/server/watcher.(*NodeManager).Tick()
      /root/ticdc/server/watcher/module_node_manager.go:98 +0xc13
  github.com/pingcap/tiflow/pkg/orchestrator.(*EtcdWorker).Run()
  /root/go/pkg/mod/github.com/pingcap/tiflow@v0.0.0-20241023094956-dd2d54ad4c19/pkg/orchestrator/etcd_worker.go:290 +0x1713
  github.com/pingcap/ticdc/server/watcher.(*EtcdWatcher).RunEtcdWorker()
      /root/ticdc/server/watcher/etcd_watcher.go:60 +0x389
  github.com/pingcap/ticdc/server/watcher.(*NodeManager).Run()
      /root/ticdc/server/watcher/module_node_manager.go:117 +0x636
  github.com/pingcap/ticdc/server.(*server).Run.(*server).Run.func3.func4()
      /root/ticdc/server/server.go:171 +0x1a3
  golang.org/x/sync/errgroup.(*Group).Go.func1()
      /root/go/pkg/mod/golang.org/x/sync@v0.8.0/errgroup/errgroup.go:78 +0x91

Goroutine 1898 (running) created at:
  github.com/pingcap/ticdc/utils/dynstream.(*stream[go.shape.int,go.shape.string,go.shape.*uint8,go.shape.*uint8,go.shape.*uint8]).start()
      /root/ticdc/utils/dynstream/stream.go:213 +0x244
  github.com/pingcap/ticdc/utils/dynstream.(*dynamicStreamImpl[go.shape.int,go.shape.string,go.shape.*uint8,go.shape.*uint8,go.shape.*uint8]).scheduler()
      /root/ticdc/utils/dynstream/dynamic_stream.go:313 +0x3f8
  github.com/pingcap/ticdc/utils/dynstream.(*dynamicStreamImpl[go.shape.int,go.shape.string,go.shape.*uint8,go.shape.*uint8,go.shape.*uint8]).Start.gowrap1()
      /root/ticdc/utils/dynstream/dynamic_stream.go:193 +0x44
      Goroutine 1781 (running) created at:
  golang.org/x/sync/errgroup.(*Group).Go()
      /root/go/pkg/mod/golang.org/x/sync@v0.8.0/errgroup/errgroup.go:75 +0x124
  github.com/pingcap/ticdc/server.(*server).Run.func3()
      /root/ticdc/server/server.go:169 +0x58d
  github.com/pingcap/ticdc/server.(*server).Run()
      /root/ticdc/server/server.go:173 +0x612
  github.com/pingcap/ticdc/cmd/server.(*options).run()
      /root/ticdc/cmd/server/server.go:117 +0xec5
  github.com/pingcap/ticdc/cmd/server.NewCmdServer.func1()
      /root/ticdc/cmd/server/server.go:232 +0x93
  github.com/spf13/cobra.(*Command).execute()
      /root/go/pkg/mod/github.com/spf13/cobra@v1.8.1/command.go:985 +0x10f3
  github.com/spf13/cobra.(*Command).ExecuteC()
      /root/go/pkg/mod/github.com/spf13/cobra@v1.8.1/command.go:1117 +0x655
  github.com/spf13/cobra.(*Command).Execute()
      /root/go/pkg/mod/github.com/spf13/cobra@v1.8.1/command.go:1041 +0x524
  github.com/pingcap/ticdc/cmd.main()
      /root/ticdc/cmd/main.go:123 +0x505
  github.com/pingcap/ticdc/cmd.TestRunMain()
      /root/ticdc/cmd/main_test.go:35 +0x1e4
  testing.tRunner()
      /root/go/pkg/mod/golang.org/toolchain@v0.0.1-go1.23.2.linux-amd64/src/testing/testing.go:1690 +0x226
  testing.(*T).Run.gowrap1()
      /root/go/pkg/mod/golang.org/toolchain@v0.0.1-go1.23.2.linux-amd64/src/testing/testing.go:1743 +0x44

Goroutine 1781 (running) created at:
  golang.org/x/sync/errgroup.(*Group).Go()
      /root/go/pkg/mod/golang.org/x/sync@v0.8.0/errgroup/errgroup.go:75 +0x124
  github.com/pingcap/ticdc/server.(*server).Run.func3()
      /root/ticdc/server/server.go:169 +0x58d
  github.com/pingcap/ticdc/server.(*server).Run()
      /root/ticdc/server/server.go:173 +0x612
  github.com/pingcap/ticdc/cmd/server.(*options).run()
      /root/ticdc/cmd/server/server.go:117 +0xec5
  github.com/pingcap/ticdc/cmd/server.NewCmdServer.func1()
      /root/ticdc/cmd/server/server.go:232 +0x93
  github.com/spf13/cobra.(*Command).execute()
      /root/go/pkg/mod/github.com/spf13/cobra@v1.8.1/command.go:985 +0x10f3
  github.com/spf13/cobra.(*Command).ExecuteC()
      /root/go/pkg/mod/github.com/spf13/cobra@v1.8.1/command.go:1117 +0x655
  github.com/spf13/cobra.(*Command).Execute()
      /root/go/pkg/mod/github.com/spf13/cobra@v1.8.1/command.go:1041 +0x524
  github.com/pingcap/ticdc/cmd.main()
      /root/ticdc/cmd/main.go:123 +0x505
  github.com/pingcap/ticdc/cmd.TestRunMain()
      /root/ticdc/cmd/main_test.go:35 +0x1e4
  testing.tRunner()
      /root/go/pkg/mod/golang.org/toolchain@v0.0.1-go1.23.2.linux-amd64/src/testing/testing.go:1690 +0x226
  testing.(*T).Run.gowrap1()
      /root/go/pkg/mod/golang.org/toolchain@v0.0.1-go1.23.2.linux-amd64/src/testing/testing.go:1743 +0x44
==================