Repository: uber/cadence
Pull request #5015 (2022-10-31 02:05:00)

Feature/min initial failover version
## Problem 

The Initial Failover version is a somewhat overloaded monotonic counter, which uses the first couple of bits to identify the origin cluster for where an action was performed. 

However, when joining separate clusters, its necessary for these to be unique, migration between clusters isn't otherwise possible. 

Unfortunately, if you're joining two clusters with fairly high traffic together, it's necessary to move one of them off to a unique initial failover version. 

## Solution 
This change allows a dynamic configuration guarded change to issue a second initialFailoverVersion from a cluster, such that traffic can gradually be moved off the previous value. 

## Testing Plan 

Testing will be done by:
[✓] Verifying locally
[ ] Migrating in test environments
[ ] Migrating against non-production traffic.



coveralls: ## Pull Request Test Coverage Report for [Build 01842be0-2cad-4762-9743-81f2d35cc28e](https://coveralls.io/builds/53733904)

* **46** of **74**   **(62.16%)**  changed or added relevant lines in **8** files are covered.
* **181** unchanged lines in **23** files lost coverage.
* Overall coverage decreased (**-0.1%**) to **57.262%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [common/cluster/metadataTest.go](https://coveralls.io/builds/53733904/source?filename=common%2Fcluster%2FmetadataTest.go#L97) | 1 | 3 | 33.33%
| [cmd/server/cadence/server.go](https://coveralls.io/builds/53733904/source?filename=cmd%2Fserver%2Fcadence%2Fserver.go#L197) | 0 | 3 | 0.0%
| [common/log/tag/tags.go](https://coveralls.io/builds/53733904/source?filename=common%2Flog%2Ftag%2Ftags.go#L868) | 0 | 3 | 0.0%
| [common/log/tag/interface.go](https://coveralls.io/builds/53733904/source?filename=common%2Flog%2Ftag%2Finterface.go#L102) | 0 | 5 | 0.0%
| [tools/cli/domainUtils.go](https://coveralls.io/builds/53733904/source?filename=tools%2Fcli%2FdomainUtils.go#L244) | 0 | 5 | 0.0%
| [common/cluster/metadata.go](https://coveralls.io/builds/53733904/source?filename=common%2Fcluster%2Fmetadata.go#L119) | 42 | 52 | 80.77%
<!-- | **Total:** | **46** | **74** | **62.16%** | -->

|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [client/history/client.go](https://coveralls.io/builds/53733904/source?filename=client%2Fhistory%2Fclient.go#L152) | 2 | 38.1% |
| [client/history/metricClient.go](https://coveralls.io/builds/53733904/source?filename=client%2Fhistory%2FmetricClient.go#L203) | 2 | 45.3% |
| [common/cache/lru.go](https://coveralls.io/builds/53733904/source?filename=common%2Fcache%2Flru.go#L259) | 2 | 90.73% |
| [common/peerprovider/ringpopprovider/config.go](https://coveralls.io/builds/53733904/source?filename=common%2Fpeerprovider%2Fringpopprovider%2Fconfig.go#L257) | 2 | 81.58% |
| [common/persistence/executionManager.go](https://coveralls.io/builds/53733904/source?filename=common%2Fpersistence%2FexecutionManager.go#L530) | 2 | 78.83% |
| [common/persistence/historyManager.go](https://coveralls.io/builds/53733904/source?filename=common%2Fpersistence%2FhistoryManager.go#L181) | 2 | 66.67% |
| [common/persistence/persistenceMetricClients.go](https://coveralls.io/builds/53733904/source?filename=common%2Fpersistence%2FpersistenceMetricClients.go#L842) | 2 | 55.45% |
| [common/task/fifoTaskScheduler.go](https://coveralls.io/builds/53733904/source?filename=common%2Ftask%2FfifoTaskScheduler.go#L136) | 2 | 85.57% |
| [service/history/execution/mutable_state_util.go](https://coveralls.io/builds/53733904/source?filename=service%2Fhistory%2Fexecution%2Fmutable_state_util.go#L166) | 2 | 36.14% |
| [service/history/handler.go](https://coveralls.io/builds/53733904/source?filename=service%2Fhistory%2Fhandler.go#L984) | 2 | 47.17% |
<!-- | **Total:** | **181** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/53733904/badge)](https://coveralls.io/builds/53733904) |
| :-- | --: |
| Change from base [Build 0183cda6-cd14-405d-936d-e99fd29c7c5a](https://coveralls.io/builds/53221509): |  -0.1% |
| Covered Lines: | 85182 |
| Relevant Lines: | 148759 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/cmd/server/cadence/server.go b/cmd/server/cadence/server.go
index b12542b44..4a6259ff0 100644
--- a/cmd/server/cadence/server.go
+++ b/cmd/server/cadence/server.go
@@ -194,6 +194,9 @@ func (s *server) startService() common.Daemon {
 		clusterGroupMetadata.PrimaryClusterName,
 		clusterGroupMetadata.CurrentClusterName,
 		clusterGroupMetadata.ClusterGroup,
+		dc.GetBoolPropertyFilteredByDomain(dynamicconfig.UseNewInitialFailoverVersion),
+		params.MetricsClient,
+		params.Logger,
 	)
 
 	advancedVisMode := dc.GetStringProperty(
diff --git a/common/cluster/metadata.go b/common/cluster/metadata.go
index 771622a89..7b249e00f 100644
--- a/common/cluster/metadata.go
+++ b/common/cluster/metadata.go
@@ -25,11 +25,18 @@ import (
 
 	"github.com/uber/cadence/common"
 	"github.com/uber/cadence/common/config"
+	"github.com/uber/cadence/common/dynamicconfig"
+	"github.com/uber/cadence/common/log"
+	"github.com/uber/cadence/common/log/tag"
+	"github.com/uber/cadence/common/metrics"
 )
 
 type (
 	// Metadata provides information about clusters
 	Metadata struct {
+		log     log.Logger
+		metrics metrics.Scope
+
 		// failoverVersionIncrement is the increment of each cluster's version when failover happen
 		failoverVersionIncrement int64
 		// primaryClusterName is the name of the primary cluster, only the primary cluster can register / update domain
@@ -45,6 +52,8 @@ type (
 		remoteClusters map[string]config.ClusterInformation
 		// versionToClusterName contains all initial version -> corresponding cluster name
 		versionToClusterName map[int64]string
+		// allows for a new failover version migration
+		useNewFailoverVersionOverride dynamicconfig.BoolPropertyFnWithDomainFilter
 	}
 )
 
@@ -54,6 +63,9 @@ func NewMetadata(
 	primaryClusterName string,
 	currentClusterName string,
 	clusterGroup map[string]config.ClusterInformation,
+	useMinFailoverVersionOverrideConfig dynamicconfig.BoolPropertyFnWithDomainFilter,
+	metricsClient metrics.Client,
+	logger log.Logger,
 ) Metadata {
 	versionToClusterName := make(map[int64]string)
 	for clusterName, info := range clusterGroup {
@@ -77,36 +89,44 @@ func NewMetadata(
 	}
 
 	return Metadata{
-		failoverVersionIncrement: failoverVersionIncrement,
-		primaryClusterName:       primaryClusterName,
-		currentClusterName:       currentClusterName,
-		allClusters:              clusterGroup,
-		enabledClusters:          enabledClusters,
-		remoteClusters:           remoteClusters,
-		versionToClusterName:     versionToClusterName,
+		log:                           logger,
+		metrics:                       metricsClient.Scope(metrics.ClusterMetadataScope),
+		failoverVersionIncrement:      failoverVersionIncrement,
+		primaryClusterName:            primaryClusterName,
+		currentClusterName:            currentClusterName,
+		allClusters:                   clusterGroup,
+		enabledClusters:               enabledClusters,
+		remoteClusters:                remoteClusters,
+		versionToClusterName:          versionToClusterName,
+		useNewFailoverVersionOverride: useMinFailoverVersionOverrideConfig,
 	}
 }
 
 // GetNextFailoverVersion return the next failover version based on input
-func (m Metadata) GetNextFailoverVersion(cluster string, currentFailoverVersion int64) int64 {
-	info, ok := m.allClusters[cluster]
-	if !ok {
-		panic(fmt.Sprintf(
-			"Unknown cluster name: %v with given cluster initial failover version map: %v.",
-			cluster,
-			m.allClusters,
-		))
-	}
-	failoverVersion := currentFailoverVersion/m.failoverVersionIncrement*m.failoverVersionIncrement + info.InitialFailoverVersion
+func (m Metadata) GetNextFailoverVersion(cluster string, currentFailoverVersion int64, domainName string) int64 {
+	initialFailoverVersion := m.getInitialFailoverVersion(cluster, domainName)
+	failoverVersion := currentFailoverVersion/m.failoverVersionIncrement*m.failoverVersionIncrement + initialFailoverVersion
 	if failoverVersion < currentFailoverVersion {
 		return failoverVersion + m.failoverVersionIncrement
 	}
 	return failoverVersion
 }
 
-// IsVersionFromSameCluster return true if 2 version are used for the same cluster
+// IsVersionFromSameCluster return true if the new version is used for the same cluster
 func (m Metadata) IsVersionFromSameCluster(version1 int64, version2 int64) bool {
-	return (version1-version2)%m.failoverVersionIncrement == 0
+	v1Server, err := m.resolveServerName(version1)
+	if err != nil {
+		// preserving old behaviour however, this should never occur
+		m.metrics.IncCounter(metrics.ClusterMetadataFailureToResolveCounter)
+		m.log.Error("could not resolve an incoming version", tag.Dynamic("failover-version", version1))
+		return false
+	}
+	v2Server, err := m.resolveServerName(version2)
+	if err != nil {
+		m.log.Error("could not resolve an incoming version", tag.Dynamic("failover-version", version2))
+		return false
+	}
+	return v1Server == v2Server
 }
 
 func (m Metadata) IsPrimaryCluster() bool {
@@ -138,16 +158,59 @@ func (m Metadata) ClusterNameForFailoverVersion(failoverVersion int64) string {
 	if failoverVersion == common.EmptyVersion {
 		return m.currentClusterName
 	}
+	server, err := m.resolveServerName(failoverVersion)
+	if err != nil {
+		m.metrics.IncCounter(metrics.ClusterMetadataResolvingFailoverVersionCounter)
+		panic(fmt.Sprintf("failed to resolve failover version: %v", err))
+	}
+	return server
+}
 
-	initialFailoverVersion := failoverVersion % m.failoverVersionIncrement
-	clusterName, ok := m.versionToClusterName[initialFailoverVersion]
+// gets the initial failover version for a cluster / domain
+// along with some helpers for a migration - should it be necessary
+func (m Metadata) getInitialFailoverVersion(cluster string, domainName string) int64 {
+	info, ok := m.allClusters[cluster]
 	if !ok {
 		panic(fmt.Sprintf(
-			"Unknown initial failover version %v with given cluster initial failover version map: %v and failover version increment %v.",
-			initialFailoverVersion,
+			"Unknown cluster name: %v with given cluster initial failover version map: %v.",
+			cluster,
 			m.allClusters,
-			m.failoverVersionIncrement,
 		))
 	}
-	return clusterName
+
+	// if using the minFailover Version during a cluster config, then return this from config
+	// (assuming it's safe to do so). This is not the normal state of things and intended only
+	// for when migrating versions.
+	usingNewFailoverVersion := m.useNewFailoverVersionOverride(domainName)
+	if usingNewFailoverVersion && info.NewInitialFailoverVersion != nil {
+		m.log.Debug("using new failover version for cluster", tag.ClusterName(cluster), tag.WorkflowDomainName(domainName))
+		m.metrics.IncCounter(metrics.ClusterMetadataGettingMinFailoverVersionCounter)
+		return *info.NewInitialFailoverVersion
+	}
+	// default behaviour - return the initial failover version - a marker to
+	// identify the cluster for all counters
+	m.log.Debug("getting failover version for cluster", tag.ClusterName(cluster), tag.WorkflowDomainName(domainName))
+	m.metrics.IncCounter(metrics.ClusterMetadataGettingFailoverVersionCounter)
+	return info.InitialFailoverVersion
+}
+
+// resolves the server name from a version number. Better to use this
+// than to check versionToClusterName directly, as this also falls back to catch
+// when there's a migration NewInitialFailoverVersion
+func (m Metadata) resolveServerName(version int64) (string, error) {
+	moddedFoVersion := version % m.failoverVersionIncrement
+	// attempt a lookup first
+	server, ok := m.versionToClusterName[moddedFoVersion]
+	if ok {
+		return server, nil
+	}
+
+	// else fall back on checking for new failover versions
+	for name, cluster := range m.allClusters {
+		if cluster.NewInitialFailoverVersion != nil && *cluster.NewInitialFailoverVersion == moddedFoVersion {
+			return name, nil
+		}
+	}
+	m.metrics.IncCounter(metrics.ClusterMetadataFailureToResolveCounter)
+	return "", fmt.Errorf("could not resolve failover version: %d", version)
 }
diff --git a/common/config/cluster.go b/common/config/cluster.go
index a55788d8c..394f56916 100644
--- a/common/config/cluster.go
+++ b/common/config/cluster.go
@@ -58,6 +58,12 @@ type (
 		Enabled bool `yaml:"enabled"`
 		// InitialFailoverVersion is the identifier of each cluster. 0 <= the value < failoverVersionIncrement
 		InitialFailoverVersion int64 `yaml:"initialFailoverVersion"`
+		// NewInitialFailoverVersion is a new failover version for an initialFailoverVersion migration
+		// for when it's necessary to migrate between two values.
+		//
+		// this is a pointer to imply optionality, it's an optional field and its lack
+		// is indicated by a nil pointer. Zero is a valid field
+		NewInitialFailoverVersion *int64 `yaml:"newInitialFailoverVersion"`
 		// RPCName indicate the remote service name
 		RPCName string `yaml:"rpcName"`
 		// Address indicate the remote service address(Host:Port). Host can be DNS name.
diff --git a/common/domain/handler.go b/common/domain/handler.go
index c5c9eb74f..859d43a93 100644
--- a/common/domain/handler.go
+++ b/common/domain/handler.go
@@ -238,7 +238,7 @@ func (d *handlerImpl) RegisterDomain(
 
 	failoverVersion := common.EmptyVersion
 	if registerRequest.GetIsGlobalDomain() {
-		failoverVersion = d.clusterMetadata.GetNextFailoverVersion(activeClusterName, 0)
+		failoverVersion = d.clusterMetadata.GetNextFailoverVersion(activeClusterName, 0, registerRequest.Name)
 	}
 
 	domainRequest := &persistence.CreateDomainRequest{
@@ -521,6 +521,7 @@ func (d *handlerImpl) UpdateDomain(
 			failoverVersion = d.clusterMetadata.GetNextFailoverVersion(
 				replicationConfig.ActiveClusterName,
 				failoverVersion,
+				updateRequest.Name,
 			)
 			failoverNotificationVersion = notificationVersion
 		}
diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go
index 8457bae83..bf0907883 100644
--- a/common/dynamicconfig/constants.go
+++ b/common/dynamicconfig/constants.go
@@ -1543,6 +1543,13 @@ const (
 	// Default value: true
 	// Allowed filters: DomainID, WorkflowID
 	EnableReplicationTaskGeneration
+	// UseNewInitialFailoverVersion is a switch to issue a failover version based on the minFailoverVersion
+	// rather than the default initialFailoverVersion. USed as a per-domain migration switch
+	// KeyName: history.useNewInitialFailoverVersion
+	// Value type: Bool
+	// Default value: false
+	// Allowed filters: N/A
+	UseNewInitialFailoverVersion
 	// EnableRecordWorkflowExecutionUninitialized enables record workflow execution uninitialized state in ElasticSearch
 	// KeyName: history.EnableRecordWorkflowExecutionUninitialized
 	// Value type: Bool
@@ -3562,6 +3569,11 @@ var BoolKeys = map[BoolKey]DynamicBool{
 		Description:  "EnableReplicationTaskGeneration is the flag to control replication generation",
 		DefaultValue: true,
 	},
+	UseNewInitialFailoverVersion: DynamicBool{
+		KeyName:      "history.useNewInitialFailoverVersion",
+		Description:  "use the minInitialFailover version",
+		DefaultValue: false,
+	},
 	AllowArchivingIncompleteHistory: DynamicBool{
 		KeyName:      "worker.AllowArchivingIncompleteHistory",
 		Description:  "AllowArchivingIncompleteHistory will continue on when seeing some error like history mutated(usually caused by database consistency issues)",
diff --git a/common/log/tag/interface.go b/common/log/tag/interface.go
index 8066b902b..dbd1ab7ed 100644
--- a/common/log/tag/interface.go
+++ b/common/log/tag/interface.go
@@ -98,3 +98,9 @@ func newPredefinedStringTag(key string, value string) Tag {
 		field: zap.String(key, value),
 	}
 }
+
+func newPredefinedDynamicTag(key string, value interface{}) Tag {
+	return Tag{
+		field: zap.Any(key, value),
+	}
+}
diff --git a/common/log/tag/tags.go b/common/log/tag/tags.go
index 04848591c..f1967c23a 100644
--- a/common/log/tag/tags.go
+++ b/common/log/tag/tags.go
@@ -862,3 +862,9 @@ func ArchivalBlobstoreContextTimeout(blobstoreContextTimeout time.Duration) Tag
 func VisibilityQuery(query string) Tag {
 	return newStringTag("visibility-query", query)
 }
+
+// Dynamic Uses reflection based logging for arbitrary values
+// for not very performant logging
+func Dynamic(key string, v interface{}) Tag {
+	return newPredefinedDynamicTag(key, v)
+}
diff --git a/common/metrics/defs.go b/common/metrics/defs.go
index 29f2bab25..25a77a1f4 100644
--- a/common/metrics/defs.go
+++ b/common/metrics/defs.go
@@ -727,6 +727,8 @@ const (
 	DomainFailoverScope
 	// DomainReplicationQueueScope is used in domainreplication queue
 	DomainReplicationQueueScope
+	// ClusterMetadataScope is used for the cluster metadata
+	ClusterMetadataScope
 
 	NumCommonScopes
 )
@@ -1526,6 +1528,7 @@ var ScopeDefs = map[ServiceIdx]map[int]scopeDefinition{
 
 		DomainFailoverScope:         {operation: "DomainFailover"},
 		DomainReplicationQueueScope: {operation: "DomainReplicationQueue"},
+		ClusterMetadataScope:        {operation: "ClusterMetadata"},
 	},
 	// Frontend Scope Names
 	Frontend: {
@@ -2028,6 +2031,12 @@ const (
 	CrossClusterTaskFetchedTimer
 	CrossClusterTaskPendingTimer
 
+	ClusterMetadataFailureToResolveCounter
+	ClusterMetadataGettingMinFailoverVersionCounter
+	ClusterMetadataGettingFailoverVersionCounter
+	ClusterMetadataResolvingFailoverVersionCounter
+	ClusterMetadataResolvingMinFailoverVersionCounter
+
 	ActivityE2ELatency
 	ActivityLostCounter
 	AckLevelUpdateCounter
@@ -2615,6 +2624,11 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		CrossClusterTaskRespondFailures:                              {metricName: "cross_cluster_fetch_errors", metricType: Counter},
 		CrossClusterTaskFetchedTimer:                                 {metricName: "cross_cluster_task_fetched", metricType: Timer},
 		CrossClusterTaskPendingTimer:                                 {metricName: "cross_cluster_task_pending", metricType: Timer},
+		ClusterMetadataFailureToResolveCounter:                       {metricName: "failed_to_resolve_failover_version", metricType: Counter},
+		ClusterMetadataGettingMinFailoverVersionCounter:              {metricName: "getting_min_failover_version_counter", metricType: Counter},
+		ClusterMetadataGettingFailoverVersionCounter:                 {metricName: "getting_failover_version_counter", metricType: Counter},
+		ClusterMetadataResolvingFailoverVersionCounter:               {metricName: "resolving_failover_version_counter", metricType: Counter},
+		ClusterMetadataResolvingMinFailoverVersionCounter:            {metricName: "resolving_min_failover_version_counter", metricType: Counter},
 		ActivityE2ELatency:                                           {metricName: "activity_end_to_end_latency", metricType: Timer},
 		ActivityLostCounter:                                          {metricName: "activity_lost", metricType: Counter},
 		AckLevelUpdateCounter:                                        {metricName: "ack_level_update", metricType: Counter},
diff --git a/config/development.yaml b/config/development.yaml
index 8f010b9e0..a62df9eb9 100644
--- a/config/development.yaml
+++ b/config/development.yaml
@@ -79,6 +79,7 @@ clusterGroupMetadata:
     cluster0:
       enabled: true
       initialFailoverVersion: 0
+      newInitialFailoverVersion: 1 # migrating to this new failover version
       rpcAddress: "localhost:7833" # this is to let worker service and XDC replicator connected to the frontend service. In cluster setup, localhost will not work
       rpcTransport: "grpc"
 
diff --git a/config/dynamicconfig/development.yaml b/config/dynamicconfig/development.yaml
index fac85a4d3..c91ea7986 100644
--- a/config/dynamicconfig/development.yaml
+++ b/config/dynamicconfig/development.yaml
@@ -10,3 +10,6 @@ history.EnableConsistentQueryByDomain:
 history.enableCrossClusterOperations:
 - value: true
   constraints: {}
+history.useNewInitialFailoverVersion:
+- value: true
+  constraints: {}
diff --git a/tools/cli/domainUtils.go b/tools/cli/domainUtils.go
index 1cd1ebd28..7a774d428 100644
--- a/tools/cli/domainUtils.go
+++ b/tools/cli/domainUtils.go
@@ -241,7 +241,7 @@ func initializeAdminDomainHandler(
 
 	metricsClient := initializeMetricsClient()
 	logger := initializeLogger(configuration)
-	clusterMetadata := initializeClusterMetadata(configuration)
+	clusterMetadata := initializeClusterMetadata(configuration, metricsClient, logger)
 	metadataMgr := initializeDomainManager(context)
 	dynamicConfig := initializeDynamicConfig(configuration, logger)
 	return initializeDomainHandler(
@@ -302,7 +302,7 @@ func initializeLogger(
 	return loggerimpl.NewLogger(zapLogger)
 }
 
-func initializeClusterMetadata(serviceConfig *config.Config) cluster.Metadata {
+func initializeClusterMetadata(serviceConfig *config.Config, metrics metrics.Client, logger log.Logger) cluster.Metadata {
 
 	clusterGroupMetadata := serviceConfig.ClusterGroupMetadata
 	return cluster.NewMetadata(
@@ -310,6 +310,9 @@ func initializeClusterMetadata(serviceConfig *config.Config) cluster.Metadata {
 		clusterGroupMetadata.PrimaryClusterName,
 		clusterGroupMetadata.CurrentClusterName,
 		clusterGroupMetadata.ClusterGroup,
+		func(d string) bool { return false },
+		metrics,
+		logger,
 	)
 }
 


Test patch:
diff --git a/common/cluster/metadataTest.go b/common/cluster/metadataTest.go
index fe5115f3f..d2c4c86e2 100644
--- a/common/cluster/metadataTest.go
+++ b/common/cluster/metadataTest.go
@@ -22,6 +22,8 @@ package cluster
 
 import (
 	"github.com/uber/cadence/common/config"
+	"github.com/uber/cadence/common/log/loggerimpl"
+	commonMetrics "github.com/uber/cadence/common/metrics"
 	"github.com/uber/cadence/common/service"
 )
 
@@ -92,6 +94,9 @@ var (
 		TestCurrentClusterName,
 		TestCurrentClusterName,
 		TestAllClusterInfo,
+		func(d string) bool { return false },
+		commonMetrics.NewNoopMetricsClient(),
+		loggerimpl.NewNopLogger(),
 	)
 
 	// TestPassiveClusterMetadata is metadata for a passive cluster
@@ -100,6 +105,9 @@ var (
 		TestCurrentClusterName,
 		TestAlternativeClusterName,
 		TestAllClusterInfo,
+		func(d string) bool { return false },
+		commonMetrics.NewNoopMetricsClient(),
+		loggerimpl.NewNopLogger(),
 	)
 )
 
@@ -115,5 +123,8 @@ func GetTestClusterMetadata(isPrimaryCluster bool) Metadata {
 		primaryClusterName,
 		TestCurrentClusterName,
 		TestAllClusterInfo,
+		func(d string) bool { return false },
+		commonMetrics.NewNoopMetricsClient(),
+		loggerimpl.NewNopLogger(),
 	)
 }
diff --git a/common/cluster/metadata_test.go b/common/cluster/metadata_test.go
new file mode 100644
index 000000000..4823b6bf5
--- /dev/null
+++ b/common/cluster/metadata_test.go
@@ -0,0 +1,943 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package cluster
+
+import (
+	"fmt"
+	"testing"
+	"testing/quick"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/uber/cadence/common/config"
+	"github.com/uber/cadence/common/log/loggerimpl"
+	"github.com/uber/cadence/common/metrics"
+)
+
+func TestMetadataBehaviour(t *testing.T) {
+
+	const clusterName1 = "c1"
+	const initialFailoverVersionC1 = 0
+	const clusterName2 = "c2"
+	const initialFailoverVersionC2 = 2
+
+	const failoverVersionIncrement = 100
+
+	tests := map[string]struct {
+		failoverCluster string
+		currentVersion  int64
+		expectedOut     int64
+	}{
+		"a new domain, created in c1 already - a failover to c1 where it already is should have no effect": {
+			failoverCluster: clusterName1,
+			currentVersion:  0,
+			expectedOut:     0,
+		},
+		"a new domain, created in c1 already - a failover to c2 should set the failover version to be based on c2": {
+			failoverCluster: clusterName2,
+			currentVersion:  0,
+			expectedOut:     2,
+		},
+		"a subsequent failover back": {
+			failoverCluster: clusterName1,
+			currentVersion:  2,
+			expectedOut:     100,
+		},
+		"and a duplicate": {
+			failoverCluster: clusterName1,
+			currentVersion:  100,
+			expectedOut:     100,
+		},
+		"and a subsequent fail back over to c2": {
+			failoverCluster: clusterName2,
+			currentVersion:  100,
+			expectedOut:     102,
+		},
+	}
+
+	for name, td := range tests {
+		t.Run(name, func(t *testing.T) {
+			m := Metadata{
+				failoverVersionIncrement: failoverVersionIncrement,
+				allClusters: map[string]config.ClusterInformation{
+					clusterName1: {
+						InitialFailoverVersion: initialFailoverVersionC1,
+					},
+					clusterName2: {
+						InitialFailoverVersion: initialFailoverVersionC2,
+					},
+				},
+				versionToClusterName: map[int64]string{
+					initialFailoverVersionC1: clusterName1,
+					initialFailoverVersionC2: clusterName2,
+				},
+				useNewFailoverVersionOverride: func(domain string) bool { return false },
+				metrics:                       metrics.NewNoopMetricsClient().Scope(0),
+				log:                           loggerimpl.NewNopLogger(),
+			}
+			assert.Equal(t, td.expectedOut, m.GetNextFailoverVersion(td.failoverCluster, td.currentVersion, "a domain"), name)
+		})
+	}
+}
+
+func TestFailoverVersionLogicIsMonotonic(t *testing.T) {
+
+	const clusterName1 = "c1"
+	const initialFailoverVersionC1 = 0
+	var minFailoverVersionC1 = int64(1)
+	const clusterName2 = "c2"
+	const initialFailoverVersionC2 = 2
+	var minFailoverVersionC2 = int64(3)
+
+	const failoverVersionIncrement = 100
+	const someDomainMigrating = "a domain migrating"
+	const someDomainNotMigrating = "a domain not migrating"
+
+	m := Metadata{
+		failoverVersionIncrement: failoverVersionIncrement,
+		allClusters: map[string]config.ClusterInformation{
+			clusterName1: {
+				InitialFailoverVersion:    initialFailoverVersionC1,
+				NewInitialFailoverVersion: &minFailoverVersionC1,
+			},
+			clusterName2: {
+				InitialFailoverVersion:    initialFailoverVersionC2,
+				NewInitialFailoverVersion: &minFailoverVersionC2,
+			},
+		},
+		versionToClusterName: map[int64]string{
+			initialFailoverVersionC1: clusterName1,
+			initialFailoverVersionC2: clusterName2,
+		},
+		useNewFailoverVersionOverride: func(domain string) bool { return someDomainMigrating == domain },
+		metrics:                       metrics.NewNoopMetricsClient().Scope(0),
+		log:                           loggerimpl.NewNopLogger(),
+	}
+
+	current := int64(0)
+	iterations := 0
+
+	for iterations < 4000 {
+		next := m.GetNextFailoverVersion(clusterName2, current, someDomainNotMigrating)
+		assert.Truef(t, next > current, "current: %d, next %d", current, next)
+		current = next
+
+		next = m.GetNextFailoverVersion(clusterName1, current, someDomainNotMigrating)
+		assert.Truef(t, next > current, "current: %d, next %d", current, next)
+		current = next
+
+		next = m.GetNextFailoverVersion(clusterName1, current, someDomainMigrating)
+		assert.Truef(t, next > current, "current: %d, next %d", current, next)
+		current = next
+		iterations++
+	}
+}
+
+func TestResolvingClusterVersion(t *testing.T) {
+	const clusterName1 = "c1"
+	const initialFailoverVersionC1 = 0
+	var newInitialFailoverVersionC1 = int64(1)
+	const clusterName2 = "c2"
+	const initialFailoverVersionC2 = 2
+	var newInitialFailoverVersionC2 = int64(3)
+	const failoverVersionIncrement = 100
+
+	tests := map[string]struct {
+		input          int64
+		expectedOutput string
+		expectedErr    error
+	}{
+		"first cluster, normal version": {
+			input:          initialFailoverVersionC1,
+			expectedOutput: clusterName1,
+		},
+		"first cluster, minFailover version": {
+			input:          newInitialFailoverVersionC1,
+			expectedOutput: clusterName1,
+		},
+		"second cluster, normal version": {
+			input:          initialFailoverVersionC2,
+			expectedOutput: clusterName2,
+		},
+		"second cluster, minFailover version": {
+			input:          newInitialFailoverVersionC2,
+			expectedOutput: clusterName2,
+		},
+
+		"first cluster, normal version - higher versions": {
+			input:          200,
+			expectedOutput: clusterName1,
+		},
+		"first cluster, minFailover version - higher versions": {
+			input:          101,
+			expectedOutput: clusterName1,
+		},
+		"second cluster, normal initialFailover version - higher versions": {
+			input:          302,
+			expectedOutput: clusterName2,
+		},
+		"second cluster, minFailover version - higher versions": {
+			input:          403,
+			expectedOutput: clusterName2,
+		},
+
+		"invalid input": {
+			input:       599,
+			expectedErr: fmt.Errorf("could not resolve failover version: 599"),
+		},
+	}
+
+	m := Metadata{
+		failoverVersionIncrement: failoverVersionIncrement,
+		allClusters: map[string]config.ClusterInformation{
+			clusterName1: {
+				InitialFailoverVersion:    initialFailoverVersionC1,
+				NewInitialFailoverVersion: &newInitialFailoverVersionC1,
+			},
+			clusterName2: {
+				InitialFailoverVersion:    initialFailoverVersionC2,
+				NewInitialFailoverVersion: &newInitialFailoverVersionC2,
+			},
+		},
+		versionToClusterName: map[int64]string{
+			initialFailoverVersionC1: clusterName1,
+			initialFailoverVersionC2: clusterName2,
+		},
+		metrics: metrics.NewNoopMetricsClient().Scope(0),
+		log:     loggerimpl.NewNopLogger(),
+	}
+
+	for name, td := range tests {
+		t.Run(name, func(*testing.T) {
+			out, err := m.resolveServerName(td.input)
+			assert.Equal(t, td.expectedOutput, out)
+			assert.Equal(t, td.expectedErr, err)
+		})
+	}
+}
+
+func TestIsPartOfTheSameCluster(t *testing.T) {
+
+	const clusterName1 = "c1"
+	const initialFailoverVersionC1 = 0
+	const clusterName2 = "c2"
+	const initialFailoverVersionC2 = 2
+	const failoverVersionIncrement = 100
+
+	tests := map[string]struct {
+		v1                             int64
+		v2                             int64
+		useMinFailoverVersionAllowance func(string) bool
+		expectedResult                 bool
+	}{
+		"normal case v1 - is from the same cluster": {
+			v1:             0,
+			v2:             0,
+			expectedResult: true,
+		},
+		"normal case v2 - is from a different cluster": {
+			v1:             0,
+			v2:             2,
+			expectedResult: false,
+		},
+		"normal case v3 - is from the same cluster": {
+			v1:             0,
+			v2:             100,
+			expectedResult: true,
+		},
+		"normal case v4 - is from a different cluster": {
+			v1:             0,
+			v2:             102,
+			expectedResult: false,
+		},
+	}
+
+	for name, td := range tests {
+		t.Run(name, func(*testing.T) {
+			m := Metadata{
+				failoverVersionIncrement: failoverVersionIncrement,
+				allClusters: map[string]config.ClusterInformation{
+					clusterName1: {
+						InitialFailoverVersion: initialFailoverVersionC1,
+					},
+					clusterName2: {
+						InitialFailoverVersion: initialFailoverVersionC2,
+					},
+				},
+				versionToClusterName: map[int64]string{
+					initialFailoverVersionC1: clusterName1,
+					initialFailoverVersionC2: clusterName2,
+				},
+				log:     loggerimpl.NewNopLogger(),
+				metrics: metrics.NewNoopMetricsClient().Scope(0),
+			}
+
+			assert.Equal(t, td.expectedResult, m.IsVersionFromSameCluster(td.v1, td.v2), name)
+		})
+
+	}
+}
+
+func TestIsPartOfTheSameClusterAPIFixing(t *testing.T) {
+
+	const clusterName1 = "c1"
+	const initialFailoverVersionC1 = 0
+	const clusterName2 = "c2"
+	const initialFailoverVersionC2 = 2
+	const failoverVersionIncrement = 100
+
+	tests := []struct {
+		v0       int64
+		v1       int64
+		expected bool
+	}{
+		{v0: 0, v1: 0, expected: true},
+		{v0: 0, v1: 1, expected: false},
+		{v0: 0, v1: 2, expected: false},
+		{v0: 0, v1: 3, expected: false},
+		{v0: 0, v1: 4, expected: false},
+		{v0: 0, v1: 5, expected: false},
+		{v0: 0, v1: 6, expected: false},
+		{v0: 0, v1: 7, expected: false},
+		{v0: 0, v1: 8, expected: false},
+		{v0: 0, v1: 9, expected: false},
+		{v0: 0, v1: 10, expected: false},
+		{v0: 0, v1: 11, expected: false},
+		{v0: 0, v1: 12, expected: false},
+		{v0: 0, v1: 13, expected: false},
+		{v0: 0, v1: 14, expected: false},
+		{v0: 0, v1: 15, expected: false},
+		{v0: 0, v1: 16, expected: false},
+		{v0: 0, v1: 17, expected: false},
+		{v0: 0, v1: 18, expected: false},
+		{v0: 0, v1: 19, expected: false},
+		{v0: 0, v1: 20, expected: false},
+		{v0: 0, v1: 21, expected: false},
+		{v0: 0, v1: 22, expected: false},
+		{v0: 0, v1: 23, expected: false},
+		{v0: 0, v1: 24, expected: false},
+		{v0: 0, v1: 25, expected: false},
+		{v0: 0, v1: 26, expected: false},
+		{v0: 0, v1: 27, expected: false},
+		{v0: 0, v1: 28, expected: false},
+		{v0: 0, v1: 29, expected: false},
+		{v0: 0, v1: 30, expected: false},
+		{v0: 0, v1: 31, expected: false},
+		{v0: 0, v1: 32, expected: false},
+		{v0: 0, v1: 33, expected: false},
+		{v0: 0, v1: 34, expected: false},
+		{v0: 0, v1: 35, expected: false},
+		{v0: 0, v1: 36, expected: false},
+		{v0: 0, v1: 37, expected: false},
+		{v0: 0, v1: 38, expected: false},
+		{v0: 0, v1: 39, expected: false},
+		{v0: 0, v1: 40, expected: false},
+		{v0: 0, v1: 41, expected: false},
+		{v0: 0, v1: 42, expected: false},
+		{v0: 0, v1: 43, expected: false},
+		{v0: 0, v1: 44, expected: false},
+		{v0: 0, v1: 45, expected: false},
+		{v0: 0, v1: 46, expected: false},
+		{v0: 0, v1: 47, expected: false},
+		{v0: 0, v1: 48, expected: false},
+		{v0: 0, v1: 49, expected: false},
+		{v0: 0, v1: 50, expected: false},
+		{v0: 0, v1: 51, expected: false},
+		{v0: 0, v1: 52, expected: false},
+		{v0: 0, v1: 53, expected: false},
+		{v0: 0, v1: 54, expected: false},
+		{v0: 0, v1: 55, expected: false},
+		{v0: 0, v1: 56, expected: false},
+		{v0: 0, v1: 57, expected: false},
+		{v0: 0, v1: 58, expected: false},
+		{v0: 0, v1: 59, expected: false},
+		{v0: 0, v1: 60, expected: false},
+		{v0: 0, v1: 61, expected: false},
+		{v0: 0, v1: 62, expected: false},
+		{v0: 0, v1: 63, expected: false},
+		{v0: 0, v1: 64, expected: false},
+		{v0: 0, v1: 65, expected: false},
+		{v0: 0, v1: 66, expected: false},
+		{v0: 0, v1: 67, expected: false},
+		{v0: 0, v1: 68, expected: false},
+		{v0: 0, v1: 69, expected: false},
+		{v0: 0, v1: 70, expected: false},
+		{v0: 0, v1: 71, expected: false},
+		{v0: 0, v1: 72, expected: false},
+		{v0: 0, v1: 73, expected: false},
+		{v0: 0, v1: 74, expected: false},
+		{v0: 0, v1: 75, expected: false},
+		{v0: 0, v1: 76, expected: false},
+		{v0: 0, v1: 77, expected: false},
+		{v0: 0, v1: 78, expected: false},
+		{v0: 0, v1: 79, expected: false},
+		{v0: 0, v1: 80, expected: false},
+		{v0: 0, v1: 81, expected: false},
+		{v0: 0, v1: 82, expected: false},
+		{v0: 0, v1: 83, expected: false},
+		{v0: 0, v1: 84, expected: false},
+		{v0: 0, v1: 85, expected: false},
+		{v0: 0, v1: 86, expected: false},
+		{v0: 0, v1: 87, expected: false},
+		{v0: 0, v1: 88, expected: false},
+		{v0: 0, v1: 89, expected: false},
+		{v0: 0, v1: 90, expected: false},
+		{v0: 0, v1: 91, expected: false},
+		{v0: 0, v1: 92, expected: false},
+		{v0: 0, v1: 93, expected: false},
+		{v0: 0, v1: 94, expected: false},
+		{v0: 0, v1: 95, expected: false},
+		{v0: 0, v1: 96, expected: false},
+		{v0: 0, v1: 97, expected: false},
+		{v0: 0, v1: 98, expected: false},
+		{v0: 0, v1: 99, expected: false},
+		{v0: 0, v1: 100, expected: true},
+		{v0: 0, v1: 101, expected: false},
+		{v0: 0, v1: 102, expected: false},
+		{v0: 0, v1: 103, expected: false},
+		{v0: 0, v1: 104, expected: false},
+		{v0: 0, v1: 105, expected: false},
+		{v0: 0, v1: 106, expected: false},
+		{v0: 0, v1: 107, expected: false},
+		{v0: 0, v1: 108, expected: false},
+		{v0: 0, v1: 109, expected: false},
+		{v0: 0, v1: 110, expected: false},
+		{v0: 0, v1: 111, expected: false},
+		{v0: 0, v1: 112, expected: false},
+		{v0: 0, v1: 113, expected: false},
+		{v0: 0, v1: 114, expected: false},
+		{v0: 0, v1: 115, expected: false},
+		{v0: 0, v1: 116, expected: false},
+		{v0: 0, v1: 117, expected: false},
+		{v0: 0, v1: 118, expected: false},
+		{v0: 0, v1: 119, expected: false},
+		{v0: 0, v1: 120, expected: false},
+		{v0: 0, v1: 121, expected: false},
+		{v0: 0, v1: 122, expected: false},
+		{v0: 0, v1: 123, expected: false},
+		{v0: 0, v1: 124, expected: false},
+		{v0: 0, v1: 125, expected: false},
+		{v0: 0, v1: 126, expected: false},
+		{v0: 0, v1: 127, expected: false},
+		{v0: 0, v1: 128, expected: false},
+		{v0: 0, v1: 129, expected: false},
+		{v0: 0, v1: 130, expected: false},
+		{v0: 0, v1: 131, expected: false},
+		{v0: 0, v1: 132, expected: false},
+		{v0: 0, v1: 133, expected: false},
+		{v0: 0, v1: 134, expected: false},
+		{v0: 0, v1: 135, expected: false},
+		{v0: 0, v1: 136, expected: false},
+		{v0: 0, v1: 137, expected: false},
+		{v0: 0, v1: 138, expected: false},
+		{v0: 0, v1: 139, expected: false},
+		{v0: 0, v1: 140, expected: false},
+		{v0: 0, v1: 141, expected: false},
+		{v0: 0, v1: 142, expected: false},
+		{v0: 0, v1: 143, expected: false},
+		{v0: 0, v1: 144, expected: false},
+		{v0: 0, v1: 145, expected: false},
+		{v0: 0, v1: 146, expected: false},
+		{v0: 0, v1: 147, expected: false},
+		{v0: 0, v1: 148, expected: false},
+		{v0: 0, v1: 149, expected: false},
+		{v0: 0, v1: 150, expected: false},
+		{v0: 0, v1: 151, expected: false},
+		{v0: 0, v1: 152, expected: false},
+		{v0: 0, v1: 153, expected: false},
+		{v0: 0, v1: 154, expected: false},
+		{v0: 0, v1: 155, expected: false},
+		{v0: 0, v1: 156, expected: false},
+		{v0: 0, v1: 157, expected: false},
+		{v0: 0, v1: 158, expected: false},
+		{v0: 0, v1: 159, expected: false},
+		{v0: 0, v1: 160, expected: false},
+		{v0: 0, v1: 161, expected: false},
+		{v0: 0, v1: 162, expected: false},
+		{v0: 0, v1: 163, expected: false},
+		{v0: 0, v1: 164, expected: false},
+		{v0: 0, v1: 165, expected: false},
+		{v0: 0, v1: 166, expected: false},
+		{v0: 0, v1: 167, expected: false},
+		{v0: 0, v1: 168, expected: false},
+		{v0: 0, v1: 169, expected: false},
+		{v0: 0, v1: 170, expected: false},
+		{v0: 0, v1: 171, expected: false},
+		{v0: 0, v1: 172, expected: false},
+		{v0: 0, v1: 173, expected: false},
+		{v0: 0, v1: 174, expected: false},
+		{v0: 0, v1: 175, expected: false},
+		{v0: 0, v1: 176, expected: false},
+		{v0: 0, v1: 177, expected: false},
+		{v0: 0, v1: 178, expected: false},
+		{v0: 0, v1: 179, expected: false},
+		{v0: 0, v1: 180, expected: false},
+		{v0: 0, v1: 181, expected: false},
+		{v0: 0, v1: 182, expected: false},
+		{v0: 0, v1: 183, expected: false},
+		{v0: 0, v1: 184, expected: false},
+		{v0: 0, v1: 185, expected: false},
+		{v0: 0, v1: 186, expected: false},
+		{v0: 0, v1: 187, expected: false},
+		{v0: 0, v1: 188, expected: false},
+		{v0: 0, v1: 189, expected: false},
+		{v0: 0, v1: 190, expected: false},
+		{v0: 0, v1: 191, expected: false},
+		{v0: 0, v1: 192, expected: false},
+		{v0: 0, v1: 193, expected: false},
+		{v0: 0, v1: 194, expected: false},
+		{v0: 0, v1: 195, expected: false},
+		{v0: 0, v1: 196, expected: false},
+		{v0: 0, v1: 197, expected: false},
+		{v0: 0, v1: 198, expected: false},
+		{v0: 0, v1: 199, expected: false},
+		{v0: 0, v1: 200, expected: true},
+	}
+
+	m := Metadata{
+		failoverVersionIncrement: failoverVersionIncrement,
+		allClusters: map[string]config.ClusterInformation{
+			clusterName1: {
+				InitialFailoverVersion: initialFailoverVersionC1,
+			},
+			clusterName2: {
+				InitialFailoverVersion: initialFailoverVersionC2,
+			},
+		},
+		versionToClusterName: map[int64]string{
+			initialFailoverVersionC1: clusterName1,
+			initialFailoverVersionC2: clusterName2,
+		},
+		useNewFailoverVersionOverride: func(domain string) bool { return false },
+		metrics:                       metrics.NewNoopMetricsClient().Scope(0),
+		log:                           loggerimpl.NewNopLogger(),
+	}
+
+	for i := range tests {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+			assert.Equal(t, tests[i].expected, m.IsVersionFromSameCluster(tests[i].v0, tests[i].v1))
+		})
+	}
+}
+
+func TestClusterNameForFailoverVersion(t *testing.T) {
+
+	const clusterName1 = "c1"
+	const initialFailoverVersionC1 = 0
+	const clusterName2 = "c2"
+	const initialFailoverVersionC2 = 2
+	const failoverVersionIncrement = 100
+
+	tests := []struct {
+		v0       int64
+		v1       int64
+		expected bool
+	}{
+		{v0: 0, v1: 0, expected: true},
+		{v0: 0, v1: 1, expected: false},
+		{v0: 0, v1: 2, expected: false},
+		{v0: 0, v1: 3, expected: false},
+		{v0: 0, v1: 4, expected: false},
+		{v0: 0, v1: 5, expected: false},
+		{v0: 0, v1: 6, expected: false},
+		{v0: 0, v1: 7, expected: false},
+		{v0: 0, v1: 8, expected: false},
+		{v0: 0, v1: 9, expected: false},
+		{v0: 0, v1: 10, expected: false},
+		{v0: 0, v1: 11, expected: false},
+		{v0: 0, v1: 12, expected: false},
+		{v0: 0, v1: 13, expected: false},
+		{v0: 0, v1: 14, expected: false},
+		{v0: 0, v1: 15, expected: false},
+		{v0: 0, v1: 16, expected: false},
+		{v0: 0, v1: 17, expected: false},
+		{v0: 0, v1: 18, expected: false},
+		{v0: 0, v1: 19, expected: false},
+		{v0: 0, v1: 20, expected: false},
+		{v0: 0, v1: 21, expected: false},
+		{v0: 0, v1: 22, expected: false},
+		{v0: 0, v1: 23, expected: false},
+		{v0: 0, v1: 24, expected: false},
+		{v0: 0, v1: 25, expected: false},
+		{v0: 0, v1: 26, expected: false},
+		{v0: 0, v1: 27, expected: false},
+		{v0: 0, v1: 28, expected: false},
+		{v0: 0, v1: 29, expected: false},
+		{v0: 0, v1: 30, expected: false},
+		{v0: 0, v1: 31, expected: false},
+		{v0: 0, v1: 32, expected: false},
+		{v0: 0, v1: 33, expected: false},
+		{v0: 0, v1: 34, expected: false},
+		{v0: 0, v1: 35, expected: false},
+		{v0: 0, v1: 36, expected: false},
+		{v0: 0, v1: 37, expected: false},
+		{v0: 0, v1: 38, expected: false},
+		{v0: 0, v1: 39, expected: false},
+		{v0: 0, v1: 40, expected: false},
+		{v0: 0, v1: 41, expected: false},
+		{v0: 0, v1: 42, expected: false},
+		{v0: 0, v1: 43, expected: false},
+		{v0: 0, v1: 44, expected: false},
+		{v0: 0, v1: 45, expected: false},
+		{v0: 0, v1: 46, expected: false},
+		{v0: 0, v1: 47, expected: false},
+		{v0: 0, v1: 48, expected: false},
+		{v0: 0, v1: 49, expected: false},
+		{v0: 0, v1: 50, expected: false},
+		{v0: 0, v1: 51, expected: false},
+		{v0: 0, v1: 52, expected: false},
+		{v0: 0, v1: 53, expected: false},
+		{v0: 0, v1: 54, expected: false},
+		{v0: 0, v1: 55, expected: false},
+		{v0: 0, v1: 56, expected: false},
+		{v0: 0, v1: 57, expected: false},
+		{v0: 0, v1: 58, expected: false},
+		{v0: 0, v1: 59, expected: false},
+		{v0: 0, v1: 60, expected: false},
+		{v0: 0, v1: 61, expected: false},
+		{v0: 0, v1: 62, expected: false},
+		{v0: 0, v1: 63, expected: false},
+		{v0: 0, v1: 64, expected: false},
+		{v0: 0, v1: 65, expected: false},
+		{v0: 0, v1: 66, expected: false},
+		{v0: 0, v1: 67, expected: false},
+		{v0: 0, v1: 68, expected: false},
+		{v0: 0, v1: 69, expected: false},
+		{v0: 0, v1: 70, expected: false},
+		{v0: 0, v1: 71, expected: false},
+		{v0: 0, v1: 72, expected: false},
+		{v0: 0, v1: 73, expected: false},
+		{v0: 0, v1: 74, expected: false},
+		{v0: 0, v1: 75, expected: false},
+		{v0: 0, v1: 76, expected: false},
+		{v0: 0, v1: 77, expected: false},
+		{v0: 0, v1: 78, expected: false},
+		{v0: 0, v1: 79, expected: false},
+		{v0: 0, v1: 80, expected: false},
+		{v0: 0, v1: 81, expected: false},
+		{v0: 0, v1: 82, expected: false},
+		{v0: 0, v1: 83, expected: false},
+		{v0: 0, v1: 84, expected: false},
+		{v0: 0, v1: 85, expected: false},
+		{v0: 0, v1: 86, expected: false},
+		{v0: 0, v1: 87, expected: false},
+		{v0: 0, v1: 88, expected: false},
+		{v0: 0, v1: 89, expected: false},
+		{v0: 0, v1: 90, expected: false},
+		{v0: 0, v1: 91, expected: false},
+		{v0: 0, v1: 92, expected: false},
+		{v0: 0, v1: 93, expected: false},
+		{v0: 0, v1: 94, expected: false},
+		{v0: 0, v1: 95, expected: false},
+		{v0: 0, v1: 96, expected: false},
+		{v0: 0, v1: 97, expected: false},
+		{v0: 0, v1: 98, expected: false},
+		{v0: 0, v1: 99, expected: false},
+		{v0: 0, v1: 100, expected: true},
+		{v0: 0, v1: 101, expected: false},
+		{v0: 0, v1: 102, expected: false},
+		{v0: 0, v1: 103, expected: false},
+		{v0: 0, v1: 104, expected: false},
+		{v0: 0, v1: 105, expected: false},
+		{v0: 0, v1: 106, expected: false},
+		{v0: 0, v1: 107, expected: false},
+		{v0: 0, v1: 108, expected: false},
+		{v0: 0, v1: 109, expected: false},
+		{v0: 0, v1: 110, expected: false},
+		{v0: 0, v1: 111, expected: false},
+		{v0: 0, v1: 112, expected: false},
+		{v0: 0, v1: 113, expected: false},
+		{v0: 0, v1: 114, expected: false},
+		{v0: 0, v1: 115, expected: false},
+		{v0: 0, v1: 116, expected: false},
+		{v0: 0, v1: 117, expected: false},
+		{v0: 0, v1: 118, expected: false},
+		{v0: 0, v1: 119, expected: false},
+		{v0: 0, v1: 120, expected: false},
+		{v0: 0, v1: 121, expected: false},
+		{v0: 0, v1: 122, expected: false},
+		{v0: 0, v1: 123, expected: false},
+		{v0: 0, v1: 124, expected: false},
+		{v0: 0, v1: 125, expected: false},
+		{v0: 0, v1: 126, expected: false},
+		{v0: 0, v1: 127, expected: false},
+		{v0: 0, v1: 128, expected: false},
+		{v0: 0, v1: 129, expected: false},
+		{v0: 0, v1: 130, expected: false},
+		{v0: 0, v1: 131, expected: false},
+		{v0: 0, v1: 132, expected: false},
+		{v0: 0, v1: 133, expected: false},
+		{v0: 0, v1: 134, expected: false},
+		{v0: 0, v1: 135, expected: false},
+		{v0: 0, v1: 136, expected: false},
+		{v0: 0, v1: 137, expected: false},
+		{v0: 0, v1: 138, expected: false},
+		{v0: 0, v1: 139, expected: false},
+		{v0: 0, v1: 140, expected: false},
+		{v0: 0, v1: 141, expected: false},
+		{v0: 0, v1: 142, expected: false},
+		{v0: 0, v1: 143, expected: false},
+		{v0: 0, v1: 144, expected: false},
+		{v0: 0, v1: 145, expected: false},
+		{v0: 0, v1: 146, expected: false},
+		{v0: 0, v1: 147, expected: false},
+		{v0: 0, v1: 148, expected: false},
+		{v0: 0, v1: 149, expected: false},
+		{v0: 0, v1: 150, expected: false},
+		{v0: 0, v1: 151, expected: false},
+		{v0: 0, v1: 152, expected: false},
+		{v0: 0, v1: 153, expected: false},
+		{v0: 0, v1: 154, expected: false},
+		{v0: 0, v1: 155, expected: false},
+		{v0: 0, v1: 156, expected: false},
+		{v0: 0, v1: 157, expected: false},
+		{v0: 0, v1: 158, expected: false},
+		{v0: 0, v1: 159, expected: false},
+		{v0: 0, v1: 160, expected: false},
+		{v0: 0, v1: 161, expected: false},
+		{v0: 0, v1: 162, expected: false},
+		{v0: 0, v1: 163, expected: false},
+		{v0: 0, v1: 164, expected: false},
+		{v0: 0, v1: 165, expected: false},
+		{v0: 0, v1: 166, expected: false},
+		{v0: 0, v1: 167, expected: false},
+		{v0: 0, v1: 168, expected: false},
+		{v0: 0, v1: 169, expected: false},
+		{v0: 0, v1: 170, expected: false},
+		{v0: 0, v1: 171, expected: false},
+		{v0: 0, v1: 172, expected: false},
+		{v0: 0, v1: 173, expected: false},
+		{v0: 0, v1: 174, expected: false},
+		{v0: 0, v1: 175, expected: false},
+		{v0: 0, v1: 176, expected: false},
+		{v0: 0, v1: 177, expected: false},
+		{v0: 0, v1: 178, expected: false},
+		{v0: 0, v1: 179, expected: false},
+		{v0: 0, v1: 180, expected: false},
+		{v0: 0, v1: 181, expected: false},
+		{v0: 0, v1: 182, expected: false},
+		{v0: 0, v1: 183, expected: false},
+		{v0: 0, v1: 184, expected: false},
+		{v0: 0, v1: 185, expected: false},
+		{v0: 0, v1: 186, expected: false},
+		{v0: 0, v1: 187, expected: false},
+		{v0: 0, v1: 188, expected: false},
+		{v0: 0, v1: 189, expected: false},
+		{v0: 0, v1: 190, expected: false},
+		{v0: 0, v1: 191, expected: false},
+		{v0: 0, v1: 192, expected: false},
+		{v0: 0, v1: 193, expected: false},
+		{v0: 0, v1: 194, expected: false},
+		{v0: 0, v1: 195, expected: false},
+		{v0: 0, v1: 196, expected: false},
+		{v0: 0, v1: 197, expected: false},
+		{v0: 0, v1: 198, expected: false},
+		{v0: 0, v1: 199, expected: false},
+		{v0: 0, v1: 200, expected: true},
+	}
+
+	m := Metadata{
+		failoverVersionIncrement: failoverVersionIncrement,
+		allClusters: map[string]config.ClusterInformation{
+			clusterName1: {
+				InitialFailoverVersion: initialFailoverVersionC1,
+			},
+			clusterName2: {
+				InitialFailoverVersion: initialFailoverVersionC2,
+			},
+		},
+		versionToClusterName: map[int64]string{
+			initialFailoverVersionC1: clusterName1,
+			initialFailoverVersionC2: clusterName2,
+		},
+		metrics:                       metrics.NewNoopMetricsClient().Scope(0),
+		useNewFailoverVersionOverride: func(domain string) bool { return false },
+		log:                           loggerimpl.NewNopLogger(),
+	}
+
+	for i := range tests {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+			assert.Equal(t, tests[i].expected, m.IsVersionFromSameCluster(tests[i].v0, tests[i].v1))
+		})
+	}
+}
+
+func TestServerResolution(t *testing.T) {
+
+	const clusterName1 = "c1"
+	const initialFailoverVersionC1 = 0
+	const clusterName2 = "c2"
+	const initialFailoverVersionC2 = 2
+	const failoverVersionIncrement = 100
+	const domainToMigrate = "some domain"
+
+	allClusters := map[string]config.ClusterInformation{
+		clusterName1: {
+			InitialFailoverVersion: initialFailoverVersionC1,
+		},
+		clusterName2: {
+			InitialFailoverVersion: initialFailoverVersionC2,
+		},
+	}
+
+	impl := Metadata{
+		failoverVersionIncrement: failoverVersionIncrement,
+		allClusters:              allClusters,
+		versionToClusterName: map[int64]string{
+			initialFailoverVersionC1: clusterName1,
+			initialFailoverVersionC2: clusterName2,
+		},
+		metrics:                       metrics.NewNoopMetricsClient().Scope(0),
+		useNewFailoverVersionOverride: func(domain string) bool { return domain == domainToMigrate },
+		log:                           loggerimpl.NewNopLogger(),
+	}
+
+	err := quick.Check(func(currentFOVersion int64, migrateDomain bool) bool {
+		var nextFailoverVersion int64
+		if migrateDomain {
+			fo := impl.GetNextFailoverVersion(clusterName1, currentFOVersion, domainToMigrate)
+			nextFailoverVersion = fo
+		} else {
+			fo := impl.GetNextFailoverVersion(clusterName1, currentFOVersion, "a different domain")
+			nextFailoverVersion = fo
+		}
+		// do a round-trip
+		clusterNameResolved := impl.ClusterNameForFailoverVersion(nextFailoverVersion)
+		return clusterName1 == clusterNameResolved
+	}, &quick.Config{})
+	assert.NoError(t, err)
+}
+
+// the point of this test is to assert that there's no clumsy errors, and
+// in an unmigrated state, the old implementation for getting versions and the new
+// one are equal
+func TestNoChangesInUnmigratedState(t *testing.T) {
+
+	const clusterName1 = "c1"
+	const initialFailoverVersionC1 = 0
+	const clusterName2 = "c2"
+	const initialFailoverVersionC2 = 2
+	const failoverVersionIncrement = 100
+
+	allClusters := map[string]config.ClusterInformation{
+		clusterName1: {
+			InitialFailoverVersion: initialFailoverVersionC1,
+		},
+		clusterName2: {
+			InitialFailoverVersion: initialFailoverVersionC2,
+		},
+	}
+
+	oldImpl := func(cluster string, currentFailoverVersion int64) int64 {
+		info, ok := allClusters[cluster]
+		if !ok {
+			panic(fmt.Sprintf(
+				"Unknown cluster name: %v with given cluster initial failover version map: %v.",
+				cluster,
+				allClusters,
+			))
+		}
+		failoverVersion := currentFailoverVersion/failoverVersionIncrement*failoverVersionIncrement + info.InitialFailoverVersion
+		if failoverVersion < currentFailoverVersion {
+			return failoverVersion + failoverVersionIncrement
+		}
+		return failoverVersion
+	}
+
+	newImpl := Metadata{
+		failoverVersionIncrement: failoverVersionIncrement,
+		allClusters:              allClusters,
+		versionToClusterName: map[int64]string{
+			initialFailoverVersionC1: clusterName1,
+			initialFailoverVersionC2: clusterName2,
+		},
+		metrics:                       metrics.NewNoopMetricsClient().Scope(0),
+		useNewFailoverVersionOverride: func(domain string) bool { return false },
+		log:                           loggerimpl.NewNopLogger(),
+	}
+
+	err := quick.CheckEqual(func(currVersion int64) int64 {
+		// partially apply the cluster-name, since the fuzzer trying to guess that is pointless
+		return oldImpl(clusterName2, currVersion)
+	}, func(currentVersion int64) int64 {
+		return newImpl.GetNextFailoverVersion(clusterName2, currentVersion, "some domain") // all domains are turned off
+	},
+		&quick.Config{})
+	assert.NoError(t, err)
+}
+
+// the point of this test is to assert that there's no clumsy errors, and
+// in an unmigrated state, the old implementation for getting versions and the new
+// one are equal
+func TestFailoverVersionResolution(t *testing.T) {
+
+	const clusterName1 = "c1"
+	const initialFailoverVersionC1 = 0
+	var newFailoverVersionC1 = int64(1)
+	const clusterName2 = "c2"
+	const initialFailoverVersionC2 = 2
+	const failoverVersionIncrement = 100
+
+	allClusters := map[string]config.ClusterInformation{
+		clusterName1: {
+			InitialFailoverVersion:    initialFailoverVersionC1,
+			NewInitialFailoverVersion: &newFailoverVersionC1,
+		},
+		clusterName2: {
+			InitialFailoverVersion: initialFailoverVersionC2,
+		},
+	}
+
+	sut := Metadata{
+		failoverVersionIncrement: failoverVersionIncrement,
+		allClusters:              allClusters,
+		versionToClusterName: map[int64]string{
+			initialFailoverVersionC1: clusterName1,
+			initialFailoverVersionC2: clusterName2,
+		},
+		metrics:                       metrics.NewNoopMetricsClient().Scope(0),
+		useNewFailoverVersionOverride: func(domain string) bool { return false },
+		log:                           loggerimpl.NewNopLogger(),
+	}
+
+	tests := map[string]struct {
+		in          int64
+		expectedOut string
+		expectedErr error
+	}{
+		"normal failover version": {
+			in:          initialFailoverVersionC1,
+			expectedOut: clusterName1,
+		},
+		"New failover version": {
+			in:          initialFailoverVersionC1,
+			expectedOut: clusterName1,
+		},
+	}
+
+	for name, td := range tests {
+		t.Run(name, func(t *testing.T) {
+			assert.Equal(t, td.expectedOut, sut.ClusterNameForFailoverVersion(td.in))
+		})
+	}
+}
diff --git a/common/domain/handler_MasterCluster_test.go b/common/domain/handler_MasterCluster_test.go
index 73e657ca5..ca2bea0d7 100644
--- a/common/domain/handler_MasterCluster_test.go
+++ b/common/domain/handler_MasterCluster_test.go
@@ -507,7 +507,7 @@ func (s *domainHandlerGlobalDomainEnabledPrimaryClusterSuite) TestRegisterGetDom
 		ActiveClusterName: s.ClusterMetadata.GetCurrentClusterName(),
 		Clusters:          clusters,
 	}, resp.ReplicationConfiguration)
-	s.Equal(s.ClusterMetadata.GetNextFailoverVersion(s.ClusterMetadata.GetCurrentClusterName(), 0), resp.GetFailoverVersion())
+	s.Equal(s.ClusterMetadata.GetNextFailoverVersion(s.ClusterMetadata.GetCurrentClusterName(), 0, "some-domaain"), resp.GetFailoverVersion())
 	s.Equal(isGlobalDomain, resp.GetIsGlobalDomain())
 }
 
@@ -575,7 +575,7 @@ func (s *domainHandlerGlobalDomainEnabledPrimaryClusterSuite) TestRegisterGetDom
 		ActiveClusterName: activeClusterName,
 		Clusters:          clusters,
 	}, resp.ReplicationConfiguration)
-	s.Equal(s.ClusterMetadata.GetNextFailoverVersion(activeClusterName, 0), resp.GetFailoverVersion())
+	s.Equal(s.ClusterMetadata.GetNextFailoverVersion(activeClusterName, 0, "some-domain"), resp.GetFailoverVersion())
 	s.Equal(isGlobalDomain, resp.GetIsGlobalDomain())
 }
 
@@ -640,7 +640,7 @@ func (s *domainHandlerGlobalDomainEnabledPrimaryClusterSuite) TestUpdateGetDomai
 			ActiveClusterName: activeClusterName,
 			Clusters:          clusters,
 		}, replicationConfig)
-		s.Equal(s.ClusterMetadata.GetNextFailoverVersion(activeClusterName, 0), failoverVersion)
+		s.Equal(s.ClusterMetadata.GetNextFailoverVersion(activeClusterName, 0, "some-domain"), failoverVersion)
 		s.Equal(isGlobalDomain, isGlobalDomain)
 	}
 
@@ -727,7 +727,7 @@ func (s *domainHandlerGlobalDomainEnabledPrimaryClusterSuite) TestUpdateGetDomai
 			ActiveClusterName: activeClusterName,
 			Clusters:          clusters,
 		}, replicationConfig)
-		s.Equal(s.ClusterMetadata.GetNextFailoverVersion(activeClusterName, 0), failoverVersion)
+		s.Equal(s.ClusterMetadata.GetNextFailoverVersion(activeClusterName, 0, "some-domain"), failoverVersion)
 		s.Equal(isGlobalDomain, isGlobalDomain)
 	}
 
@@ -832,8 +832,8 @@ func (s *domainHandlerGlobalDomainEnabledPrimaryClusterSuite) TestUpdateGetDomai
 		}, replicationConfig)
 		s.Equal(s.ClusterMetadata.GetNextFailoverVersion(
 			nextActiveClusterName,
-			s.ClusterMetadata.GetNextFailoverVersion(prevActiveClusterName, 0),
-		), failoverVersion)
+			s.ClusterMetadata.GetNextFailoverVersion(prevActiveClusterName, 0, "some-domain"),
+			"some-domain"), failoverVersion)
 		s.Equal(isGlobalDomain, isGlobalDomain)
 	}
 
@@ -927,7 +927,7 @@ func (s *domainHandlerGlobalDomainEnabledPrimaryClusterSuite) TestUpdateDomain_C
 		ActiveClusterName: s.ClusterMetadata.GetCurrentClusterName(),
 		Clusters:          clusters,
 	}, resp.ReplicationConfiguration)
-	s.Equal(s.ClusterMetadata.GetNextFailoverVersion(s.ClusterMetadata.GetCurrentClusterName(), 0), resp.GetFailoverVersion())
+	s.Equal(s.ClusterMetadata.GetNextFailoverVersion(s.ClusterMetadata.GetCurrentClusterName(), 0, "some-domain"), resp.GetFailoverVersion())
 	s.Equal(isGlobalDomain, resp.GetIsGlobalDomain())
 
 	_, err = s.handler.UpdateDomain(context.Background(), &types.UpdateDomainRequest{
diff --git a/common/domain/handler_NotMasterCluster_test.go b/common/domain/handler_NotMasterCluster_test.go
index 3ea96478f..4f291416c 100644
--- a/common/domain/handler_NotMasterCluster_test.go
+++ b/common/domain/handler_NotMasterCluster_test.go
@@ -539,7 +539,7 @@ func (s *domainHandlerGlobalDomainEnabledNotPrimaryClusterSuite) TestUpdateGetDo
 		},
 		IsGlobalDomain:  isGlobalDomain,
 		ConfigVersion:   0,
-		FailoverVersion: s.ClusterMetadata.GetNextFailoverVersion(activeClusterName, 0),
+		FailoverVersion: s.ClusterMetadata.GetNextFailoverVersion(activeClusterName, 0, "some-domain"),
 	})
 	s.Nil(err)
 
@@ -599,7 +599,7 @@ func (s *domainHandlerGlobalDomainEnabledNotPrimaryClusterSuite) TestUpdateGetDo
 		},
 		IsGlobalDomain:  isGlobalDomain,
 		ConfigVersion:   0,
-		FailoverVersion: s.ClusterMetadata.GetNextFailoverVersion(activeClusterName, 0),
+		FailoverVersion: s.ClusterMetadata.GetNextFailoverVersion(activeClusterName, 0, "some-domain"),
 	})
 	s.Nil(err)
 
@@ -672,7 +672,7 @@ func (s *domainHandlerGlobalDomainEnabledNotPrimaryClusterSuite) TestUpdateGetDo
 		},
 		IsGlobalDomain:  isGlobalDomain,
 		ConfigVersion:   0,
-		FailoverVersion: s.ClusterMetadata.GetNextFailoverVersion(prevActiveClusterName, 0),
+		FailoverVersion: s.ClusterMetadata.GetNextFailoverVersion(prevActiveClusterName, 0, "some-domain"),
 	})
 	s.Nil(err)
 
