Fix: Import data not considering the initiated cutover during resumption

yb-voyager/cmd/cutover.go

@@ -171,3 +171,4 @@ func ExitIfAlreadyCutover(importerOrExporterRole string) {
 		panic(fmt.Sprintf("invalid role %s", importerOrExporterRole))
 	}
 }
+

yb-voyager/cmd/eventQueue.go

@@ -27,8 +27,8 @@ import (
 	"time"
 
 	"github.com/goccy/go-json"
-
 	log "github.com/sirupsen/logrus"
+
 	"github.com/yugabyte/yb-voyager/yb-voyager/src/metadb"
 	"github.com/yugabyte/yb-voyager/yb-voyager/src/tgtdb"
 	"github.com/yugabyte/yb-voyager/yb-voyager/src/utils"

yb-voyager/cmd/exportDataDebezium.go

@@ -350,6 +350,7 @@ func isOracleJDBCWalletLocationSet(s srcdb.Source) (bool, error) {
 // ---------------------------------------------- Export Data ---------------------------------------//
 
 func debeziumExportData(ctx context.Context, config *dbzm.Config, tableNameToApproxRowCountMap map[string]int64) error {
+
 	if config.SnapshotMode != "never" {
 		err := metaDB.UpdateMigrationStatusRecord(func(record *metadb.MigrationStatusRecord) {
 			record.SnapshotMechanism = "debezium"

yb-voyager/cmd/live_migration.go

@@ -55,11 +55,36 @@ func init() {
 	MAX_INTERVAL_BETWEEN_BATCHES = utils.GetEnvAsInt("MAX_INTERVAL_BETWEEN_BATCHES", 2000)
 }
 
+func cutoverInitiatedAndCutoverEventProcessed() (bool, error) {
+	msr, err := metaDB.GetMigrationStatusRecord()
+	if err != nil {
+		return false, fmt.Errorf("getting migration status record: %v", err)
+	}
+	switch importerRole {
+	case TARGET_DB_IMPORTER_ROLE:
+		return msr.CutoverToTargetRequested && msr.CutoverDetectedByTargetImporter, nil
+	case SOURCE_REPLICA_DB_IMPORTER_ROLE:
+		return msr.CutoverToSourceReplicaRequested && msr.CutoverDetectedBySourceReplicaImporter, nil
+	case SOURCE_DB_IMPORTER_ROLE:
+		return msr.CutoverToSourceRequested && msr.CutoverDetectedBySourceImporter, nil
+	}
+
+	return false, nil
+}
+
 func streamChanges(state *ImportDataState, tableNames []sqlname.NameTuple) error {
+	ok, err := cutoverInitiatedAndCutoverEventProcessed()
+	if err != nil {
+		return err
+	}
+	if ok {
+		log.Info("cutover is initiated and the event is detected..")
+		return nil
+	}
 	log.Infof("NUM_EVENT_CHANNELS: %d, EVENT_CHANNEL_SIZE: %d, MAX_EVENTS_PER_BATCH: %d, MAX_INTERVAL_BETWEEN_BATCHES: %d",
 		NUM_EVENT_CHANNELS, EVENT_CHANNEL_SIZE, MAX_EVENTS_PER_BATCH, MAX_INTERVAL_BETWEEN_BATCHES)
 	// re-initilizing name registry in case it hadn't picked up the names registered on source/target/source-replica
-	err := namereg.NameReg.Init()
+	err = namereg.NameReg.Init()
 	if err != nil {
 		return fmt.Errorf("init name registry again: %v", err)
 	}
@@ -178,6 +203,19 @@ func streamChangesFromSegment(
 			event.IsCutoverToSourceReplica() && importerRole == SOURCE_REPLICA_DB_IMPORTER_ROLE ||
 			event.IsCutoverToSource() && importerRole == SOURCE_DB_IMPORTER_ROLE { // cutover or fall-forward command
 
+			err := metaDB.UpdateMigrationStatusRecord(func(record *metadb.MigrationStatusRecord) {
+				switch importerRole {
+				case TARGET_DB_IMPORTER_ROLE:
+					record.CutoverDetectedByTargetImporter = true
+				case SOURCE_REPLICA_DB_IMPORTER_ROLE:
+					record.CutoverDetectedBySourceReplicaImporter = true
+				case SOURCE_DB_IMPORTER_ROLE:
+					record.CutoverDetectedBySourceImporter = true
+				}
+			})
+			if err != nil {
+				return fmt.Errorf("error updating the migration status record for cutover detected case: %v", err)
+			}
 			updateCallhomeImportPhase(event)
 
 			eventQueue.EndOfQueue = true

yb-voyager/src/metadb/migrationStatus.go

@@ -24,18 +24,27 @@ type MigrationStatusRecord struct {
 	TableListExportedFromSource []string          `json:"TableListExportedFromSource"`
 	SourceDBConf                *srcdb.Source     `json:"SourceDBConf"`
 
-	CutoverToTargetRequested                        bool `json:"CutoverToTargetRequested"`
+	//All the cutover requested flags by initiate cutover command
+	CutoverToTargetRequested        bool `json:"CutoverToTargetRequested"`
+	CutoverToSourceRequested        bool `json:"CutoverToSourceRequested"`
+	CutoverToSourceReplicaRequested bool `json:"CutoverToSourceReplicaRequested"`
+
+	//All the cutover detected by importer flags (marked when the cutover event is recieved by the importer)
+	CutoverDetectedByTargetImporter        bool `json:"CutoverDetectedByTargetImporter"`
+	CutoverDetectedBySourceImporter        bool `json:"CutoverDetectedBySourceImporter"`
+	CutoverDetectedBySourceReplicaImporter bool `json:"CutoverDetectedBySourceReplicaImporter"`
+
+	//All the cutover processed by importer/exporter flags - indicating that the cutover is completed by that command.
 	CutoverProcessedBySourceExporter                bool `json:"CutoverProcessedBySourceExporter"`
-	CutoverProcessedByTargetImporter                bool `json:"CutoverProcessedByTargetImporter"`
-	ExportFromTargetFallForwardStarted              bool `json:"ExportFromTargetFallForwardStarted"`
-	CutoverToSourceReplicaRequested                 bool `json:"CutoverToSourceReplicaRequested"`
+	CutoverToSourceProcessedByTargetExporter        bool `json:"CutoverToSourceProcessedByTargetExporter"`
 	CutoverToSourceReplicaProcessedByTargetExporter bool `json:"CutoverToSourceReplicaProcessedByTargetExporter"`
+	CutoverProcessedByTargetImporter                bool `json:"CutoverProcessedByTargetImporter"`
 	CutoverToSourceReplicaProcessedBySRImporter     bool `json:"CutoverToSourceReplicaProcessedBySRImporter"`
-	ExportFromTargetFallBackStarted                 bool `json:"ExportFromTargetFallBackStarted"`
-	CutoverToSourceRequested                        bool `json:"CutoverToSourceRequested"`
-	CutoverToSourceProcessedByTargetExporter        bool `json:"CutoverToSourceProcessedByTargetExporter"`
 	CutoverToSourceProcessedBySourceImporter        bool `json:"CutoverToSourceProcessedBySourceImporter"`
 
+	ExportFromTargetFallForwardStarted bool `json:"ExportFromTargetFallForwardStarted"`
+	ExportFromTargetFallBackStarted    bool `json:"ExportFromTargetFallBackStarted"`
+
 	ExportSchemaDone                bool `json:"ExportSchemaDone"`
 	ExportDataDone                  bool `json:"ExportDataDone"` // to be interpreted as export of snapshot data from source is complete
 	ExportDataSourceDebeziumStarted bool `json:"ExportDataSourceDebeziumStarted"`