# CDC RCU List API + Kafka Smoke Validation 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:** Add unified CDC RCU-list retrieval for both ConsoleAPI/OpenAPI modes and validate it in dedicated `kafka_smoke` E2E flow.

**Architecture:** Add mode-specific `ListChangefeedRCUs` methods in dedicated wrapper layers (console and openapi), then expose a normalized API in `cdcapi/client_wrapper.go` that always returns `console_model.DataflowChangefeedInstanceTypes`. Use this unified method in `kafka_smoke.go` with a single non-empty assertion step.

**Tech Stack:** Go, go-swagger generated clients, testify, dataflow-service-e2e wrappers.

---

### Task 1: Add mode-specific RCU list wrapper methods

**Files:**
- Modify: `test/e2e/internal/tidbcloud/dedicated/consoleapi/cdc_client_wrapper.go`
- Modify: `test/e2e/internal/tidbcloud/dedicated/openapi/open_api_client_wrapper.go`
- Test: `test/e2e/internal/tidbcloud/dedicated/openapi/open_api_client_wrapper_test.go`

- [ ] **Step 1: Write failing test for OpenAPI RCU list wrapper call path**

```go
func TestListChangefeedRCUs_CallsOpenAPI(t *testing.T) {
	testLogger := noopLogger{}
	var called bool
	fakeService := &fakeChangefeedOpenAPIService{
		listRCUFn: func(params *changefeed_service.ListChangefeedRCUsParams, opts ...changefeed_service.ClientOption) (*changefeed_service.ListChangefeedRCUsOK, error) {
			called = true
			require.NotNil(t, params.Context)
			assert.NotNil(t, log.FromContext(params.Context))
			return &changefeed_service.ListChangefeedRCUsOK{}, nil
		},
	}
	wrapper := newChangefeedOpenApiClientWrapperWithService(testLogger, fakeService)

	_, err := wrapper.ListChangefeedRCUs()
	require.NoError(t, err)
	assert.True(t, called)
}
```

- [ ] **Step 2: Run test to verify it fails**

Run:

```bash
cd /home/nil/pingcap/workbench/repos/dataflow-service/test/e2e
go test ./internal/tidbcloud/dedicated/openapi -run TestListChangefeedRCUs_CallsOpenAPI -count=1
```

Expected: FAIL because wrapper/fake does not yet implement the tested path.

- [ ] **Step 3: Implement console wrapper method**

```go
func (c *CDCClientWrapper) ListChangefeedRCUs() (*changefeed_instance_type_service.ListChangefeedInstanceTypesOK, error) {
	params := changefeed_instance_type_service.NewListChangefeedInstanceTypesParams()
	resp, err := BizChangefeedClient().ChangefeedInstanceTypeService.ListChangefeedInstanceTypes(params, nil)
	if err != nil {
		c.lgr.Warn("ListChangefeedInstanceTypes failed, params: %s, error: %s", misc.JSON(params), err.Error())
		return nil, err
	}
	c.lgr.Info("ListChangefeedInstanceTypes successfully, params: %s, resp: %s", misc.JSON(params), misc.JSON(resp))
	return resp, nil
}
```

- [ ] **Step 4: Implement openapi wrapper interface + method**

```go
type changefeedOpenApiService interface {
	ListChangefeedRCUs(params *changefeed_service2.ListChangefeedRCUsParams, opts ...changefeed_service2.ClientOption) (*changefeed_service2.ListChangefeedRCUsOK, error)
}

func (c *ChangefeedOpenApiClientWrapper) ListChangefeedRCUs() (*changefeed_service2.ListChangefeedRCUsOK, error) {
	params := changefeed_service2.NewListChangefeedRCUsParamsWithContext(c.ctx())
	resp, err := c.service.ListChangefeedRCUs(params)
	if err != nil {
		c.lgr.Warn("list changefeed RCUs failed, params: %s, error: %s", misc.JSON(params), err.Error())
		return nil, err
	}
	c.lgr.Info("list changefeed RCUs successfully, params: %s, resp: %s", misc.JSON(params), misc.JSON(resp))
	return resp, nil
}
```

