Skip to content
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

ddl: refine logic of OwnerCheckAllVersions #39070

Merged
merged 3 commits into from
Nov 11, 2022
Merged

ddl: refine logic of OwnerCheckAllVersions #39070

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
184e7f5
done
wjhuang2016 Nov 10, 2022
7407a13
fix
wjhuang2016 Nov 10, 2022
70e2df8
Merge branch 'master' into refine_checkOwner
ti-chi-bot Nov 11, 2022
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
25 changes: 21 additions & 4 deletions ddl/syncer/syncer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,7 +262,7 @@ func (s *schemaVersionSyncer) OwnerCheckAllVersions(ctx context.Context, jobID i
// If MDL is enabled, updatedMap is used to check if all the servers report the least version.
// updatedMap is initialed to record all the server in every loop. We delete a server from the map if it gets the metadata lock(the key version equal the given version.
// updatedMap should be empty if all the servers get the metadata lock.
updatedMap := make(map[string]struct{})
updatedMap := make(map[string]string)
for {
if util.IsContextDone(ctx) {
// ctx is canceled or timeout.
Expand All @@ -278,9 +278,23 @@ func (s *schemaVersionSyncer) OwnerCheckAllVersions(ctx context.Context, jobID i
if err != nil {
return err
}
updatedMap = make(map[string]struct{})
updatedMap = make(map[string]string)
instance2id := make(map[string]string)

// Set updatedMap according to the serverInfos, and remove some invalid serverInfos.
for _, info := range serverInfos {
updatedMap[info.ID] = struct{}{}
instance := fmt.Sprintf("%s:%d", info.IP, info.Port)
if id, ok := instance2id[instance]; ok {
if info.StartTimestamp > serverInfos[id].StartTimestamp {
// Replace it.
delete(updatedMap, id)
updatedMap[info.ID] = fmt.Sprintf("instance ip %s, port %d, id %s", info.IP, info.Port, info.ID)
instance2id[instance] = info.ID
}
} else {
updatedMap[info.ID] = fmt.Sprintf("instance ip %s, port %d, id %s", info.IP, info.Port, info.ID)
instance2id[instance] = info.ID
}
}
}

Expand Down Expand Up @@ -315,6 +329,9 @@ func (s *schemaVersionSyncer) OwnerCheckAllVersions(ctx context.Context, jobID i
}
if len(updatedMap) > 0 {
succ = false
for _, info := range updatedMap {
logutil.BgLogger().Info("[ddl] syncer check all versions, someone is not synced", zap.String("info", info), zap.Any("ddl id", jobID), zap.Any("ver", latestVer))
}
}
} else {
for _, kv := range resp.Kvs {
Expand All @@ -337,7 +354,7 @@ func (s *schemaVersionSyncer) OwnerCheckAllVersions(ctx context.Context, jobID i
notMatchVerCnt++
break
}
updatedMap[string(kv.Key)] = struct{}{}
updatedMap[string(kv.Key)] = ""
}
}

Expand Down