drainer: make sure to sync binlog in ascending commitTS order (#123)

* drainer: make sure to sync binlog in ascending commitTS order
pingcap · Feb 27, 2017 · c79d23f · c79d23f
1 parent 8cce9ed
commit c79d23f
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 1 deletion.
diff --git a/drainer/executor.go b/drainer/executor.go
@@ -478,10 +478,15 @@ func (d *Executor) run(b *binlogItem) error {
 		go d.sync(d.toDBs[i], d.jobCh[i])
 	}
 
+	maxCommitTS := d.initCommitTS
 	for {
 		binlog := b.binlog
 		commitTS := binlog.GetCommitTs()
 		jobID := binlog.GetDdlJobId()
+		if commitTS <= maxCommitTS {
+			continue
+		}
+		maxCommitTS = commitTS
 
 		if jobID == 0 {
 			preWriteValue := binlog.GetPrewriteValue()

diff --git a/drainer/pump.go b/drainer/pump.go
@@ -182,7 +182,7 @@ func (p *Pump) mustFindCommitBinlog(t *tikv.LockResolver, startTS int64) {
 		default:
 		}
 
-		 b, ok := p.getPrewriteBinlogEntity(startTS)
+		b, ok := p.getPrewriteBinlogEntity(startTS)
 		if ok {
 			time.Sleep(waitTime)
 			// check again after sleep a moment