@@ -21,6 +21,7 @@ import (
2121 "path"
2222 "reflect"
2323 "regexp"
24+ "runtime"
2425 "slices"
2526 "sort"
2627 "strconv"
@@ -13084,3 +13085,117 @@ func TestDatabaseLevelChangefeedWithInitialScanOptions(t *testing.T) {
1308413085 })
1308513086 }
1308613087}
13088+
13089+ // TestChangefeedNumSinkWorkersPrecedence verifies that the sink_io_workers
13090+ // option works correctly and that the precedence logic (changefeed option is used when set)
13091+ // is applied when both cluster setting and per-changefeed option are set.
13092+ func TestChangefeedNumSinkWorkersPrecedence (t * testing.T ) {
13093+ defer leaktest .AfterTest (t )()
13094+ defer log .Scope (t ).Close (t )
13095+
13096+ testFn := func (t * testing.T , s TestServer , f cdctest.TestFeedFactory ) {
13097+ registry := s .Server .JobRegistry ().(* jobs.Registry )
13098+ metrics := registry .MetricsStruct ().Changefeed .(* Metrics ).AggMetrics
13099+ sqlDB := sqlutils .MakeSQLRunner (s .DB )
13100+
13101+ sqlDB .Exec (t , `CREATE TABLE foo (a INT PRIMARY KEY)` )
13102+ sqlDB .Exec (t , `INSERT INTO foo VALUES (1)` )
13103+ KafkaV2Enabled .Override (context .Background (), & s .Server .ClusterSettings ().SV , true )
13104+
13105+ t .Run ("uses per-changefeed option when cluster setting is zero" , func (t * testing.T ) {
13106+ sqlDB .Exec (t , `SET CLUSTER SETTING changefeed.sink_io_workers = 0` )
13107+ cf := feed (t , f , `CREATE CHANGEFEED FOR foo WITH sink_io_workers = '5'` )
13108+ defer closeFeed (t , cf )
13109+
13110+ testutils .SucceedsSoon (t , func () error {
13111+ workers := metrics .ParallelIOWorkers .Value ()
13112+ if workers != 5 {
13113+ return errors .Newf ("expected 5 workers, got %d" , workers )
13114+ }
13115+ return nil
13116+ })
13117+ })
13118+
13119+ t .Run ("uses per-changefeed option when it is set" , func (t * testing.T ) {
13120+ sqlDB .Exec (t , `SET CLUSTER SETTING changefeed.sink_io_workers = 10` )
13121+ cf := feed (t , f , `CREATE CHANGEFEED FOR foo WITH sink_io_workers = '3'` )
13122+ defer closeFeed (t , cf )
13123+
13124+ testutils .SucceedsSoon (t , func () error {
13125+ workers := metrics .ParallelIOWorkers .Value ()
13126+ if workers != 3 {
13127+ return errors .Newf ("expected 3 workers (from changefeed setting), got %d" , workers )
13128+ }
13129+ return nil
13130+ })
13131+ })
13132+
13133+ t .Run ("uses cluster setting when per-changefeed option is not set" , func (t * testing.T ) {
13134+ sqlDB .Exec (t , `SET CLUSTER SETTING changefeed.sink_io_workers = 7` )
13135+ cf := feed (t , f , `CREATE CHANGEFEED FOR foo` )
13136+ defer closeFeed (t , cf )
13137+
13138+ testutils .SucceedsSoon (t , func () error {
13139+ workers := metrics .ParallelIOWorkers .Value ()
13140+ if workers != 7 {
13141+ return errors .Newf ("expected 7 workers (from cluster setting), got %d" , workers )
13142+ }
13143+ return nil
13144+ })
13145+ })
13146+
13147+ t .Run ("uses GOMAXPROCS default when changefeed setting is negative" , func (t * testing.T ) {
13148+ sqlDB .Exec (t , `SET CLUSTER SETTING changefeed.sink_io_workers = 10` )
13149+ cf := feed (t , f , `CREATE CHANGEFEED FOR foo WITH sink_io_workers = '-1'` )
13150+ defer closeFeed (t , cf )
13151+
13152+ testutils .SucceedsSoon (t , func () error {
13153+ workers := metrics .ParallelIOWorkers .Value ()
13154+ expectedWorkers := runtime .GOMAXPROCS (0 )
13155+ if workers != int64 (expectedWorkers ) {
13156+ return errors .Newf ("expected %d workers (GOMAXPROCS-based), got %d" , expectedWorkers , workers )
13157+ }
13158+ return nil
13159+ })
13160+ })
13161+
13162+ t .Run ("uses GOMAXPROCS default when cluster setting is negative and no changefeed option is set" , func (t * testing.T ) {
13163+ sqlDB .Exec (t , `SET CLUSTER SETTING changefeed.sink_io_workers = -1` )
13164+ cf := feed (t , f , `CREATE CHANGEFEED FOR foo` )
13165+ defer closeFeed (t , cf )
13166+
13167+ testutils .SucceedsSoon (t , func () error {
13168+ workers := metrics .ParallelIOWorkers .Value ()
13169+ expectedWorkers := runtime .GOMAXPROCS (0 )
13170+ if workers != int64 (expectedWorkers ) {
13171+ return errors .Newf ("expected %d workers (GOMAXPROCS-based), got %d" , expectedWorkers , workers )
13172+ }
13173+ return nil
13174+ })
13175+ })
13176+
13177+ t .Run ("uses GOMAXPROCS default when both sink_io_workers settings are negative" , func (t * testing.T ) {
13178+ sqlDB .Exec (t , `SET CLUSTER SETTING changefeed.sink_io_workers = -1` )
13179+ cf := feed (t , f , `CREATE CHANGEFEED FOR foo WITH sink_io_workers = '-1'` )
13180+ defer closeFeed (t , cf )
13181+
13182+ testutils .SucceedsSoon (t , func () error {
13183+ workers := metrics .ParallelIOWorkers .Value ()
13184+ expectedWorkers := runtime .GOMAXPROCS (0 )
13185+ if workers != int64 (expectedWorkers ) {
13186+ return errors .Newf ("expected %d workers (GOMAXPROCS-based), got %d" , expectedWorkers , workers )
13187+ }
13188+ return nil
13189+ })
13190+ })
13191+
13192+ t .Run ("rejects invalid sink_io_workers value" , func (t * testing.T ) {
13193+ sqlDB .Exec (t , `SET CLUSTER SETTING changefeed.sink_io_workers = 5` )
13194+ sqlDB .ExpectErrWithTimeout (t , "invalid integer value" ,
13195+ `CREATE CHANGEFEED FOR foo WITH sink_io_workers = 'invalid'` )
13196+ })
13197+ }
13198+
13199+ // Test with sinks that support parallel IO.
13200+ cdcTest (t , testFn , feedTestRestrictSinks ("pubsub" , "webhook" , "kafka" ))
13201+ }
0 commit comments