cdc: drain waits for owner resign and table move out

Makefile

@@ -267,6 +267,8 @@ swagger-spec: tools/bin/swag
 
 generate_mock: tools/bin/mockgen
 	tools/bin/mockgen -source cdc/owner/owner.go -destination cdc/owner/mock/owner_mock.go
+	tools/bin/mockgen -source cdc/processor/manager.go -destination cdc/processor/mock/manager_mock.go
+	tools/bin/mockgen -source cdc/capture/capture.go -destination cdc/capture/mock/capture_mock.go
 
 clean:
 	go clean -i ./...

cdc/api/owner/owner.go

@@ -78,11 +78,11 @@ func (c ChangefeedResp) MarshalJSON() ([]byte, error) {
 
 // ownerAPI provides owner APIs.
 type ownerAPI struct {
-	capture *capture.Capture
+	capture capture.Capture
 }
 
 // RegisterOwnerAPIRoutes registers routes for owner APIs.
-func RegisterOwnerAPIRoutes(router *gin.Engine, capture *capture.Capture) {
+func RegisterOwnerAPIRoutes(router *gin.Engine, capture capture.Capture) {
 	ownerAPI := ownerAPI{capture: capture}
 	owner := router.Group("/capture/owner")
 
@@ -241,13 +241,13 @@ func (h *ownerAPI) handleChangefeedQuery(w http.ResponseWriter, req *http.Reques
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	cfInfo, err := h.capture.EtcdClient.GetChangeFeedInfo(ctx, changefeedID)
+	cfInfo, err := h.capture.GetEtcdClient().GetChangeFeedInfo(ctx, changefeedID)
 	if err != nil && cerror.ErrChangeFeedNotExists.NotEqual(err) {
 		api.WriteError(w, http.StatusBadRequest,
 			cerror.ErrAPIInvalidParam.GenWithStack("invalid changefeed id: %s", changefeedID))
 		return
 	}
-	cfStatus, _, err := h.capture.EtcdClient.GetChangeFeedStatus(ctx, changefeedID)
+	cfStatus, _, err := h.capture.GetEtcdClient().GetChangeFeedStatus(ctx, changefeedID)
 	if err != nil && cerror.ErrChangeFeedNotExists.NotEqual(err) {
 		api.WriteError(w, http.StatusBadRequest, err)
 		return

cdc/api/status/status.go

@@ -38,11 +38,11 @@ type status struct {
 }
 
 type statusAPI struct {
-	capture *capture.Capture
+	capture capture.Capture
 }
 
 // RegisterStatusAPIRoutes registers routes for status.
-func RegisterStatusAPIRoutes(router *gin.Engine, capture *capture.Capture) {
+func RegisterStatusAPIRoutes(router *gin.Engine, capture capture.Capture) {
 	statusAPI := statusAPI{capture: capture}
 	router.GET("/status", gin.WrapF(statusAPI.handleStatus))
 	router.GET("/debug/info", gin.WrapF(statusAPI.handleDebugInfo))
@@ -64,7 +64,7 @@ func (h *statusAPI) handleDebugInfo(w http.ResponseWriter, req *http.Request) {
 	ctx := req.Context()
 	h.capture.WriteDebugInfo(ctx, w)
 	fmt.Fprintf(w, "\n\n*** etcd info ***:\n\n")
-	h.writeEtcdInfo(ctx, h.capture.EtcdClient, w)
+	h.writeEtcdInfo(ctx, h.capture.GetEtcdClient(), w)
 }
 
 func (h *statusAPI) handleStatus(w http.ResponseWriter, req *http.Request) {

cdc/api/util.go

@@ -94,7 +94,7 @@ func WriteData(w http.ResponseWriter, data interface{}) {
 
 // HandleOwnerJob enqueue the admin job
 func HandleOwnerJob(
-	ctx context.Context, capture *capture.Capture, job model.AdminJob,
+	ctx context.Context, capture capture.Capture, job model.AdminJob,
 ) error {
 	// Use buffered channel to prevernt blocking owner.
 	done := make(chan error, 1)
@@ -113,7 +113,7 @@ func HandleOwnerJob(
 
 // HandleOwnerBalance balance the changefeed tables
 func HandleOwnerBalance(
-	ctx context.Context, capture *capture.Capture, changefeedID model.ChangeFeedID,
+	ctx context.Context, capture capture.Capture, changefeedID model.ChangeFeedID,
 ) error {
 	// Use buffered channel to prevernt blocking owner.
 	done := make(chan error, 1)
@@ -132,7 +132,7 @@ func HandleOwnerBalance(
 
 // HandleOwnerScheduleTable schedule tables
 func HandleOwnerScheduleTable(
-	ctx context.Context, capture *capture.Capture,
+	ctx context.Context, capture capture.Capture,
 	changefeedID model.ChangeFeedID, captureID string, tableID int64,
 ) error {
 	// Use buffered channel to prevent blocking owner.
@@ -233,7 +233,7 @@ func ForwardToOwner(c *gin.Context, p CaptureInfoProvider) {
 
 // HandleOwnerDrainCapture schedule drain the target capture
 func HandleOwnerDrainCapture(
-	ctx context.Context, capture *capture.Capture, captureID string,
+	ctx context.Context, capture capture.Capture, captureID string,
 ) (*model.DrainCaptureResp, error) {
 	// Use buffered channel to prevent blocking owner.
 	done := make(chan error, 1)

cdc/api/v1/api.go

@@ -44,18 +44,18 @@ const (
 
 // OpenAPI provides capture APIs.
 type OpenAPI struct {
-	capture *capture.Capture
+	capture capture.Capture
 	// use for unit test only
 	testStatusProvider owner.StatusProvider
 }
 
 // NewOpenAPI creates a new OpenAPI.
-func NewOpenAPI(c *capture.Capture) OpenAPI {
+func NewOpenAPI(c capture.Capture) OpenAPI {
 	return OpenAPI{capture: c}
 }
 
 // NewOpenAPI4Test return a OpenAPI for test
-func NewOpenAPI4Test(c *capture.Capture, p owner.StatusProvider) OpenAPI {
+func NewOpenAPI4Test(c capture.Capture, p owner.StatusProvider) OpenAPI {
 	return OpenAPI{capture: c, testStatusProvider: p}
 }
 
@@ -268,8 +268,8 @@ func (h *OpenAPI) CreateChangefeed(c *gin.Context) {
 		return
 	}
 
-	err = h.capture.EtcdClient.CreateChangefeedInfo(ctx, info,
-		model.DefaultChangeFeedID(changefeedConfig.ID))
+	err = h.capture.GetEtcdClient().CreateChangefeedInfo(
+		ctx, info, model.DefaultChangeFeedID(changefeedConfig.ID))
 	if err != nil {
 		_ = c.Error(err)
 		return
@@ -403,7 +403,7 @@ func (h *OpenAPI) UpdateChangefeed(c *gin.Context) {
 		return
 	}
 
-	err = h.capture.EtcdClient.SaveChangeFeedInfo(ctx, newInfo, changefeedID)
+	err = h.capture.GetEtcdClient().SaveChangeFeedInfo(ctx, newInfo, changefeedID)
 	if err != nil {
 		_ = c.Error(err)
 		return

cdc/api/v1/api_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/capture"
+	mock_capture "github.com/pingcap/tiflow/cdc/capture/mock"
 	"github.com/pingcap/tiflow/cdc/model"
 	mock_owner "github.com/pingcap/tiflow/cdc/owner/mock"
 	"github.com/pingcap/tiflow/cdc/scheduler"
@@ -91,7 +92,7 @@ func (p *mockStatusProvider) GetCaptures(ctx context.Context) ([]*model.CaptureI
 	return args.Get(0).([]*model.CaptureInfo), args.Error(1)
 }
 
-func newRouter(c *capture.Capture, p *mockStatusProvider) *gin.Engine {
+func newRouter(c capture.Capture, p *mockStatusProvider) *gin.Engine {
 	router := gin.New()
 	RegisterOpenAPIRoutes(router, NewOpenAPI4Test(c, p))
 	return router
@@ -749,12 +750,21 @@ func TestServerStatusLiveness(t *testing.T) {
 	t.Parallel()
 	// capture is owner
 	ctrl := gomock.NewController(t)
-	mo := mock_owner.NewMockOwner(ctrl)
-	cp := capture.NewCapture4Test(mo)
+	cp := mock_capture.NewMockCapture(ctrl)
 	ownerRouter := newRouter(cp, newStatusProvider())
 	api := testCase{url: "/api/v1/status", method: "GET"}
 
+	cp.EXPECT().Info().DoAndReturn(func() (model.CaptureInfo, error) {
+		return model.CaptureInfo{}, nil
+	}).AnyTimes()
+	cp.EXPECT().IsOwner().DoAndReturn(func() bool {
+		return true
+	}).AnyTimes()
+
 	// Alive.
+	alive := cp.EXPECT().Liveness().DoAndReturn(func() model.Liveness {
+		return model.LivenessCaptureAlive
+	})
 	w := httptest.NewRecorder()
 	req, _ := http.NewRequestWithContext(context.Background(), api.method, api.url, nil)
 	ownerRouter.ServeHTTP(w, req)
@@ -765,7 +775,9 @@ func TestServerStatusLiveness(t *testing.T) {
 	require.EqualValues(t, model.LivenessCaptureAlive, resp.Liveness)
 
 	// Draining the capture.
-	cp.Drain()
+	cp.EXPECT().Liveness().DoAndReturn(func() model.Liveness {
+		return model.LivenessCaptureStopping
+	}).After(alive)
 	w = httptest.NewRecorder()
 	req, _ = http.NewRequestWithContext(context.Background(), api.method, api.url, nil)
 	ownerRouter.ServeHTTP(w, req)

cdc/api/v2/api.go

@@ -21,11 +21,11 @@ import (
 
 // OpenAPIV2 provides CDC v2 APIs
 type OpenAPIV2 struct {
-	capture *capture.Capture
+	capture capture.Capture
 }
 
 // NewOpenAPIV2 creates a new OpenAPIV2.
-func NewOpenAPIV2(c *capture.Capture) OpenAPIV2 {
+func NewOpenAPIV2(c capture.Capture) OpenAPIV2 {
 	return OpenAPIV2{capture: c}
 }
 

cdc/api/v2/api_test.go

@@ -37,7 +37,7 @@ type testCase struct {
 	method string
 }
 
-func newRouter(c *capture.Capture) *gin.Engine {
+func newRouter(c capture.Capture) *gin.Engine {
 	router := gin.New()
 	RegisterOpenAPIV2Routes(router, NewOpenAPIV2(c))
 	return router

cdc/api/v2/tso.go

@@ -24,11 +24,11 @@ import (
 // GetTso request and returns a TSO from PD
 func (h *OpenAPIV2) GetTso(c *gin.Context) {
 	ctx := c.Request.Context()
-	if h.capture.UpstreamManager == nil {
+	if h.capture.GetUpstreamManager() == nil {
 		c.Status(http.StatusServiceUnavailable)
 		return
 	}
-	pdClient := h.capture.UpstreamManager.Get(upstream.DefaultUpstreamID).PDClient
+	pdClient := h.capture.GetUpstreamManager().Get(upstream.DefaultUpstreamID).PDClient
 	if pdClient == nil {
 		c.Status(http.StatusServiceUnavailable)
 		return

cdc/api/validator/validator.go

@@ -41,10 +41,10 @@ import (
 func VerifyCreateChangefeedConfig(
 	ctx context.Context,
 	changefeedConfig model.ChangefeedConfig,
-	capture *capture.Capture,
+	capture capture.Capture,
 ) (*model.ChangeFeedInfo, error) {
 	// TODO(dongmen): we should pass ClusterID in ChangefeedConfig in the upcoming future
-	up := capture.UpstreamManager.Get(upstream.DefaultUpstreamID)
+	up := capture.GetUpstreamManager().Get(upstream.DefaultUpstreamID)
 	defer up.Release()
 
 	// verify sinkURI