api,owner(ticdc): return error when api fails

Makefile

@@ -112,7 +112,7 @@ kafka_consumer:
 install:
 	go install ./...
 
-unit_test: check_failpoint_ctl
+unit_test: check_failpoint_ctl generate_mock
 	mkdir -p "$(TEST_DIR)"
 	$(FAILPOINT_ENABLE)
 	@export log_level=error;\
@@ -167,7 +167,7 @@ integration_test_mysql:
 integration_test_kafka: check_third_party_binary
 	tests/integration_tests/run.sh kafka "$(CASE)" "$(START_AT)"
 
-fmt: tools/bin/gofumports tools/bin/shfmt
+fmt: tools/bin/gofumports tools/bin/shfmt generate_mock
 	@echo "gofmt (simplify)"
 	tools/bin/gofumports -l -w $(FILES) 2>&1 | $(FAIL_ON_STDOUT)
 	@echo "run shfmt"
@@ -234,6 +234,9 @@ data-flow-diagram: docs/data-flow.dot
 swagger-spec: tools/bin/swag
 	tools/bin/swag init --parseVendor -generalInfo cdc/api/open.go --output docs/swagger
 
+generate_mock: tools/bin/mockgen
+	tools/bin/mockgen -source cdc/owner/owner.go -destination cdc/owner/mock/owner_mock.go
+
 clean:
 	go clean -i ./...
 	rm -rf *.out

cdc/api/open.go

@@ -319,11 +319,16 @@ func (h *openAPI) PauseChangefeed(c *gin.Context) {
 		Type: model.AdminStop,
 	}
 
-	_ = h.capture.OperateOwnerUnderLock(func(owner *owner.Owner) error {
-		owner.EnqueueJob(job)
+	// Use buffered channel to prevernt blocking owner.
+	done := make(chan error, 1)
+	_ = h.capture.OperateOwnerUnderLock(func(owner owner.Owner) error {
+		owner.EnqueueJob(job, done)
 		return nil
 	})
-
+	if err := waitDone(ctx, done); err != nil {
+		_ = c.Error(err)
+		return
+	}
 	c.Status(http.StatusAccepted)
 }
 
@@ -361,11 +366,16 @@ func (h *openAPI) ResumeChangefeed(c *gin.Context) {
 		Type: model.AdminResume,
 	}
 
-	_ = h.capture.OperateOwnerUnderLock(func(owner *owner.Owner) error {
-		owner.EnqueueJob(job)
+	// Use buffered channel to prevernt blocking owner.
+	done := make(chan error, 1)
+	_ = h.capture.OperateOwnerUnderLock(func(owner owner.Owner) error {
+		owner.EnqueueJob(job, done)
 		return nil
 	})
-
+	if err := waitDone(ctx, done); err != nil {
+		_ = c.Error(err)
+		return
+	}
 	c.Status(http.StatusAccepted)
 }
 
@@ -465,11 +475,16 @@ func (h *openAPI) RemoveChangefeed(c *gin.Context) {
 		Type: model.AdminRemove,
 	}
 
-	_ = h.capture.OperateOwnerUnderLock(func(owner *owner.Owner) error {
-		owner.EnqueueJob(job)
+	// Use buffered channel to prevernt blocking owner.
+	done := make(chan error, 1)
+	_ = h.capture.OperateOwnerUnderLock(func(owner owner.Owner) error {
+		owner.EnqueueJob(job, done)
 		return nil
 	})
-
+	if err := waitDone(ctx, done); err != nil {
+		_ = c.Error(err)
+		return
+	}
 	c.Status(http.StatusAccepted)
 }
 
@@ -503,11 +518,16 @@ func (h *openAPI) RebalanceTable(c *gin.Context) {
 		return
 	}
 
-	_ = h.capture.OperateOwnerUnderLock(func(owner *owner.Owner) error {
-		owner.TriggerRebalance(changefeedID)
+	// Use buffered channel to prevernt blocking owner.
+	done := make(chan error, 1)
+	_ = h.capture.OperateOwnerUnderLock(func(owner owner.Owner) error {
+		owner.TriggerRebalance(changefeedID, done)
 		return nil
 	})
-
+	if err := waitDone(ctx, done); err != nil {
+		_ = c.Error(err)
+		return
+	}
 	c.Status(http.StatusAccepted)
 }
 
@@ -557,11 +577,16 @@ func (h *openAPI) MoveTable(c *gin.Context) {
 		return
 	}
 
-	_ = h.capture.OperateOwnerUnderLock(func(owner *owner.Owner) error {
-		owner.ManualSchedule(changefeedID, data.CaptureID, data.TableID)
+	// Use buffered channel to prevernt blocking owner.
+	done := make(chan error, 1)
+	_ = h.capture.OperateOwnerUnderLock(func(owner owner.Owner) error {
+		owner.ManualSchedule(changefeedID, data.CaptureID, data.TableID, done)
 		return nil
 	})
-
+	if err := waitDone(ctx, done); err != nil {
+		_ = c.Error(err)
+		return
+	}
 	c.Status(http.StatusAccepted)
 }
 
@@ -580,7 +605,7 @@ func (h *openAPI) ResignOwner(c *gin.Context) {
 		return
 	}
 
-	_ = h.capture.OperateOwnerUnderLock(func(owner *owner.Owner) error {
+	_ = h.capture.OperateOwnerUnderLock(func(owner owner.Owner) error {
 		owner.AsyncStop()
 		return nil
 	})