fix cr suggestion

Change-Id: I89c4ec04fc28efcca6fc911eca51bfbcbff924b3

go/logic/migrator.go

@@ -1280,18 +1280,8 @@ func (this *Migrator) executeWriteFuncs() error {
 				}
 			}
 		case copyRowsFunc := <-this.copyRowsQueue:
-		priority:
+			if err := this.drainApplierEventQueue(); err != nil {
-			for {
+				return this.migrationContext.Log.Errore(err)
-				select {
-				case eventStruct := <-this.applyEventsQueue:
-					{
-						if err := this.onApplyEventStruct(eventStruct); err != nil {
-							return err
-						}
-					}
-				default:
-					break priority
-				}
 			}
 			{
 				copyRowsStartTime := time.Now()
@@ -1302,7 +1292,7 @@ func (this *Migrator) executeWriteFuncs() error {
 				if niceRatio := this.migrationContext.GetNiceRatio(); niceRatio > 0 {
 					copyRowsDuration := time.Since(copyRowsStartTime)
 					sleepTimeNanosecondFloat64 := niceRatio * float64(copyRowsDuration.Nanoseconds())
-					sleepTime := time.Duration(time.Duration(int64(sleepTimeNanosecondFloat64)) * time.Nanosecond)
+					sleepTime := time.Duration(sleepTimeNanosecondFloat64) * time.Nanosecond
 					time.Sleep(sleepTime)
 				}
 			}
@@ -1310,6 +1300,22 @@ func (this *Migrator) executeWriteFuncs() error {
 	}
 }
 
+func (this *Migrator) drainApplierEventQueue() error {
+	for {
+		this.throttler.throttle(nil)
+		select {
+		case eventStruct := <-this.applyEventsQueue:
+			{
+				if err := this.onApplyEventStruct(eventStruct); err != nil {
+					return err
+				}
+			}
+		default:
+			return nil
+		}
+	}
+}
+
 // finalCleanup takes actions at very end of migration, dropping tables etc.
 func (this *Migrator) finalCleanup() error {
 	atomic.StoreInt64(&this.migrationContext.CleanupImminentFlag, 1)