processor(ticdc): clean up dead code

cdc/processor/manager.go

@@ -38,9 +38,6 @@ const (
 	commandTpUnknown commandTp = iota
 	commandTpClose
 	commandTpWriteDebugInfo
-	// Query the number of tables in the manager.
-	// command payload is a buffer channel of int, make(chan int, 1).
-	commandTpQueryTableCount
 	processorLogsWarnDuration = 1 * time.Second
 )
 
@@ -53,7 +50,6 @@ type command struct {
 // Manager is a manager of processor, which maintains the state and behavior of processors
 type Manager interface {
 	orchestrator.Reactor
-	QueryTableCount(ctx context.Context, tableCh chan int, done chan<- error)
 	WriteDebugInfo(ctx context.Context, w io.Writer, done chan<- error)
 	AsyncClose()
 }
@@ -199,16 +195,6 @@ func (m *managerImpl) AsyncClose() {
 	}
 }
 
-// QueryTableCount query the number of tables in the manager.
-func (m *managerImpl) QueryTableCount(
-	ctx context.Context, tableCh chan int, done chan<- error,
-) {
-	err := m.sendCommand(ctx, commandTpQueryTableCount, tableCh, done)
-	if err != nil {
-		log.Warn("send command commandTpQueryTableCount failed", zap.Error(err))
-	}
-}
-
 // WriteDebugInfo write the debug info to Writer
 func (m *managerImpl) WriteDebugInfo(
 	ctx context.Context, w io.Writer, done chan<- error,
@@ -256,15 +242,6 @@ func (m *managerImpl) handleCommand(ctx cdcContext.Context) error {
 		if err != nil {
 			cmd.done <- err
 		}
-	case commandTpQueryTableCount:
-		count := 0
-		for _, p := range m.processors {
-			count += p.GetTableSpanCount()
-		}
-		select {
-		case cmd.payload.(chan int) <- count:
-		default:
-		}
 	default:
 		log.Warn("Unknown command in processor manager", zap.Any("command", cmd))
 	}

cdc/processor/manager_test.go

@@ -291,25 +291,3 @@ func TestManagerLiveness(t *testing.T) {
 	s.liveness.Store(model.LivenessCaptureStopping)
 	require.Equal(t, model.LivenessCaptureStopping, p.liveness.Load())
 }
-
-func TestQueryTableCount(t *testing.T) {
-	t.Skip("FIXME: add tables")
-	liveness := model.LivenessCaptureAlive
-	cfg := config.NewDefaultSchedulerConfig()
-	m := NewManager(&model.CaptureInfo{ID: "capture-test"}, nil, &liveness, cfg).(*managerImpl)
-	ctx := context.TODO()
-	m.processors[model.ChangeFeedID{ID: "test"}] = &processor{}
-
-	done := make(chan error, 1)
-	tableCh := make(chan int, 1)
-	err := m.sendCommand(ctx, commandTpQueryTableCount, tableCh, done)
-	require.Nil(t, err)
-	err = m.handleCommand(nil)
-	require.Nil(t, err)
-	select {
-	case count := <-tableCh:
-		require.Equal(t, 2, count)
-	case <-time.After(time.Second):
-		require.FailNow(t, "done must be closed")
-	}
-}