@@ -703,8 +703,8 @@ func (s *domainHandlerGlobalDomainEnabledNotPrimaryClusterSuite) TestUpdateGetDo
 		}, replicationConfig)
 		s.Equal(s.ClusterMetadata.GetNextFailoverVersion(
 			nextActiveClusterName,
-			s.ClusterMetadata.GetNextFailoverVersion(prevActiveClusterName, 0),
-		), failoverVersion)
+			s.ClusterMetadata.GetNextFailoverVersion(prevActiveClusterName, 0, "some-domain"),
+			"some-domain"), failoverVersion)
 		s.Equal(isGlobalDomain, isGlobalDomain)
 	}
 
@@ -801,7 +801,7 @@ func setupGlobalDomainWithMetadataManager(s suite.Suite, handler *handlerImpl, c
 		},
 		IsGlobalDomain:  isGlobalDomain,
 		ConfigVersion:   0,
-		FailoverVersion: clusterMetadata.GetNextFailoverVersion(activeClusterName, 0),
+		FailoverVersion: clusterMetadata.GetNextFailoverVersion(activeClusterName, 0, "some-domain"),
 	})
 	s.Nil(err)
 
diff --git a/common/domain/handler_test.go b/common/domain/handler_test.go
index ae077f50c..47606a509 100644
--- a/common/domain/handler_test.go
+++ b/common/domain/handler_test.go
@@ -386,7 +386,7 @@ func (s *domainHandlerCommonSuite) TestListDomain() {
 				ActiveClusterName: activeClusterName2,
 				Clusters:          cluster2,
 			},
