v2.0.11: add checkpoint update max try times, #323

ChangeLog

@@ -1,3 +1,7 @@
+2020-04-08 Alibaba Cloud.
+	* version: 2.0.11
+	* BUGFIX: add checkpoint flush max try times to 10. see #323.
+
 2020-03-27 Alibaba Cloud.
 	* version: 2.0.10
 	* BUGFIX: data may lost when using batch update meets duplicate error.

src/mongoshake/collector/syncer.go

@@ -29,6 +29,7 @@ const (
 	DDLCheckpointInterval         = 300  // unit: ms.
 	FilterCheckpointGap           = 180  // unit: seconds. no checkpoint update, flush checkpoint mandatory
 	FilterCheckpointCheckInterval = 180  // unit: seconds.
+	CheckCheckpointUpdateTimes    = 10   // at most times of time check
 )
 
 type OplogHandler interface {
@@ -263,8 +264,9 @@ func (sync *OplogSyncer) checkCheckpointUpdate(barrier bool, newestTs bson.Mongo
 	// if barrier == true, we should check whether the checkpoint is updated to `newestTs`.
 	if barrier && newestTs > 0 && conf.Options.WorkerNum > 1 {
 		LOG.Info("find barrier")
-		for {
-			checkpointTs := sync.ckptManager.Get().Timestamp
+		var checkpointTs bson.MongoTimestamp
+		for i := 0; i < CheckCheckpointUpdateTimes; i++ {
+			checkpointTs = sync.ckptManager.Get().Timestamp
 			LOG.Info("compare remote checkpoint[%v(%v)] to local newestTs[%v(%v)]",
 				checkpointTs, utils.ExtractMongoTimestamp(checkpointTs), newestTs, utils.ExtractMongoTimestamp(newestTs))
 			if checkpointTs >= newestTs {
@@ -276,6 +278,15 @@ func (sync *OplogSyncer) checkCheckpointUpdate(barrier bool, newestTs bson.Mongo
 			// re-flush
 			sync.checkpoint(true, 0)
 		}
+
+		/*
+		 * if code hits here, it means the checkpoint has not been updated(usually DDL).
+		 * it's ok because the checkpoint can still forward on the next time.
+		 * However, if MongoShake crashes here and restarts, there maybe a conflict when the
+		 * oplog is DDL that has been applied but checkpoint not updated.
+		 */
+		LOG.Warn("check checkpoint[%v] update[%v] failed, but do worry",
+			utils.ExtractTimestampForLog(checkpointTs), utils.ExtractTimestampForLog(newestTs))
 	}
 }
 

src/mongoshake/common/http.go

@@ -9,5 +9,5 @@ var (
 )
 
 func InitHttpApi(port int) {
-	HttpApi = nimo.NewHttpRestProvdier(port)
+	HttpApi = nimo.NewHttpRestProvider(port)
 }

src/vendor/vendor.json

@@ -45,10 +45,10 @@
 			"revisionTime": "2019-09-04T06:34:54Z"
 		},
 		{
-			"checksumSHA1": "52vjznwP2OHubLvZkLCUCC9Gp5o=",
+			"checksumSHA1": "PA8OcnbR5oMK1TI+eA8CIuO7gLY=",
 			"path": "github.com/gugemichael/nimo4go",
-			"revision": "32795d80f83ab8c37b47589c5068dfbb14545c34",
-			"revisionTime": "2019-09-04T07:30:57Z"
+			"revision": "647883f3a053414d7dc4f3771c6c69b5c4a8ea11",
+			"revisionTime": "2020-04-03T10:17:49Z"
 		},
 		{
 			"checksumSHA1": "5AxXPtBqAKyFGcttFzxT5hp/3Tk=",