Skip to content

maintainer: make dispatcher operator admission atomic #6070

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wk989898 wants to merge 3 commits into
base: master
Choose a base branch
from
Open

maintainer: make dispatcher operator admission atomic #6070

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ab3745b
init
wk989898 Aug 24, 2026
b093196
update
wk989898 Aug 24, 2026
e8413d0
debug
wk989898 Aug 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 20 additions & 9 deletions maintainer/barrier.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,16 +88,27 @@ func (b *Barrier) HandleStatus(from node.ID,
for _, status := range request.BlockStatuses {
// only receive block status from the replicating dispatcher
dispatcherID := common.NewDispatcherIDFromPB(status.ID)
task := b.spanController.GetTaskByID(dispatcherID)
if task == nil {
log.Info("Get block status from unexisted dispatcher, ignore it",
zap.String("changefeed", request.ChangefeedID.GetName()),
zap.String("dispatcher", dispatcherID.String()),
zap.Uint64("commitTs", status.State.BlockTs),
zap.Int64("mode", b.mode))
continue
}
ownerNodeID := task.GetNodeID()
if ownerNodeID != from {
log.Warn("ignore block status from non-owner dispatcher",
zap.String("changefeed", request.ChangefeedID.GetName()),
zap.String("dispatcherID", dispatcherID.String()),
zap.String("ownerNodeID", ownerNodeID.String()),
zap.String("fromNodeID", from.String()),
zap.Uint64("commitTs", status.State.BlockTs),
zap.Int64("mode", b.mode))
continue
}
if dispatcherID != b.spanController.GetDDLDispatcherID() {
task := b.spanController.GetTaskByID(dispatcherID)
if task == nil {
log.Info("Get block status from unexisted dispatcher, ignore it",
zap.String("changefeed", request.ChangefeedID.GetName()),
zap.String("dispatcher", dispatcherID.String()),
zap.Uint64("commitTs", status.State.BlockTs),
zap.Int64("mode", b.mode))
continue
}
if !b.spanController.IsReplicating(task) {
log.Info("Get block status from unreplicating dispatcher, ignore it",
zap.String("changefeed", request.ChangefeedID.GetName()),
Expand Down
53 changes: 51 additions & 2 deletions maintainer/barrier_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,49 @@ func TestOneBlockEvent(t *testing.T) {
require.Len(t, resp.DispatcherStatuses, 0)
}

func TestBarrierIgnoresBlockStatusFromNonOwner(t *testing.T) {
testutil.SetUpTestServices(t)
tableTriggerEventDispatcherID := common.NewDispatcherID()
cfID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
ddlSpan := replica.NewWorkingSpanReplication(cfID, tableTriggerEventDispatcherID,
common.DDLSpanSchemaID,
common.KeyspaceDDLSpan(common.DefaultKeyspaceID), &heartbeatpb.TableSpanStatus{
ID: tableTriggerEventDispatcherID.ToPB(),
ComponentStatus: heartbeatpb.ComponentState_Working,
CheckpointTs: 1,
}, "node1", false)
spanController := span.NewController(cfID, ddlSpan, nil, nil, nil, common.DefaultKeyspaceID, common.DefaultMode)
operatorController := operator.NewOperatorController(cfID, spanController, 1000, common.DefaultMode)
spanController.AddNewTable(commonEvent.Table{SchemaID: 1, TableID: 1}, 10)
stm := spanController.GetTasksByTableID(1)[0]
spanController.BindSpanToNode("", "node1", stm)
spanController.MarkSpanReplicating(stm)

barrier := NewBarrier(spanController, operatorController, false, nil, common.DefaultMode, nil)
msgs := barrier.HandleStatus("node2", &heartbeatpb.BlockStatusRequest{
ChangefeedID: cfID.ToPB(),
BlockStatuses: []*heartbeatpb.TableSpanBlockStatus{
{
ID: stm.ID.ToPB(),
State: &heartbeatpb.State{
IsBlocked: true,
BlockTs: 10,
BlockTables: &heartbeatpb.InfluencedTables{
InfluenceType: heartbeatpb.InfluenceType_Normal,
TableIDs: []int64{1},
},
},
},
},
})

require.Len(t, msgs, 1)
resp := msgs[0].Message[0].(*heartbeatpb.HeartBeatResponse)
require.Empty(t, resp.DispatcherStatuses)
require.Empty(t, barrier.blockedEvents.m)
require.Equal(t, uint64(10), stm.GetStatus().CheckpointTs)
}

func TestNormalBlock(t *testing.T) {
testutil.SetUpTestServices(t)
tableTriggerEventDispatcherID := common.NewDispatcherID()
Expand Down Expand Up @@ -1101,7 +1144,7 @@ func TestSchemaBlock(t *testing.T) {
require.Len(t, resp.DispatcherStatuses, 1)

// selected node write done
_ = barrier.HandleStatus("node2", &heartbeatpb.BlockStatusRequest{
_ = barrier.HandleStatus("node1", &heartbeatpb.BlockStatusRequest{
ChangefeedID: cfID.ToPB(),
BlockStatuses: []*heartbeatpb.TableSpanBlockStatus{
{
Expand Down Expand Up @@ -1253,7 +1296,7 @@ func TestSyncPointBlock(t *testing.T) {
require.Equal(t, event.writerDispatcher, spanController.GetDDLDispatcherID())

// selected node write done
_ = barrier.HandleStatus("node2", &heartbeatpb.BlockStatusRequest{
_ = barrier.HandleStatus("node1", &heartbeatpb.BlockStatusRequest{
ChangefeedID: cfID.ToPB(),
BlockStatuses: []*heartbeatpb.TableSpanBlockStatus{
{
Expand Down Expand Up @@ -1293,6 +1336,11 @@ func TestSyncPointBlock(t *testing.T) {
IsSyncPoint: true,
},
},
},
})
_ = barrier.HandleStatus("node2", &heartbeatpb.BlockStatusRequest{
ChangefeedID: cfID.ToPB(),
BlockStatuses: []*heartbeatpb.TableSpanBlockStatus{
{
ID: dispatcherIDs[2],
State: &heartbeatpb.State{
Expand Down Expand Up @@ -1328,6 +1376,7 @@ func TestNonBlocked(t *testing.T) {
stm := spanController.GetTasksByTableID(int64(id))[0]
dispatcherID := stm.ID
blockedDispatcherIDS = append(blockedDispatcherIDS, dispatcherID.ToPB())
spanController.BindSpanToNode("", "node1", stm)
spanController.MarkSpanReplicating(stm)
}
msgs := barrier.HandleStatus("node1", &heartbeatpb.BlockStatusRequest{
Expand Down
42 changes: 20 additions & 22 deletions maintainer/operator/operator_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,8 @@ type Controller struct {
nodeManager *watcher.NodeManager
maintainerEpoch atomic.Uint64

// admissionMu serializes removing-mode quiesce with normal operator side effects.
// admissionMu serializes removing-mode quiesce and remove-operator replacement
// with normal operator side effects.
// A normal operator must hold the read side from its final allow check through
// Start or Schedule/SendCommand so it cannot cross the handoff boundary after
// QuiesceExcept has made the controller quiescing.
Expand Down Expand Up @@ -259,7 +260,7 @@ func (oc *Controller) AddOperator(op operator.Operator[common.DispatcherID, *hea
zap.String("operator", op.String()))
return false
}
return oc.pushOperatorWithAdmission(op)
return oc.pushOperatorWithAdmission(op, false)
}

func (oc *Controller) UpdateOperatorStatus(id common.DispatcherID, from node.ID, status *heartbeatpb.TableSpanStatus) {
Expand Down Expand Up @@ -465,8 +466,8 @@ func (oc *Controller) cancelOperator(opID common.DispatcherID) {
}

func (oc *Controller) removeReplicaSet(op *removeDispatcherOperator) {
oc.admissionMu.RLock()
defer oc.admissionMu.RUnlock()
oc.admissionMu.Lock()
defer oc.admissionMu.Unlock()

if !oc.isOperatorAllowed(op.ID()) {
log.Info("skip remove operator while controller is quiescing",
Expand All @@ -488,35 +489,32 @@ func (oc *Controller) removeReplicaSet(op *removeDispatcherOperator) {
old.OP.OnTaskRemoved()
oc.finalizeOperator(old, op.ID())
}
oc.pushOperatorWithAdmission(op)
oc.pushOperatorWithAdmission(op, true)
}

// pushOperator add an operator to the controller queue.
func (oc *Controller) pushOperator(op operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus]) bool {
oc.admissionMu.RLock()
defer oc.admissionMu.RUnlock()
func (oc *Controller) pushOperatorWithAdmission(
op operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus],
replaceExisting bool,
) bool {
withTime := operator.NewOperatorWithTime(op, time.Now())
opID := op.ID()

if !oc.isOperatorAllowed(op.ID()) {
log.Info("skip operator while controller is quiescing",
oc.mu.Lock()
if old, ok := oc.operators[opID]; ok && !replaceExisting {
oc.mu.Unlock()
log.Info("add operator failed, operator already exists",
zap.String("role", oc.role),
zap.Stringer("changefeedID", oc.changefeedID),
zap.String("dispatcherID", op.ID().String()),
zap.String("operator", op.String()))
zap.String("operator", op.String()),
zap.String("oldOperator", old.OP.String()))
return false
}
return oc.pushOperatorWithAdmission(op)
}

func (oc *Controller) pushOperatorWithAdmission(op operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus]) bool {
oc.operators[opID] = withTime
oc.mu.Unlock()
Comment thread
coderabbitai[bot] marked this conversation as resolved.
log.Info("add operator to running queue",
zap.String("role", oc.role),
zap.Stringer("changefeedID", oc.changefeedID),
zap.String("operator", op.String()))
withTime := operator.NewOperatorWithTime(op, time.Now())

oc.mu.Lock()
oc.operators[op.ID()] = withTime
oc.mu.Unlock()

op.Start()
// Check affected nodes after Start to avoid operators being forced into terminal states
Expand Down
Loading
Loading