Skip to content

maintainer: add mode information in barrier logs (#4412) #5236

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

Closed
ti-chi-bot wants to merge 1 commit into from
Closed

maintainer: add mode information in barrier logs (#4412) #5236

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
55 changes: 46 additions & 9 deletions maintainer/barrier.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,15 @@
spanController *span.Controller
operatorController *operator.Controller
splitTableEnabled bool
<<<<<<< HEAD

Check failure on line 45 in maintainer/barrier.go

View workflow job for this annotation

GitHub Actions / Mac OS Build 

syntax error: unexpected <<, expected field name or embedded type
mode int64
=======

Check failure on line 47 in maintainer/barrier.go

View workflow job for this annotation

GitHub Actions / Mac OS Build 

syntax error: unexpected ==, expected field name or embedded type
flushEnabled bool
// mode identifies which replication pipeline this barrier belongs to
// (common.DefaultMode or common.RedoMode). Barrier state, resend messages,
// and logs must stay in the same mode.
mode int64
>>>>>>> c1f39ed2e (maintainer: add mode information in barrier logs (#4412))

Check failure on line 53 in maintainer/barrier.go

View workflow job for this annotation

GitHub Actions / Mac OS Build 

invalid character U+0023 '#'

Check failure on line 53 in maintainer/barrier.go

View workflow job for this annotation

GitHub Actions / Mac OS Build 

syntax error: unexpected >>, expected field name or embedded type
Comment on lines +45 to +53

@gemini-code-assist gemini-code-assist Bot Jun 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

There is an unresolved git conflict marker here. Since the flush logic is not supported on this branch and the corresponding fields are missing from BarrierEvent, we should resolve this conflict by keeping only the mode field and discarding flushEnabled.

	// mode identifies which replication pipeline this barrier belongs to
	// (common.DefaultMode or common.RedoMode). Barrier state, resend messages,
	// and logs must stay in the same mode.
	mode int64

}

// NewBarrier create a new barrier for the changefeed
Expand Down Expand Up @@ -80,11 +88,11 @@
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))
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
} else {
if !b.spanController.IsReplicating(task) {
log.Info("Get block status from unreplicating dispatcher, ignore it", zap.String("changefeed", request.ChangefeedID.GetName()), zap.String("dispatcher", dispatcherID.String()), zap.Uint64("commitTs", status.State.BlockTs))
log.Info("Get block status from unreplicating 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
}
}
Expand All @@ -100,7 +108,8 @@
zap.String("from", from.String()),
zap.String("changefeed", request.ChangefeedID.GetName()),
zap.String("detail", status.String()),
zap.Uint64("commitTs", status.State.BlockTs))
zap.Uint64("commitTs", status.State.BlockTs),
zap.Int64("mode", b.mode))
continue
}
if needACK {
Expand All @@ -123,7 +132,8 @@
if len(dispatcherStatus) <= 0 {
log.Warn("no dispatcher status to send",
zap.String("from", from.String()),
zap.String("changefeed", request.ChangefeedID.String()))
zap.String("changefeed", request.ChangefeedID.String()),
zap.Int64("mode", b.mode))
}

// send ack or write action message to dispatcher
Expand Down Expand Up @@ -167,10 +177,16 @@
key := getEventKey(blockState.BlockTs, blockState.IsSyncPoint)
event, ok := b.blockedEvents.Get(key)
if !ok {
<<<<<<< HEAD

Check failure on line 180 in maintainer/barrier.go

View workflow job for this annotation

GitHub Actions / Mac OS Build 

syntax error: unexpected <<, expected }
event = NewBlockEvent(common.NewChangefeedIDFromPB(resp.ChangefeedID), common.NewDispatcherIDFromPB(span.ID), b.spanController, b.operatorController, blockState, b.splitTableEnabled)
=======

Check failure on line 182 in maintainer/barrier.go

View workflow job for this annotation

GitHub Actions / Mac OS Build 