-			FailoverVersion: s.ClusterMetadata.GetNextFailoverVersion(activeClusterName2, 0),
+			FailoverVersion: s.ClusterMetadata.GetNextFailoverVersion(activeClusterName2, 0, "some-domain"),
 			IsGlobalDomain:  isGlobalDomain2,
 		},
 	}, domains)
diff --git a/host/testcluster.go b/host/testcluster.go
index a109b96cb..8415a211a 100644
--- a/host/testcluster.go
+++ b/host/testcluster.go
@@ -27,6 +27,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/uber/cadence/common/log/loggerimpl"
+
 	"github.com/uber-go/tally"
 
 	adminClient "github.com/uber/cadence/client/admin"
@@ -186,6 +188,9 @@ func NewClusterMetadata(options *TestClusterConfig) cluster.Metadata {
 			options.ClusterGroupMetadata.PrimaryClusterName,
 			options.ClusterGroupMetadata.CurrentClusterName,
 			options.ClusterGroupMetadata.ClusterGroup,
+			func(domain string) bool { return false },
+			metrics.NewNoopMetricsClient(),
+			loggerimpl.NewNopLogger(),
 		)
 	}
 	return clusterMetadata
diff --git a/service/history/replication/metrics_emitter_test.go b/service/history/replication/metrics_emitter_test.go
index f8721d44f..3ea00746d 100644
--- a/service/history/replication/metrics_emitter_test.go
+++ b/service/history/replication/metrics_emitter_test.go
@@ -26,6 +26,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/uber/cadence/common/log/loggerimpl"
+
 	"github.com/stretchr/testify/assert"
 
 	"github.com/uber/cadence/common/clock"