- [ ] **Step 5: Update fake service in test to support RCU-list hook**

```go
type fakeChangefeedOpenAPIService struct {
	listRCUFn func(params *changefeed_service.ListChangefeedRCUsParams, opts ...changefeed_service.ClientOption) (*changefeed_service.ListChangefeedRCUsOK, error)
}

func (f *fakeChangefeedOpenAPIService) ListChangefeedRCUs(params *changefeed_service.ListChangefeedRCUsParams, opts ...changefeed_service.ClientOption) (*changefeed_service.ListChangefeedRCUsOK, error) {
	if f.listRCUFn != nil {
		return f.listRCUFn(params, opts...)
	}
	return &changefeed_service.ListChangefeedRCUsOK{}, nil
}
```

- [ ] **Step 6: Re-run test to verify it passes**

Run:

```bash
cd /home/nil/pingcap/workbench/repos/dataflow-service/test/e2e
go test ./internal/tidbcloud/dedicated/openapi -run TestListChangefeedRCUs_CallsOpenAPI -count=1
```

Expected: PASS.

- [ ] **Step 7: Commit**

```bash
cd /home/nil/pingcap/workbench/repos/dataflow-service
git add test/e2e/internal/tidbcloud/dedicated/consoleapi/cdc_client_wrapper.go \
        test/e2e/internal/tidbcloud/dedicated/openapi/open_api_client_wrapper.go \
        test/e2e/internal/tidbcloud/dedicated/openapi/open_api_client_wrapper_test.go
git commit -m "feat(e2e): add changefeed RCU list wrapper methods"
```

### Task 2: Add unified RCU list API in cdcapi wrapper

**Files:**
- Modify: `test/e2e/internal/tidbcloud/dedicated/cdcapi/client_wrapper.go`
- Test: `test/e2e/internal/tidbcloud/dedicated/cdcapi/client_wrapper_test.go`

- [ ] **Step 1: Write failing tests for cdcapi wrapper dispatch + mapping**

```go
func TestListChangefeedRCUs_ConsoleModeUsesConsolePayload(t *testing.T) {
	withAPIMode("consoleapi", func() {
		consoleResp := &changefeed_instance_type_service.ListChangefeedInstanceTypesOK{
			Payload: &console_model.DataflowChangefeedInstanceTypes{
				Items: []*console_model.DataflowChangefeedInstanceType{
					{Name: "2rcu", RcuNum: 2, RpsLimit: 3000},
				},
				Total: swag.Int64(1),
			},
		}
		console := &fakeConsoleClient{listChangefeedRCUResp: consoleResp}
		openapi := &fakeOpenAPIClient{}
		wrapper := newCDCClientWrapperWithClients(noopLogger{}, console, openapi, nil)

		resp, err := wrapper.ListChangefeedRCUs(&TiDBDedicatedEntity{ClusterID: "cluster-1"})
		require.NoError(t, err)
		require.NotNil(t, resp)
		require.NotNil(t, resp.Total)
		assert.Equal(t, int64(1), *resp.Total)
		assert.Equal(t, 1, console.listChangefeedRCUCalls)
		assert.Equal(t, 0, openapi.listRCUCalls)
	})
}

func TestListChangefeedRCUs_OpenAPIModeMapsToConsoleModel(t *testing.T) {
	withAPIMode("openapi", func() {
		openapi := &fakeOpenAPIClient{
			listRCUResp: &openapi_changefeed_service.ListChangefeedRCUsOK{
				Payload: &openapi_model.V1beta1ListChangefeedRCUsResponse{
					Rcus: []*openapi_model.V1beta1ChangefeedRCU{
						{Name: "2rcu", RcuNum: 2, RpsLimit: 3000},
					},
				},
			},
		}
		console := &fakeConsoleClient{}
		wrapper := newCDCClientWrapperWithClients(noopLogger{}, console, openapi, nil)

		resp, err := wrapper.ListChangefeedRCUs(&TiDBDedicatedEntity{ClusterID: "cluster-1"})
		require.NoError(t, err)
		require.NotNil(t, resp)
		require.Len(t, resp.Items, 1)
		assert.Equal(t, "2rcu", resp.Items[0].Name)
		assert.Equal(t, int64(2), resp.Items[0].RcuNum)
		assert.Equal(t, int64(3000), resp.Items[0].RpsLimit)
		require.NotNil(t, resp.Total)
		assert.Equal(t, int64(1), *resp.Total)
		assert.Equal(t, 0, console.listChangefeedRCUCalls)
		assert.Equal(t, 1, openapi.listRCUCalls)
	})
}
```

