diff --git a/go.mod b/go.mod index 372f022fa3..f3cbc48363 100644 --- a/go.mod +++ b/go.mod @@ -52,7 +52,7 @@ require ( github.com/pierrec/lz4/v4 v4.1.26 github.com/pingcap/errors v0.11.5-0.20260508054701-306e305bcf41 github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 - github.com/pingcap/kvproto v0.0.0-20260601035955-b2b3bb492278 + github.com/pingcap/kvproto v0.0.0-20260707053407-18552472891c github.com/pingcap/log v1.1.1-0.20250917021125-19901e015dc9 github.com/pingcap/sysutil v1.0.1-0.20240311050922-ae81ee01f3a5 github.com/pingcap/tidb v1.1.0-beta.0.20260604031706-f9faeaf4828f diff --git a/go.sum b/go.sum index a3e32540d5..562b19881a 100644 --- a/go.sum +++ b/go.sum @@ -750,8 +750,8 @@ github.com/pingcap/fn v1.0.0/go.mod h1:u9WZ1ZiOD1RpNhcI42RucFh/lBuzTu6rw88a+oF2Z github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989 h1:surzm05a8C9dN8dIUmo4Be2+pMRb6f55i+UIYrluu2E= github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989/go.mod h1:O17XtbryoCJhkKGbT62+L2OlrniwqiGLSqrmdHCMzZw= github.com/pingcap/kvproto v0.0.0-20191211054548-3c6b38ea5107/go.mod h1:WWLmULLO7l8IOcQG+t+ItJ3fEcrL5FxF0Wu+HrMy26w= -github.com/pingcap/kvproto v0.0.0-20260601035955-b2b3bb492278 h1:VV03wSSG2XhOwhF79lvT88Z83lwzMT+aZHXrcCIN9B0= -github.com/pingcap/kvproto v0.0.0-20260601035955-b2b3bb492278/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE= +github.com/pingcap/kvproto v0.0.0-20260707053407-18552472891c h1:jeuQge1aBQp9VRboYbh+k7CCMlLI+8WT1Zustk5jpHU= +github.com/pingcap/kvproto v0.0.0-20260707053407-18552472891c/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE= github.com/pingcap/log v0.0.0-20191012051959-b742a5d432e9/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8= github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM= github.com/pingcap/log v1.1.0/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4= diff --git a/logservice/eventstore/event_store.go b/logservice/eventstore/event_store.go index f5b3e78986..1ac83f26df 100644 --- a/logservice/eventstore/event_store.go +++ b/logservice/eventstore/event_store.go @@ -78,9 +78,7 @@ type Subscriber struct { type EventStore interface { common.SubModule - // Note: changefeedID is just a tag for dispatcher, avoid abuse it RegisterDispatcher( - changefeedID common.ChangeFeedID, dispatcherID common.DispatcherID, span *heartbeatpb.TableSpan, startTS uint64, @@ -460,7 +458,6 @@ func (e *eventStore) Close(_ context.Context) error { } func (e *eventStore) RegisterDispatcher( - _ common.ChangeFeedID, dispatcherID common.DispatcherID, dispatcherSpan *heartbeatpb.TableSpan, startTs uint64, diff --git a/logservice/eventstore/event_store_test.go b/logservice/eventstore/event_store_test.go index 4ac22e504d..e324de858f 100644 --- a/logservice/eventstore/event_store_test.go +++ b/logservice/eventstore/event_store_test.go @@ -191,7 +191,6 @@ func TestEventStoreInteractionWithSubClient(t *testing.T) { dispatcherID1 := common.NewDispatcherID() dispatcherID2 := common.NewDispatcherID() dispatcherID3 := common.NewDispatcherID() - cfID := common.NewChangefeedID4Test("default", "test-cf") { span := &heartbeatpb.TableSpan{ @@ -199,7 +198,7 @@ func TestEventStoreInteractionWithSubClient(t *testing.T) { StartKey: []byte("a"), EndKey: []byte("e"), } - ok := store.RegisterDispatcher(cfID, dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } // add a dispatcher with the same span @@ -209,7 +208,7 @@ func TestEventStoreInteractionWithSubClient(t *testing.T) { StartKey: []byte("a"), EndKey: []byte("e"), } - ok := store.RegisterDispatcher(cfID, dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } // check there is only one subscription in subClient @@ -226,7 +225,7 @@ func TestEventStoreInteractionWithSubClient(t *testing.T) { StartKey: []byte("a"), EndKey: []byte("b"), } - ok := store.RegisterDispatcher(cfID, dispatcherID3, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID3, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } // check a new subscription is created in subClient @@ -247,14 +246,13 @@ func TestEventStoreUsesKeyspaceIDForEncryption(t *testing.T) { es.encryptionManager = spy dispatcherID := common.NewDispatcherID() - cfID := common.NewChangefeedID4Test("default", "test-cf") span := &heartbeatpb.TableSpan{ TableID: 1, StartKey: []byte("a"), EndKey: []byte("z"), KeyspaceID: 42, } - ok := store.RegisterDispatcher(cfID, dispatcherID, span, 0, func(uint64, uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID, span, 0, func(uint64, uint64) {}, false, false) require.True(t, ok) es.dispatcherMeta.RLock() @@ -341,14 +339,13 @@ func TestEventStoreHandlesUnencryptedValuesFromEncryptionLayer(t *testing.T) { es.encryptionManager = encryption.NewEncryptionManager(&unencryptedMetaManager{}) dispatcherID := common.NewDispatcherID() - cfID := common.NewChangefeedID4Test("default", "test-cf") span := &heartbeatpb.TableSpan{ TableID: 1, StartKey: []byte("a"), EndKey: []byte("z"), KeyspaceID: 42, } - ok := store.RegisterDispatcher(cfID, dispatcherID, span, 0, func(uint64, uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID, span, 0, func(uint64, uint64) {}, false, false) require.True(t, ok) es.dispatcherMeta.RLock() @@ -437,7 +434,6 @@ func TestEventStoreOnlyReuseDispatcher(t *testing.T) { dispatcherID2 := common.NewDispatcherID() dispatcherID3 := common.NewDispatcherID() tableID := int64(1) - cfID := common.NewChangefeedID4Test("default", "test-cf") // add a dispatcher to create a subscription { span := &heartbeatpb.TableSpan{ @@ -445,7 +441,7 @@ func TestEventStoreOnlyReuseDispatcher(t *testing.T) { StartKey: []byte("a"), EndKey: []byte("h"), } - ok := store.RegisterDispatcher(cfID, dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } // add a dispatcher(onlyReuse=true) with a non-containing span which should fail @@ -455,7 +451,7 @@ func TestEventStoreOnlyReuseDispatcher(t *testing.T) { StartKey: []byte("b"), EndKey: []byte("i"), } - ok := store.RegisterDispatcher(cfID, dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false) + ok := store.RegisterDispatcher(dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false) require.False(t, ok) } // when the existing subscription is not initialized, add a dispatcher(onlyReuse=true) should fail @@ -465,7 +461,7 @@ func TestEventStoreOnlyReuseDispatcher(t *testing.T) { StartKey: []byte("b"), EndKey: []byte("h"), } - ok := store.RegisterDispatcher(cfID, dispatcherID3, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false) + ok := store.RegisterDispatcher(dispatcherID3, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false) require.False(t, ok) } // mark existing subscription as initialized @@ -477,11 +473,11 @@ func TestEventStoreOnlyReuseDispatcher(t *testing.T) { StartKey: []byte("b"), EndKey: []byte("h"), } - ok := store.RegisterDispatcher(cfID, dispatcherID3, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false) + ok := store.RegisterDispatcher(dispatcherID3, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false) require.True(t, ok) } { - store.UnregisterDispatcher(cfID, dispatcherID1) + store.UnregisterDispatcher(common.NewChangefeedID4Test("default", "test-cf"), dispatcherID1) subStats := store.(*eventStore).dispatcherMeta.tableStats[tableID] require.Equal(t, 1, len(subStats)) // because there is only one subStat, we know its subID is 1 @@ -491,7 +487,7 @@ func TestEventStoreOnlyReuseDispatcher(t *testing.T) { require.NotNil(t, subData) require.Equal(t, 1, len(subData.subscribers)) require.Equal(t, int64(0), subData.idleTime) - store.UnregisterDispatcher(cfID, dispatcherID3) + store.UnregisterDispatcher(common.NewChangefeedID4Test("default", "test-cf"), dispatcherID3) subData = subStat.subscribers.Load() require.NotNil(t, subData) require.Equal(t, 0, len(subData.subscribers)) @@ -510,7 +506,6 @@ func TestEventStoreOnlyReuseDispatcherSuccess(t *testing.T) { dispatcherID2 := common.NewDispatcherID() dispatcherID3 := common.NewDispatcherID() tableID := int64(1) - cfID := common.NewChangefeedID4Test("default", "test-cf") // 1. Register a dispatcher to create a large subscription. { @@ -519,7 +514,7 @@ func TestEventStoreOnlyReuseDispatcherSuccess(t *testing.T) { StartKey: []byte("a"), EndKey: []byte("z"), } - ok := es.RegisterDispatcher(cfID, dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := es.RegisterDispatcher(dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } markSubStatsInitializedForTest(store, tableID) @@ -532,7 +527,7 @@ func TestEventStoreOnlyReuseDispatcherSuccess(t *testing.T) { StartKey: []byte("b"), EndKey: []byte("y"), } - ok := es.RegisterDispatcher(cfID, dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false) + ok := es.RegisterDispatcher(dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false) require.True(t, ok) } @@ -544,7 +539,7 @@ func TestEventStoreOnlyReuseDispatcherSuccess(t *testing.T) { StartKey: []byte("a"), EndKey: []byte("z"), } - ok := es.RegisterDispatcher(cfID, dispatcherID3, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false) + ok := es.RegisterDispatcher(dispatcherID3, span, 100, func(watermark uint64, latestCommitTs uint64) {}, true, false) require.True(t, ok) } } @@ -560,7 +555,6 @@ func TestEventStoreNonOnlyReuseDispatcher(t *testing.T) { dispatcherID3 := common.NewDispatcherID() dispatcherID4 := common.NewDispatcherID() tableID := int64(1) - cfID := common.NewChangefeedID4Test("default", "test-cf") // add a subscription to create a subscription { span := &heartbeatpb.TableSpan{ @@ -568,7 +562,7 @@ func TestEventStoreNonOnlyReuseDispatcher(t *testing.T) { StartKey: []byte("a"), EndKey: []byte("h"), } - ok := store.RegisterDispatcher(cfID, dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } // add a dispatcher(onlyReuse=false) with a non-containing span @@ -578,7 +572,7 @@ func TestEventStoreNonOnlyReuseDispatcher(t *testing.T) { StartKey: []byte("c"), EndKey: []byte("i"), } - ok := store.RegisterDispatcher(cfID, dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } // do some check @@ -594,7 +588,7 @@ func TestEventStoreNonOnlyReuseDispatcher(t *testing.T) { StartKey: []byte("b"), EndKey: []byte("h"), } - ok := store.RegisterDispatcher(cfID, dispatcherID3, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID3, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } // do some check @@ -616,7 +610,7 @@ func TestEventStoreNonOnlyReuseDispatcher(t *testing.T) { StartKey: []byte("a"), EndKey: []byte("h"), } - ok := store.RegisterDispatcher(cfID, dispatcherID4, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID4, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) subStats := store.(*eventStore).dispatcherMeta.tableStats[tableID] require.Equal(t, 3, len(subStats)) @@ -629,7 +623,7 @@ func TestEventStoreNonOnlyReuseDispatcher(t *testing.T) { } // test unregister dispatcherID3 can remove its dependency on two subscriptions { - store.UnregisterDispatcher(cfID, dispatcherID3) + store.UnregisterDispatcher(common.NewChangefeedID4Test("default", "test-cf"), dispatcherID3) subStats := store.(*eventStore).dispatcherMeta.tableStats[tableID] require.Equal(t, 3, len(subStats)) { @@ -657,7 +651,6 @@ func TestEventStoreRegisterDispatcherWithoutDataSharing(t *testing.T) { es := store.(*eventStore) tableID := int64(1) - cfID := common.NewChangefeedID4Test("default", "test-cf") dispatcherID1 := common.NewDispatcherID() dispatcherID2 := common.NewDispatcherID() dispatcherID3 := common.NewDispatcherID() @@ -668,16 +661,16 @@ func TestEventStoreRegisterDispatcherWithoutDataSharing(t *testing.T) { StartKey: []byte("a"), EndKey: []byte("h"), } - require.True(t, store.RegisterDispatcher(cfID, dispatcherID1, spanFull, 100, func(uint64, uint64) {}, false, false)) + require.True(t, store.RegisterDispatcher(dispatcherID1, spanFull, 100, func(uint64, uint64) {}, false, false)) - require.True(t, store.RegisterDispatcher(cfID, dispatcherID2, spanFull, 100, func(uint64, uint64) {}, false, false)) + require.True(t, store.RegisterDispatcher(dispatcherID2, spanFull, 100, func(uint64, uint64) {}, false, false)) spanSubset := &heartbeatpb.TableSpan{ TableID: tableID, StartKey: []byte("b"), EndKey: []byte("g"), } - require.True(t, store.RegisterDispatcher(cfID, dispatcherID3, spanSubset, 100, func(uint64, uint64) {}, false, false)) + require.True(t, store.RegisterDispatcher(dispatcherID3, spanSubset, 100, func(uint64, uint64) {}, false, false)) mockSubClient := subClient.(*mockSubscriptionClient) mockSubClient.mu.Lock() @@ -691,7 +684,7 @@ func TestEventStoreRegisterDispatcherWithoutDataSharing(t *testing.T) { require.Nil(t, es.dispatcherMeta.dispatcherStats[dispatcherID3].pendingSubStat) es.dispatcherMeta.RUnlock() - ok := store.RegisterDispatcher(cfID, dispatcherID4, spanFull, 100, func(uint64, uint64) {}, true, false) + ok := store.RegisterDispatcher(dispatcherID4, spanFull, 100, func(uint64, uint64) {}, true, false) require.False(t, ok) es.dispatcherMeta.RLock() @@ -711,8 +704,6 @@ func TestGetIteratorPanicWhenStartLessThanCheckpoint(t *testing.T) { defer func() { require.NoError(t, store.Close(context.Background())) }() - - cfID := common.NewChangefeedID4Test("default", "test") dispatcherID := common.NewDispatcherID() span := &heartbeatpb.TableSpan{ TableID: 1, @@ -720,7 +711,7 @@ func TestGetIteratorPanicWhenStartLessThanCheckpoint(t *testing.T) { EndKey: []byte("z"), } - require.True(t, store.RegisterDispatcher(cfID, dispatcherID, span, 100, func(uint64, uint64) {}, false, false)) + require.True(t, store.RegisterDispatcher(dispatcherID, span, 100, func(uint64, uint64) {}, false, false)) stat := store.dispatcherMeta.dispatcherStats[dispatcherID] require.NotNil(t, stat) @@ -746,21 +737,20 @@ func TestEventStoreUnregisterDispatcherWithoutDataSharingRemovesSubscription(t * es := store.(*eventStore) tableID := int64(1) - cfID := common.NewChangefeedID4Test("default", "test-cf") dispatcherID := common.NewDispatcherID() span := &heartbeatpb.TableSpan{ TableID: tableID, StartKey: []byte("a"), EndKey: []byte("h"), } - require.True(t, store.RegisterDispatcher(cfID, dispatcherID, span, 100, func(uint64, uint64) {}, false, false)) + require.True(t, store.RegisterDispatcher(dispatcherID, span, 100, func(uint64, uint64) {}, false, false)) mockSubClient := subClient.(*mockSubscriptionClient) mockSubClient.mu.Lock() require.Equal(t, 1, len(mockSubClient.subscriptions)) mockSubClient.mu.Unlock() - store.UnregisterDispatcher(cfID, dispatcherID) + store.UnregisterDispatcher(common.NewChangefeedID4Test("default", "test-cf"), dispatcherID) mockSubClient.mu.Lock() require.Equal(t, 1, len(mockSubClient.subscriptions)) @@ -788,16 +778,15 @@ func TestEventStoreUnregisterDispatcherWithDataSharingKeepsSubscriptionForTTL(t es := store.(*eventStore) tableID := int64(1) - cfID := common.NewChangefeedID4Test("default", "test-cf") dispatcherID := common.NewDispatcherID() span := &heartbeatpb.TableSpan{ TableID: tableID, StartKey: []byte("a"), EndKey: []byte("h"), } - require.True(t, store.RegisterDispatcher(cfID, dispatcherID, span, 100, func(uint64, uint64) {}, false, false)) + require.True(t, store.RegisterDispatcher(dispatcherID, span, 100, func(uint64, uint64) {}, false, false)) - store.UnregisterDispatcher(cfID, dispatcherID) + store.UnregisterDispatcher(common.NewChangefeedID4Test("default", "test-cf"), dispatcherID) mockSubClient := subClient.(*mockSubscriptionClient) mockSubClient.mu.Lock() @@ -833,7 +822,6 @@ func TestEventStoreUpdateCheckpointTs(t *testing.T) { dispatcherID1 := common.NewDispatcherID() dispatcherID2 := common.NewDispatcherID() tableID := int64(1) - cfID := common.NewChangefeedID4Test("default", "test-cf") // add first dispatcher { span := &heartbeatpb.TableSpan{ @@ -841,7 +829,7 @@ func TestEventStoreUpdateCheckpointTs(t *testing.T) { StartKey: []byte("a"), EndKey: []byte("h"), } - ok := store.RegisterDispatcher(cfID, dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } // add a dispatcher(onlyReuse=false) with a containing span @@ -851,7 +839,7 @@ func TestEventStoreUpdateCheckpointTs(t *testing.T) { StartKey: []byte("b"), EndKey: []byte("h"), } - ok := store.RegisterDispatcher(cfID, dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } // check subStat checkpointTs cannot advance when their resolved ts is not advanced @@ -920,15 +908,14 @@ func TestEventStoreUpdateCheckpointTsConcurrentStaleUpdates(t *testing.T) { dispatcherID1 := common.NewDispatcherID() dispatcherID2 := common.NewDispatcherID() tableID := int64(1) - cfID := common.NewChangefeedID4Test("default", "test-cf") span := &heartbeatpb.TableSpan{ TableID: tableID, StartKey: []byte("a"), EndKey: []byte("h"), } - require.True(t, store.RegisterDispatcher(cfID, dispatcherID1, span, 100, func(uint64, uint64) {}, false, false)) - require.True(t, store.RegisterDispatcher(cfID, dispatcherID2, span, 100, func(uint64, uint64) {}, false, false)) + require.True(t, store.RegisterDispatcher(dispatcherID1, span, 100, func(uint64, uint64) {}, false, false)) + require.True(t, store.RegisterDispatcher(dispatcherID2, span, 100, func(uint64, uint64) {}, false, false)) es.dispatcherMeta.RLock() stat1 := es.dispatcherMeta.dispatcherStats[dispatcherID1] @@ -975,7 +962,6 @@ func TestEventStoreSwitchSubStat(t *testing.T) { dispatcherID1 := common.NewDispatcherID() dispatcherID2 := common.NewDispatcherID() tableID := int64(1) - cfID := common.NewChangefeedID4Test("default", "test-cf") updateSubStatResolvedTs := func(subID logpuller.SubscriptionID, ts uint64) { subStats := store.(*eventStore).dispatcherMeta.tableStats[tableID] @@ -1007,7 +993,7 @@ func TestEventStoreSwitchSubStat(t *testing.T) { StartKey: []byte("a"), EndKey: []byte("h"), } - ok := store.RegisterDispatcher(cfID, dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID1, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } // add a dispatcher(onlyReuse=false) with a containing span @@ -1018,7 +1004,7 @@ func TestEventStoreSwitchSubStat(t *testing.T) { StartKey: []byte("b"), EndKey: []byte("h"), } - ok := store.RegisterDispatcher(cfID, dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) + ok := store.RegisterDispatcher(dispatcherID2, span, 100, func(watermark uint64, latestCommitTs uint64) {}, false, false) require.True(t, ok) } @@ -1528,12 +1514,11 @@ func TestEventStoreGetIteratorConcurrently(t *testing.T) { // 1. Register a dispatcher. dispatcherID := common.NewDispatcherID() - cfID := common.NewChangefeedID4Test("default", "test-cf") span := &heartbeatpb.TableSpan{TableID: 1, StartKey: []byte("a"), EndKey: []byte("z")} startTs := uint64(100) var resolvedTs atomic.Uint64 resolvedTs.Store(startTs) - ok := store.RegisterDispatcher(cfID, dispatcherID, span, startTs, func(watermark, latestCommitTs uint64) { + ok := store.RegisterDispatcher(dispatcherID, span, startTs, func(watermark, latestCommitTs uint64) { resolvedTs.Store(watermark) }, false, false) require.True(t, ok) diff --git a/logservice/logpuller/priority_task.go b/logservice/logpuller/priority_task.go index 68ef9e885d..56bc7f72cc 100644 --- a/logservice/logpuller/priority_task.go +++ b/logservice/logpuller/priority_task.go @@ -14,9 +14,9 @@ package logpuller import ( - "fmt" "time" + "github.com/pingcap/kvproto/pkg/cdcpb" "github.com/tikv/client-go/v2/oracle" ) @@ -39,7 +39,36 @@ const ( ) func (t TaskType) String() string { - return fmt.Sprintf("%d", t) + switch t { + case TaskHighPrior: + return "high" + case TaskLowPrior: + return "low" + default: + return "unknown" + } +} + +func (t TaskType) scanPriority() cdcpb.ScanPriority { + switch t { + case TaskHighPrior: + return cdcpb.ScanPriority_SCAN_PRIORITY_HIGH + case TaskLowPrior: + return cdcpb.ScanPriority_SCAN_PRIORITY_LOW + default: + return cdcpb.ScanPriority_SCAN_PRIORITY_LOW + } +} + +func taskTypeFromScanPriority(priority cdcpb.ScanPriority) TaskType { + if priority == cdcpb.ScanPriority_SCAN_PRIORITY_HIGH { + return TaskHighPrior + } + return TaskLowPrior +} + +func normalizeScanPriority(priority cdcpb.ScanPriority) cdcpb.ScanPriority { + return taskTypeFromScanPriority(priority).scanPriority() } // PriorityTask is the interface for priority-based tasks diff --git a/logservice/logpuller/priority_task_test.go b/logservice/logpuller/priority_task_test.go index d8b3f26d48..921f227102 100644 --- a/logservice/logpuller/priority_task_test.go +++ b/logservice/logpuller/priority_task_test.go @@ -18,6 +18,7 @@ import ( "testing" "time" + "github.com/pingcap/kvproto/pkg/cdcpb" "github.com/pingcap/ticdc/heartbeatpb" "github.com/pingcap/ticdc/utils/priorityqueue" "github.com/stretchr/testify/require" @@ -25,6 +26,15 @@ import ( "github.com/tikv/client-go/v2/tikv" ) +func TestTaskTypeScanPriorityMapping(t *testing.T) { + require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, TaskHighPrior.scanPriority()) + require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_LOW, TaskLowPrior.scanPriority()) + require.Equal(t, TaskHighPrior, taskTypeFromScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_HIGH)) + require.Equal(t, TaskLowPrior, taskTypeFromScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_LOW)) + require.Equal(t, TaskLowPrior, taskTypeFromScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_UNKNOWN)) + require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_LOW, normalizeScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_UNKNOWN)) +} + // TestPriorityCalculationLogic tests the priority calculation logic in isolation func TestPriorityCalculationLogic(t *testing.T) { currentTime := time.Now() diff --git a/logservice/logpuller/region_event_handler.go b/logservice/logpuller/region_event_handler.go index 7e96008c77..d8feaae8cb 100644 --- a/logservice/logpuller/region_event_handler.go +++ b/logservice/logpuller/region_event_handler.go @@ -142,6 +142,9 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent) log.Panic("should not reach", zap.Any("event", event), zap.Any("events", events)) } } + if newResolvedTs > 0 && h.subClient != nil { + h.subClient.maybeEnableRealtimeScanPriority(span, newResolvedTs) + } tryAdvanceResolvedTs := func() { if newResolvedTs != 0 { span.advanceResolvedTs(newResolvedTs) diff --git a/logservice/logpuller/region_request_worker.go b/logservice/logpuller/region_request_worker.go index 142f4c6493..367ddc2561 100644 --- a/logservice/logpuller/region_request_worker.go +++ b/logservice/logpuller/region_request_worker.go @@ -448,6 +448,7 @@ func (s *regionRequestWorker) createRegionRequest(region regionInfo) *cdcpb.Chan EndKey: region.span.EndKey, ExtraOp: kvrpcpb.ExtraOp_ReadOldValue, FilterLoop: region.filterLoop, + ScanPriority: normalizeScanPriority(region.scanPriority), } } diff --git a/logservice/logpuller/region_request_worker_test.go b/logservice/logpuller/region_request_worker_test.go index f9752a0eb3..a7231da794 100644 --- a/logservice/logpuller/region_request_worker_test.go +++ b/logservice/logpuller/region_request_worker_test.go @@ -57,6 +57,42 @@ func prepareRegionForSendTest(region regionInfo) regionInfo { return region } +func TestCreateRegionRequestScanPriority(t *testing.T) { + worker := ®ionRequestWorker{ + client: &subscriptionClient{clusterID: 1}, + } + + for _, tc := range []struct { + name string + priority cdcpb.ScanPriority + expected cdcpb.ScanPriority + }{ + { + name: "high", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + expected: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + }, + { + name: "low", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + expected: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + }, + { + name: "unknown defaults to low", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_UNKNOWN, + expected: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + }, + } { + t.Run(tc.name, func(t *testing.T) { + region := prepareRegionForSendTest(createTestRegionInfo(1, 1)) + region.scanPriority = tc.priority + + req := worker.createRegionRequest(region) + require.Equal(t, tc.expected, req.GetScanPriority()) + }) + } +} + func TestRegionStatesOperation(t *testing.T) { worker := ®ionRequestWorker{} worker.requestedRegions.subscriptions = make(map[SubscriptionID]regionFeedStates) diff --git a/logservice/logpuller/region_state.go b/logservice/logpuller/region_state.go index e9c21a7aad..40e98bdb26 100644 --- a/logservice/logpuller/region_state.go +++ b/logservice/logpuller/region_state.go @@ -16,6 +16,7 @@ package logpuller import ( "sync" + "github.com/pingcap/kvproto/pkg/cdcpb" "github.com/pingcap/ticdc/heartbeatpb" "github.com/pingcap/ticdc/logservice/logpuller/regionlock" "github.com/tikv/client-go/v2/tikv" @@ -46,6 +47,9 @@ type regionInfo struct { // Whether to filter out the value write by cdc itself. // It should be `true` in BDR mode filterLoop bool + // scanPriority is sent to TiKV/CSE so remote incremental scan admission can + // preserve TiCDC's business priority across retries. + scanPriority cdcpb.ScanPriority } func (s *regionInfo) isStopped() bool { @@ -66,6 +70,7 @@ func newRegionInfo( rpcCtx: rpcCtx, subscribedSpan: subscribedSpan, filterLoop: filterLoop, + scanPriority: TaskLowPrior.scanPriority(), } } diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index fbcffdc4f1..107da949d0 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -41,6 +41,7 @@ import ( "github.com/tikv/client-go/v2/tikv" pd "github.com/tikv/pd/client" "go.uber.org/zap" + "go.uber.org/zap/zapcore" "golang.org/x/sync/errgroup" ) @@ -141,6 +142,10 @@ type subscribedSpan struct { initialized atomic.Bool resolvedTsUpdated atomic.Int64 resolvedTs atomic.Uint64 + // realtimeScanPriority is set after this subscription catches up once. + // It is sticky so later recovery scans can protect realtime changefeeds + // even if historical catch-up scans temporarily push their lag up again. + realtimeScanPriority atomic.Bool } func (span *subscribedSpan) clearKVEventsCache() { @@ -370,16 +375,63 @@ func (s *subscriptionClient) Subscribe( areaSetting := dynstream.NewAreaSettingsWithMaxPendingSize(1*1024*1024*1024, dynstream.MemoryControlForPuller, "logPuller") // 1GB s.ds.AddPath(rt.subID, rt, areaSetting) + initialPriority := s.initialScanTaskPriority(startTs) select { case <-s.ctx.Done(): log.Warn("subscribes span failed, the subscription client has closed") - case s.rangeTaskCh <- rangeTask{span: span, subscribedSpan: rt, filterLoop: rt.filterLoop, priority: TaskLowPrior}: - log.Info("subscribes span done", zap.Uint64("subscriptionID", uint64(subID)), + case s.rangeTaskCh <- rangeTask{span: span, subscribedSpan: rt, filterLoop: rt.filterLoop, priority: initialPriority}: + log.Info("subscribes span done", + zap.Uint64("subscriptionID", uint64(subID)), zap.Int64("tableID", span.TableID), zap.Uint64("startTs", startTs), + zap.String("initialScanPriority", initialPriority.String()), zap.String("startKey", spanz.HexKey(span.StartKey)), zap.String("endKey", spanz.HexKey(span.EndKey))) } } +func (s *subscriptionClient) initialScanTaskPriority(startTs uint64) TaskType { + if s.isTsCloseToCurrent(startTs) { + return TaskHighPrior + } + return TaskLowPrior +} + +func (s *subscriptionClient) oldStartTsScanLowPriorityThreshold() time.Duration { + threshold := time.Duration(config.GetGlobalServerConfig().Debug.Puller.OldStartTsScanLowPriorityThreshold) + if threshold > 0 { + return threshold + } + return config.DefaultOldStartTsScanLowPriorityThreshold +} + +func (s *subscriptionClient) isTsCloseToCurrent(ts uint64) bool { + if ts == 0 { + return false + } + return s.pdClock.CurrentTime().Sub(oracle.GetTimeFromTS(ts)) <= s.oldStartTsScanLowPriorityThreshold() +} + +func (s *subscriptionClient) maybeEnableRealtimeScanPriority(span *subscribedSpan, resolvedTs uint64) { + if span == nil || !span.initialized.Load() || span.realtimeScanPriority.Load() { + return + } + if !s.isTsCloseToCurrent(resolvedTs) { + return + } + if span.realtimeScanPriority.CompareAndSwap(false, true) { + log.Info("subscription client enables realtime scan priority", + zap.Uint64("subscriptionID", uint64(span.subID)), + zap.Uint64("resolvedTs", resolvedTs), + zap.Duration("threshold", s.oldStartTsScanLowPriorityThreshold())) + } +} + +func (s *subscriptionClient) effectiveScanTaskPriority(subscribedSpan *subscribedSpan, priority TaskType) TaskType { + if subscribedSpan != nil && subscribedSpan.realtimeScanPriority.Load() { + return TaskHighPrior + } + return priority +} + // Unsubscribe the given table span. All covered regions will be deregistered asynchronously. // NOTE: `span.TableID` must be set correctly. func (s *subscriptionClient) Unsubscribe(subID SubscriptionID) { @@ -803,6 +855,8 @@ func (s *subscriptionClient) divideSpanAndScheduleRegionRequests( // scheduleRegionRequest locks the region's range and send the region to regionTaskQueue, // which will be handled by handleRegions. func (s *subscriptionClient) scheduleRegionRequest(ctx context.Context, region regionInfo, priority TaskType) { + priority = s.effectiveScanTaskPriority(region.subscribedSpan, priority) + region.scanPriority = priority.scanPriority() lockRangeResult := region.subscribedSpan.rangeLock.LockRange( ctx, region.span.StartKey, region.span.EndKey, region.verID.GetID(), region.verID.GetVer()) @@ -814,6 +868,18 @@ func (s *subscriptionClient) scheduleRegionRequest(ctx context.Context, region r case regionlock.LockRangeStatusSuccess: region.lockedRangeState = lockRangeResult.LockedRangeState s.regionTaskQueue.Push(NewRegionPriorityTask(priority, region, s.pdClock.CurrentTS())) + if log.GetLevel() <= zapcore.DebugLevel { + log.Debug("cdc region scan task enqueued", + zap.Uint64("subscriptionID", uint64(region.subscribedSpan.subID)), + zap.Int64("tableID", region.subscribedSpan.span.TableID), + zap.Uint64("startTs", region.subscribedSpan.startTs), + zap.Uint64("regionID", region.verID.GetID()), + zap.Uint64("regionEpochVersion", region.verID.GetVer()), + zap.Uint64("regionEpochConfVer", region.verID.GetConfVer()), + zap.String("priority", priority.String()), + zap.String("scanPriority", region.scanPriority.String()), + zap.String("span", common.FormatTableSpan(®ion.span))) + } case regionlock.LockRangeStatusStale: for _, r := range lockRangeResult.RetryRanges { s.scheduleRangeRequest(ctx, r, region.subscribedSpan, region.filterLoop, priority) @@ -829,6 +895,7 @@ func (s *subscriptionClient) scheduleRangeRequest( filterLoop bool, priority TaskType, ) { + priority = s.effectiveScanTaskPriority(subscribedSpan, priority) select { case <-ctx.Done(): case s.rangeTaskCh <- rangeTask{span: span, subscribedSpan: subscribedSpan, filterLoop: filterLoop, priority: priority}: @@ -851,6 +918,7 @@ func (s *subscriptionClient) handleErrors(ctx context.Context) error { func (s *subscriptionClient) doHandleError(ctx context.Context, errInfo regionErrorInfo) error { err := errors.Cause(errInfo.err) + retryPriority := taskTypeFromScanPriority(errInfo.scanPriority) if _, requestCancelled := err.(*requestCancelledErr); !requestCancelled { log.Debug("cdc region error", zap.Uint64("subscriptionID", uint64(errInfo.subscribedSpan.subID)), @@ -865,27 +933,27 @@ func (s *subscriptionClient) doHandleError(ctx context.Context, errInfo regionEr if notLeader := innerErr.GetNotLeader(); notLeader != nil { metricFeedNotLeaderCounter.Inc() s.regionCache.UpdateLeader(errInfo.verID, notLeader.GetLeader(), errInfo.rpcCtx.AccessIdx) - s.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskHighPrior) + s.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority) return nil } if innerErr.GetEpochNotMatch() != nil { metricFeedEpochNotMatchCounter.Inc() - s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior) + s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, retryPriority) return nil } if innerErr.GetRegionNotFound() != nil { metricFeedRegionNotFoundCounter.Inc() - s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior) + s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, retryPriority) return nil } if innerErr.GetCongested() != nil { metricKvCongestedCounter.Inc() - s.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior) + s.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority) return nil } if innerErr.GetServerIsBusy() != nil { metricKvIsBusyCounter.Inc() - s.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior) + s.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority) return nil } if duplicated := innerErr.GetDuplicateRequest(); duplicated != nil { @@ -904,24 +972,24 @@ func (s *subscriptionClient) doHandleError(ctx context.Context, errInfo regionEr zap.Uint64("subscriptionID", uint64(errInfo.subscribedSpan.subID)), zap.Stringer("error", innerErr)) metricFeedUnknownErrorCounter.Inc() - s.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskHighPrior) + s.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority) return nil case *rpcCtxUnavailableErr: metricFeedRPCCtxUnavailable.Inc() - s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior) + s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, retryPriority) return nil case *getStoreErr: metricGetStoreErr.Inc() bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff) // cannot get the store the region belongs to, so we need to reload the region. s.regionCache.OnSendFail(bo, errInfo.rpcCtx, true, err) - s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior) + s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, retryPriority) return nil case *storeStreamErr: metricStoreSendRequestErr.Inc() bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff) s.regionCache.OnSendFail(bo, errInfo.rpcCtx, regionScheduleReload, err) - s.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskHighPrior) + s.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority) return nil case *requestCancelledErr: // the corresponding subscription has been unsubscribed, just ignore. diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index b502c4d7d1..e093b5a637 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -22,10 +22,12 @@ import ( "github.com/pingcap/errors" "github.com/pingcap/kvproto/pkg/cdcpb" + "github.com/pingcap/kvproto/pkg/errorpb" "github.com/pingcap/ticdc/heartbeatpb" "github.com/pingcap/ticdc/logservice/logpuller/regionlock" "github.com/pingcap/ticdc/pkg/common" appcontext "github.com/pingcap/ticdc/pkg/common/context" + "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/pdutil" "github.com/pingcap/ticdc/pkg/security" @@ -171,7 +173,7 @@ func TestResolveLockTaskDeduplicatedAcrossSubscribedSpans(t *testing.T) { } func TestHandleResolveLockTasksMetrics(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(t.Context()) defer cancel() resolver := &mockLockResolver{} @@ -352,6 +354,138 @@ func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { require.NotContains(t, client.totalSpans.spanMap, span.subID) } +func TestBusyRetryPreservesScanPriority(t *testing.T) { + for _, tc := range []struct { + name string + priority cdcpb.ScanPriority + cdcErr *cdcpb.Error + expected TaskType + }{ + { + name: "server is busy high", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + cdcErr: &cdcpb.Error{ServerIsBusy: &errorpb.ServerIsBusy{}}, + expected: TaskHighPrior, + }, + { + name: "server is busy low", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + cdcErr: &cdcpb.Error{ServerIsBusy: &errorpb.ServerIsBusy{}}, + expected: TaskLowPrior, + }, + { + name: "congested high", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + cdcErr: &cdcpb.Error{Congested: &cdcpb.Congested{}}, + expected: TaskHighPrior, + }, + { + name: "congested low", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + cdcErr: &cdcpb.Error{Congested: &cdcpb.Congested{}}, + expected: TaskLowPrior, + }, + { + name: "unknown retry high", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + cdcErr: &cdcpb.Error{}, + expected: TaskHighPrior, + }, + { + name: "unknown retry low", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + cdcErr: &cdcpb.Error{}, + expected: TaskLowPrior, + }, + } { + t.Run(tc.name, func(t *testing.T) { + client := &subscriptionClient{ + regionTaskQueue: priorityqueue.New[PriorityTask](), + } + client.pdClock = pdutil.NewClock4Test() + _, span := newScanPriorityTestSpan() + region := newScanPriorityTestRegion(span) + region.scanPriority = tc.priority + + err := client.doHandleError(context.Background(), newRegionErrorInfo(region, &eventError{err: tc.cdcErr})) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + task, err := client.regionTaskQueue.Pop(ctx) + require.NoError(t, err) + require.Equal(t, tc.expected, task.(*regionPriorityTask).taskType) + require.Equal(t, tc.priority, task.GetRegionInfo().scanPriority) + }) + } +} + +func TestRangeRetryPreservesScanPriority(t *testing.T) { + for _, tc := range []struct { + name string + priority cdcpb.ScanPriority + err error + expected TaskType + }{ + { + name: "epoch not match high", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + err: &eventError{err: &cdcpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}}}, + expected: TaskHighPrior, + }, + { + name: "epoch not match low", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + err: &eventError{err: &cdcpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}}}, + expected: TaskLowPrior, + }, + { + name: "region not found high", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + err: &eventError{err: &cdcpb.Error{RegionNotFound: &errorpb.RegionNotFound{}}}, + expected: TaskHighPrior, + }, + { + name: "region not found low", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + err: &eventError{err: &cdcpb.Error{RegionNotFound: &errorpb.RegionNotFound{}}}, + expected: TaskLowPrior, + }, + { + name: "rpc context unavailable high", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, + err: &rpcCtxUnavailableErr{verID: tikv.NewRegionVerID(1, 1, 1)}, + expected: TaskHighPrior, + }, + { + name: "rpc context unavailable low", + priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW, + err: &rpcCtxUnavailableErr{verID: tikv.NewRegionVerID(1, 1, 1)}, + expected: TaskLowPrior, + }, + } { + t.Run(tc.name, func(t *testing.T) { + client := &subscriptionClient{ + rangeTaskCh: make(chan rangeTask, 1), + } + rawSpan, span := newScanPriorityTestSpan() + region := newScanPriorityTestRegion(span) + region.scanPriority = tc.priority + + err := client.doHandleError(context.Background(), newRegionErrorInfo(region, tc.err)) + require.NoError(t, err) + + select { + case task := <-client.rangeTaskCh: + require.Equal(t, tc.expected, task.priority) + require.Equal(t, rawSpan, task.span) + case <-time.After(time.Second): + require.Fail(t, "expected range retry task") + } + }) + } +} + type mockDynamicStream struct{} func (s *mockDynamicStream) Start() {} @@ -382,6 +516,194 @@ func (s *mockDynamicStream) GetMetrics() dynstream.Metrics[int, SubscriptionID] return dynstream.Metrics[int, SubscriptionID]{} } +func TestInitialScanTaskPriority(t *testing.T) { + restore := setInitialScanLowPriorityThresholdForTest(t, 30*time.Minute) + defer restore() + + currentTime := time.Date(2026, time.June, 27, 12, 0, 0, 0, time.UTC) + pdClock := pdutil.NewClock4Test() + pdClock.(*pdutil.Clock4Test).SetTS(oracle.GoTimeToTS(currentTime)) + client := &subscriptionClient{ + pdClock: pdClock, + } + + for _, tc := range []struct { + name string + startTs uint64 + expected TaskType + }{ + { + name: "zero start ts", + startTs: 0, + expected: TaskLowPrior, + }, + { + name: "recent start ts", + startTs: oracle.GoTimeToTS(currentTime.Add(-29 * time.Minute)), + expected: TaskHighPrior, + }, + { + name: "threshold boundary", + startTs: oracle.GoTimeToTS(currentTime.Add(-30 * time.Minute)), + expected: TaskHighPrior, + }, + { + name: "old start ts", + startTs: oracle.GoTimeToTS(currentTime.Add(-31 * time.Minute)), + expected: TaskLowPrior, + }, + { + name: "future start ts", + startTs: oracle.GoTimeToTS(currentTime.Add(time.Minute)), + expected: TaskHighPrior, + }, + } { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.expected, client.initialScanTaskPriority(tc.startTs)) + }) + } +} + +func TestSubscribeUsesInitialScanTaskPriority(t *testing.T) { + restore := setInitialScanLowPriorityThresholdForTest(t, 30*time.Minute) + defer restore() + + ctx := t.Context() + + currentTime := time.Date(2026, time.June, 27, 12, 0, 0, 0, time.UTC) + pdClock := pdutil.NewClock4Test() + pdClock.(*pdutil.Clock4Test).SetTS(oracle.GoTimeToTS(currentTime)) + client := &subscriptionClient{ + ctx: ctx, + ds: &mockDynamicStream{}, + rangeTaskCh: make(chan rangeTask, 2), + pdClock: pdClock, + } + client.totalSpans.spanMap = make(map[SubscriptionID]*subscribedSpan) + + span := heartbeatpb.TableSpan{TableID: 1, StartKey: []byte("a"), EndKey: []byte("z")} + consumeKVEvents := func(_ []common.RawKVEntry, _ func()) bool { return false } + advanceResolvedTs := func(uint64) {} + + client.Subscribe( + SubscriptionID(1), + span, + oracle.GoTimeToTS(currentTime.Add(-time.Minute)), + consumeKVEvents, + advanceResolvedTs, + 0, + false, + ) + client.Subscribe( + SubscriptionID(2), + span, + oracle.GoTimeToTS(currentTime.Add(-31*time.Minute)), + consumeKVEvents, + advanceResolvedTs, + 0, + false, + ) + + require.Equal(t, TaskHighPrior, (<-client.rangeTaskCh).priority) + require.Equal(t, TaskLowPrior, (<-client.rangeTaskCh).priority) +} + +func TestRealtimeScanPriorityEnabledAfterSubscriptionCatchesUp(t *testing.T) { + restore := setInitialScanLowPriorityThresholdForTest(t, 30*time.Minute) + defer restore() + + currentTime := time.Date(2026, time.June, 27, 12, 0, 0, 0, time.UTC) + pdClock := pdutil.NewClock4Test() + pdClock.(*pdutil.Clock4Test).SetTS(oracle.GoTimeToTS(currentTime)) + client := &subscriptionClient{ + pdClock: pdClock, + } + + _, span := newScanPriorityTestSpan() + + client.maybeEnableRealtimeScanPriority(span, oracle.GoTimeToTS(currentTime.Add(-time.Minute))) + require.False(t, span.realtimeScanPriority.Load()) + + span.initialized.Store(true) + client.maybeEnableRealtimeScanPriority(span, oracle.GoTimeToTS(currentTime.Add(-31*time.Minute))) + require.False(t, span.realtimeScanPriority.Load()) + + client.maybeEnableRealtimeScanPriority(span, oracle.GoTimeToTS(currentTime.Add(-time.Minute))) + require.True(t, span.realtimeScanPriority.Load()) + require.Equal(t, TaskHighPrior, client.effectiveScanTaskPriority(span, TaskLowPrior)) +} + +func TestRealtimeScanPriorityUpgradesRegionRetry(t *testing.T) { + client := &subscriptionClient{ + regionTaskQueue: priorityqueue.New[PriorityTask](), + } + client.pdClock = pdutil.NewClock4Test() + _, span := newScanPriorityTestSpan() + span.realtimeScanPriority.Store(true) + region := newScanPriorityTestRegion(span) + region.scanPriority = cdcpb.ScanPriority_SCAN_PRIORITY_LOW + + err := client.doHandleError(context.Background(), newRegionErrorInfo(region, &eventError{err: &cdcpb.Error{ServerIsBusy: &errorpb.ServerIsBusy{}}})) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + task, err := client.regionTaskQueue.Pop(ctx) + require.NoError(t, err) + require.Equal(t, TaskHighPrior, task.(*regionPriorityTask).taskType) + require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, task.GetRegionInfo().scanPriority) +} + +func TestRealtimeScanPriorityUpgradesRangeRetry(t *testing.T) { + client := &subscriptionClient{ + rangeTaskCh: make(chan rangeTask, 1), + } + rawSpan, span := newScanPriorityTestSpan() + span.realtimeScanPriority.Store(true) + region := newScanPriorityTestRegion(span) + region.scanPriority = cdcpb.ScanPriority_SCAN_PRIORITY_LOW + + err := client.doHandleError(context.Background(), newRegionErrorInfo(region, &eventError{err: &cdcpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}}})) + require.NoError(t, err) + + select { + case task := <-client.rangeTaskCh: + require.Equal(t, TaskHighPrior, task.priority) + require.Equal(t, rawSpan, task.span) + case <-time.After(time.Second): + require.Fail(t, "expected range retry task") + } +} + +func newScanPriorityTestSpan() (heartbeatpb.TableSpan, *subscribedSpan) { + rawSpan := heartbeatpb.TableSpan{ + TableID: 1, + StartKey: []byte("a"), + EndKey: []byte("z"), + } + span := &subscribedSpan{ + subID: SubscriptionID(1), + span: rawSpan, + rangeLock: regionlock.NewRangeLock(1, rawSpan.StartKey, rawSpan.EndKey, 100), + } + return rawSpan, span +} + +func newScanPriorityTestRegion(span *subscribedSpan) regionInfo { + return newRegionInfo(tikv.NewRegionVerID(1, 1, 1), span.span, nil, span, false) +} + +func setInitialScanLowPriorityThresholdForTest(t *testing.T, threshold time.Duration) func() { + t.Helper() + oldConfig := config.GetGlobalServerConfig() + testConfig := oldConfig.Clone() + testConfig.Debug.Puller.OldStartTsScanLowPriorityThreshold = config.TomlDuration(threshold) + config.StoreGlobalServerConfig(testConfig) + return func() { + config.StoreGlobalServerConfig(oldConfig) + } +} + func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { client := &subscriptionClient{ ds: &mockDynamicStream{}, diff --git a/logservice/schemastore/ddl_job_fetcher.go b/logservice/schemastore/ddl_job_fetcher.go index 16e079e290..0e943be723 100644 --- a/logservice/schemastore/ddl_job_fetcher.go +++ b/logservice/schemastore/ddl_job_fetcher.go @@ -96,7 +96,15 @@ func (p *ddlJobFetcher) run(startTs uint64) error { advanceSubSpanResolvedTs := func(ts uint64) { p.tryAdvanceResolvedTs(subID, ts) } - p.subClient.Subscribe(subID, span, startTs, p.input, advanceSubSpanResolvedTs, 0, ddlPullerFilterLoop) + p.subClient.Subscribe( + subID, + span, + startTs, + p.input, + advanceSubSpanResolvedTs, + 0, + ddlPullerFilterLoop, + ) } return nil } diff --git a/metrics/grafana/ticdc_new_arch.json b/metrics/grafana/ticdc_new_arch.json index 5b6a19db31..d90057fb1f 100644 --- a/metrics/grafana/ticdc_new_arch.json +++ b/metrics/grafana/ticdc_new_arch.json @@ -22644,7 +22644,7 @@ "dashes": false, "datasource": "${DS_TEST-CLUSTER}", "decimals": 1, - "description": "", + "description": "Queued CDC incremental scan jobs by priority per TiKV instance.", "fieldConfig": { "defaults": {}, "overrides": [] @@ -22658,6 +22658,335 @@ "y": 41 }, "hiddenSeries": false, + "id": 62047, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "hideEmpty": false, + "hideZero": true, + "max": true, + "min": false, + "rightSide": false, + "show": true, + "sideWidth": null, + "sort": "current", + "sortDesc": true, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [], + "nullPointMode": "null", + "options": { + "alertThreshold": true + }, + "paceLength": 10, + "percentage": false, + "pluginVersion": "7.5.17", + "pointradius": 5, + "points": false, + "renderer": "flot", + "seriesOverrides": [], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "expr": "sum(tikv_cdc_scan_scheduler_queue_length{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tikv_instance\"}) by (instance, priority)", + "format": "time_series", + "hide": false, + "intervalFactor": 1, + "legendFormat": "{{instance}}-{{priority}}", + "refId": "A" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "CDC Scan Scheduler Queue Length", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "none", + "label": null, + "logBase": 1, + "max": null, + "min": "0", + "show": true + }, + { + "format": "short", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": false + } + ], + "yaxis": { + "align": false, + "alignLevel": null + } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "decimals": 1, + "description": "Running CDC incremental scan jobs by slot kind per TiKV instance.", + "fieldConfig": { + "defaults": {}, + "overrides": [] + }, + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 12, + "x": 12, + "y": 41 + }, + "hiddenSeries": false, + "id": 62048, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "hideEmpty": false, + "hideZero": true, + "max": true, + "min": false, + "rightSide": false, + "show": true, + "sideWidth": null, + "sort": "current", + "sortDesc": true, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [], + "nullPointMode": "null", + "options": { + "alertThreshold": true + }, + "paceLength": 10, + "percentage": false, + "pluginVersion": "7.5.17", + "pointradius": 5, + "points": false, + "renderer": "flot", + "seriesOverrides": [], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "expr": "sum(tikv_cdc_scan_scheduler_running{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tikv_instance\"}) by (instance, slot)", + "format": "time_series", + "hide": false, + "intervalFactor": 1, + "legendFormat": "{{instance}}-{{slot}}", + "refId": "A" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "CDC Scan Scheduler Running", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "none", + "label": null, + "logBase": 1, + "max": null, + "min": "0", + "show": true + }, + { + "format": "short", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": false + } + ], + "yaxis": { + "align": false, + "alignLevel": null + } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "decimals": 1, + "description": "Time spent waiting in the CDC scan scheduler before dispatch.", + "fieldConfig": { + "defaults": {}, + "overrides": [] + }, + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 24, + "x": 0, + "y": 48 + }, + "hiddenSeries": false, + "id": 62049, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "hideEmpty": false, + "hideZero": true, + "max": true, + "min": false, + "rightSide": false, + "show": true, + "sideWidth": null, + "sort": "current", + "sortDesc": true, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [], + "nullPointMode": "null", + "options": { + "alertThreshold": true + }, + "paceLength": 10, + "percentage": false, + "pluginVersion": "7.5.17", + "pointradius": 5, + "points": false, + "renderer": "flot", + "seriesOverrides": [], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "exemplar": true, + "expr": "histogram_quantile(0.99, sum(rate(tikv_cdc_scan_scheduler_dispatch_latency_seconds_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tikv_instance\"}[1m])) by (le, instance, priority, slot))", + "format": "time_series", + "hide": false, + "interval": "", + "intervalFactor": 1, + "legendFormat": "{{instance}}-{{priority}}-{{slot}}-p99", + "refId": "A" + }, + { + "exemplar": true, + "expr": "sum(rate(tikv_cdc_scan_scheduler_dispatch_latency_seconds_sum{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tikv_instance\"}[1m])) by (instance, priority, slot) / sum(rate(tikv_cdc_scan_scheduler_dispatch_latency_seconds_count{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tikv_instance\"}[1m])) by (instance, priority, slot)", + "format": "time_series", + "hide": false, + "interval": "", + "legendFormat": "{{instance}}-{{priority}}-{{slot}}-avg", + "refId": "B" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "CDC Scan Scheduler Dispatch Latency", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "s", + "label": null, + "logBase": 1, + "max": null, + "min": "0", + "show": true + }, + { + "format": "short", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": false + } + ], + "yaxis": { + "align": false, + "alignLevel": null + } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "decimals": 1, + "description": "", + "fieldConfig": { + "defaults": {}, + "overrides": [] + }, + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 12, + "x": 0, + "y": 55 + }, + "hiddenSeries": false, "id": 72, "legend": { "alignAsTable": true, @@ -22776,7 +23105,7 @@ "h": 7, "w": 12, "x": 12, - "y": 41 + "y": 55 }, "heatmap": {}, "hideZeroBuckets": true, @@ -22853,7 +23182,7 @@ "h": 7, "w": 12, "x": 0, - "y": 48 + "y": 62 }, "hiddenSeries": false, "id": 22297, @@ -22971,7 +23300,7 @@ "h": 7, "w": 12, "x": 12, - "y": 48 + "y": 62 }, "hiddenSeries": false, "id": 139, @@ -23082,7 +23411,7 @@ "h": 7, "w": 12, "x": 0, - "y": 55 + "y": 69 }, "hiddenSeries": false, "id": 76, @@ -23193,7 +23522,7 @@ "h": 7, "w": 12, "x": 12, - "y": 55 + "y": 69 }, "hiddenSeries": false, "id": 22298, @@ -23306,7 +23635,7 @@ "h": 7, "w": 12, "x": 0, - "y": 62 + "y": 76 }, "hiddenSeries": false, "id": 152, @@ -23444,7 +23773,7 @@ "h": 7, "w": 12, "x": 12, - "y": 62 + "y": 76 }, "hiddenSeries": false, "id": 70, @@ -23549,7 +23878,7 @@ "h": 7, "w": 12, "x": 0, - "y": 69 + "y": 83 }, "hiddenSeries": false, "id": 143, @@ -23674,7 +24003,7 @@ "h": 7, "w": 12, "x": 12, - "y": 69 + "y": 83 }, "hiddenSeries": false, "id": 153, @@ -23788,7 +24117,7 @@ "h": 7, "w": 12, "x": 0, - "y": 76 + "y": 90 }, "heatmap": {}, "hideZeroBuckets": true, @@ -23864,7 +24193,7 @@ "h": 7, "w": 12, "x": 12, - "y": 76 + "y": 90 }, "hiddenSeries": false, "id": 145, @@ -24002,7 +24331,7 @@ "h": 7, "w": 12, "x": 0, - "y": 83 + "y": 97 }, "hiddenSeries": false, "id": 142, @@ -24119,7 +24448,7 @@ "h": 7, "w": 12, "x": 12, - "y": 83 + "y": 97 }, "hiddenSeries": false, "id": 141, diff --git a/metrics/nextgengrafana/ticdc_new_arch_next_gen.json b/metrics/nextgengrafana/ticdc_new_arch_next_gen.json index 77a3a6a48a..50027729e6 100644 --- a/metrics/nextgengrafana/ticdc_new_arch_next_gen.json +++ b/metrics/nextgengrafana/ticdc_new_arch_next_gen.json @@ -22644,7 +22644,7 @@ "dashes": false, "datasource": "${DS_TEST-CLUSTER}", "decimals": 1, - "description": "", + "description": "Queued CDC incremental scan jobs by priority per TiKV instance.", "fieldConfig": { "defaults": {}, "overrides": [] @@ -22658,6 +22658,335 @@ "y": 41 }, "hiddenSeries": false, + "id": 62047, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "hideEmpty": false, + "hideZero": true, + "max": true, + "min": false, + "rightSide": false, + "show": true, + "sideWidth": null, + "sort": "current", + "sortDesc": true, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [], + "nullPointMode": "null", + "options": { + "alertThreshold": true + }, + "paceLength": 10, + "percentage": false, + "pluginVersion": "7.5.17", + "pointradius": 5, + "points": false, + "renderer": "flot", + "seriesOverrides": [], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "expr": "sum(tikv_cdc_scan_scheduler_queue_length{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$tikv_instance\"}) by (instance, priority)", + "format": "time_series", + "hide": false, + "intervalFactor": 1, + "legendFormat": "{{instance}}-{{priority}}", + "refId": "A" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "CDC Scan Scheduler Queue Length", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "none", + "label": null, + "logBase": 1, + "max": null, + "min": "0", + "show": true + }, + { + "format": "short", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": false + } + ], + "yaxis": { + "align": false, + "alignLevel": null + } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "decimals": 1, + "description": "Running CDC incremental scan jobs by slot kind per TiKV instance.", + "fieldConfig": { + "defaults": {}, + "overrides": [] + }, + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 12, + "x": 12, + "y": 41 + }, + "hiddenSeries": false, + "id": 62048, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "hideEmpty": false, + "hideZero": true, + "max": true, + "min": false, + "rightSide": false, + "show": true, + "sideWidth": null, + "sort": "current", + "sortDesc": true, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [], + "nullPointMode": "null", + "options": { + "alertThreshold": true + }, + "paceLength": 10, + "percentage": false, + "pluginVersion": "7.5.17", + "pointradius": 5, + "points": false, + "renderer": "flot", + "seriesOverrides": [], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "expr": "sum(tikv_cdc_scan_scheduler_running{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$tikv_instance\"}) by (instance, slot)", + "format": "time_series", + "hide": false, + "intervalFactor": 1, + "legendFormat": "{{instance}}-{{slot}}", + "refId": "A" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "CDC Scan Scheduler Running", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "none", + "label": null, + "logBase": 1, + "max": null, + "min": "0", + "show": true + }, + { + "format": "short", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": false + } + ], + "yaxis": { + "align": false, + "alignLevel": null + } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "decimals": 1, + "description": "Time spent waiting in the CDC scan scheduler before dispatch.", + "fieldConfig": { + "defaults": {}, + "overrides": [] + }, + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 24, + "x": 0, + "y": 48 + }, + "hiddenSeries": false, + "id": 62049, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "hideEmpty": false, + "hideZero": true, + "max": true, + "min": false, + "rightSide": false, + "show": true, + "sideWidth": null, + "sort": "current", + "sortDesc": true, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [], + "nullPointMode": "null", + "options": { + "alertThreshold": true + }, + "paceLength": 10, + "percentage": false, + "pluginVersion": "7.5.17", + "pointradius": 5, + "points": false, + "renderer": "flot", + "seriesOverrides": [], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "exemplar": true, + "expr": "histogram_quantile(0.99, sum(rate(tikv_cdc_scan_scheduler_dispatch_latency_seconds_bucket{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$tikv_instance\"}[1m])) by (le, instance, priority, slot))", + "format": "time_series", + "hide": false, + "interval": "", + "intervalFactor": 1, + "legendFormat": "{{instance}}-{{priority}}-{{slot}}-p99", + "refId": "A" + }, + { + "exemplar": true, + "expr": "sum(rate(tikv_cdc_scan_scheduler_dispatch_latency_seconds_sum{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$tikv_instance\"}[1m])) by (instance, priority, slot) / sum(rate(tikv_cdc_scan_scheduler_dispatch_latency_seconds_count{k8s_cluster=\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$tikv_instance\"}[1m])) by (instance, priority, slot)", + "format": "time_series", + "hide": false, + "interval": "", + "legendFormat": "{{instance}}-{{priority}}-{{slot}}-avg", + "refId": "B" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "CDC Scan Scheduler Dispatch Latency", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "s", + "label": null, + "logBase": 1, + "max": null, + "min": "0", + "show": true + }, + { + "format": "short", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": false + } + ], + "yaxis": { + "align": false, + "alignLevel": null + } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "decimals": 1, + "description": "", + "fieldConfig": { + "defaults": {}, + "overrides": [] + }, + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 12, + "x": 0, + "y": 55 + }, + "hiddenSeries": false, "id": 72, "legend": { "alignAsTable": true, @@ -22776,7 +23105,7 @@ "h": 7, "w": 12, "x": 12, - "y": 41 + "y": 55 }, "heatmap": {}, "hideZeroBuckets": true, @@ -22853,7 +23182,7 @@ "h": 7, "w": 12, "x": 0, - "y": 48 + "y": 62 }, "hiddenSeries": false, "id": 22297, @@ -22971,7 +23300,7 @@ "h": 7, "w": 12, "x": 12, - "y": 48 + "y": 62 }, "hiddenSeries": false, "id": 139, @@ -23082,7 +23411,7 @@ "h": 7, "w": 12, "x": 0, - "y": 55 + "y": 69 }, "hiddenSeries": false, "id": 76, @@ -23193,7 +23522,7 @@ "h": 7, "w": 12, "x": 12, - "y": 55 + "y": 69 }, "hiddenSeries": false, "id": 22298, @@ -23306,7 +23635,7 @@ "h": 7, "w": 12, "x": 0, - "y": 62 + "y": 76 }, "hiddenSeries": false, "id": 152, @@ -23444,7 +23773,7 @@ "h": 7, "w": 12, "x": 12, - "y": 62 + "y": 76 }, "hiddenSeries": false, "id": 70, @@ -23549,7 +23878,7 @@ "h": 7, "w": 12, "x": 0, - "y": 69 + "y": 83 }, "hiddenSeries": false, "id": 143, @@ -23674,7 +24003,7 @@ "h": 7, "w": 12, "x": 12, - "y": 69 + "y": 83 }, "hiddenSeries": false, "id": 153, @@ -23788,7 +24117,7 @@ "h": 7, "w": 12, "x": 0, - "y": 76 + "y": 90 }, "heatmap": {}, "hideZeroBuckets": true, @@ -23864,7 +24193,7 @@ "h": 7, "w": 12, "x": 12, - "y": 76 + "y": 90 }, "hiddenSeries": false, "id": 145, @@ -24002,7 +24331,7 @@ "h": 7, "w": 12, "x": 0, - "y": 83 + "y": 97 }, "hiddenSeries": false, "id": 142, @@ -24119,7 +24448,7 @@ "h": 7, "w": 12, "x": 12, - "y": 83 + "y": 97 }, "hiddenSeries": false, "id": 141, diff --git a/pkg/config/debug.go b/pkg/config/debug.go index 1aa6f9b924..5dd8b56a22 100644 --- a/pkg/config/debug.go +++ b/pkg/config/debug.go @@ -19,6 +19,12 @@ import ( "github.com/pingcap/errors" ) +const ( + // DefaultOldStartTsScanLowPriorityThreshold is the default age threshold for + // classifying initial scan tasks as low priority. + DefaultOldStartTsScanLowPriorityThreshold = 30 * time.Minute +) + // DebugConfig represents config for ticdc unexposed feature configurations type DebugConfig struct { DB *DBConfig `toml:"db" json:"db"` @@ -49,6 +55,7 @@ func (c *DebugConfig) ValidateAndAdjust() error { if err := c.Scheduler.ValidateAndAdjust(); err != nil { return errors.Trace(err) } + c.Puller.ValidateAndAdjust() return nil } @@ -67,6 +74,9 @@ type PullerConfig struct { // For example, if PendingRegionRequestQueueSize is 32 and there are 8 workers connecting to the same store, // each worker's queue size will be 32 / 8 = 4. PendingRegionRequestQueueSize int `toml:"pending-region-request-queue-size" json:"pending_region_request_queue_size"` + // OldStartTsScanLowPriorityThreshold is the startTs age threshold for initial scans. + // Initial scans older than this threshold are scheduled as low priority. + OldStartTsScanLowPriorityThreshold TomlDuration `toml:"old-start-ts-scan-low-priority-threshold" json:"old_start_ts_scan_low_priority_threshold"` } // NewDefaultPullerConfig return the default puller configuration @@ -76,6 +86,15 @@ func NewDefaultPullerConfig() *PullerConfig { ResolvedTsStuckInterval: TomlDuration(5 * time.Minute), LogRegionDetails: false, PendingRegionRequestQueueSize: 32, // This value is chosen to reduce the impact of new changefeeds on existing ones. + OldStartTsScanLowPriorityThreshold: TomlDuration( + DefaultOldStartTsScanLowPriorityThreshold), + } +} + +// ValidateAndAdjust validates and adjusts puller configuration. +func (c *PullerConfig) ValidateAndAdjust() { + if c.OldStartTsScanLowPriorityThreshold <= 0 { + c.OldStartTsScanLowPriorityThreshold = TomlDuration(DefaultOldStartTsScanLowPriorityThreshold) } } diff --git a/pkg/eventservice/event_broker.go b/pkg/eventservice/event_broker.go index 8acc8d9232..48e6233808 100644 --- a/pkg/eventservice/event_broker.go +++ b/pkg/eventservice/event_broker.go @@ -1023,7 +1023,6 @@ func (c *eventBroker) addDispatcher(info DispatcherInfo) error { start := time.Now() success := c.eventStore.RegisterDispatcher( - changefeedID, id, span, info.GetStartTs(), diff --git a/pkg/eventservice/event_service_test.go b/pkg/eventservice/event_service_test.go index 166df9429e..d9090a7b38 100644 --- a/pkg/eventservice/event_service_test.go +++ b/pkg/eventservice/event_service_test.go @@ -287,7 +287,6 @@ func (m *mockEventStore) GetLogCoordinatorNodeID() node.ID { } func (m *mockEventStore) RegisterDispatcher( - changefeedID common.ChangeFeedID, dispatcherID common.DispatcherID, span *heartbeatpb.TableSpan, startTS common.Ts,