# Changefeed OpenAPI Review Round 2 Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Fix approved Review Round 2 issues for Changefeed OpenAPI in a backward-compatible way, including pagination correctness, route exposure, validation semantics, and clearer Kafka/MySQL request behavior. **Architecture:** Keep the existing OpenAPI adapter + console-controller pipeline intact, and apply surgical fixes at three boundaries: HTTP route mapping, OpenAPI adapter validation logic, and proto validation/doc metadata. Use generated artifacts as the source of truth for API contract/output behavior, and drive behavior changes with focused controller/bootstrap tests. **Tech Stack:** Go, gRPC-Gateway, protobuf (`protoc` + PGV), Gin, Testify, Makefile (`make generate-idl`), Swagger JSON generation. --- ## File Structure and Responsibilities - **Modify:** `internal/cmd/bootstrap/dataflow_http.go` Add explicit v1beta1 passthrough route for `GET /v1beta1/changefeedRCUs` and extract v1beta1 passthrough route registration into a testable helper. - **Create:** `internal/cmd/bootstrap/dataflow_http_routes_test.go` Route-regression tests for explicit v1beta1 passthrough routes (including `changefeedRCUs`). - **Modify:** `app/controllers/open/changefeed/adapter_model.go` Fix `next_page_token` emission logic and add Kafka create pre-check for public/VPC network broker endpoint requirements. - **Modify:** `app/controllers/open/changefeed/service_test.go` Add/adjust tests for pagination terminal behavior, proto validation expectations, and Kafka/MySQL create-request semantics. - **Modify:** `idl/idl/dataflow/openapi/v1beta1/changefeed.proto` Update validation/doc semantics for `page_size`, `connection.endpoint` (PrivateLink flow), and `StorageFlushConfig.interval` default text. - **Modify (generated):** - `idl/pbgen/proto/openapi/v1beta1/changefeed.pb.go` - `idl/pbgen/proto/openapi/v1beta1/changefeed.pb.validate.go` - `idl/pbgen/proto/openapi/v1beta1/changefeed.pb.gw.go` - `idl/pbgen/proto/openapi/v1beta1/changefeed_grpc.pb.go` - `api/changefeed.swagger.json` ## Task 1: Fix terminal-page `next_page_token` behavior **Files:** - Modify: `app/controllers/open/changefeed/service_test.go` - Modify: `app/controllers/open/changefeed/adapter_model.go` - Test: `app/controllers/open/changefeed/service_test.go` - [ ] **Step 1: Update pagination test expectations to encode desired behavior (failing first)** ```go func TestListChangefeeds_PaginationTokenRoundTrip(t *testing.T) { scope := changefeedScope{TenantID: 10, ProjectID: 20, ClusterID: 30} t.Run("default page size", func(t *testing.T) { stub := &connectorServiceStub{ listConnectors: func(ctx context.Context, req *dataflow.ConnectorListReq) (*dataflow.Connectors, error) { require.Equal(t, defaultListChangefeedsPageSize, req.GetPerPage()) return &dataflow.Connectors{ Total: 1, Items: []*dataflow.Connector{connectorForList(101, "cf-1", scope)}, }, nil }, } resp, err := newConnectorChangefeedAdapter(stub, nil, newNoopBizChangefeedModel()). ListChangefeeds(context.Background(), &changefeedopenbp.ListChangefeedsRequest{ ClusterId: fmt.Sprint(scope.ClusterID), }, scope) require.NoError(t, err) require.Len(t, resp.GetChangefeeds(), 1) require.Empty(t, resp.GetNextPageToken()) // changed from NotEmpty }) t.Run("token round trip", func(t *testing.T) { var pages []uint32 stub := &connectorServiceStub{ listConnectors: func(ctx context.Context, req *dataflow.ConnectorListReq) (*dataflow.Connectors, error) { pages = append(pages, req.GetPage()) require.Equal(t, uint32(1), req.GetPerPage()) switch req.GetPage() { case 0: return &dataflow.Connectors{ Total: 2, Items: []*dataflow.Connector{connectorForList(101, "cf-1", scope)}, }, nil case 1: return &dataflow.Connectors{ Total: 2, Items: []*dataflow.Connector{connectorForList(102, "cf-2", scope)}, }, nil default: t.Fatalf("unexpected page %d", req.GetPage()) return nil, nil } }, } adapter := newConnectorChangefeedAdapter(stub, nil, newNoopBizChangefeedModel()) first, err := adapter.ListChangefeeds(context.Background(), &changefeedopenbp.ListChangefeedsRequest{ ClusterId: fmt.Sprint(scope.ClusterID), PageSize: 1, }, scope) require.NoError(t, err) require.NotEmpty(t, first.GetNextPageToken()) second, err := adapter.ListChangefeeds(context.Background(), &changefeedopenbp.ListChangefeedsRequest{ ClusterId: fmt.Sprint(scope.ClusterID), PageSize: 1, PageToken: first.GetNextPageToken(), }, scope) require.NoError(t, err) require.Equal(t, []uint32{0, 1}, pages) require.Empty(t, second.GetNextPageToken()) }) } ``` - [ ] **Step 2: Run focused test to confirm failure** Run: ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service go test ./app/controllers/open/changefeed -run TestListChangefeeds_PaginationTokenRoundTrip -count=1 ``` Expected: FAIL because adapter currently emits token even on terminal pages. - [ ] **Step 3: Implement page-aware `hasNextPage` logic in adapter** ```go func (a *connectorChangefeedAdapter) ListChangefeeds( ctx context.Context, req *changefeedopenbp.ListChangefeedsRequest, scope changefeedScope, ) (*changefeedopenbp.ListChangefeedsResponse, error) { pageSize := normalizeListPageSize(req.GetPageSize()) page, err := decodePageToken(req.GetPageToken()) if err != nil { return nil, err } connectors, err := a.connectorAPI.ListConnectors(ctx, &dataflow.ConnectorListReq{ ClusterId: scope.ClusterID, ProjectId: scope.ProjectID, OrgId: scope.TenantID, Page: page, PerPage: pageSize, }) if err != nil { return nil, fmt.Errorf("list changefeeds: %w", err) } if connectors == nil { connectors = &dataflow.Connectors{} } out := &changefeedopenbp.ListChangefeedsResponse{ Changefeeds: make([]*changefeedopenbp.Changefeed, 0, len(connectors.GetItems())), TotalSize: int32(connectors.GetTotal()), } for _, connector := range connectors.GetItems() { out.Changefeeds = append(out.Changefeeds, convertConnectorToChangefeed(connector)) } if hasNextPage(page, pageSize, connectors.GetTotal()) { token, err := encodePageToken(page + 1) if err != nil { return nil, err } out.NextPageToken = token } return out, nil } func hasNextPage(page, pageSize uint32, total int64) bool { if pageSize == 0 || total <= 0 { return false } shown := uint64(page+1) * uint64(pageSize) return shown < uint64(total) } ``` - [ ] **Step 4: Re-run focused test** Run: ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service go test ./app/controllers/open/changefeed -run TestListChangefeeds_PaginationTokenRoundTrip -count=1 ``` Expected: PASS. - [ ] **Step 5: Commit** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service git add app/controllers/open/changefeed/adapter_model.go app/controllers/open/changefeed/service_test.go git commit -m "fix(openapi): emit next_page_token only when next page exists" ``` ## Task 2: Update proto validation/doc semantics (`page_size`, MySQL PrivateLink endpoint, S3 flush default) **Files:** - Modify: `app/controllers/open/changefeed/service_test.go` - Modify: `idl/idl/dataflow/openapi/v1beta1/changefeed.proto` - Modify (generated): `idl/pbgen/proto/openapi/v1beta1/*`, `api/changefeed.swagger.json` - Test: `app/controllers/open/changefeed/service_test.go` - [ ] **Step 1: Add failing tests for new validation expectations** ```go func TestListChangefeedsRequest_PageSizeOptional(t *testing.T) { req := &changefeedopenbp.ListChangefeedsRequest{ ClusterId: "1234567890123", // PageSize intentionally omitted } require.NoError(t, req.Validate()) } func TestCreateChangefeedRequest_MySQLPrivateLink_AllowsEmptyEndpoint(t *testing.T) { sinkEndpointID := uint64(12345) req := &changefeedopenbp.CreateChangefeedRequest{ Changefeed: &changefeedopenbp.Changefeed{ ClusterId: "1234567890123", Name: "cf-mysql-pl", ReplicationCapacity: "2rcu", DownstreamType: changefeedopenbp.DownstreamType_MYSQL, NetworkInfo: &changefeedopenbp.NetworkInfo{ NetworkType: changefeedopenbp.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK, SinkEndpointId: &sinkEndpointID, Ports: []uint32{3306}, }, TableConfig: &changefeedopenbp.TableConfig{ FilterRules: []string{"*.*"}, Mode: changefeedopenbp.TableConfig_IGNORE_NOT_SUPPORT_TABLE, }, StartPosition: &changefeedopenbp.StartPosition{ Mode: changefeedopenbp.StartPosition_FROM_NOW, }, DownstreamConfig: &changefeedopenbp.Changefeed_Mysql{ Mysql: &changefeedopenbp.MySQLDownstreamConfig{ Connection: &changefeedopenbp.Connection{ Endpoint: "", Username: "user", Password: "pass", }, }, }, }, } require.NoError(t, req.Validate()) } ``` - [ ] **Step 2: Run focused tests to confirm failure** Run: ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service go test ./app/controllers/open/changefeed -run "TestListChangefeedsRequest_PageSizeOptional|TestCreateChangefeedRequest_MySQLPrivateLink_AllowsEmptyEndpoint" -count=1 ``` Expected: FAIL on current generated validation (`page_size` range check and `connection.endpoint` min length). - [ ] **Step 3: Update proto field rules and docs** ```proto // ListChangefeedsRequest.page_size int32 page_size = 100 [ (google.api.field_behavior) = OPTIONAL, (validate.rules).int32 = { gte: 1 lte: 100 ignore_empty: true }, (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = { description: "The maximum number of changefeeds to return. If not specified, 20 items are returned by default." example: "20" minimum: 1 maximum: 100 } ]; // Connection.endpoint string endpoint = 1 [ (google.api.field_behavior) = OPTIONAL, (validate.rules).string = { min_len: 1 ignore_empty: true }, (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = { description: "Endpoint address (host:port). Required for PUBLIC/VPC_PEERING. Optional for PRIVATE_LINK/PRIVATE_SERVICE_CONNECT when endpoint is assembled server-side from sink endpoint and ports." example: "\"mysql.example.com:3306\"" } ]; // StorageFlushConfig.interval docs // Range: 2s to 10m. Default: "1m". // Example: "1m" string interval = 1 [ (google.api.field_behavior) = OPTIONAL, (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = { description: "Flush interval. Range: 2s to 10m. Default: \"1m\". Files are flushed when either interval or file_size threshold is met." example: "\"1m\"" } ]; ``` - [ ] **Step 4: Regenerate IDL artifacts and rerun focused tests** Run: ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service make generate-idl go test ./app/controllers/open/changefeed -run "TestListChangefeedsRequest_PageSizeOptional|TestCreateChangefeedRequest_MySQLPrivateLink_AllowsEmptyEndpoint|TestListChangefeeds_PaginationTokenRoundTrip" -count=1 ``` Expected: `make generate-idl` succeeds; all targeted tests PASS. - [ ] **Step 5: Commit** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service git add idl/idl/dataflow/openapi/v1beta1/changefeed.proto \ idl/pbgen/proto/openapi/v1beta1 \ api/changefeed.swagger.json \ app/controllers/open/changefeed/service_test.go git commit -m "fix(openapi): align validation and docs for pagination and private-link mysql" ``` ## Task 3: Expose `/v1beta1/changefeedRCUs` via explicit v1beta1 passthrough routes **Files:** - Create: `internal/cmd/bootstrap/dataflow_http_routes_test.go` - Modify: `internal/cmd/bootstrap/dataflow_http.go` - Test: `internal/cmd/bootstrap/dataflow_http_routes_test.go` - [ ] **Step 1: Add failing route regression test (expects helper function + route)** ```go package bootstrap import ( "net/http" "net/http/httptest" "testing" "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" ) func TestRegisterOpenAPIV1beta1PassthroughRoutes_IncludesChangefeedRCUs(t *testing.T) { gin.SetMode(gin.TestMode) eng := gin.New() registerOpenAPIV1beta1PassthroughRoutes(eng, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) })) req := httptest.NewRequest(http.MethodGet, "/v1beta1/changefeedRCUs", nil) w := httptest.NewRecorder() eng.ServeHTTP(w, req) require.NotEqual(t, http.StatusNotFound, w.Code) } ``` - [ ] **Step 2: Run focused test to confirm compile/runtime failure before implementation** Run: ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service go test ./internal/cmd/bootstrap -run TestRegisterOpenAPIV1beta1PassthroughRoutes_IncludesChangefeedRCUs -count=1 ``` Expected: FAIL (missing helper / route not yet registered). - [ ] **Step 3: Implement helper and route registration** ```go import "net/http" func registerOpenAPIV1beta1PassthroughRoutes(eng *gin.Engine, gateway http.Handler) { eng.Any("/v1beta1/backups", gin.WrapH(gateway)) eng.Any("/v1beta1/backups/*any", gin.WrapH(gateway)) eng.Any("/v1beta1/changefeeds", gin.WrapH(gateway)) eng.Any("/v1beta1/changefeeds/*any", gin.WrapH(gateway)) eng.Any("/v1beta1/changefeedRCUs", gin.WrapH(gateway)) eng.Any("/v1beta1/clusters/*any", gin.WrapH(gateway)) } // In registerRequestMapping, replace the inline v1beta1 route list with: registerOpenAPIV1beta1PassthroughRoutes(eng, gwmuvV1beta1) ``` - [ ] **Step 4: Re-run focused test** Run: ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service go test ./internal/cmd/bootstrap -run TestRegisterOpenAPIV1beta1PassthroughRoutes_IncludesChangefeedRCUs -count=1 ``` Expected: PASS. - [ ] **Step 5: Commit** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service git add internal/cmd/bootstrap/dataflow_http.go internal/cmd/bootstrap/dataflow_http_routes_test.go git commit -m "fix(http): route v1beta1 changefeedRCUs to openapi gateway" ``` ## Task 4: Improve Kafka public-create error semantics in OpenAPI adapter **Files:** - Modify: `app/controllers/open/changefeed/service_test.go` - Modify: `app/controllers/open/changefeed/adapter_model.go` - Test: `app/controllers/open/changefeed/service_test.go` - [ ] **Step 1: Add failing tests for Kafka create pre-check** ```go func TestToConnectorCreateReq_KafkaPublicMissingBrokerEndpoints(t *testing.T) { adapter := &connectorChangefeedAdapter{} scope := changefeedScope{TenantID: 10, ProjectID: 20, ClusterID: 30} _, err := adapter.toConnectorCreateReq(&changefeedopenbp.CreateChangefeedRequest{ Changefeed: &changefeedopenbp.Changefeed{ ClusterId: "30", Name: "cf-kafka", ReplicationCapacity: "2rcu", DownstreamType: changefeedopenbp.DownstreamType_KAFKA, NetworkInfo: &changefeedopenbp.NetworkInfo{NetworkType: changefeedopenbp.NetworkInfo_NETWORK_TYPE_PUBLIC}, TableConfig: &changefeedopenbp.TableConfig{FilterRules: []string{"*.*"}, Mode: changefeedopenbp.TableConfig_IGNORE_NOT_SUPPORT_TABLE}, StartPosition: &changefeedopenbp.StartPosition{Mode: changefeedopenbp.StartPosition_FROM_NOW}, DownstreamConfig: &changefeedopenbp.Changefeed_Kafka{ Kafka: &changefeedopenbp.KafkaDownstreamConfig{ Broker: &changefeedopenbp.KafkaDownstreamConfig_Broker{}, // no brokerEndpoints DataFormat: &changefeedopenbp.KafkaDownstreamConfig_DataFormat{Protocol: changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_CANAL_JSON}, TopicPartitionConfig: &changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig{DispatchType: changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_ONE_TOPIC, DefaultTopic: "cdc-events", ReplicationFactor: 1, PartitionNum: 1}, }, }, }, }, scope) require.Error(t, err) require.Contains(t, err.Error(), "kafka.broker.broker_endpoints") } func TestToConnectorCreateReq_KafkaPrivateLinkAllowsEmptyBrokerEndpoints(t *testing.T) { adapter := &connectorChangefeedAdapter{} scope := changefeedScope{TenantID: 10, ProjectID: 20, ClusterID: 30} sinkEndpointID := uint64(12345) _, err := adapter.toConnectorCreateReq(&changefeedopenbp.CreateChangefeedRequest{ Changefeed: &changefeedopenbp.Changefeed{ ClusterId: "30", Name: "cf-kafka-pl", ReplicationCapacity: "2rcu", DownstreamType: changefeedopenbp.DownstreamType_KAFKA, NetworkInfo: &changefeedopenbp.NetworkInfo{ NetworkType: changefeedopenbp.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK, SinkEndpointId: &sinkEndpointID, Ports: []uint32{9092}, }, TableConfig: &changefeedopenbp.TableConfig{FilterRules: []string{"*.*"}, Mode: changefeedopenbp.TableConfig_IGNORE_NOT_SUPPORT_TABLE}, StartPosition: &changefeedopenbp.StartPosition{Mode: changefeedopenbp.StartPosition_FROM_NOW}, DownstreamConfig: &changefeedopenbp.Changefeed_Kafka{ Kafka: &changefeedopenbp.KafkaDownstreamConfig{ Broker: &changefeedopenbp.KafkaDownstreamConfig_Broker{}, // allowed for PRIVATE_LINK DataFormat: &changefeedopenbp.KafkaDownstreamConfig_DataFormat{Protocol: changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_CANAL_JSON}, TopicPartitionConfig: &changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig{DispatchType: changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_ONE_TOPIC, DefaultTopic: "cdc-events", ReplicationFactor: 1, PartitionNum: 1}, }, }, }, }, scope) require.NoError(t, err) } ``` - [ ] **Step 2: Run focused tests to verify failure** Run: ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service go test ./app/controllers/open/changefeed -run "TestToConnectorCreateReq_KafkaPublicMissingBrokerEndpoints|TestToConnectorCreateReq_KafkaPrivateLinkAllowsEmptyBrokerEndpoints" -count=1 ``` Expected: FAIL because public missing-endpoint case currently is not pre-validated in adapter. - [ ] **Step 3: Implement Kafka network-aware broker-endpoint pre-check** ```go func requiresKafkaBrokerEndpoints(networkInfo *dataflow.NetworkInfo) bool { if networkInfo == nil { return true } switch networkInfo.GetNetworkType() { case dataflow.NetworkType_NetworkType_Private_Link, dataflow.NetworkType_NetworkType_Private_Service_Connect: return false default: return true } } func validateKafkaCreateInput(req *dataflow.ConnectorCreateReq) error { kafkaInfo := req.GetKafkaInfo() if kafkaInfo == nil || kafkaInfo.GetBroker() == nil { return bizerr.New(bizerr.ErrCodeParams, "kafka.broker is required") } if requiresKafkaBrokerEndpoints(req.GetNetworkInfo()) && strings.TrimSpace(kafkaInfo.GetBroker().GetBrokerEndpoints()) == "" { return bizerr.New(bizerr.ErrCodeParams, "kafka.broker.broker_endpoints is required when network_info.network_type is PUBLIC or VPC_PEERING") } return nil } // In toConnectorCreateReq, Kafka branch: case dataflow.ConnectorType_KAFKA: req.KafkaInfo = toDataflowKafkaInfo(cf, scope) if req.KafkaInfo == nil { return nil, bizerr.New(bizerr.ErrCodeParams, "kafka configuration is required") } if err := validateKafkaCreateInput(req); err != nil { return nil, err } ``` - [ ] **Step 4: Re-run focused and adjacent tests** Run: ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service go test ./app/controllers/open/changefeed -run "TestToConnectorCreateReq_KafkaPublicMissingBrokerEndpoints|TestToConnectorCreateReq_KafkaPrivateLinkAllowsEmptyBrokerEndpoints|TestCreateChangefeed_DryRunSkipsGetConnector|TestListChangefeedRCUs_MapsInstanceTypes" -count=1 ``` Expected: PASS. - [ ] **Step 5: Commit** ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service git add app/controllers/open/changefeed/adapter_model.go app/controllers/open/changefeed/service_test.go git commit -m "fix(openapi): clarify kafka public create validation errors" ``` ## Task 5: Final verification pass **Files:** - Verify all modified files from Tasks 1-4 - [ ] **Step 1: Run targeted package test suites** Run: ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service go test ./app/controllers/open/changefeed ./internal/cmd/bootstrap -count=1 ``` Expected: PASS. - [ ] **Step 2: Validate generated API contract outputs** Run: ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service git --no-pager diff -- idl/idl/dataflow/openapi/v1beta1/changefeed.proto api/changefeed.swagger.json | cat ``` Expected: Diff clearly reflects only approved semantic changes (`page_size`, MySQL endpoint validation semantics, `StorageFlushConfig.interval` default text). - [ ] **Step 3: Optional full app quick regression for touched area** Run: ```bash cd /home/nil/pingcap/workbench/repos/dataflow-service make app-unit-test-local ``` Expected: PASS (or report unrelated pre-existing failures separately if any).