- [ ] **Step 2: Run tests to verify they fail**

Run:

```bash
cd /home/nil/pingcap/workbench/repos/dataflow-service/test/e2e
go test ./internal/tidbcloud/dedicated/cdcapi -run TestListChangefeedRCUs -count=1
```

Expected: FAIL because method does not exist yet.

- [ ] **Step 3: Extend interfaces in `cdcapi/client_wrapper.go`**

```go
type consoleClient interface {
	ListChangefeedRCUs() (*changefeed_instance_type_service.ListChangefeedInstanceTypesOK, error)
}

type openAPIClient interface {
	ListChangefeedRCUs() (*openapi_changefeed_service.ListChangefeedRCUsOK, error)
}
```

- [ ] **Step 4: Add unified wrapper method + converter**

```go
func (c *CDCClientWrapper) ListChangefeedRCUs(tidbCluster *TiDBDedicatedEntity) (*console_model.DataflowChangefeedInstanceTypes, error) {
	if !isOpenAPIMode() {
		resp, err := c.console.ListChangefeedRCUs()
		if err != nil {
			return nil, err
		}
		if resp == nil || resp.Payload == nil {
			return &console_model.DataflowChangefeedInstanceTypes{Items: []*console_model.DataflowChangefeedInstanceType{}, Total: swag.Int64(0)}, nil
		}
		return resp.Payload, nil
	}

	resp, err := c.openapi.ListChangefeedRCUs()
	if err != nil {
		return nil, err
	}
	return toConsoleChangefeedInstanceTypes(resp), nil
}
```

```go
func toConsoleChangefeedInstanceTypes(resp *openapi_changefeed_service.ListChangefeedRCUsOK) *console_model.DataflowChangefeedInstanceTypes {
	items := make([]*console_model.DataflowChangefeedInstanceType, 0)
	if resp == nil || resp.Payload == nil || len(resp.Payload.Rcus) == 0 {
		return &console_model.DataflowChangefeedInstanceTypes{Items: items, Total: swag.Int64(0)}
	}
	for _, r := range resp.Payload.Rcus {
		if r == nil {
			continue
		}
		items = append(items, &console_model.DataflowChangefeedInstanceType{
			Name:     r.Name,
			RcuNum:   r.RcuNum,
			RpsLimit: r.RpsLimit,
		})
	}
	return &console_model.DataflowChangefeedInstanceTypes{Items: items, Total: swag.Int64(int64(len(items)))}
}
```

- [ ] **Step 5: Extend test fakes with new method behavior**

```go
type fakeConsoleClient struct {
	describeAdminProfileCalls        int
	createCalls                      int
	editCalls                        int
	deleteCalls                      int
	scaleCalls                       int
	pauseCalls                       int
	resumeCalls                      int
	getCalls                         int
	listCalls                        int
	describeSchemaTableCalls         int
	getChangefeedKernelConfigCalls   int
	getTiCDCClusterByChangefeedCalls int
	getTiCDCClustersCalls            int
	calChangefeedFeeCalls            int
	listChangefeedRCUCalls           int
	testCalls                        int

	listChangefeedRCUResp  *changefeed_instance_type_service.ListChangefeedInstanceTypesOK
	listChangefeedRCUErr   error
}

func (f *fakeConsoleClient) ListChangefeedRCUs() (*changefeed_instance_type_service.ListChangefeedInstanceTypesOK, error) {
	f.listChangefeedRCUCalls++
	return f.listChangefeedRCUResp, f.listChangefeedRCUErr
}
```

