rangefeed: remove LegacyProcessor

The ScheduledProcessor has been in default use since 24.1,
substantially out performs the LegacyProcessor, and is the likely
place that new improvements will be implemented.

Here, we remove the LegacyProcessor, retiring the associated cluster
setting.

This is similar to cockroachdb#114410, but reconstructed from scratch.

Epic: none

Release note (ops change): The setting kv.rangefeed.scheduler.enabled
has been retired. The rangefeed scheduler is now unconditionally
enabled.

Co-authored-by: Erik Grinaker <[email protected]>

pkg/cmd/roachtest/tests/cdc_bench.go

@@ -56,15 +56,17 @@ const (
 	// practice it can.
 	cdcBenchColdCatchupScan cdcBenchScanType = "catchup-cold"
 
-	cdcBenchNoServer        cdcBenchServer = ""
-	cdcBenchProcessorServer cdcBenchServer = "processor" // legacy processor
+	cdcBenchNoServer cdcBenchServer = ""
+	// The legacy processor was removed in 25.1+. In such
+	// timeseries, "processor" refers to the now defunct legacy
+	// processor.
 	cdcBenchSchedulerServer cdcBenchServer = "scheduler" // new scheduler
 )
 
 var (
 	cdcBenchScanTypes = []cdcBenchScanType{
 		cdcBenchInitialScan, cdcBenchCatchupScan, cdcBenchColdCatchupScan}
-	cdcBenchServers = []cdcBenchServer{cdcBenchProcessorServer, cdcBenchSchedulerServer}
+	cdcBenchServers = []cdcBenchServer{cdcBenchSchedulerServer}
 )
 
 func registerCDCBench(r registry.Registry) {
@@ -416,8 +418,6 @@ func runCDCBenchWorkload(
 	settings.ClusterSettings["server.child_metrics.enabled"] = "true"
 
 	switch server {
-	case cdcBenchProcessorServer:
-		settings.ClusterSettings["kv.rangefeed.scheduler.enabled"] = "false"
 	case cdcBenchSchedulerServer:
 		settings.ClusterSettings["kv.rangefeed.scheduler.enabled"] = "true"
 	case cdcBenchNoServer:

pkg/kv/kvclient/rangefeed/rangefeed_external_test.go

@@ -65,9 +65,6 @@ func (t feedProcessorType) String() string {
 }
 
 var procTypes = []feedProcessorType{
-	{
-		useScheduler: false,
-	},
 	{
 		useScheduler: true,
 	},

pkg/kv/kvserver/BUILD.bazel

@@ -397,7 +397,6 @@ go_test(
         "//pkg/kv",
         "//pkg/kv/kvclient",
         "//pkg/kv/kvclient/kvcoord",
-        "//pkg/kv/kvclient/rangefeed",
         "//pkg/kv/kvclient/rangefeed/rangefeedcache",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/abortspan",

pkg/kv/kvserver/rangefeed/BUILD.bazel

@@ -20,7 +20,6 @@ go_library(
         "stream.go",
         "task.go",
         "test_helpers.go",
-        "testutil.go",
         "unbuffered_sender.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/rangefeed",

pkg/kv/kvserver/rangefeed/bench_test.go

@@ -24,10 +24,10 @@ import (
 )
 
 type benchmarkRangefeedOpts struct {
-	procType         procType
-	opType           opType
-	numRegistrations int
-	budget           int64
+	rangefeedTestType rangefeedTestType
+	opType            opType
+	numRegistrations  int
+	budget            int64
 }
 
 type opType string
@@ -47,10 +47,10 @@ func BenchmarkRangefeed(b *testing.B) {
 				name := fmt.Sprintf("procType=%s/opType=%s/numRegs=%d", procType, opType, numRegistrations)
 				b.Run(name, func(b *testing.B) {
 					runBenchmarkRangefeed(b, benchmarkRangefeedOpts{
-						procType:         procType,
-						opType:           opType,
-						numRegistrations: numRegistrations,
-						budget:           math.MaxInt64,
+						rangefeedTestType: procType,
+						opType:            opType,
+						numRegistrations:  numRegistrations,
+						budget:            math.MaxInt64,
 					})
 				})
 			}
@@ -90,7 +90,7 @@ func runBenchmarkRangefeed(b *testing.B, opts benchmarkRangefeedOpts) {
 	span := roachpb.RSpan{Key: roachpb.RKey("a"), EndKey: roachpb.RKey("z")}
 
 	p, h, stopper := newTestProcessor(b, withSpan(span), withBudget(budget), withChanCap(b.N),
-		withEventTimeout(time.Hour), withProcType(opts.procType))
+		withEventTimeout(time.Hour), withRangefeedTestType(opts.rangefeedTestType))
 	defer stopper.Stop(ctx)
 
 	// Add registrations.