Change Default Eventstore Version to 2 (#2758)

uber · Nov 1, 2019 · f58eae7 · f58eae7
1 parent a4561ec
commit f58eae7
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 26 deletions.
diff --git a/common/persistence/cassandra/cassandraPersistence.go b/common/persistence/cassandra/cassandraPersistence.go
@@ -2724,10 +2724,10 @@ func (d *cassandraPersistence) PutReplicationTaskToDLQ(request *p.PutReplication
 		task.Version,
 		task.LastReplicationInfo,
 		task.ScheduledID,
-		defaultEventStoreVersionValue,
+		p.EventStoreVersion,
 		task.BranchToken,
 		task.ResetWorkflow,
-		defaultEventStoreVersionValue,
+		p.EventStoreVersion,
 		task.NewRunBranchToken,
 		defaultVisibilityTimestamp,
 		task.GetTaskID())

diff --git a/common/persistence/cassandra/cassandraPersistenceUtil.go b/common/persistence/cassandra/cassandraPersistenceUtil.go
@@ -31,10 +31,6 @@ import (
 	p "github.com/uber/cadence/common/persistence"
 )
 
-// eventStoreVersion is already deprecated, this is just a constant for place holder.
-// TODO we can remove it after fixing all the query templates
-const defaultEventStoreVersionValue = -1
-
 func applyWorkflowMutationBatch(
 	batch *gocql.Batch,
 	shardID int,
@@ -456,7 +452,7 @@ func createExecution(
 			executionInfo.ExpirationTime,
 			executionInfo.MaximumAttempts,
 			executionInfo.NonRetriableErrors,
-			defaultEventStoreVersionValue,
+			p.EventStoreVersion,
 			executionInfo.BranchToken,
 			executionInfo.CronSchedule,
 			executionInfo.ExpirationSeconds,
@@ -526,7 +522,7 @@ func createExecution(
 			executionInfo.ExpirationTime,
 			executionInfo.MaximumAttempts,
 			executionInfo.NonRetriableErrors,
-			defaultEventStoreVersionValue,
+			p.EventStoreVersion,
 			executionInfo.BranchToken,
 			executionInfo.CronSchedule,
 			executionInfo.ExpirationSeconds,
@@ -601,7 +597,7 @@ func createExecution(
 			executionInfo.ExpirationTime,
 			executionInfo.MaximumAttempts,
 			executionInfo.NonRetriableErrors,
-			defaultEventStoreVersionValue,
+			p.EventStoreVersion,
 			executionInfo.BranchToken,
 			executionInfo.CronSchedule,
 			executionInfo.ExpirationSeconds,
@@ -714,7 +710,7 @@ func updateExecution(
 			executionInfo.ExpirationTime,
 			executionInfo.MaximumAttempts,
 			executionInfo.NonRetriableErrors,
-			defaultEventStoreVersionValue,
+			p.EventStoreVersion,
 			executionInfo.BranchToken,
 			executionInfo.CronSchedule,
 			executionInfo.ExpirationSeconds,
@@ -785,7 +781,7 @@ func updateExecution(
 			executionInfo.ExpirationTime,
 			executionInfo.MaximumAttempts,
 			executionInfo.NonRetriableErrors,
-			defaultEventStoreVersionValue,
+			p.EventStoreVersion,
 			executionInfo.BranchToken,
 			executionInfo.CronSchedule,
 			executionInfo.ExpirationSeconds,
@@ -861,7 +857,7 @@ func updateExecution(
 			executionInfo.ExpirationTime,
 			executionInfo.MaximumAttempts,
 			executionInfo.NonRetriableErrors,
-			defaultEventStoreVersionValue,
+			p.EventStoreVersion,
 			executionInfo.BranchToken,
 			executionInfo.CronSchedule,
 			executionInfo.ExpirationSeconds,
@@ -1088,10 +1084,10 @@ func createReplicationTasks(
 			version,
 			lastReplicationInfo,
 			activityScheduleID,
-			defaultEventStoreVersionValue,
+			p.EventStoreVersion,
 			branchToken,
 			resetWorkflow,
-			defaultEventStoreVersionValue,
+			p.EventStoreVersion,
 			newRunBranchToken,
 			defaultVisibilityTimestamp,
 			task.GetTaskID())

diff --git a/common/persistence/dataInterfaces.go b/common/persistence/dataInterfaces.go
@@ -40,6 +40,14 @@ const (
 	DomainStatusDeleted
 )
 
+const (
+	// EventStoreVersion is already deprecated, this is used for forward
+	// compatibility (so that rollback is possible).
+	// TODO we can remove it after fixing all the query templates and when
+	// we decide the compatibility is no longer needed.
+	EventStoreVersion = 2
+)
+
 // CreateWorkflowMode workflow creation mode
 type CreateWorkflowMode int
 

diff --git a/common/persistence/sql/sqlExecutionManagerUtil.go b/common/persistence/sql/sqlExecutionManagerUtil.go
@@ -889,18 +889,20 @@ func createReplicationTasks(
 		}
 
 		blob, err := replicationTaskInfoToBlob(&sqlblobs.ReplicationTaskInfo{
-			DomainID:            domainID,
-			WorkflowID:          &workflowID,
-			RunID:               runID,
-			TaskType:            common.Int16Ptr(int16(task.GetType())),
-			FirstEventID:        &firstEventID,
-			NextEventID:         &nextEventID,
-			Version:             &version,
-			LastReplicationInfo: lastReplicationInfo,
-			ScheduledID:         &activityScheduleID,
-			BranchToken:         branchToken,
-			NewRunBranchToken:   newRunBranchToken,
-			ResetWorkflow:       &resetWorkflow,
+			DomainID:                domainID,
+			WorkflowID:              &workflowID,
+			RunID:                   runID,
+			TaskType:                common.Int16Ptr(int16(task.GetType())),
+			FirstEventID:            &firstEventID,
+			NextEventID:             &nextEventID,
+			Version:                 &version,
+			LastReplicationInfo:     lastReplicationInfo,
+			ScheduledID:             &activityScheduleID,
+			EventStoreVersion:       common.Int32Ptr(p.EventStoreVersion),
+			NewRunEventStoreVersion: common.Int32Ptr(p.EventStoreVersion),
+			BranchToken:             branchToken,
+			NewRunBranchToken:       newRunBranchToken,
+			ResetWorkflow:           &resetWorkflow,
 		})
 		if err != nil {
 			return err
@@ -1242,6 +1244,7 @@ func buildExecutionRow(
 		RetryExpirationSeconds:                  &executionInfo.ExpirationSeconds,
 		RetryExpirationTimeNanos:                common.Int64Ptr(executionInfo.ExpirationTime.UnixNano()),
 		RetryNonRetryableErrors:                 executionInfo.NonRetriableErrors,
+		EventStoreVersion:                       common.Int32Ptr(p.EventStoreVersion),
 		EventBranchToken:                        executionInfo.BranchToken,
 		AutoResetPoints:                         executionInfo.AutoResetPoints.Data,
 		AutoResetPointsEncoding:                 common.StringPtr(string(executionInfo.AutoResetPoints.GetEncoding())),