```go
type fakeOpenAPIClient struct {
	createCalls  int
	editCalls    int
	deleteCalls  int
	scaleCalls   int
	pauseCalls   int
	resumeCalls  int
	getCalls     int
	listCalls    int
	listRCUCalls int

	listRCUResp  *openapi_changefeed_service.ListChangefeedRCUsOK
	listRCUErr   error
}

func (f *fakeOpenAPIClient) ListChangefeedRCUs() (*openapi_changefeed_service.ListChangefeedRCUsOK, error) {
	f.listRCUCalls++
	return f.listRCUResp, f.listRCUErr
}
```

- [ ] **Step 6: Re-run tests to verify they pass**

Run:

```bash
cd /home/nil/pingcap/workbench/repos/dataflow-service/test/e2e
go test ./internal/tidbcloud/dedicated/cdcapi -run TestListChangefeedRCUs -count=1
```

Expected: PASS.

- [ ] **Step 7: Commit**

```bash
cd /home/nil/pingcap/workbench/repos/dataflow-service
git add test/e2e/internal/tidbcloud/dedicated/cdcapi/client_wrapper.go \
        test/e2e/internal/tidbcloud/dedicated/cdcapi/client_wrapper_test.go
git commit -m "feat(e2e): add unified cdcapi RCU list adapter"
```

### Task 3: Add RCU list verification step in kafka smoke case

**Files:**
- Modify: `test/e2e/cases/dedicated/changefeed/kafka_smoke.go`
- Test: `test/e2e/cases/dedicated/changefeed/kafka_smoke.go` (package compile check)

- [ ] **Step 1: Add failing assertion step in chain**

```go
.Step("list changefeed rcus and ensure non-empty", func() error {
	rcuList, err := cdcApiWrapper.ListChangefeedRCUs(tidbCluster)
	if err != nil {
		return err
	}
	if rcuList == nil || rcuList.Items == nil || len(rcuList.Items) == 0 {
		return fmt.Errorf("changefeed rcu list is empty")
	}
	return nil
})
```

- [ ] **Step 2: Run package test/compile check**

Run:

```bash
cd /home/nil/pingcap/workbench/repos/dataflow-service/test/e2e
go test ./cases/dedicated/changefeed -count=1
```

Expected: PASS (or `ok` with no test files but successful compile).

- [ ] **Step 3: Commit**

```bash
cd /home/nil/pingcap/workbench/repos/dataflow-service
git add test/e2e/cases/dedicated/changefeed/kafka_smoke.go
git commit -m "test(e2e): validate changefeed RCU list in kafka smoke"
```

### Task 4: Final verification and integration commit strategy

**Files:**
- Test: `test/e2e/internal/tidbcloud/dedicated/openapi/open_api_client_wrapper_test.go`
- Test: `test/e2e/internal/tidbcloud/dedicated/cdcapi/client_wrapper_test.go`
- Test: `test/e2e/cases/dedicated/changefeed/kafka_smoke.go` (compile check)

- [ ] **Step 1: Run targeted tests in one pass**

Run:

```bash
cd /home/nil/pingcap/workbench/repos/dataflow-service/test/e2e
go test ./internal/tidbcloud/dedicated/openapi ./internal/tidbcloud/dedicated/cdcapi ./cases/dedicated/changefeed -count=1
```

Expected: all selected packages pass.

- [ ] **Step 2: Inspect final diff for generated-file safety**

Run:

```bash
cd /home/nil/pingcap/workbench/repos/dataflow-service
git --no-pager diff -- test/e2e/internal/tidbcloud/dedicated/openapi/changefeed/client \
                     test/e2e/internal/tidbcloud/dedicated/consoleapi/cdc/client
```

Expected: no diff output (no generated files touched).

- [ ] **Step 3: Squash strategy decision**

If preserving per-task commits is preferred, keep all commits.  
If a single PR commit is preferred, interactive squash before PR creation (no content change).