@@ -48,7 +50,11 @@ func TestMetricsEmitter(t *testing.T) {
 		cluster1: {Enabled: true},
 		cluster2: {Enabled: true},
 		cluster3: {Enabled: true},
-	})
+	},
+		func(d string) bool { return false },
+		metrics.NewNoopMetricsClient(),
+		loggerimpl.NewNopLogger(),
+	)
 	testShardData := newTestShardData(timeSource, metadata)
 	timeSource.Update(time.Unix(10000, 0))
 
diff --git a/service/history/replication/task_store_test.go b/service/history/replication/task_store_test.go
index ff7abfce9..1f7716e96 100644
--- a/service/history/replication/task_store_test.go
+++ b/service/history/replication/task_store_test.go
@@ -26,6 +26,8 @@ import (
 	"context"
 	"testing"
 
+	"github.com/uber/cadence/common/log/loggerimpl"
+
 	"github.com/stretchr/testify/assert"
 
 	"github.com/uber/cadence/common/cache"
@@ -153,7 +155,11 @@ func createTestTaskStore(domains domainCache, hydrator taskHydrator) *TaskStore
 		testClusterA: {Enabled: true},
 		testClusterB: {Enabled: true},
 		testClusterC: {Enabled: true},
-	})
+	},
+		func(d string) bool { return false },
+		metrics.NewNoopMetricsClient(),
+		loggerimpl.NewNopLogger(),
+	)
 
 	return NewTaskStore(&cfg, clusterMetadata, domains, metrics.NewNoopMetricsClient(), log.NewNoop(), hydrator)
 }