Method waitForEventsUpToLock

go/logic/migrator.go:912–950

Inject the "AllEventsUpToLockProcessed" state hint, wait for it to appear in the binary logs, make sure the queue is drained.

()

// Inject the "AllEventsUpToLockProcessed" state hint, wait for it to appear in the binary logs,
// make sure the queue is drained.
func (mgtr *Migrator) waitForEventsUpToLock() error {
	timeout := time.NewTimer(time.Second * time.Duration(mgtr.migrationContext.CutOverLockTimeoutSeconds))

	mgtr.migrationContext.MarkPointOfInterest()
	waitForEventsUpToLockStartTime := time.Now()

	allEventsUpToLockProcessedChallenge := fmt.Sprintf("%s:%d", string(AllEventsUpToLockProcessed), waitForEventsUpToLockStartTime.UnixNano())
	mgtr.migrationContext.Log.Infof("Writing changelog state: %+v", allEventsUpToLockProcessedChallenge)
	if _, err := mgtr.applier.WriteChangelogState(allEventsUpToLockProcessedChallenge); err != nil {
		return err
	}
	mgtr.migrationContext.Log.Infof("Waiting for events up to lock")
	atomic.StoreInt64(&mgtr.migrationContext.AllEventsUpToLockProcessedInjectedFlag, 1)
	var lockProcessed *lockProcessedStruct
	for found := false; !found; {
		select {
		case <-timeout.C:
			{
				return mgtr.migrationContext.Log.Errorf("timeout while waiting for events up to lock")
			}
		case lockProcessed = <-mgtr.allEventsUpToLockProcessed:
			{
				if lockProcessed.state == allEventsUpToLockProcessedChallenge {
					mgtr.migrationContext.Log.Infof("Waiting for events up to lock: got %s", lockProcessed.state)
					found = true
					mgtr.lastLockProcessed = lockProcessed
				} else {
					mgtr.migrationContext.Log.Infof("Waiting for events up to lock: skipping %s", lockProcessed.state)
				}
			}
		}
	}
	waitForEventsUpToLockDuration := time.Since(waitForEventsUpToLockStartTime)

	mgtr.migrationContext.Log.Infof("Done waiting for events up to lock; duration=%+v", waitForEventsUpToLockDuration)
	mgtr.printStatus(ForcePrintStatusAndHintRule)

	return nil
}

// cutOverTwoStep will lock down the original table, execute
// what's left of last DML entries, and **non-atomically** swap original->old, then new->original.