syntax error: unexpected ==, expected }
event = NewBlockEvent(common.NewChangefeedIDFromPB(resp.ChangefeedID), common.NewDispatcherIDFromPB(span.ID), b.spanController, b.operatorController, blockState, b.splitTableEnabled, b.mode)
event.flushEnabled = b.flushEnabled
event.flushDispatcherAdvanced = !b.flushEnabled
>>>>>>> c1f39ed2e (maintainer: add mode information in barrier logs (#4412))

Check failure on line 186 in maintainer/barrier.go

View workflow job for this annotation

GitHub Actions / Mac OS Build 

invalid character U+0023 '#'

Check failure on line 186 in maintainer/barrier.go

View workflow job for this annotation

GitHub Actions / Mac OS Build 

syntax error: unexpected >>, expected }
Comment on lines +180 to +186

@gemini-code-assist gemini-code-assist Bot Jun 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

Unresolved git conflict marker. Since flushEnabled and flushDispatcherAdvanced are not defined on BarrierEvent in this branch, we should resolve this conflict by only passing b.mode to NewBlockEvent and discarding the flush-related fields.

				event = NewBlockEvent(common.NewChangefeedIDFromPB(resp.ChangefeedID), common.NewDispatcherIDFromPB(span.ID), b.spanController, b.operatorController, blockState, b.splitTableEnabled, b.mode)

b.blockedEvents.Set(key, event)
}
switch blockState.Stage {

Check failure on line 189 in maintainer/barrier.go

View workflow job for this annotation

GitHub Actions / Mac OS Build 

syntax error: non-declaration statement outside function body
case heartbeatpb.BlockStage_WAITING:
// it's the dispatcher's responsibility to resend the block event
case heartbeatpb.BlockStage_WRITING:
Expand All @@ -197,7 +213,7 @@
// so it will not block by the ddl, and can continue to handle the following events.
// While for the table1 in NodeB, it's still wait the pass action.
// So we need to check the block event when the maintainer is restarted to help block event decide its state.
b.blockedEvents.Range(func(key eventKey, barrierEvent *BarrierEvent) bool {

Check failure on line 216 in maintainer/barrier.go

View workflow job for this annotation

GitHub Actions / Mac OS Build 

method has multiple receivers
if barrierEvent.allDispatcherReported() {
// it means the dispatchers involved in the block event are all in the cached resp, not restarted.
// so we don't do speical check for this event
Expand Down Expand Up @@ -339,7 +355,8 @@
log.Info("the block event is sent by ddl dispatcher",
zap.String("changefeed", changefeedID.Name()),
zap.String("dispatcher", dispatcherID.String()),
zap.Uint64("commitTs", blockState.BlockTs))
zap.Uint64("commitTs", blockState.BlockTs),
zap.Int64("mode", b.mode))
event.tableTriggerDispatcherRelated = true
}
if event.selected.Load() {
Expand All @@ -348,6 +365,7 @@
zap.String("changefeed", changefeedID.Name()),
zap.String("dispatcher", dispatcherID.String()),
zap.Uint64("commitTs", blockState.BlockTs),
zap.Int64("mode", b.mode),
)
// check whether the event can be finished.
b.checkEventFinish(event)
Expand All @@ -374,7 +392,8 @@
zap.String("dispatcher", dispatcherID.String()),
zap.Uint64("commitTs", blockState.BlockTs),
zap.Bool("isSyncPoint", blockState.IsSyncPoint),
zap.Int("pendingScheduleEvents", pending))
zap.Int("pendingScheduleEvents", pending),
zap.Int64("mode", b.mode))
return event, nil, "", false
}
}
Expand Down Expand Up @@ -421,7 +440,13 @@
) *BarrierEvent {
event, ok := b.blockedEvents.Get(key)
if !ok {
<<<<<<< HEAD
event = NewBlockEvent(changefeedID, dispatcherID, b.spanController, b.operatorController, blockState, b.splitTableEnabled)
=======
event = NewBlockEvent(changefeedID, dispatcherID, b.spanController, b.operatorController, blockState, b.splitTableEnabled, b.mode)
event.flushEnabled = b.flushEnabled
event.flushDispatcherAdvanced = !b.flushEnabled
>>>>>>> c1f39ed2e (maintainer: add mode information in barrier logs (#4412))
Comment on lines +443 to +449

@gemini-code-assist gemini-code-assist Bot Jun 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

Unresolved git conflict marker. Resolve this by only passing b.mode to NewBlockEvent and discarding the unsupported flush fields.

		event = NewBlockEvent(changefeedID, dispatcherID, b.spanController, b.operatorController, blockState, b.splitTableEnabled, b.mode)

b.blockedEvents.Set(key, event)
}
return event
Expand All @@ -433,13 +458,22 @@
if !be.allDispatcherReported() {
return
}
<<<<<<< HEAD
if be.selected.Load() {
log.Info("all dispatchers reported event done, remove event",
zap.String("changefeed", be.cfID.Name()),
zap.Uint64("committs", be.commitTs))
// already selected a dispatcher to write, now all dispatchers reported the block event
b.blockedEvents.Delete(getEventKey(be.commitTs, be.isSyncPoint))
}
=======

log.Info("all dispatchers reported event done, remove event",
zap.String("changefeed", be.cfID.Name()),
zap.Uint64("committs", be.commitTs),
zap.Int64("mode", b.mode))
b.blockedEvents.Delete(getEventKey(be.commitTs, be.isSyncPoint))
>>>>>>> c1f39ed2e (maintainer: add mode information in barrier logs (#4412))
Comment on lines +461 to +476

@gemini-code-assist gemini-code-assist Bot Jun 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

Unresolved git conflict marker. Resolve this by keeping the original be.selected.Load() check while adding the mode field to the log message.

	if be.selected.Load() {
		log.Info("all dispatchers reported event done, remove event",
			zap.String("changefeed", be.cfID.Name()),
			zap.Uint64("committs", be.commitTs),
			zap.Int64("mode", b.mode))
		// already selected a dispatcher to write, now all dispatchers reported the block event
		b.blockedEvents.Delete(getEventKey(be.commitTs, be.isSyncPoint))
	}

}

func (b *Barrier) tryScheduleEvent(event *BarrierEvent) bool {
Expand All @@ -449,7 +483,8 @@
log.Info("event trySchedule",
zap.String("changefeed", event.cfID.Name()),
zap.String("writerDispatcher", event.writerDispatcher.String()),
zap.Uint64("EventCommitTs", event.commitTs))
zap.Uint64("EventCommitTs", event.commitTs),
zap.Int64("mode", b.mode))
// pending queue ensures ddl with the same eventKey only schedules once and in order
ready, candidate := b.pendingEvents.popIfHead(event)
if !ready {
Expand All @@ -458,15 +493,17 @@
zap.String("changefeed", event.cfID.Name()),
zap.String("writerDispatcher", event.writerDispatcher.String()),
zap.Uint64("EventCommitTs", event.commitTs),
zap.Bool("isSyncPoint", event.isSyncPoint))
zap.Bool("isSyncPoint", event.isSyncPoint),
zap.Int64("mode", b.mode))
} else {
log.Info("event waits for a smaller commitTs before scheduling",
zap.String("changefeed", event.cfID.Name()),
zap.String("writerDispatcher", event.writerDispatcher.String()),
zap.Uint64("EventCommitTs", event.commitTs),
zap.Bool("isSyncPoint", event.isSyncPoint),
zap.Uint64("blockingEventCommitTs", candidate.commitTs),
zap.Bool("blockingEventIsSyncPoint", candidate.isSyncPoint))
zap.Bool("blockingEventIsSyncPoint", candidate.isSyncPoint),
zap.Int64("mode", b.mode))
}
return false
}
Expand Down
Loading
Loading