capture(ticdc): disable best effort graceful shutdown (#6647)

Because TiCDC release-6.2 does not support cross version upgrade and stops service during upgrade, it is impossible to perform graceful shutdown. Signed-off-by: Neil Shen <[email protected]>
pingcap · Aug 8, 2022 · c0eaa00 · c0eaa00
1 parent bd21a6e
commit c0eaa00
Show file tree

Hide file tree

Showing 2 changed files with 50 additions and 1 deletion.
diff --git a/cdc/capture/capture.go b/cdc/capture/capture.go
@@ -76,6 +76,7 @@ type captureImpl struct {
 	processorManager processor.Manager
 	liveness         model.Liveness
 	config           *config.ServerConfig
+	disableDrain     bool
 
 	pdEndpoints     []string
 	ownerMu         sync.Mutex
@@ -114,13 +115,19 @@ type captureImpl struct {
 	newOwner func(upstreamManager *upstream.Manager) owner.Owner
 }
 
+// disableDrain is a boolean that disable capture drain feature.
+// Because TiCDC release-6.2 does not support cross version upgrade and stops
+// service during upgrade, it is impossible to perform graceful shutdown.
+const disableDrain = true
+
 // NewCapture returns a new Capture instance
 func NewCapture(pdEndpoints []string,
 	etcdClient etcd.CDCEtcdClient,
 	grpcService *p2p.ServerWrapper,
 ) Capture {
 	conf := config.GetGlobalServerConfig()
 	return &captureImpl{
+		disableDrain:        disableDrain,
 		config:              config.GetGlobalServerConfig(),
 		liveness:            model.LivenessCaptureAlive,
 		EtcdClient:          etcdClient,
@@ -666,7 +673,7 @@ func (c *captureImpl) Drain(ctx context.Context) <-chan struct{} {
 }
 
 func (c *captureImpl) drainImpl(ctx context.Context) bool {
-	if !c.config.Debug.EnableSchedulerV3 {
+	if !c.config.Debug.EnableSchedulerV3 || c.disableDrain {
 		// Skip drain as two phase scheduler is disabled.
 		return true
 	}

diff --git a/cdc/capture/capture_test.go b/cdc/capture/capture_test.go
@@ -461,3 +461,45 @@ func TestCampaignLiveness(t *testing.T) {
 
 	wg.Wait()
 }
+
+func TestDisableDrainByDefault(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	mo := mock_owner.NewMockOwner(ctrl)
+	mm := mock_processor.NewMockManager(ctrl)
+	me := mock_etcd.NewMockCDCEtcdClient(ctrl)
+	cp := &captureImpl{
+		EtcdClient: me,
+		info: &model.CaptureInfo{
+			ID:            "capture-for-test",
+			AdvertiseAddr: "127.0.0.1", Version: "test",
+		},
+		processorManager: mm,
+		owner:            mo,
+		config:           config.GetDefaultServerConfig(),
+		disableDrain:     disableDrain,
+	}
+	cp.config.Debug.EnableSchedulerV3 = true
+	require.Equal(t, model.LivenessCaptureAlive, cp.Liveness())
+
+	// Owner can not be resigned.
+	mo.EXPECT().Query(gomock.Any(), gomock.Any()).Do(func(
+		query *owner.Query, done chan<- error,
+	) {
+		// Two captures to allow owner resign.
+		query.Data = []*model.CaptureInfo{{}, {}}
+		close(done)
+	}).AnyTimes()
+	mo.EXPECT().AsyncStop().AnyTimes()
+
+	done := cp.Drain(ctx)
+
+	// Must skipping drain by default.
+	select {
+	case <-time.After(3 * time.Second):
+		require.Fail(t, "timeout")
+	case <-done:
+	}
+}