Skip to content

Commit

Permalink
update some filed name (#368)
Browse files Browse the repository at this point in the history
  • Loading branch information
@sdojjy
sdojjy authored Oct 14, 2024
1 parent 0fe6b9d commit 5d8bf8a
Show file tree
Hide file tree
Showing 11 changed files with 200 additions and 183 deletions.
16 changes: 8 additions & 8 deletions maintainer/barrier_event_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ func TestScheduleEvent(t *testing.T) {
})
event.scheduleBlockEvent()
//drop table will be executed first
require.Equal(t, 2, controller.db.GetAbsentSize())
require.Equal(t, 2, controller.replicationDB.GetAbsentSize())

event = NewBlockEvent("test", controller, &heartbeatpb.State{
IsBlocked: true,
Expand All @@ -55,7 +55,7 @@ func TestScheduleEvent(t *testing.T) {
})
event.scheduleBlockEvent()
//drop table will be executed first, then add the new table
require.Equal(t, 1, controller.db.GetAbsentSize())
require.Equal(t, 1, controller.replicationDB.GetAbsentSize())

event = NewBlockEvent("test", controller, &heartbeatpb.State{
IsBlocked: true,
Expand All @@ -68,7 +68,7 @@ func TestScheduleEvent(t *testing.T) {
})
event.scheduleBlockEvent()
//drop table will be executed first, then add the new table
require.Equal(t, 1, controller.db.GetAbsentSize())
require.Equal(t, 1, controller.replicationDB.GetAbsentSize())
}

func TestResendAction(t *testing.T) {
Expand All @@ -79,10 +79,10 @@ func TestResendAction(t *testing.T) {
controller.AddNewTable(commonEvent.Table{SchemaID: 1, TableID: 1}, 1)
controller.AddNewTable(commonEvent.Table{SchemaID: 1, TableID: 2}, 1)
var dispatcherIDs []common.DispatcherID
absents, _ := controller.db.GetScheduleSate(make([]*replica.SpanReplication, 0), 100)
absents, _ := controller.replicationDB.GetScheduleSate(make([]*replica.SpanReplication, 0), 100)
for _, stm := range absents {
controller.db.BindSpanToNode("", "node1", stm)
controller.db.MarkSpanReplicating(stm)
controller.replicationDB.BindSpanToNode("", "node1", stm)
controller.replicationDB.MarkSpanReplicating(stm)
dispatcherIDs = append(dispatcherIDs, stm.ID)
}
event := NewBlockEvent("test", controller, &heartbeatpb.State{
Expand Down Expand Up @@ -172,7 +172,7 @@ func TestUpdateSchemaID(t *testing.T) {
setNodeManagerAndMessageCenter()
controller := NewController("test", 1, nil, nil, nil, nil, 1000, 0)
controller.AddNewTable(commonEvent.Table{SchemaID: 1, TableID: 1}, 1)
require.Equal(t, 1, controller.db.GetAbsentSize())
require.Equal(t, 1, controller.replicationDB.GetAbsentSize())
require.Len(t, controller.GetTasksBySchemaID(1), 1)
event := NewBlockEvent("test", controller, &heartbeatpb.State{
IsBlocked: true,
Expand All @@ -189,7 +189,7 @@ func TestUpdateSchemaID(t *testing.T) {
}},
)
event.scheduleBlockEvent()
require.Equal(t, 1, controller.db.GetAbsentSize())
require.Equal(t, 1, controller.replicationDB.GetAbsentSize())
// check the schema id and map is updated
require.Len(t, controller.GetTasksBySchemaID(1), 0)
require.Len(t, controller.GetTasksBySchemaID(2), 1)
Expand Down
46 changes: 23 additions & 23 deletions maintainer/barrier_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,8 @@ func TestOneBlockEvent(t *testing.T) {
controller := NewController("test", 1, nil, nil, nil, nil, 1000, 0)
controller.AddNewTable(commonEvent.Table{SchemaID: 1, TableID: 1}, 0)
stm := controller.GetTasksByTableIDs(1)[0]
controller.db.BindSpanToNode("", "node1", stm)
controller.db.MarkSpanReplicating(stm)
controller.replicationDB.BindSpanToNode("", "node1", stm)
controller.replicationDB.MarkSpanReplicating(stm)
barrier := NewBarrier(controller)
msg := barrier.HandleStatus("node1", &heartbeatpb.BlockStatusRequest{
ChangefeedID: "test",
Expand Down Expand Up @@ -101,14 +101,14 @@ func TestNormalBlock(t *testing.T) {
controller.AddNewTable(commonEvent.Table{SchemaID: 1, TableID: int64(id)}, 0)
stm := controller.GetTasksByTableIDs(int64(id))[0]
blockedDispatcherIDS = append(blockedDispatcherIDS, stm.ID.ToPB())
controller.db.BindSpanToNode("", "node1", stm)
controller.db.MarkSpanReplicating(stm)
controller.replicationDB.BindSpanToNode("", "node1", stm)
controller.replicationDB.MarkSpanReplicating(stm)
}

// the last one is the writer
var selectDispatcherID = common.NewDispatcherIDFromPB(blockedDispatcherIDS[2])
selectedRep := controller.GetTask(selectDispatcherID)
controller.db.BindSpanToNode("node1", "node2", selectedRep)
controller.replicationDB.BindSpanToNode("node1", "node2", selectedRep)
dropID := selectedRep.Span.TableID

newSpan := &heartbeatpb.Table{TableID: 10, SchemaID: 1}
Expand Down Expand Up @@ -284,13 +284,13 @@ func TestSchemaBlock(t *testing.T) {
controller.AddNewTable(commonEvent.Table{SchemaID: 2, TableID: 3}, 1)
var dispatcherIDs []*heartbeatpb.DispatcherID
var dropTables = []int64{1, 2}
absents, _ := controller.db.GetScheduleSate(make([]*replica.SpanReplication, 0), 100)
absents, _ := controller.replicationDB.GetScheduleSate(make([]*replica.SpanReplication, 0), 100)
for _, stm := range absents {
if stm.SchemaID == 1 {
if stm.GetSchemaID() == 1 {
dispatcherIDs = append(dispatcherIDs, stm.ID.ToPB())
}
controller.db.BindSpanToNode("", "node1", stm)
controller.db.MarkSpanReplicating(stm)
controller.replicationDB.BindSpanToNode("", "node1", stm)
controller.replicationDB.MarkSpanReplicating(stm)
}

newTable := &heartbeatpb.Table{TableID: 10, SchemaID: 2}
Expand Down Expand Up @@ -412,17 +412,17 @@ func TestSchemaBlock(t *testing.T) {
require.Len(t, barrier.blockedTs, 1)
// the writer already advanced
require.Len(t, event.reportedDispatchers, 1)
require.Equal(t, 1, controller.db.GetAbsentSize())
require.Equal(t, 2, controller.oc.OperatorSize())
require.Equal(t, 1, controller.replicationDB.GetAbsentSize())
require.Equal(t, 2, controller.operatorController.OperatorSize())
// two dispatcher and moved to operator queue, operator will be removed after ack
require.Equal(t, 3, controller.db.GetReplicatingSize())
for _, task := range controller.db.GetReplicating() {
op := controller.oc.GetOperator(task.ID)
require.Equal(t, 3, controller.replicationDB.GetReplicatingSize())
for _, task := range controller.replicationDB.GetReplicating() {
op := controller.operatorController.GetOperator(task.ID)
if op != nil {
op.PostFinish()
}
}
require.Equal(t, 1, controller.db.GetReplicatingSize())
require.Equal(t, 1, controller.replicationDB.GetReplicatingSize())

// other dispatcher advanced checkpoint ts
msg = barrier.HandleStatus("node1", &heartbeatpb.BlockStatusRequest{
Expand Down Expand Up @@ -455,15 +455,15 @@ func TestSyncPointBlock(t *testing.T) {
controller.AddNewTable(commonEvent.Table{SchemaID: 2, TableID: 3}, 1)
var dispatcherIDs []*heartbeatpb.DispatcherID
var dropTables = []int64{1, 2, 3}
absents, _ := controller.db.GetScheduleSate(make([]*replica.SpanReplication, 0), 10000)
absents, _ := controller.replicationDB.GetScheduleSate(make([]*replica.SpanReplication, 0), 10000)
for _, stm := range absents {
dispatcherIDs = append(dispatcherIDs, stm.ID.ToPB())
controller.db.BindSpanToNode("", "node1", stm)
controller.db.MarkSpanReplicating(stm)
controller.replicationDB.BindSpanToNode("", "node1", stm)
controller.replicationDB.MarkSpanReplicating(stm)
}
var selectDispatcherID = common.NewDispatcherIDFromPB(dispatcherIDs[2])
selectedRep := controller.GetTask(selectDispatcherID)
controller.db.BindSpanToNode("node1", "node2", selectedRep)
controller.replicationDB.BindSpanToNode("node1", "node2", selectedRep)

newSpan := &heartbeatpb.Table{TableID: 10, SchemaID: 2}
barrier := NewBarrier(controller)
Expand Down Expand Up @@ -631,7 +631,7 @@ func TestNonBlocked(t *testing.T) {
require.True(t, heartbeatpb.InfluenceType_Normal == resp.DispatcherStatuses[0].InfluencedDispatchers.InfluenceType)
require.Equal(t, resp.DispatcherStatuses[0].InfluencedDispatchers.DispatcherIDs[0], blockedDispatcherIDS[0])
require.Len(t, barrier.blockedTs, 0)
require.Equal(t, 2, barrier.controller.db.GetAbsentSize(), 2)
require.Equal(t, 2, barrier.controller.replicationDB.GetAbsentSize(), 2)
}

func TestSyncPointBlockPerf(t *testing.T) {
Expand All @@ -642,10 +642,10 @@ func TestSyncPointBlockPerf(t *testing.T) {
controller.AddNewTable(commonEvent.Table{SchemaID: 1, TableID: int64(id)}, 1)
}
var dispatcherIDs []*heartbeatpb.DispatcherID
absent, _ := controller.db.GetScheduleSate(make([]*replica.SpanReplication, 0), 10000)
absent, _ := controller.replicationDB.GetScheduleSate(make([]*replica.SpanReplication, 0), 10000)
for _, stm := range absent {
controller.db.BindSpanToNode("", "node1", stm)
controller.db.MarkSpanReplicating(stm)
controller.replicationDB.BindSpanToNode("", "node1", stm)
controller.replicationDB.MarkSpanReplicating(stm)
dispatcherIDs = append(dispatcherIDs, stm.ID.ToPB())
}
var blockStatus []*heartbeatpb.TableSpanBlockStatus
Expand Down
11 changes: 5 additions & 6 deletions maintainer/maintainer.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -304,7 +304,7 @@ func (m *Maintainer) onInit() bool {
func (m *Maintainer) onMessage(msg *messaging.TargetMessage) error {
switch msg.Type {
case messaging.TypeHeartBeatRequest:
return m.onHeartBeatRequest(msg)
m.onHeartBeatRequest(msg)
case messaging.TypeBlockStatusRequest:
m.onBlockStateRequest(msg)
case messaging.TypeMaintainerBootstrapResponse:
Expand Down Expand Up @@ -436,7 +436,7 @@ func (m *Maintainer) sendMessages(msgs []*messaging.TargetMessage) {
}
}

func (m *Maintainer) onHeartBeatRequest(msg *messaging.TargetMessage) error {
func (m *Maintainer) onHeartBeatRequest(msg *messaging.TargetMessage) {
req := msg.Message[0].(*heartbeatpb.HeartBeatRequest)
if req.Watermark != nil {
m.checkpointTsByCapture[msg.From] = *req.Watermark
Expand All @@ -452,7 +452,6 @@ func (m *Maintainer) onHeartBeatRequest(msg *messaging.TargetMessage) error {
m.runningErrors[msg.From] = req.Err
m.errLock.Unlock()
}
return nil
}

func (m *Maintainer) onBlockStateRequest(msg *messaging.TargetMessage) {
Expand Down Expand Up @@ -639,9 +638,9 @@ func (m *Maintainer) onPeriodTask() {
func (m *Maintainer) collectMetrics() {
if time.Since(m.lastPrintStatusTime) > time.Second*20 {
total := m.controller.TaskSize()
scheduling := m.controller.db.GetAbsentSize()
working := m.controller.db.GetReplicatingSize()
absent := m.controller.db.GetAbsentSize()
scheduling := m.controller.replicationDB.GetAbsentSize()
working := m.controller.replicationDB.GetReplicatingSize()
absent := m.controller.replicationDB.GetAbsentSize()

m.tableCountGauge.Set(float64(total))
m.scheduledTaskGauge.Set(float64(scheduling))
Expand Down
Loading

0 comments on commit 5d8bf8a

Please sign in to comment.