Repository: FoundationDB/fdb-kubernetes-operator
Pull request #1368 (2022-10-03 13:01:12)

Initial doc for replacement buckets
# Description

Fixes: https://github.com/FoundationDB/fdb-kubernetes-operator/issues/1316

## Type of change


- New feature (non-breaking change which adds functionality)

## Discussion

-

## Testing

-

## Documentation

-

## Follow-up

I will write another design doc for allowing replacements on a fault domain basis.

johscheuer: CC @saintstack @RenxuanW please provide feedback on this idea.
foundationdb-ci: 
### Result of fdb-kubernetes-operator-pr on Linux CentOS 7

* Commit ID: 89cb2a32e2530d84a215ce6f20c2a70b416fc9b4
* Duration 2:16:48
* Result: :x: **FAILED**
* Error: `Error while executing command: if $(grep -q -- "--- FAIL:" logs/*.log); then echo "TESTS FAILED SEE THESE LOGS:"; echo ; grep -l -- "--- FAIL:" logs/*.log; exit 1; fi. Reason: exit status 1`
* [Build Logs](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator/fdb-kubernetes-operator-pr/150b69d9-f8ca-42a3-9c29-1d91006f3d82.gz) (available for 30 days)
* [Build Artifact](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator-pr/150b69d9-f8ca-42a3-9c29-1d91006f3d82/workspace.zip) (available for 30 days)

foundationdb-ci: 
### Result of fdb-kubernetes-operator-pr on Linux CentOS 7

* Commit ID: f0caed38f76d6ee6261e8755c676c93f4faf2be8
* Duration 2:47:13
* Result: :white_check_mark: **SUCCEEDED**
* Error: `N/A`
* [Build Log](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator/fdb-kubernetes-operator-pr/d505645c-5bc1-4a5a-af8a-4dbb735f5b2c.gz) terminal output (available for 30 days)
* [Build Workspace](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator-pr/d505645c-5bc1-4a5a-af8a-4dbb735f5b2c/workspace.zip) zip file of the working directory (available for 30 days)


Patch:
diff --git a/api/v1beta2/foundationdb_database_configuration.go b/api/v1beta2/foundationdb_database_configuration.go
index fbf94708..103edc71 100644
--- a/api/v1beta2/foundationdb_database_configuration.go
+++ b/api/v1beta2/foundationdb_database_configuration.go
@@ -33,7 +33,7 @@ import (
 type DatabaseConfiguration struct {
 	// RedundancyMode defines the core replication factor for the database.
 	// +kubebuilder:validation:Optional
-	// +kubebuilder:validation:Enum=single;double;triple
+	// +kubebuilder:validation:Enum=single;double;triple;three_data_hall
 	// +kubebuilder:default:double
 	RedundancyMode RedundancyMode `json:"redundancy_mode,omitempty"`
 
@@ -211,6 +211,19 @@ func (configuration DatabaseConfiguration) NormalizeConfiguration() DatabaseConf
 	return *result
 }
 
+// CountUniqueDataCenters returns the number of unique data centers based on the desired DatabaseConfiguration.
+func (configuration *DatabaseConfiguration) CountUniqueDataCenters() int {
+	uniqueDataCenters := map[string]None{}
+
+	for _, region := range configuration.Regions {
+		for _, dc := range region.DataCenters {
+			uniqueDataCenters[dc.ID] = None{}
+		}
+	}
+
+	return len(uniqueDataCenters)
+}
+
 // NormalizeConfigurationWithSeparatedProxies ensures a standardized
 // format and defaults when comparing database configuration in the
 // cluster spec with database configuration in the cluster status,
@@ -284,7 +297,7 @@ func (configuration *DatabaseConfiguration) GetRoleCountsWithDefaults(version Ve
 		counts.Storage = 2*faultTolerance + 1
 	}
 	if counts.Logs == 0 {
-		counts.Logs = 3
+		counts.Logs = configuration.RedundancyMode.getDefaultLogCount()
 	}
 
 	if version.HasSeparatedProxies() {
@@ -597,9 +610,11 @@ func (configuration DatabaseConfiguration) FillInDefaultsFromStatus() DatabaseCo
 	if result.RemoteLogs == 0 {
 		result.RemoteLogs = -1
 	}
+
 	if result.LogRouters == 0 {
 		result.LogRouters = -1
 	}
+
 	return *result
 }
 
@@ -678,7 +693,7 @@ func DesiredFaultTolerance(redundancyMode RedundancyMode) int {
 		return 0
 	case RedundancyModeDouble, RedundancyModeUnset:
 		return 1
-	case RedundancyModeTriple:
+	case RedundancyModeTriple, RedundancyModeThreeDataHall:
 		return 2
 	default:
 		return 0
@@ -694,6 +709,8 @@ func MinimumFaultDomains(redundancyMode RedundancyMode) int {
 		return 2
 	case RedundancyModeTriple:
 		return 3
+	case RedundancyModeThreeDataHall:
+		return 4
 	default:
 		return 1
 	}
@@ -710,6 +727,8 @@ const (
 	RedundancyModeDouble RedundancyMode = "double"
 	// RedundancyModeTriple defines the replication factor 3.
 	RedundancyModeTriple RedundancyMode = "triple"
+	// RedundancyModeThreeDataHall defines the replication factor three_data_hall.
+	RedundancyModeThreeDataHall RedundancyMode = "three_data_hall"
 	// RedundancyModeOneSatelliteSingle defines the replication factor one_satellite_single.
 	RedundancyModeOneSatelliteSingle RedundancyMode = "one_satellite_single"
 	// RedundancyModeOneSatelliteDouble  defines the replication factor one_satellite_double.
@@ -718,6 +737,15 @@ const (
 	RedundancyModeUnset RedundancyMode = ""
 )
 
+// getDefaultLogCount returns the default log count for the provided redundancy mode
+func (redundancyMode RedundancyMode) getDefaultLogCount() int {
+	if redundancyMode == RedundancyModeThreeDataHall {
+		return 4
+	}
+
+	return 3
+}
+
 // StorageEngine defines the storage engine for the database
 // +kubebuilder:validation:MaxLength=100
 type StorageEngine string
diff --git a/api/v1beta2/foundationdb_labels.go b/api/v1beta2/foundationdb_labels.go
index 21a4386d..98ddac4d 100644
--- a/api/v1beta2/foundationdb_labels.go
+++ b/api/v1beta2/foundationdb_labels.go
@@ -72,6 +72,10 @@ const (
 	// the zone ID.
 	FDBLocalityZoneIDKey = "zoneid"
 
+	// FDBLocalityMachineIDKey represents the key in the locality map that holds
+	// the machine ID.
+	FDBLocalityMachineIDKey = "machineid"
+
 	// FDBLocalityDCIDKey represents the key in the locality map that holds
 	// the DC ID.
 	FDBLocalityDCIDKey = "dcid"
@@ -86,4 +90,12 @@ const (
 
 	// FDBLocalityExclusionPrefix represents the exclusion prefix for locality based exclusions.
 	FDBLocalityExclusionPrefix = "locality_instance_id"
+
+	// FDBLocalityDataHallKey represents the key in the locality map that holds
+	// the data hall.
+	FDBLocalityDataHallKey = "data_hall"
+
+	// FDBLocalityDCIDlKey represents the key in the locality map that holds
+	// the data center ID.
+	FDBLocalityDCIDlKey = "dcid"
 )
diff --git a/api/v1beta2/foundationdb_version.go b/api/v1beta2/foundationdb_version.go
index 11e41643..b68aff8a 100644
--- a/api/v1beta2/foundationdb_version.go
+++ b/api/v1beta2/foundationdb_version.go
@@ -214,6 +214,11 @@ func (version Version) SupportsDNSInClusterFile() bool {
 	return version.IsAtLeast(Versions.SupportsDNSInClusterFile)
 }
 
+// SupportsVersionChange returns true if the current version can be downgraded or upgraded to provided other version.
+func (version Version) SupportsVersionChange(other Version) bool {
+	return version.IsProtocolCompatible(other) || other.IsAtLeast(version)
+}
+
 // Versions provides a shorthand for known versions.
 // This is only to be used in testing.
 var Versions = struct {
diff --git a/api/v1beta2/foundationdbcluster_types.go b/api/v1beta2/foundationdbcluster_types.go
index 53117790..5ec6b307 100644
--- a/api/v1beta2/foundationdbcluster_types.go
+++ b/api/v1beta2/foundationdbcluster_types.go
@@ -480,15 +480,19 @@ func (processGroupStatus *ProcessGroupStatus) GetPodName(cluster *FoundationDBCl
 	return sb.String()
 }
 
-// NeedsReplacement checks if the ProcessGroupStatus has conditions so that it should be removed
-func (processGroupStatus *ProcessGroupStatus) NeedsReplacement(failureTime int, taintReplacementTime int) (bool, int64) {
+// NeedsReplacement checks if the ProcessGroupStatus has conditions that require a replacement of the failed Process Group.
+// The method will return the failure condition and the timestamp. If no failure is detected an empty condition and a 0
+// will be returned.
+func (processGroupStatus *ProcessGroupStatus) NeedsReplacement(failureTime int, taintReplacementTime int) (ProcessGroupConditionType, int64) {
 	var earliestFailureTime int64 = math.MaxInt64
 	var earliestTaintReplacementTime int64 = math.MaxInt64
 
+	// If the process group is already marked for removal we can ignore it.
 	if processGroupStatus.IsMarkedForRemoval() {
-		return false, 0
+		return "", 0
 	}
 
+	var failureCondition ProcessGroupConditionType
 	for _, conditionType := range conditionsThatNeedReplacement {
 		conditionTimePtr := processGroupStatus.GetConditionTime(conditionType)
 		if conditionTimePtr == nil {
@@ -500,23 +504,29 @@ func (processGroupStatus *ProcessGroupStatus) NeedsReplacement(failureTime int,
 			if earliestTaintReplacementTime > conditionTime {
 				earliestTaintReplacementTime = conditionTime
 			}
-		} else {
-			if earliestFailureTime > conditionTime {
-				earliestFailureTime = conditionTime
-			}
+
+			failureCondition = conditionType
+			continue
+		}
+
+		if earliestFailureTime > conditionTime {
+			earliestFailureTime = conditionTime
+			failureCondition = conditionType
 		}
 	}
 
 	failureWindowStart := time.Now().Add(-1 * time.Duration(failureTime) * time.Second).Unix()
 	if earliestFailureTime < failureWindowStart {
-		return true, earliestFailureTime
+		return failureCondition, earliestFailureTime
 	}
+
 	taintWindowStart := time.Now().Add(-1 * time.Duration(taintReplacementTime) * time.Second).Unix()
 	if earliestTaintReplacementTime < taintWindowStart {
-		return true, earliestTaintReplacementTime
+		return failureCondition, earliestTaintReplacementTime
 	}
 
-	return false, 0
+	// No failure detected.
+	return "", 0
 }
 
 // AddAddresses adds the new address to the ProcessGroupStatus and removes duplicates and old addresses
@@ -809,6 +819,11 @@ func (processGroupStatus *ProcessGroupStatus) GetConditionTime(conditionType Pro
 
 // IsUnderMaintenance checks if the process is in maintenance zone.
 func (processGroupStatus *ProcessGroupStatus) IsUnderMaintenance(maintenanceZone FaultDomain) bool {
+	// Only storage processes are affected by the maintenance zone.
+	if processGroupStatus.ProcessClass != ProcessClassStorage {
+		return false
+	}
+
 	// If the maintenanceZone is not set or the Process Group has not fault domain set, return false.
 	if maintenanceZone == "" || processGroupStatus.FaultDomain == "" {
 		return false
@@ -1253,7 +1268,7 @@ func (cluster *FoundationDBCluster) GetProcessSettings(processClass ProcessClass
 // The default Storage value will be 2F + 1, where F is the cluster's fault
 // tolerance.
 //
-// The default Logs value will be 3.
+// The default Logs value will be 3 or 4 for three_data_hall.
 //
 // The default Proxies value will be 3.
 //
@@ -1415,10 +1430,9 @@ func (cluster *FoundationDBCluster) MinimumFaultDomains() int {
 	return MinimumFaultDomains(cluster.Spec.DatabaseConfiguration.RedundancyMode)
 }
 
-// DesiredCoordinatorCount returns the number of coordinators to recruit for
-// a cluster.
+// DesiredCoordinatorCount returns the number of coordinators to recruit for a cluster.
 func (cluster *FoundationDBCluster) DesiredCoordinatorCount() int {
-	if cluster.Spec.DatabaseConfiguration.UsableRegions > 1 {
+	if cluster.Spec.DatabaseConfiguration.UsableRegions > 1 || cluster.Spec.DatabaseConfiguration.RedundancyMode == RedundancyModeThreeDataHall {
 		return 9
 	}
 
@@ -2730,6 +2744,10 @@ func (cluster *FoundationDBCluster) Validate() error {
 		return err
 	}
 
+	if !version.IsSupported() {
+		return fmt.Errorf("version: %s is not supported, minimum supported version is: %s", version.String(), Versions.MinimumVersion.String())
+	}
+
 	if !version.IsStorageEngineSupported(cluster.Spec.DatabaseConfiguration.StorageEngine) {
 		validations = append(validations, fmt.Sprintf("storage engine %s is not supported on version %s", cluster.Spec.DatabaseConfiguration.StorageEngine, cluster.Spec.Version))
 	}
diff --git a/config/crd/bases/apps.foundationdb.org_foundationdbclusters.yaml b/config/crd/bases/apps.foundationdb.org_foundationdbclusters.yaml
index 46a0f5af..7c2371cb 100644
--- a/config/crd/bases/apps.foundationdb.org_foundationdbclusters.yaml
+++ b/config/crd/bases/apps.foundationdb.org_foundationdbclusters.yaml
@@ -10355,6 +10355,7 @@ spec:
                     - single
                     - double
                     - triple
+                    - three_data_hall
                     maxLength: 100
                     type: string
                   regions:
@@ -14090,6 +14091,7 @@ spec:
                     - single
                     - double
                     - triple
+                    - three_data_hall
                     maxLength: 100
                     type: string
                   regions:
diff --git a/controllers/change_coordinators.go b/controllers/change_coordinators.go
index e37ec8f7..4961af48 100644
--- a/controllers/change_coordinators.go
+++ b/controllers/change_coordinators.go
@@ -161,7 +161,7 @@ func selectCoordinators(logger logr.Logger, cluster *fdbv1beta2.FoundationDBClus
 		HardLimits: locality.GetHardLimits(cluster),
 	})
 
-	logger.Info("Current coordinators", "coordinators", coordinators)
+	logger.Info("Current coordinators", "coordinators", coordinators, "error", err)
 	if err != nil {
 		return candidates, err
 	}
diff --git a/controllers/check_client_compatibility.go b/controllers/check_client_compatibility.go
index c025582d..cbaf86f6 100644
--- a/controllers/check_client_compatibility.go
+++ b/controllers/check_client_compatibility.go
@@ -53,8 +53,8 @@ func (c checkClientCompatibility) reconcile(_ context.Context, r *FoundationDBCl
 		return &requeue{curError: err}
 	}
 
-	if !version.IsAtLeast(runningVersion) && !version.IsProtocolCompatible(runningVersion) {
-		return &requeue{message: fmt.Sprintf("cluster downgrade operation is only supported for protocol compatible versions, running version %s and desired version %s are not compatible", runningVersion, version)}
+	if !runningVersion.SupportsVersionChange(version) {
+		return &requeue{message: fmt.Sprintf("cluster version change from version %s to version %s is not supported", runningVersion, version)}
 	}
 
 	if version.IsProtocolCompatible(runningVersion) {
diff --git a/controllers/cluster_controller.go b/controllers/cluster_controller.go
index 2bd098b6..5ef48034 100644
--- a/controllers/cluster_controller.go
+++ b/controllers/cluster_controller.go
@@ -117,6 +117,12 @@ func (r *FoundationDBClusterReconciler) Reconcile(ctx context.Context, request c
 	}
 	defer adminClient.Close()
 
+	err = cluster.Validate()
+	if err != nil {
+		r.Recorder.Event(cluster, corev1.EventTypeWarning, "ClusterSpec not valid", err.Error())
+		return ctrl.Result{}, fmt.Errorf("ClusterSpec is not valid: %w", err)
+	}
+
 	supportedVersion, err := adminClient.VersionSupported(cluster.Spec.Version)
 	if err != nil {
 		return ctrl.Result{}, err
@@ -125,12 +131,6 @@ func (r *FoundationDBClusterReconciler) Reconcile(ctx context.Context, request c
 		return ctrl.Result{}, fmt.Errorf("version %s is not supported", cluster.Spec.Version)
 	}
 
-	err = cluster.Validate()
-	if err != nil {
-		r.Recorder.Event(cluster, corev1.EventTypeWarning, "ClusterSpec not valid", err.Error())
-		return ctrl.Result{}, fmt.Errorf("ClusterSpec is not valid: %w", err)
-	}
-
 	var status *fdbv1beta2.FoundationDBStatus
 	if cacheStatus {
 		clusterLog.Info("Fetch machine-readable status for reconcilitation loop", "cacheStatus", cacheStatus)
diff --git a/docs/design/replacement_buckets.md b/docs/design/replacement_buckets.md
new file mode 100644
index 00000000..6fba4e5f
--- /dev/null
+++ b/docs/design/replacement_buckets.md
@@ -0,0 +1,147 @@
+# Replacement Buckets For Automatic Replacements
+
+## Metadata
+
+* Authors: @johscheuer
+* Created: 2022-10-03
+* Updated: 2023-10-10
+
+## Background
+
+The operator supports to replace process groups with different conditions like `MissingProcess`.
+This allows the operator to bring the cluster into a better state if some process groups are failing.
+A human operator can currently limit the number of automatic replacements, by default the operator will only replace one process group at a time.
+The drawback of this approach is that all replacements will be accounted in the same way.
+That brings the risk that some replacements are blocked for a longer time by a single (or small) number of replacements.
+Replacements for stateless or log process groups are typically fast, but a single replacement of a storage process group could block those replacements for a long time.
+
+## General Design Goals
+
+Allow a human operator to define replacement buckets which allow to set limits on a per process class base.
+This should allow the operator to be more flexible during replacements with a lower risk of being blocked by another replacement of a different process class.
+
+## Current Implementation
+
+The current implementation counts the number of the currently ongoing replacements and if this number is higher than the limit the operator won't replace any faulty process groups.
+
+## Proposed Design
+
+The operator should be extended to support replacement buckets that allow limits on a process class level.
+The `AutomaticReplacementOptions` struct can be extended to contain a replacement bucket struct like this:
+
+
+```go
+// ReplacementBucketConfiguration contains the configuration for the ReplacementBuckets if enabled.
+type ReplacementBucketConfiguration struct {
+	// Enabled, if set the ReplacementBucketConfiguration will be used. If the ReplacementBucketConfiguration is disabled all replacements are counted into the same bucket,
+	// this basically is the same behaviour with the current maxConcurrentReplacements value. Defaults to false.
+	Enabled bool
+	// ReplacementBuckets defines the configuration for each bucket. A bucket can be configured per process class.
+	// If no specific bucket for a process class is configured the bucket for the general process class will be used.
+	// If no general process class bucket is defined the value from cluster.spec.automationOptions.replacements.maxConcurrentReplacements will be used.
+	ReplacementBuckets []ReplacementBucket
+}
+
+// ReplacementBucket defines the configuration for a specific Replacement bucket.
+type ReplacementBucket struct {
+	// ProcessClass the process class this bucket applies too.
+	ProcessClass ProcessClass
+	// MaxConcurrentReplacements the number of concurrent replacements that are allowed by the operator.
+	MaxConcurrentReplacements *int
+}
+```
+
+The `getMaxReplacements(cluster *fdbv1beta2.FoundationDBCluster, maxReplacements int) int` method would be adjusted to return a `map[fdbv1beta2.ProcessClass]int`:
+
+```go
+func getMaxReplacements(cluster *fdbv1beta2.FoundationDBCluster, maxReplacements int) map[fdbv1beta2.ProcessClass]int {
+	if cluster.ReplacementBucketIsEnabled() {
+		return cluster.getReplacementBuckets()
+    }
+	
+	// The maximum number of replacements will be the defined number in the cluster spec
+	// minus all currently ongoing replacements e.g. process groups marked for removal but
+	// not fully excluded.
+	removalCount := 0
+	for _, processGroupStatus := range cluster.Status.ProcessGroups {
+		if processGroupStatus.IsMarkedForRemoval() && !processGroupStatus.IsExcluded() {
+			// If we already have a removal in-flight, we should not try
+			// replacing more failed process groups.
+			removalCount++
+		}
+	}
+
+	return map[fdbv1beta2.ProcessClass]int{fdbv1beta2.ProcessClassGeneral: maxReplacements - removalCount}
+}
+```
+
+The `getReplacementBucket()` could be implemented like this:
+
+```go
+func (cluster *FoundationDBCluster) getReplacementBuckets() map[fdbv1beta2.ProcessClass]int {
+	// The maximum number of replacements will be the defined number in the cluster spec
+	// minus all currently ongoing replacements e.g. process groups marked for removal but
+	// not fully excluded.
+	replacementBuckets := cluster.getReplacementBucketWithDefaults()
+	for _, processGroup := range cluster.Status.ProcessGroups {
+		if processGroup.IsMarkedForRemoval() && !processGroup.IsExcluded() {
+			// If we already have a removal in-flight, we should not try
+			// replacing more failed process groups.
+			_, ok := replacementBuckets[processGroup.ProcessClass.ReplacementBucket()]
+			!ok {
+				replacementBuckets[processGroup.ProcessClass.ReplacementBucket()]--
+				continue
+			}
+
+			replacementBuckets[fdbv1beta2.ProcessClassGeneral]--
+		}
+	}
+
+	return replacementBuckets
+}
+```
+
+The method `getReplacementBucketWithDefaults()` will return a map of the type `map[fdbv1beta2.ProcessClass]int` with the defaults configured in `ReplacementBucketConfiguration`.
+To allow the correct selection of the replacement bucket we need to adjust the `ReplaceFailedProcessGroups` method:
+
+```go
+func ReplaceFailedProcessGroups(log logr.Logger, cluster *fdbv1beta2.FoundationDBCluster, adminClient fdbadminclient.AdminClient) bool {
+    ...
+	replacementBucket := getMaxReplacements(cluster, cluster.GetMaxConcurrentAutomaticReplacements())
+	replacementBucketIsEnabled := cluster.GetReplacementBucketEnabled()
+
+	hasReplacement := false
+	for _, processGroupStatus := range cluster.Status.ProcessGroups {
+		// ...
+        if !needsReplacement {
+			continue
+        }
+
+        var replacementClass fdbv1beta2.ProcessClass
+        _, ok := replacementBucket[processGroupStatus.ProcessClass]
+        if ok {
+            replacementClass = processGroupStatus.ProcessClas
+        } else {
+			// If there is no specific bucket, make use of the general bucket.
+            replacementClass = fdbv1beta2.ProcessClassGeneral
+        }
+
+        if replacementBucket[replacementClass] <= 0 {
+            // Add log statement
+            continue
+        }
+        // ...
+        hasReplacement = true
+        replacementBucket[replacementClass]--
+	}
+
+	return hasReplacement
+}
+```
+
+This allows to replace different process classes concurrently without blocking each other.
+
+## Related Links
+
+- [Automatic Replacements](https://github.com/FoundationDB/fdb-kubernetes-operator/blob/main/docs/design/implemented/automatic_replacements.md)
+- [Replacements and Deletions](https://github.com/FoundationDB/fdb-kubernetes-operator/blob/main/docs/manual/replacements_and_deletions.md)
diff --git a/docs/manual/fault_domains.md b/docs/manual/fault_domains.md
index 8d970ee9..7b39b84f 100644
--- a/docs/manual/fault_domains.md
+++ b/docs/manual/fault_domains.md
@@ -98,6 +98,67 @@ spec:
 
 This strategy uses the pod name as the fault domain, which allows each process to act as a separate failure domain. Any hardware failure could lead to a complete loss of the cluster. This configuration should not be used in any production environment.
 
+## Three-Data-Hall Replication
+
+**NOTE**: The support for this redundancy mode is new and might have issues. Please make sure you test this configuration in our test/QA environment.
+The [three-data-hall](https://apple.github.io/foundationdb/configuration.html#single-datacenter-modes) replication can be use to replicate data across three data halls, or availability zones.
+This requires that your fault domains are properly labeled on the Kubernetes nodes.
+Most cloud-providers will use the well-known label [topology.kubernetes.io/zone](https://kubernetes.io/docs/reference/labels-annotations-taints/#topologykubernetesiozone) for this.
+When creating a three-data-hall replicated FoundationDBCluster on Kubernetes we have to create 3 `FoundationDBCluster` resources.
+**NOTE**: This is a limitation of the current approach not to read any information from the Kubernetes nodes and simplify the scheduling logic of the operator.
+In the future, this might change and the deployment model for a three-data-hall FoundationDB cluster will be simplified.
+
+We have to start with a simple `FoundationDBCluster` that is running in one single availability zone, e.g. `az1`:
+
+```yaml
+apiVersion: apps.foundationdb.org/v1beta2
+kind: FoundationDBCluster
+metadata:
+  name: sample-cluster-az1
+spec:
+  version: 7.1.26
+  spec:
+    processGroupIDPrefix: az1
+    dataHall: az1
+    databaseConfiguration:
+      redundancyMode: triple
+    processes:
+      general:
+        podTemplate:
+          spec:
+            nodeSelector:
+              "topology.kubernetes.io/zone": "az1"
+```
+
+Once the cluster in `az1` is reconciled and running we can change the `redundancyMode` to `three_data_hall`.
+For the other two created `FoundationDBCluster` resources you have to set the `seedConnectionString` to the current connection string of the `FoundationDBCluster` resource in az1.
+The cluster will be stuck in a reconciling state until the other two `FoundationDBClusters`'s in `az2` and `az3` are created:
+
+```yaml
+apiVersion: apps.foundationdb.org/v1beta2
+kind: FoundationDBCluster
+metadata:
+  name: sample-cluster-az1
+spec:
+  version: 7.1.26
+  spec:
+    dataHall: az1
+    processGroupIDPrefix: az1
+    databaseConfiguration:
+      redundancyMode: three_data_hall
+    seedConnectionString: ""
+    processes:
+      general:
+        podTemplate:
+          spec:
+            nodeSelector:
+              "topology.kubernetes.io/zone": "az1"
+```
+
+Once all three `FoundationDBCluster` resources are marked as reconciled the FoundationDB cluster is up and running.
+You can run this configuration in the same namespace, different namespaces or even across multiple different Kubernetes clusters.
+Operations across the different `FoundationDBCluster` resources are [coordinated](#coordinating-global-operations). 
+
 ## Multi-Region Replication
 
 The replication strategies above all describe how data is replicated within a data center.
@@ -191,7 +252,8 @@ spec:
 
 ## Coordinating Global Operations
 
-When running a FoundationDB cluster that is deployed across multiple Kubernetes clusters, each Kubernetes cluster will have its own instance of the operator working on the processes in its cluster. There will be some operations that cannot be scoped to a single Kubernetes cluster, such as changing the database configuration. The operator provides a locking system to ensure that only one instance of the operator can perform these operations at a time. You can enable this locking system by setting `lockOptions.disableLocks = false` in the cluster spec. The locking system is automatically enabled by default for any cluster that has multiple regions in its database configuration, or a `zoneCount` greater than 1 in its fault domain configuration.
+When running a FoundationDB cluster that is deployed across multiple Kubernetes clusters, each Kubernetes cluster will have its own instance of the operator working on the processes in its cluster. There will be some operations that cannot be scoped to a single Kubernetes cluster, such as changing the database configuration.
+The operator provides a locking system to ensure that only one instance of the operator can perform these operations at a time. You can enable this locking system by setting `lockOptions.disableLocks = false` in the cluster spec. The locking system is automatically enabled by default for any cluster that has multiple regions in its database configuration, or a `zoneCount` greater than 1 in its fault domain configuration.
 
 The locking system uses the `processGroupIDPrefix` from the cluster spec to identify an process group of the operator.
 Make sure to set this to a unique value for each Kubernetes cluster, both to support the locking system and to prevent duplicate process group IDs.
diff --git a/fdbclient/admin_client.go b/fdbclient/admin_client.go
index 15ac014a..01a84775 100644
--- a/fdbclient/admin_client.go
+++ b/fdbclient/admin_client.go
@@ -505,16 +505,7 @@ func (client *cliAdminClient) GetConnectionString() (string, error) {
 // VersionSupported reports whether we can support a cluster with a given
 // version.
 func (client *cliAdminClient) VersionSupported(versionString string) (bool, error) {
-	version, err := fdbv1beta2.ParseFdbVersion(versionString)
-	if err != nil {
-		return false, err
-	}
-
-	if !version.IsSupported() {
-		return false, nil
-	}
-
-	_, err = os.Stat(getBinaryPath(fdbcliStr, versionString))
+	_, err := os.Stat(getBinaryPath(fdbcliStr, versionString))
 	if err != nil {
 		return false, err
 	}
diff --git a/go.mod b/go.mod
index ad9ee8cc..17350722 100644
--- a/go.mod
+++ b/go.mod
@@ -91,11 +91,11 @@ require (
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
 	go.uber.org/zap v1.24.0 // indirect
-	golang.org/x/net v0.10.0 // indirect
+	golang.org/x/net v0.17.0 // indirect
 	golang.org/x/oauth2 v0.4.0 // indirect
-	golang.org/x/sys v0.8.0 // indirect
-	golang.org/x/term v0.8.0 // indirect
-	golang.org/x/text v0.9.0 // indirect
+	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/term v0.13.0 // indirect
+	golang.org/x/text v0.13.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 	golang.org/x/tools v0.9.1 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.2.0 // indirect
diff --git a/go.sum b/go.sum
index 90ab3488..b0f95dc9 100644
--- a/go.sum
+++ b/go.sum
@@ -470,8 +470,8 @@ golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwY
 golang.org/x/net v0.0.0-20201209123823-ac852fbbde11/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
-golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
-golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
+golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -537,19 +537,19 @@ golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.8.0 h1:EBmGv8NaZBZTWvrbjNoL6HVt+IVy3QDQpJs7VRIw3tU=
-golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
-golang.org/x/term v0.8.0 h1:n5xxQn2i3PC0yLAbjTpNT85q/Kgzcr2gIoX9OrJUols=
-golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
+golang.org/x/term v0.13.0 h1:bb+I9cTfFazGW51MZqBVmZy7+JEJMouUHTUSKVQLBek=
+golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
-golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
diff --git a/internal/locality/locality.go b/internal/locality/locality.go
index 16d66ebb..acf73058 100644
--- a/internal/locality/locality.go
+++ b/internal/locality/locality.go
@@ -155,6 +155,9 @@ func ChooseDistributedProcesses(cluster *fdbv1beta2.FoundationDBCluster, process
 	fields := constraint.Fields
 	if len(fields) == 0 {
 		fields = []string{fdbv1beta2.FDBLocalityZoneIDKey, fdbv1beta2.FDBLocalityDCIDKey}
+		if cluster.Spec.DatabaseConfiguration.RedundancyMode == fdbv1beta2.RedundancyModeThreeDataHall {
+			fields = append(fields, fdbv1beta2.FDBLocalityDataHallKey)
+		}
 	}
 
 	chosenCounts := make(map[string]map[string]int, len(fields))
@@ -230,19 +233,35 @@ func ChooseDistributedProcesses(cluster *fdbv1beta2.FoundationDBCluster, process
 		}
 	}
 
+	if len(chosen) != count {
+		return chosen, notEnoughProcessesError{Desired: count, Chosen: len(chosen), Options: processes}
+	}
+
 	return chosen, nil
 }
 
 // GetHardLimits returns the distribution of localities.
 func GetHardLimits(cluster *fdbv1beta2.FoundationDBCluster) map[string]int {
 	if cluster.Spec.DatabaseConfiguration.UsableRegions <= 1 {
+		// For the three_data_hall redundancy mode we will recruit 9 coordinators and those hard limits are only used
+		// for selecting coordinators. We want to make sure we select coordinators across as many fault domains as possible.
+		if cluster.Spec.DatabaseConfiguration.RedundancyMode == fdbv1beta2.RedundancyModeThreeDataHall {
+			return map[string]int{
+				// Assumption here is that we have 3 data halls and we want to spread the coordinators
+				// equally across those 3 data halls.
+				fdbv1beta2.FDBLocalityDataHallKey: 3,
+				fdbv1beta2.FDBLocalityZoneIDKey:   1,
+			}
+		}
+
 		return map[string]int{fdbv1beta2.FDBLocalityZoneIDKey: 1}
 	}
 
-	// TODO (johscheuer): should we calculate that based on the number of DCs?
-	maxCoordinatorsPerDC := int(math.Floor(float64(cluster.DesiredCoordinatorCount()) / 2.0))
-
-	return map[string]int{fdbv1beta2.FDBLocalityZoneIDKey: 1, fdbv1beta2.FDBLocalityDCIDKey: maxCoordinatorsPerDC}
+	maxCoordinatorsPerDC := int(math.Ceil(float64(cluster.DesiredCoordinatorCount()) / float64(cluster.Spec.DatabaseConfiguration.CountUniqueDataCenters())))
+	return map[string]int{
+		fdbv1beta2.FDBLocalityDCIDKey:   maxCoordinatorsPerDC,
+		fdbv1beta2.FDBLocalityZoneIDKey: 1,
+	}
 }
 
 // CheckCoordinatorValidity determines if the cluster's current coordinators
@@ -261,16 +280,21 @@ func CheckCoordinatorValidity(logger logr.Logger, cluster *fdbv1beta2.Foundation
 	allAddressesValid := true
 	allEligible := true
 	allUsingCorrectAddress := true
+	hardLimits := GetHardLimits(cluster)
+	coordinatorLocalities := make(map[string]map[string]int)
+	// Track what fields should be validated.
+	fieldsToValidate := make([]string, 0, len(hardLimits))
+	for field := range hardLimits {
+		fieldsToValidate = append(fieldsToValidate, field)
+	}
 
-	coordinatorZones := make(map[string]int, len(coordinatorStatus))
-	coordinatorDCs := make(map[string]int, len(coordinatorStatus))
-	processGroups := make(map[fdbv1beta2.ProcessGroupID]*fdbv1beta2.ProcessGroupStatus)
-	for _, processGroup := range cluster.Status.ProcessGroups {
-		processGroups[processGroup.ProcessGroupID] = processGroup
+	// Track the coordinator localities to verify if hard limits satisfy the fault tolerance requirements.
+	for _, field := range fieldsToValidate {
+		coordinatorLocalities[field] = make(map[string]int)
 	}
 
 	for _, process := range status.Cluster.Processes {
-		processGroupID, ok := process.Locality[fdbv1beta2.FDBLocalityInstanceIDKey]
+		processGroupID := process.Locality[fdbv1beta2.FDBLocalityInstanceIDKey]
 
 		pLogger := logger.WithValues("process", processGroupID)
 		if process.Address.IsEmpty() {
@@ -278,7 +302,7 @@ func CheckCoordinatorValidity(logger logr.Logger, cluster *fdbv1beta2.Foundation
 			continue
 		}
 
-		if !ok {
+		if len(process.Locality) == 0 {
 			pLogger.Info("Skip process with empty localities")
 			continue
 		}
@@ -288,23 +312,11 @@ func CheckCoordinatorValidity(logger logr.Logger, cluster *fdbv1beta2.Foundation
 			continue
 		}
 
-		processGroupStatus := processGroups[fdbv1beta2.ProcessGroupID(processGroupID)]
-		pendingRemoval := processGroupStatus != nil && processGroupStatus.IsMarkedForRemoval()
-		if processGroupStatus != nil && cluster.SkipProcessGroup(processGroupStatus) {
-			pLogger.Info("Skipping process group with pending Pod",
-				"namespace", cluster.Namespace,
-				"cluster", cluster.Name,
-				"processGroupID", processGroupID,
-				"class", process.ProcessClass)
-			continue
-		}
-
 		addresses, err := fdbv1beta2.ParseProcessAddressesFromCmdline(process.CommandLine)
 		if err != nil {
 			// We will end here in the error case when the address
 			// is not parsable e.g. no IP address is assigned.
 			allAddressesValid = false
-			// add: command_line
 			pLogger.Info("Could not parse address from command_line", "command_line", process.CommandLine)
 			continue
 		}
@@ -341,13 +353,19 @@ func CheckCoordinatorValidity(logger logr.Logger, cluster *fdbv1beta2.Foundation
 			coordinatorAddress = dnsAddress.String()
 		}
 
-		if coordinatorAddress != "" && !process.Excluded && !pendingRemoval && !process.UnderMaintenance {
+		if coordinatorAddress != "" && !process.Excluded && !process.UnderMaintenance {
 			coordinatorStatus[coordinatorAddress] = true
 		}
 
 		if coordinatorAddress != "" {
-			coordinatorZones[process.Locality[fdbv1beta2.FDBLocalityZoneIDKey]]++
-			coordinatorDCs[process.Locality[fdbv1beta2.FDBLocalityDCIDKey]]++
+			for _, field := range fieldsToValidate {
+				locality, ok := process.Locality[field]
+				// If the field is not set ignore it.
+				if !ok {
+					continue
+				}
+				coordinatorLocalities[field][locality]++
+			}
 
 			if !cluster.IsEligibleAsCandidate(process.ProcessClass) {
 				pLogger.Info("Process class of process is not eligible as coordinator", "class", process.ProcessClass, "address", coordinatorAddress)
@@ -367,21 +385,13 @@ func CheckCoordinatorValidity(logger logr.Logger, cluster *fdbv1beta2.Foundation
 		}
 	}
 
-	desiredCount := cluster.DesiredCoordinatorCount()
-	hasEnoughZones := len(coordinatorZones) == desiredCount
-	if !hasEnoughZones {
-		logger.Info("Cluster does not have coordinators in the correct number of zones", "desiredCount", desiredCount, "coordinatorZones", coordinatorZones)
-	}
-
-	var maxCoordinatorsPerDC int
-	hasEnoughDCs := true
-	if cluster.Spec.DatabaseConfiguration.UsableRegions > 1 {
-		maxCoordinatorsPerDC = int(math.Floor(float64(desiredCount) / 2.0))
-
-		for dc, count := range coordinatorDCs {
-			if count > maxCoordinatorsPerDC {
-				logger.Info("Cluster has too many coordinators in a single DC", "DC", dc, "count", count, "max", maxCoordinatorsPerDC)
-				hasEnoughDCs = false
+	// Check if the coordinators are distributed across the localities based on the hard limit requirements.
+	hasCorrectLocalityDistribution := true
+	for field, maxValue := range hardLimits {
+		for locality, currentValue := range coordinatorLocalities[field] {
+			if currentValue > maxValue {
+				logger.Info("Cluster does not have coordinators in the correct number of localities", "desiredCount", maxValue, "currentCount", currentValue, "locality", locality)
+				hasCorrectLocalityDistribution = false
 			}
 		}
 	}
@@ -394,5 +404,13 @@ func CheckCoordinatorValidity(logger logr.Logger, cluster *fdbv1beta2.Foundation
 		}
 	}
 
-	return hasEnoughDCs && hasEnoughZones && allHealthy && allUsingCorrectAddress && allEligible, allAddressesValid, nil
+	// Verify that enough coordinators are running.
+	desiredCoordinatorCount := cluster.DesiredCoordinatorCount()
+	runningCoordinators := len(coordinatorLocalities[fdbv1beta2.FDBLocalityZoneIDKey])
+	hasEnoughCoordinators := runningCoordinators == desiredCoordinatorCount
+	if !hasEnoughCoordinators {
+		logger.Info("Cluster has not enough running coordinators", "runningCoordinators", runningCoordinators, "desiredCount", desiredCoordinatorCount)
+	}
+
+	return hasEnoughCoordinators && hasCorrectLocalityDistribution && allHealthy && allUsingCorrectAddress && allEligible, allAddressesValid, nil
 }
diff --git a/internal/monitor_conf.go b/internal/monitor_conf.go
index a221467c..660f4466 100644
--- a/internal/monitor_conf.go
+++ b/internal/monitor_conf.go
@@ -205,7 +205,7 @@ func GetMonitorProcessConfiguration(cluster *fdbv1beta2.FoundationDBCluster, pro
 			},
 		})
 		configuration.Arguments = append(configuration.Arguments, monitorapi.Argument{ArgumentType: monitorapi.ConcatenateArgumentType, Values: []monitorapi.Argument{
-			{Value: "--locality_process_id="},
+			{Value: getKnobParameter(fdbv1beta2.FDBLocalityProcessIDKey, true)},
 			{ArgumentType: monitorapi.EnvironmentArgumentType, Source: "FDB_INSTANCE_ID"},
 			{Value: "-"},
 			{ArgumentType: monitorapi.ProcessNumberArgumentType},
@@ -216,15 +216,15 @@ func GetMonitorProcessConfiguration(cluster *fdbv1beta2.FoundationDBCluster, pro
 
 	configuration.Arguments = append(configuration.Arguments,
 		monitorapi.Argument{ArgumentType: monitorapi.ConcatenateArgumentType, Values: []monitorapi.Argument{
-			{Value: "--locality_instance_id="},
+			{Value: getKnobParameter(fdbv1beta2.FDBLocalityInstanceIDKey, true)},
 			{ArgumentType: monitorapi.EnvironmentArgumentType, Source: "FDB_INSTANCE_ID"},
 		}},
 		monitorapi.Argument{ArgumentType: monitorapi.ConcatenateArgumentType, Values: []monitorapi.Argument{
-			{Value: "--locality_machineid="},
+			{Value: getKnobParameter(fdbv1beta2.FDBLocalityMachineIDKey, true)},
 			{ArgumentType: monitorapi.EnvironmentArgumentType, Source: "FDB_MACHINE_ID"},
 		}},
 		monitorapi.Argument{ArgumentType: monitorapi.ConcatenateArgumentType, Values: []monitorapi.Argument{
-			{Value: "--locality_zoneid="},
+			{Value: getKnobParameter(fdbv1beta2.FDBLocalityZoneIDKey, true)},
 			{ArgumentType: monitorapi.EnvironmentArgumentType, Source: zoneVariable},
 		}},
 	)
@@ -234,7 +234,7 @@ func GetMonitorProcessConfiguration(cluster *fdbv1beta2.FoundationDBCluster, pro
 	}
 
 	if cluster.Spec.MainContainer.PeerVerificationRules != "" {
-		configuration.Arguments = append(configuration.Arguments, monitorapi.Argument{Value: fmt.Sprintf("--tls_verify_peers=%s", cluster.Spec.MainContainer.PeerVerificationRules)})
+		configuration.Arguments = append(configuration.Arguments, monitorapi.Argument{Value: getKnobParameterWithValue("tls_verify_peers", cluster.Spec.MainContainer.PeerVerificationRules, false)})
 	}
 
 	podSettings := cluster.GetProcessSettings(processClass)
@@ -254,11 +254,11 @@ func GetMonitorProcessConfiguration(cluster *fdbv1beta2.FoundationDBCluster, pro
 	}
 
 	if cluster.Spec.DataCenter != "" {
-		configuration.Arguments = append(configuration.Arguments, monitorapi.Argument{Value: fmt.Sprintf("--locality_dcid=%s", cluster.Spec.DataCenter)})
+		configuration.Arguments = append(configuration.Arguments, monitorapi.Argument{Value: getKnobParameterWithValue(fdbv1beta2.FDBLocalityDCIDlKey, cluster.Spec.DataCenter, true)})
 	}
 
 	if cluster.Spec.DataHall != "" {
-		configuration.Arguments = append(configuration.Arguments, monitorapi.Argument{Value: fmt.Sprintf("--locality_data_hall=%s", cluster.Spec.DataHall)})
+		configuration.Arguments = append(configuration.Arguments, monitorapi.Argument{Value: getKnobParameterWithValue(fdbv1beta2.FDBLocalityDataHallKey, cluster.Spec.DataHall, true)})
 	}
 
 	if cluster.DefineDNSLocalityFields() {
@@ -271,6 +271,25 @@ func GetMonitorProcessConfiguration(cluster *fdbv1beta2.FoundationDBCluster, pro
 	return configuration, nil
 }
 
+// getKnobParameter will return the knob parameter with a trailing =. If the provided knob is a locality the key will be
+// prefixed with "locality_".
+func getKnobParameter(key string, isLocality bool) string {
+	var sb strings.Builder
+	sb.WriteString("--")
+	if isLocality {
+		sb.WriteString("locality_")
+	}
+	sb.WriteString(key)
+	sb.WriteString("=")
+
+	return sb.String()
+}
+
+// getKnobParameterWithValue is the same as getKnobParameter but will append the value at the end.
+func getKnobParameterWithValue(key string, value string, isLocality bool) string {
+	return getKnobParameter(key, isLocality) + value
+}
+
 // buildIPArgument builds an argument that takes an IP address from an environment variable
 func buildIPArgument(parameter string, environmentVariable string, imageType FDBImageType, sampleAddresses []fdbv1beta2.ProcessAddress) []monitorapi.Argument {
 	var leftIPWrap string
diff --git a/internal/replacements/replace_failed_process_groups.go b/internal/replacements/replace_failed_process_groups.go
index a321ad2a..4d173ba5 100644
--- a/internal/replacements/replace_failed_process_groups.go
+++ b/internal/replacements/replace_failed_process_groups.go
@@ -55,7 +55,6 @@ func ReplaceFailedProcessGroups(log logr.Logger, cluster *fdbv1beta2.FoundationD
 	hasReplacement := false
 	crashLoopContainerProcessGroups := cluster.GetCrashLoopContainerProcessGroups()
 
-ProcessGroupLoop:
 	for _, processGroupStatus := range cluster.Status.ProcessGroups {
 		// If a process group is already marked for removal we can skip it here.
 		if processGroupStatus.IsMarkedForRemoval() {
@@ -70,18 +69,25 @@ ProcessGroupLoop:
 			continue
 		}
 
-		canReplace := maxReplacements > 0
-
+		var shouldBeIgnored bool
 		for _, targets := range crashLoopContainerProcessGroups {
 			if _, ok := targets[processGroupStatus.ProcessGroupID]; ok {
-				continue ProcessGroupLoop
-			} else if _, ok := targets["*"]; ok {
-				continue ProcessGroupLoop
+				shouldBeIgnored = true
+				break
+			}
+
+			if _, ok := targets["*"]; ok {
+				shouldBeIgnored = true
+				break
 			}
 		}
 
-		needsReplacement, missingTime := processGroupStatus.NeedsReplacement(cluster.GetFailureDetectionTimeSeconds(), cluster.GetTaintReplacementTimeSeconds())
-		if !needsReplacement {
+		if shouldBeIgnored {
+			continue
+		}
+
+		failureCondition, failureTime := processGroupStatus.NeedsReplacement(cluster.GetFailureDetectionTimeSeconds(), cluster.GetTaintReplacementTimeSeconds())
+		if failureTime == 0 {
 			continue
 		}
 
@@ -91,7 +97,7 @@ ProcessGroupLoop:
 				log.Info(
 					"Skip process group with missing address",
 					"processGroupID", processGroupStatus.ProcessGroupID,
-					"failureTime", time.Unix(missingTime, 0).UTC().String())
+					"failureTime", time.Unix(failureTime, 0).UTC().String())
 				continue
 			}
 
@@ -102,20 +108,24 @@ ProcessGroupLoop:
 			log.Info(
 				"Replace process group with missing address",
 				"processGroupID", processGroupStatus.ProcessGroupID,
-				"failureTime", time.Unix(missingTime, 0).UTC().String())
+				"failureTime", time.Unix(failureTime, 0).UTC().String())
 		}
 
-		// We are not allowed to replace additional process groups
-		if !canReplace {
+		// We are not allowed to replace additional process groups.
+		if maxReplacements <= 0 {
 			log.Info("Detected replace process group but cannot replace it because we hit the replacement limit",
 				"processGroupID", processGroupStatus.ProcessGroupID,
-				"reason", fmt.Sprintf("automatic replacement detected failure time: %s", time.Unix(missingTime, 0).UTC().String()))
+				"failureCondition", failureCondition,
+				"faultDomain", processGroupStatus.FaultDomain,
+				"reason", fmt.Sprintf("automatic replacement detected failure time: %s", time.Unix(failureTime, 0).UTC().String()))
 			continue
 		}
 
 		log.Info("Replace process group",
 			"processGroupID", processGroupStatus.ProcessGroupID,
-			"reason", fmt.Sprintf("automatic replacement detected failure time: %s", time.Unix(missingTime, 0).UTC().String()))
+			"failureCondition", failureCondition,
+			"faultDomain", processGroupStatus.FaultDomain,
+			"reason", fmt.Sprintf("automatic replacement detected failure time: %s", time.Unix(failureTime, 0).UTC().String()))
 
 		processGroupStatus.MarkForRemoval()
 		hasReplacement = true
diff --git a/kubectl-fdb/cmd/analyze.go b/kubectl-fdb/cmd/analyze.go
index 88a82597..ff15ae35 100644
--- a/kubectl-fdb/cmd/analyze.go
+++ b/kubectl-fdb/cmd/analyze.go
@@ -293,9 +293,11 @@ func analyzeCluster(cmd *cobra.Command, kubeClient client.Client, cluster *fdbv1
 			printStatement(cmd, statement, errorMessage)
 		}
 
-		needsReplacement, _ := processGroup.NeedsReplacement(0, 0)
-		if needsReplacement && autoFix {
-			failedProcessGroups = append(failedProcessGroups, string(processGroup.ProcessGroupID))
+		if autoFix {
+			_, failureTime := processGroup.NeedsReplacement(0, 0)
+			if failureTime > 0 {
+				failedProcessGroups = append(failedProcessGroups, string(processGroup.ProcessGroupID))
+			}
 		}
 	}
 
diff --git a/kubectl-fdb/cmd/remove_process_group.go b/kubectl-fdb/cmd/remove_process_group.go
index bb04e6d2..fcade941 100644
--- a/kubectl-fdb/cmd/remove_process_group.go
+++ b/kubectl-fdb/cmd/remove_process_group.go
@@ -88,7 +88,7 @@ kubectl fdb -n default remove process-group -c cluster --remove-all-failed
 `,
 	}
 
-	cmd.Flags().StringP("fdb-cluster", "c", "", "remove process groupss from the provided cluster.")
+	cmd.Flags().StringP("fdb-cluster", "c", "", "remove process groups from the provided cluster.")
 	cmd.Flags().BoolP("exclusion", "e", true, "define if the process groups should be removed with exclusion.")
 	cmd.Flags().Bool("remove-all-failed", false, "define if all failed processes should be replaced.")
 	cmd.Flags().Bool("use-process-group-id", false, "define if the process-group should be used instead of the Pod name.")
@@ -147,8 +147,8 @@ func replaceProcessGroups(kubeClient client.Client, clusterName string, ids []st
 				continue
 			}
 
-			needsReplacement, _ := processGroupStatus.NeedsReplacement(0, 0)
-			if needsReplacement {
+			_, failureTime := processGroupStatus.NeedsReplacement(0, 0)
+			if failureTime > 0 {
 				processGroupIDs = append(processGroupIDs, processGroupStatus.ProcessGroupID)
 			}
 		}
diff --git a/sample-apps/data-loader/Dockerfile b/sample-apps/data-loader/Dockerfile
index 9497bc01..01c0f2c7 100644
--- a/sample-apps/data-loader/Dockerfile
+++ b/sample-apps/data-loader/Dockerfile
@@ -6,7 +6,16 @@ RUN pip install foundationdb==6.2.10
 RUN groupadd --gid 4059 fdb && \
 	useradd --gid 4059 --uid 4059 --shell /usr/sbin/nologin fdb
 
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends curl && \
+    curl -Ls https://github.com/krallin/tini/releases/download/v0.19.0/tini-amd64 -o tini  && \
+    echo "93dcc18adc78c65a028a84799ecf8ad40c936fdfc5f2a57b1acda5a8117fa82c  tini" > tini-sha.txt && \
+    sha256sum --quiet -c tini-sha.txt && \
+    chmod +x tini && \
+    mv tini /usr/bin/ && \
+    rm -rf /tmp/*
+
 # Set to the numeric UID of fdb user to satisfy PodSecurityPolices which enforce runAsNonRoot
 USER 4059
 
-ENTRYPOINT [ "python", "/usr/local/bin/app.py" ]
+ENTRYPOINT [ "/usr/bin/tini", "-g", "--", "python", "/usr/local/bin/app.py" ]


Test patch:
diff --git a/api/v1beta2/foundationdb_database_configuration_test.go b/api/v1beta2/foundationdb_database_configuration_test.go
index ec63884b..4e875837 100644
--- a/api/v1beta2/foundationdb_database_configuration_test.go
+++ b/api/v1beta2/foundationdb_database_configuration_test.go
@@ -158,6 +158,42 @@ var _ = Describe("DatabaseConfiguration", func() {
 		})
 	})
 
+	When("a three_data_hall cluster with the default values is provided", func() {
+		var cluster *FoundationDBCluster
+
+		BeforeEach(func() {
+			cluster = &FoundationDBCluster{
+				Spec: FoundationDBClusterSpec{
+					Version:  "7.1.33",
+					DataHall: "az1",
+					ProcessCounts: ProcessCounts{
+						Stateless: -1,
+					},
+					DatabaseConfiguration: DatabaseConfiguration{
+						StorageEngine:  StorageEngineSSD,
+						RedundancyMode: RedundancyModeThreeDataHall,
+						UsableRegions:  1,
+					},
+				},
+			}
+		})
+
+		When("getting the default process counts", func() {
+			var err error
+			var counts ProcessCounts
+
+			BeforeEach(func() {
+				counts, err = cluster.GetProcessCountsWithDefaults()
+			})
+
+			It("It should calculate the default process counts", func() {
+				Expect(err).NotTo(HaveOccurred())
+				Expect(counts.Log).To(BeNumerically("==", 6)) // 4 required + 2 additional
+				Expect(counts.Storage).To(BeNumerically("==", 5))
+			})
+		})
+	})
+
 	When("using ProcessCounts", func() {
 		When("calculating the total number of processes", func() {
 			var counts ProcessCounts
diff --git a/api/v1beta2/foundationdb_version_test.go b/api/v1beta2/foundationdb_version_test.go
index aa559d1e..132a2314 100644
--- a/api/v1beta2/foundationdb_version_test.go
+++ b/api/v1beta2/foundationdb_version_test.go
@@ -154,6 +154,101 @@ var _ = Describe("[api] FDBVersion", func() {
 					expectedResult:         true,
 				}),
 		)
-
 	})
+
+	DescribeTable("validating in a version check is allowed", func(version Version, targetVersion Version, expected bool) {
+		Expect(version.SupportsVersionChange(targetVersion)).To(Equal(expected))
+	},
+		Entry("Same version",
+			Version{
+				Major: 7,
+				Minor: 1,
+				Patch: 0,
+			},
+			Version{
+				Major: 7,
+				Minor: 1,
+				Patch: 0,
+			},
+			true,
+		),
+		Entry("Patch upgrade",
+			Version{
+				Major: 7,
+				Minor: 1,
+				Patch: 0,
+			},
+			Version{
+				Major: 7,
+				Minor: 1,
+				Patch: 1,
+			},
+			true,
+		),
+		Entry("Minor upgrade",
+			Version{
+				Major: 7,
+				Minor: 1,
+				Patch: 0,
+			},
+			Version{
+				Major: 7,
+				Minor: 2,
+				Patch: 0,
+			},
+			true,
+		),
+		Entry("Major upgrade",
+			Version{
+				Major: 7,
+				Minor: 1,
+				Patch: 0,
+			},
+			Version{
+				Major: 8,
+				Minor: 1,
+				Patch: 0,
+			},
+			true,
+		),
+		Entry("Patch downgrade",
+			Version{
+				Major: 7,
+				Minor: 1,
+				Patch: 1,
+			},
+			Version{
+				Major: 7,
+				Minor: 1,
+				Patch: 0,
+			},
+			true,
+		),
+		Entry("Minor downgrade",
+			Version{
+				Major: 7,
+				Minor: 1,
+				Patch: 1,
+			},
+			Version{
+				Major: 7,
+				Minor: 0,
+				Patch: 0,
+			},
+			false,
+		),
+		Entry("Major downgrade",
+			Version{
+				Major: 7,
+				Minor: 1,
+				Patch: 1,
+			},
+			Version{
+				Major: 6,
+				Minor: 1,
+				Patch: 0,
+			},
+			false,
+		),
+	)
 })
diff --git a/api/v1beta2/foundationdbcluster_types_test.go b/api/v1beta2/foundationdbcluster_types_test.go
index 27c52e04..aa747e2a 100644
--- a/api/v1beta2/foundationdbcluster_types_test.go
+++ b/api/v1beta2/foundationdbcluster_types_test.go
@@ -4065,8 +4065,8 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 
 	When("checking if the process group needs a replacement", func() {
 		var processGroup *ProcessGroupStatus
-		var needsReplacement bool
-		var timestamp int64
+		var failureCondition ProcessGroupConditionType
+		var failureTime int64
 		var oldTimestamp int64
 
 		BeforeEach(func() {
@@ -4075,12 +4075,13 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 		})
 
 		JustBeforeEach(func() {
-			needsReplacement, timestamp = processGroup.NeedsReplacement(60, 60)
+			failureCondition, failureTime = processGroup.NeedsReplacement(60, 60)
 		})
 
 		Context("with no conditions", func() {
 			It("should not need replacement", func() {
-				Expect(needsReplacement).To(BeFalse())
+				Expect(failureTime).To(BeZero())
+				Expect(failureCondition).To(Equal(ProcessGroupConditionType("")))
 			})
 		})
 
@@ -4090,7 +4091,8 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			})
 
 			It("should not need replacement", func() {
-				Expect(needsReplacement).To(BeFalse())
+				Expect(failureTime).To(BeZero())
+				Expect(failureCondition).To(Equal(ProcessGroupConditionType("")))
 			})
 		})
 
@@ -4103,8 +4105,8 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			})
 
 			It("should need replacement", func() {
-				Expect(needsReplacement).To(BeTrue())
-				Expect(timestamp).To(Equal(targetTimestamp))
+				Expect(failureTime).To(BeNumerically("==", targetTimestamp))
+				Expect(failureCondition).To(Equal(MissingProcesses))
 			})
 		})
 
@@ -4119,8 +4121,8 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			})
 
 			It("should use the older timestamp", func() {
-				Expect(needsReplacement).To(BeTrue())
-				Expect(timestamp).To(Equal(targetTimestamp - 60))
+				Expect(failureTime).To(BeNumerically("==", targetTimestamp-60))
+				Expect(failureCondition).To(Equal(PodFailing))
 			})
 		})
 
@@ -4131,8 +4133,8 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			})
 
 			It("should need replacement", func() {
-				Expect(needsReplacement).To(BeTrue())
-				Expect(timestamp).To(Equal(oldTimestamp))
+				Expect(failureTime).To(BeNumerically("==", oldTimestamp))
+				Expect(failureCondition).To(Equal(PodFailing))
 			})
 		})
 
@@ -4143,8 +4145,8 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			})
 
 			It("should need replacement", func() {
-				Expect(needsReplacement).To(BeTrue())
-				Expect(timestamp).To(Equal(oldTimestamp))
+				Expect(failureTime).To(BeNumerically("==", oldTimestamp))
+				Expect(failureCondition).To(Equal(MissingPod))
 			})
 		})
 
@@ -4155,8 +4157,8 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			})
 
 			It("should need replacement", func() {
-				Expect(needsReplacement).To(BeTrue())
-				Expect(timestamp).To(Equal(oldTimestamp))
+				Expect(failureTime).To(BeNumerically("==", oldTimestamp))
+				Expect(failureCondition).To(Equal(MissingPVC))
 			})
 		})
 
@@ -4167,8 +4169,8 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			})
 
 			It("should need replacement", func() {
-				Expect(needsReplacement).To(BeTrue())
-				Expect(timestamp).To(Equal(oldTimestamp))
+				Expect(failureTime).To(BeNumerically("==", oldTimestamp))
+				Expect(failureCondition).To(Equal(MissingService))
 			})
 		})
 
@@ -4179,8 +4181,20 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			})
 
 			It("should need replacement", func() {
-				Expect(needsReplacement).To(BeTrue())
-				Expect(timestamp).To(Equal(oldTimestamp))
+				Expect(failureTime).To(BeNumerically("==", oldTimestamp))
+				Expect(failureCondition).To(Equal(PodPending))
+			})
+		})
+
+		When("process group is excluded without being marked for removal", func() {
+			BeforeEach(func() {
+				processGroup.UpdateCondition(ProcessIsMarkedAsExcluded, true)
+				processGroup.ProcessGroupConditions[0].Timestamp = oldTimestamp
+			})
+
+			It("should need replacement", func() {
+				Expect(failureTime).To(BeNumerically("==", oldTimestamp))
+				Expect(failureCondition).To(Equal(ProcessIsMarkedAsExcluded))
 			})
 		})
 
@@ -4191,7 +4205,8 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			})
 
 			It("should not need replacement", func() {
-				Expect(needsReplacement).To(BeFalse())
+				Expect(failureTime).To(BeZero())
+				Expect(failureCondition).To(Equal(ProcessGroupConditionType("")))
 			})
 		})
 	})
@@ -4746,10 +4761,9 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			func(cluster *FoundationDBCluster, expected error) {
 				if expected == nil {
 					Expect(cluster.Validate()).NotTo(HaveOccurred())
-				} else {
-					Expect(cluster.Validate()).To(Equal(expected))
+					return
 				}
-
+				Expect(cluster.Validate()).To(Equal(expected))
 			},
 			Entry("valid cluster spec",
 				&FoundationDBCluster{
@@ -4851,7 +4865,7 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			Entry("multiple validations",
 				&FoundationDBCluster{
 					Spec: FoundationDBClusterSpec{
-						Version: "6.1.3",
+						Version: "6.2.20",
 						DatabaseConfiguration: DatabaseConfiguration{
 							StorageEngine: StorageEngineRocksDbV1,
 						},
@@ -4862,7 +4876,7 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 						},
 					},
 				},
-				fmt.Errorf("storage engine ssd-rocksdb-v1 is not supported on version 6.1.3, stateless is not a valid process class for coordinators"),
+				fmt.Errorf("storage engine ssd-rocksdb-v1 is not supported on version 6.2.20, stateless is not a valid process class for coordinators"),
 			),
 			Entry("using invalid version for sharded rocksdb",
 				&FoundationDBCluster{
@@ -4886,6 +4900,17 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 				},
 				nil,
 			),
+			Entry("using an unsupported FDB version",
+				&FoundationDBCluster{
+					Spec: FoundationDBClusterSpec{
+						Version: Versions.IncompatibleVersion.String(),
+						DatabaseConfiguration: DatabaseConfiguration{
+							StorageEngine: StorageEngineSSD2,
+						},
+					},
+				},
+				fmt.Errorf("version: 6.1.0 is not supported, minimum supported version is: 6.2.20"),
+			),
 		)
 	})
 
diff --git a/config/tests/three_data_hall/Readme.md b/config/tests/three_data_hall/Readme.md
new file mode 100644
index 00000000..8f3a14a3
--- /dev/null
+++ b/config/tests/three_data_hall/Readme.md
@@ -0,0 +1,21 @@
+# Three-Data-hall example
+
+This example requires that your Kubernetes cluster has nodes which are labeled with `topology.kubernetes.io/zone`.
+The example requires at least 3 unique zones, those can be faked for testing, by adding the labels to a node.
+If you want to use cloud provider specific zone label values you can set the `AZ1`, `AZ2` and `AZ3` environment variables.
+
+## Create the Three-Data-Hall cluster
+
+This will bring up a FDB cluster using the three-data-hall redundancy mode.
+
+```bash
+./create.bash
+```
+
+## Delete
+
+This will remove all created resources:
+
+```bash
+./delete.bash
+```
diff --git a/config/tests/three_data_hall/create.bash b/config/tests/three_data_hall/create.bash
new file mode 100755
index 00000000..e35f54b5
--- /dev/null
+++ b/config/tests/three_data_hall/create.bash
@@ -0,0 +1,34 @@
+#!/usr/bin/env bash
+
+set -eu
+
+# This directory provides an example of creating a cluster using the three_data_hall
+# replication topology.
+#
+# This example is built for local testing, so it will create all of the Pods
+# within a single Kubernetes cluster, but will give false locality information for the zones
+# to make the processes believe they are in different locations.
+#
+# You can use this script to bootstrap the cluster. Once it finishes, you can
+# make changes to the cluster by editing the final.yaml file and running the
+# apply.bash script. You can clean up the clusters by running the delete.bash
+# script.
+DIR="${BASH_SOURCE%/*}"
+
+. $DIR/functions.bash
+
+AZ1=${AZ1:-"az1"}
+AZ2=${AZ2:-"az2"}
+AZ3=${AZ3:-"az3"}
+
+applyFile "${DIR}/stage_1.yaml" "${AZ1}" '""'
+checkReconciliationLoop test-cluster-${AZ1}
+connectionString=$(getConnectionString test-cluster-${AZ1})
+
+applyFile "${DIR}/final.yaml" "${AZ1}" "${connectionString}"
+applyFile "${DIR}/final.yaml" ${AZ2} "${connectionString}"
+applyFile "${DIR}/final.yaml" ${AZ3} "${connectionString}"
+
+checkReconciliationLoop test-cluster-${AZ1}
+checkReconciliationLoop test-cluster-${AZ2}
+checkReconciliationLoop test-cluster-${AZ3}
diff --git a/config/tests/three_data_hall/delete.bash b/config/tests/three_data_hall/delete.bash
new file mode 100755
index 00000000..387266c1
--- /dev/null
+++ b/config/tests/three_data_hall/delete.bash
@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+
+set -eu
+
+kubectl delete fdb -l cluster-group=test-cluster
diff --git a/config/tests/three_data_hall/final.yaml b/config/tests/three_data_hall/final.yaml
new file mode 100644
index 00000000..96df4add
--- /dev/null
+++ b/config/tests/three_data_hall/final.yaml
@@ -0,0 +1,40 @@
+# This file provides an example of a cluster you can run in a local testing
+# environment to create a simulation of a three_data_hall cluster.
+#
+# This requires variables to be interpolated for $az and $connectionString
+apiVersion: apps.foundationdb.org/v1beta2
+kind: FoundationDBCluster
+metadata:
+  labels:
+    cluster-group: test-cluster
+  name: test-cluster-$az
+spec:
+  version: 7.1.26
+  faultDomain:
+    key: foundationdb.org/none
+  processGroupIDPrefix: $az
+  dataHall: $az
+  processCounts:
+    stateless: -1
+  seedConnectionString: $connectionString
+  databaseConfiguration:
+    redundancy_mode: "three_data_hall"
+  processes:
+    general:
+      customParameters:
+      - "knob_disable_posix_kernel_aio=1"
+      volumeClaimTemplate:
+        spec:
+          resources:
+            requests:
+              storage: "16G"
+      podTemplate:
+        spec:
+          nodeSelector:
+            "topology.kubernetes.io/zone": "$az"
+          containers:
+            - name: foundationdb
+              resources:
+                requests:
+                  cpu: 250m
+                  memory: 128Mi
diff --git a/config/tests/three_data_hall/functions.bash b/config/tests/three_data_hall/functions.bash
new file mode 100644
index 00000000..a2fc3626
--- /dev/null
+++ b/config/tests/three_data_hall/functions.bash
@@ -0,0 +1,29 @@
+function applyFile() {
+	az=${2}
+
+	az="${az}" connectionString="${3}" envsubst < "${1}"| kubectl apply -f -
+}
+
+function checkReconciliation() {
+	clusterName=$1
+
+	generationsOutput=$(kubectl get fdb "${clusterName}" -o jsonpath='{.metadata.generation} {.status.generations.reconciled}')
+	read -ra generations <<< "${generationsOutput}"
+	if [[ ("${#generations[@]}" -ge 2) && ("${generations[0]}" == "${generations[1]}") ]]; then
+		return 1
+	else
+		echo "Latest generations for $clusterName: $generationsOutput"
+		return 0
+	fi
+}
+
+function getConnectionString() {
+	kubectl get fdb "${1}" -o jsonpath='{.status.connectionString}'
+}
+
+function checkReconciliationLoop() {
+	while checkReconciliation "${1}" ; do
+		echo "Waiting for reconciliation"
+		sleep 5
+	done
+}
diff --git a/config/tests/three_data_hall/stage_1.yaml b/config/tests/three_data_hall/stage_1.yaml
new file mode 100644
index 00000000..5e272ff1
--- /dev/null
+++ b/config/tests/three_data_hall/stage_1.yaml
@@ -0,0 +1,40 @@
+# This file provides an example of a cluster you can run in a local testing
+# environment to create a simulation of a three_data_hall cluster.
+#
+# This requires variables to be interpolated for $az and $connectionString
+apiVersion: apps.foundationdb.org/v1beta2
+kind: FoundationDBCluster
+metadata:
+  labels:
+    cluster-group: test-cluster
+  name: test-cluster-$az
+spec:
+  version: 7.1.26
+  faultDomain:
+    key: foundationdb.org/none
+  processGroupIDPrefix: $az
+  dataHall: $az
+  processCounts:
+    stateless: -1
+  seedConnectionString: $connectionString
+  databaseConfiguration:
+    redundancy_mode: "triple"
+  processes:
+    general:
+      customParameters:
+      - "knob_disable_posix_kernel_aio=1"
+      volumeClaimTemplate:
+        spec:
+          resources:
+            requests:
+              storage: "16G"
+      podTemplate:
+        spec:
+          nodeSelector:
+            "topology.kubernetes.io/zone": "$az"
+          containers:
+            - name: foundationdb
+              resources:
+                requests:
+                  cpu: 250m
+                  memory: 128Mi
diff --git a/controllers/change_coordinators_test.go b/controllers/change_coordinators_test.go
index 56cc0125..8e52efc8 100644
--- a/controllers/change_coordinators_test.go
+++ b/controllers/change_coordinators_test.go
@@ -241,17 +241,16 @@ var _ = Describe("Change coordinators", func() {
 
 			JustBeforeEach(func() {
 				cluster.Spec.DatabaseConfiguration.UsableRegions = 2
-				cluster.Spec.DataCenter = "primary"
+				cluster.Spec.DataCenter = "dc0"
 				cluster.Spec.ProcessGroupsToRemove = removals
+				setDatabaseConfiguration(cluster, satCnt)
 
 				var err error
 				status, err = adminClient.GetStatus()
 				Expect(err).NotTo(HaveOccurred())
+				status.Cluster.Processes = generateProcessInfoForMultiRegion(dcCnt, satCnt, excludes)
 
-				// generate status for 2 dcs and 1 sate
-				status.Cluster.Processes = generateProcessInfo(dcCnt, satCnt, excludes)
-
-				candidates, err = selectCoordinators(logr.Discard(), cluster, status)
+				candidates, err = selectCoordinators(testLogger, cluster, status)
 				if shouldFail {
 					Expect(err).To(HaveOccurred())
 				} else {
@@ -629,6 +628,74 @@ var _ = Describe("Change coordinators", func() {
 				})
 			})
 		})
+
+		When("using a FDB cluster with three_data_hall", func() {
+			var status *fdbv1beta2.FoundationDBStatus
+			var candidates []locality.Info
+
+			JustBeforeEach(func() {
+				cluster.Spec.DataHall = "az1"
+				cluster.Spec.DatabaseConfiguration.RedundancyMode = fdbv1beta2.RedundancyModeThreeDataHall
+
+				var err error
+				status, err = adminClient.GetStatus()
+				Expect(err).NotTo(HaveOccurred())
+
+				status.Cluster.Processes = generateProcessInfoForThreeDataHall(3, nil)
+
+				candidates, err = selectCoordinators(logr.Discard(), cluster, status)
+				Expect(err).NotTo(HaveOccurred())
+			})
+
+			When("all processes are healthy", func() {
+				It("should only select storage processes", func() {
+					Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 9))
+					Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+
+					dataHallCount := map[string]int{}
+					for _, candidate := range candidates {
+						Expect(candidate.ID).To(ContainSubstring("storage"))
+						dataHallCount[strings.Split(candidate.ID, "-")[0]]++
+					}
+
+					Expect(dataHallCount).To(Equal(map[string]int{
+						"datahall0": 3,
+						"datahall1": 3,
+						"datahall2": 3,
+					}))
+				})
+			})
+
+			When("when one storage process is marked for removal", func() {
+				removedProcess := fdbv1beta2.ProcessGroupID("storage-2")
+
+				BeforeEach(func() {
+					cluster.Spec.ProcessGroupsToRemove = []fdbv1beta2.ProcessGroupID{
+						removedProcess,
+					}
+					Expect(cluster.ProcessGroupIsBeingRemoved(removedProcess)).To(BeTrue())
+				})
+
+				It("should only select storage processes and exclude the removed process", func() {
+					Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 9))
+					Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+
+					// Only select Storage processes since we select 3 processes and we have 4 storage processes
+					dataHallCount := map[string]int{}
+					for _, candidate := range candidates {
+						Expect(candidate.ID).NotTo(Equal(removedProcess))
+						Expect(candidate.ID).To(ContainSubstring("storage"))
+						dataHallCount[strings.Split(candidate.ID, "-")[0]]++
+					}
+
+					Expect(dataHallCount).To(Equal(map[string]int{
+						"datahall0": 3,
+						"datahall1": 3,
+						"datahall2": 3,
+					}))
+				})
+			})
+		})
 	})
 
 	Describe("reconcile", func() {
@@ -725,30 +792,49 @@ var _ = Describe("Change coordinators", func() {
 	})
 })
 
-func generateProcessInfo(dcCount int, satCount int, excludes []string) map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo {
+func generateProcessInfoForMultiRegion(dcCount int, satCount int, excludes []string) map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo {
 	res := map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo{}
 	logCnt := 4
 
 	for i := 0; i < dcCount; i++ {
 		dcid := fmt.Sprintf("dc%d", i)
 
-		generateProcessInfoDetails(res, dcid, 8, excludes, fdbv1beta2.ProcessClassStorage)
-		generateProcessInfoDetails(res, dcid, logCnt, excludes, fdbv1beta2.ProcessClassLog)
+		generateProcessInfoDetails(res, dcid, "", 8, excludes, fdbv1beta2.ProcessClassStorage)
+		generateProcessInfoDetails(res, dcid, "", logCnt, excludes, fdbv1beta2.ProcessClassLog)
 	}
 
 	for i := 0; i < satCount; i++ {
-		dcid := fmt.Sprintf("sat%d", i)
+		generateProcessInfoDetails(res, fmt.Sprintf("sat%d", i), "", logCnt, excludes, fdbv1beta2.ProcessClassLog)
+	}
+
+	return res
+}
+
+func generateProcessInfoForThreeDataHall(dataHallCount int, excludes []string) map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo {
+	res := map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo{}
+	logCnt := 4
 
-		generateProcessInfoDetails(res, dcid, logCnt, excludes, fdbv1beta2.ProcessClassLog)
+	for i := 0; i < dataHallCount; i++ {
+		dataHallID := fmt.Sprintf("datahall%d", i)
+		generateProcessInfoDetails(res, "", dataHallID, 8, excludes, fdbv1beta2.ProcessClassStorage)
+		generateProcessInfoDetails(res, "", dataHallID, logCnt, excludes, fdbv1beta2.ProcessClassLog)
 	}
 
 	return res
 }
 
-func generateProcessInfoDetails(res map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo, dcID string, cnt int, excludes []string, pClass fdbv1beta2.ProcessClass) {
+func generateProcessInfoDetails(res map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo, dcID string, dataHall string, cnt int, excludes []string, pClass fdbv1beta2.ProcessClass) {
 	for idx := 0; idx < cnt; idx++ {
 		excluded := false
-		zoneID := fmt.Sprintf("%s-%s-%d", dcID, pClass, idx)
+		var zoneID string
+
+		if dcID != "" {
+			zoneID = fmt.Sprintf("%s-%s-%d", dcID, pClass, idx)
+		}
+
+		if dataHall != "" {
+			zoneID = fmt.Sprintf("%s-%s-%d", dataHall, pClass, idx)
+		}
 
 		for _, exclude := range excludes {
 			if exclude != zoneID {
@@ -760,12 +846,11 @@ func generateProcessInfoDetails(res map[fdbv1beta2.ProcessGroupID]fdbv1beta2.Fou
 		}
 
 		addr := fmt.Sprintf("1.1.1.%d:4501", len(res))
-		res[fdbv1beta2.ProcessGroupID(zoneID)] = fdbv1beta2.FoundationDBStatusProcessInfo{
+		processInfo := fdbv1beta2.FoundationDBStatusProcessInfo{
 			ProcessClass: pClass,
 			Locality: map[string]string{
 				fdbv1beta2.FDBLocalityInstanceIDKey: zoneID,
 				fdbv1beta2.FDBLocalityZoneIDKey:     zoneID,
-				fdbv1beta2.FDBLocalityDCIDKey:       dcID,
 			},
 			Excluded: excluded,
 			Address: fdbv1beta2.ProcessAddress{
@@ -774,5 +859,83 @@ func generateProcessInfoDetails(res map[fdbv1beta2.ProcessGroupID]fdbv1beta2.Fou
 			},
 			CommandLine: fmt.Sprintf("/fdbserver --public_address=%s", addr),
 		}
+
+		if dcID != "" {
+			processInfo.Locality[fdbv1beta2.FDBLocalityDCIDKey] = dcID
+		}
+
+		if dataHall != "" {
+			processInfo.Locality[fdbv1beta2.FDBLocalityDataHallKey] = dataHall
+		}
+
+		res[fdbv1beta2.ProcessGroupID(zoneID)] = processInfo
+	}
+}
+
+func setDatabaseConfiguration(cluster *fdbv1beta2.FoundationDBCluster, satCnt int) {
+	if satCnt == 1 {
+		cluster.Spec.DatabaseConfiguration.Regions = []fdbv1beta2.Region{
+			{
+				DataCenters: []fdbv1beta2.DataCenter{
+					{
+						ID: "dc0",
+					},
+					{
+						ID:        "sat0",
+						Satellite: 1,
+					},
+					{
+						ID:        "dc1",
+						Satellite: 1,
+						Priority:  1,
+					},
+				},
+			},
+			{
+				DataCenters: []fdbv1beta2.DataCenter{
+					{
+						ID: "dc1",
+					},
+					{
+						ID:        "sat0",
+						Satellite: 1,
+					},
+					{
+						ID:        "dc0",
+						Satellite: 1,
+						Priority:  1,
+					},
+				},
+			},
+		}
+		return
+	}
+
+	if satCnt == 2 {
+		cluster.Spec.DatabaseConfiguration.Regions = []fdbv1beta2.Region{
+			{
+				DataCenters: []fdbv1beta2.DataCenter{
+					{
+						ID: "dc0",
+					},
+					{
+						ID:        "sat0",
+						Satellite: 1,
+					},
+				},
+			},
+			{
+				DataCenters: []fdbv1beta2.DataCenter{
+					{
+						ID:       "dc1",
+						Priority: 1,
+					},
+					{
+						ID:        "sat1",
+						Satellite: 1,
+					},
+				},
+			},
+		}
 	}
 }
diff --git a/controllers/exclude_processes_test.go b/controllers/exclude_processes_test.go
index 231960cc..8e7cfc62 100644
--- a/controllers/exclude_processes_test.go
+++ b/controllers/exclude_processes_test.go
@@ -63,7 +63,7 @@ var _ = Describe("exclude_processes", func() {
 			allowedExclusions, missingProcesses = canExcludeNewProcesses(globalControllerLogger, cluster, fdbv1beta2.ProcessClassStorage, processCounts.Storage, ongoingExclusions, false)
 		})
 
-		FWhen("using a small cluster", func() {
+		When("using a small cluster", func() {
 			When("all processes are healthy", func() {
 				When("no additional processes are running", func() {
 					It("should not allow the exclusion", func() {
diff --git a/controllers/suite_test.go b/controllers/suite_test.go
index 415a7601..a479ce43 100644
--- a/controllers/suite_test.go
+++ b/controllers/suite_test.go
@@ -22,6 +22,7 @@ package controllers
 
 import (
 	"context"
+	"github.com/go-logr/logr"
 	"testing"
 	"time"
 
@@ -67,9 +68,11 @@ func TestAPIs(t *testing.T) {
 	RunSpecs(t, "FDB Controllers")
 }
 
+var testLogger logr.Logger
+
 var _ = BeforeSuite(func() {
 	logf.SetLogger(zap.New(zap.UseDevMode(true), zap.WriteTo(GinkgoWriter)))
-
+	testLogger = logf.Log
 	Expect(scheme.AddToScheme(scheme.Scheme)).NotTo(HaveOccurred())
 	Expect(fdbv1beta2.AddToScheme(scheme.Scheme)).NotTo(HaveOccurred())
 
diff --git a/e2e/fixtures/factory.go b/e2e/fixtures/factory.go
index 8e350712..0c268202 100644
--- a/e2e/fixtures/factory.go
+++ b/e2e/fixtures/factory.go
@@ -824,6 +824,12 @@ func (factory *Factory) GetOperatorImage() string {
 	return prependRegistry(factory.options.registry, factory.options.operatorImage)
 }
 
+// GetDataLoaderImage returns the dataloader image provided via command line. If a registry was defined the registry will be
+// prepended.
+func (factory *Factory) GetDataLoaderImage() string {
+	return prependRegistry(factory.options.registry, factory.options.dataLoaderImage)
+}
+
 // GetSidecarImage returns the sidecar image provided via command line. If a registry was defined the registry will be
 // prepended.
 func (factory *Factory) GetSidecarImage() string {
diff --git a/e2e/fixtures/fdb_cluster.go b/e2e/fixtures/fdb_cluster.go
index ee57bd67..87e369da 100644
--- a/e2e/fixtures/fdb_cluster.go
+++ b/e2e/fixtures/fdb_cluster.go
@@ -25,6 +25,7 @@ import (
 	"fmt"
 	"golang.org/x/sync/errgroup"
 	"log"
+	"math"
 	"strconv"
 	"strings"
 	"sync"
@@ -1320,3 +1321,33 @@ func (fdbCluster *FdbCluster) UpgradeAndVerify(version string) {
 	gomega.Expect(fdbCluster.UpgradeCluster(version, true)).NotTo(gomega.HaveOccurred())
 	fdbCluster.VerifyVersion(version)
 }
+
+// EnsureTeamTrackersAreHealthy will check if the machine-readable status suggest that the team trackers are healthy
+// and all data is present.
+func (fdbCluster *FdbCluster) EnsureTeamTrackersAreHealthy() {
+	gomega.Eventually(func() bool {
+		for _, tracker := range fdbCluster.GetStatus().Cluster.Data.TeamTrackers {
+			if !tracker.State.Healthy {
+				return false
+			}
+		}
+
+		return true
+	}).WithTimeout(1 * time.Minute).WithPolling(1 * time.Second).MustPassRepeatedly(5).Should(gomega.BeTrue())
+}
+
+// EnsureTeamTrackersHaveMinReplicas will check if the machine-readable status suggest that the team trackers min_replicas
+// match the expected replicas.
+func (fdbCluster *FdbCluster) EnsureTeamTrackersHaveMinReplicas() {
+	desiredFaultTolerance := fdbCluster.GetCachedCluster().DesiredFaultTolerance()
+	gomega.Eventually(func() int {
+		minReplicas := math.MaxInt
+		for _, tracker := range fdbCluster.GetStatus().Cluster.Data.TeamTrackers {
+			if minReplicas > tracker.State.MinReplicasRemaining {
+				minReplicas = tracker.State.MinReplicasRemaining
+			}
+		}
+
+		return minReplicas
+	}).WithTimeout(1 * time.Minute).WithPolling(1 * time.Second).MustPassRepeatedly(5).Should(gomega.BeNumerically(">=", desiredFaultTolerance))
+}
diff --git a/e2e/fixtures/fdb_data_loader.go b/e2e/fixtures/fdb_data_loader.go
new file mode 100644
index 00000000..123ced2d
--- /dev/null
+++ b/e2e/fixtures/fdb_data_loader.go
@@ -0,0 +1,273 @@
+/*
+ * fdb_data_loader.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2023 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package fixtures
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"github.com/onsi/gomega"
+	"io"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/serializer/yaml"
+	yamlutil "k8s.io/apimachinery/pkg/util/yaml"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"text/template"
+	"time"
+)
+
+const (
+	// The name of the data loader Job.
+	dataLoaderName = "fdb-data-loader"
+
+	// For now we only load 2GB into the cluster, we can increase this later if we want.
+	dataLoaderJob = `apiVersion: batch/v1
+kind: Job
+metadata:
+  name: {{ .Name }}
+  namespace: {{ .Namespace }}
+  labels:
+    app: {{ .Name }}
+spec:
+  backoffLimit: 2
+  completions: 2
+  parallelism: 2
+  template:
+    spec:
+      containers:
+      - image: {{ .Image }}
+        imagePullPolicy: Always
+        name: {{ .Name }}
+        # This configuration will load ~1GB per data loader.
+        args:
+        - --keys=1000000
+        - --batch-size=50
+        - --value-size=1000
+        env:
+          - name: FDB_CLUSTER_FILE
+            value: /var/dynamic/fdb/fdb.cluster
+          - name: FDB_TLS_CERTIFICATE_FILE
+            value: /tmp/fdb-certs/tls.crt
+          - name: FDB_TLS_CA_FILE
+            value: /tmp/fdb-certs/ca.pem
+          - name: FDB_TLS_KEY_FILE
+            value: /tmp/fdb-certs/tls.key
+          # FDB 7.3 adds a check for loading external client library, which doesn't work with 6.3.
+          # Consider remove this option once 6.3 is no longer being used.
+          - name: FDB_NETWORK_OPTION_IGNORE_EXTERNAL_CLIENT_FAILURES
+            value: ""
+          - name: LD_LIBRARY_PATH
+            value: /var/dynamic/fdb/primary/lib
+          - name: FDB_NETWORK_OPTION_TRACE_LOG_GROUP
+            value: {{ .Name }}
+          - name: FDB_NETWORK_OPTION_EXTERNAL_CLIENT_DIRECTORY
+            value: /var/dynamic/fdb/libs
+          - name: PYTHONUNBUFFERED
+            value: "on"
+        volumeMounts:
+          - name: config-map
+            mountPath: /var/dynamic-conf
+          - name: fdb-libs
+            mountPath: /var/dynamic/fdb
+          - name: fdb-certs
+            mountPath: /tmp/fdb-certs
+            readOnly: true
+        resources:
+         requests:
+           cpu: "1"
+           memory: 4Gi
+      initContainers:
+        {{ range $index, $version := .SidecarVersions }}
+        - name: foundationdb-kubernetes-init-{{ $index }}
+          image: {{ .BaseImage }}:{{ .SidecarTag}}
+          imagePullPolicy: Always
+          command:
+            - /bin/bash
+          # This is a workaround for a change of the version schema that was never tested/supported
+          args:
+            - -c
+            - echo "{{ .FDBVersion.String }}" > /var/fdb/version && runuser -u fdb -g fdb -- /entrypoint.bash --copy-library {{ .FDBVersion.Compact }} --output-dir /var/output-files/{{ .FDBVersion.Compact }} --init-mode
+          volumeMounts:
+            - name: fdb-libs
+              mountPath: /var/output-files
+          securityContext:
+            runAsUser: 0
+            runAsGroup: 0
+        # Install this library in a special location to force the operator to use it as the primary library.
+        {{ if eq .FDBVersion.Compact "7.1" }}
+        - name: foundationdb-kubernetes-init-7-1-primary
+          image: {{ .BaseImage }}:{{ .SidecarTag}}
+          imagePullPolicy: {{ .ImagePullPolicy }}
+          args:
+            # Note that we are only copying a library, rather than copying any binaries. 
+            - "--copy-library"
+            - "{{ .FDBVersion.Compact }}"
+            - "--output-dir"
+            - "/var/output-files/primary" # Note that we use primary as the subdirectory rather than specifying the FoundationDB version like we did in the other examples.
+            - "--init-mode"
+          volumeMounts:
+            - name: fdb-libs
+              mountPath: /var/output-files
+        {{ end }}
+        {{ end }}
+        - image: {{ .Image }}
+          imagePullPolicy: Always
+          name: fdb-lib-copy
+          command:
+            - /bin/bash
+          args:
+            - -c
+            - mkdir -p /var/dynamic/fdb/libs && {{ range $index, $version := .SidecarVersions -}} cp /var/dynamic/fdb/{{ .FDBVersion.Compact }}/lib/libfdb_c.so /var/dynamic/fdb/libs/libfdb_{{ .FDBVersion.Compact }}_c.so && {{ end }} cp /var/dynamic-conf/fdb.cluster /var/dynamic/fdb/fdb.cluster
+          volumeMounts:
+          - name: config-map
+            mountPath: /var/dynamic-conf
+          - name: fdb-libs
+            mountPath: /var/dynamic/fdb
+          - name: fdb-certs
+            mountPath: /tmp/fdb-certs
+            readOnly: true
+      restartPolicy: Never
+      volumes:
+        - name: config-map
+          configMap:
+            name: {{ .ClusterName }}-config
+            items:
+              - key: cluster-file
+                path: fdb.cluster
+        - name: fdb-libs
+          emptyDir: {}
+        - name: fdb-certs
+          secret:
+            secretName: {{ .SecretName }}`
+)
+
+// dataLoaderConfig represents the configuration of the Dataloader Job.
+type dataLoaderConfig struct {
+	// Name of the data loader Job.
+	Name string
+	// Image represents the data loader image that should be used in the Job.
+	Image string
+	// SidecarVersions represents the sidecar configurations for different FoundationDB versions.
+	SidecarVersions []SidecarConfig
+	// Namespace represents the namespace for the Deployment and all associated resources
+	Namespace string
+	// ClusterName the name of the cluster to load data into.
+	ClusterName string
+	// SecretName represents the Kubernetes secret that contains the certificates for communicating with the FoundationDB
+	// cluster.
+	SecretName string
+}
+
+func (factory *Factory) getDataLoaderConfig(cluster *FdbCluster) *dataLoaderConfig {
+	return &dataLoaderConfig{
+		Name:            dataLoaderName,
+		Image:           factory.GetDataLoaderImage(),
+		Namespace:       cluster.Namespace(),
+		SidecarVersions: factory.GetSidecarConfigs(),
+		ClusterName:     cluster.Name(),
+		SecretName:      factory.GetSecretName(),
+	}
+}
+
+// CreateDataLoaderIfAbsent will create the data loader for the provided cluster and load some random data into the cluster.
+func (factory *Factory) CreateDataLoaderIfAbsent(cluster *FdbCluster) {
+	if !factory.options.enableDataLoading {
+		return
+	}
+
+	t, err := template.New("dataLoaderJob").Parse(dataLoaderJob)
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+	buf := bytes.Buffer{}
+	gomega.Expect(t.Execute(&buf, factory.getDataLoaderConfig(cluster))).NotTo(gomega.HaveOccurred())
+	decoder := yamlutil.NewYAMLOrJSONDecoder(&buf, 100000)
+	for {
+		var rawObj runtime.RawExtension
+		err := decoder.Decode(&rawObj)
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				break
+			}
+			gomega.Expect(err).NotTo(gomega.HaveOccurred())
+		}
+
+		obj, _, err := yaml.NewDecodingSerializer(unstructured.UnstructuredJSONScheme).
+			Decode(rawObj.Raw, nil, nil)
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+		unstructuredMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
+		unstructuredObj := &unstructured.Unstructured{Object: unstructuredMap}
+
+		gomega.Expect(
+			factory.CreateIfAbsent(unstructuredObj),
+		).NotTo(gomega.HaveOccurred())
+	}
+
+	factory.WaitUntilDataLoaderIsDone(cluster)
+}
+
+// WaitUntilDataLoaderIsDone will wait until the data loader Job has finished.
+func (factory *Factory) WaitUntilDataLoaderIsDone(cluster *FdbCluster) {
+	gomega.Eventually(func() int {
+		pods := &corev1.PodList{}
+		gomega.Expect(
+			factory.controllerRuntimeClient.List(
+				context.Background(),
+				pods,
+				client.InNamespace(cluster.Namespace()),
+				client.MatchingLabels(map[string]string{"job-name": dataLoaderName}),
+			),
+		).NotTo(gomega.HaveOccurred())
+
+		var runningPods int
+		for _, pod := range pods.Items {
+			if pod.Status.Phase == corev1.PodRunning {
+				runningPods++
+			}
+		}
+
+		return runningPods
+	}).WithTimeout(5 * time.Minute).WithPolling(5 * time.Second).Should(gomega.BeNumerically(">", 0))
+
+	// Wait for at most 15 minutes to let the data load complete.
+	gomega.Eventually(func() corev1.ConditionStatus {
+		job := &batchv1.Job{}
+		gomega.Expect(
+			factory.controllerRuntimeClient.Get(
+				context.Background(),
+				client.ObjectKey{
+					Namespace: cluster.Namespace(),
+					Name:      dataLoaderName,
+				},
+				job),
+		).NotTo(gomega.HaveOccurred())
+
+		for _, condition := range job.Status.Conditions {
+			if condition.Type == batchv1.JobComplete {
+				return condition.Status
+			}
+		}
+
+		return corev1.ConditionUnknown
+	}).WithTimeout(15 * time.Minute).WithPolling(5 * time.Second).Should(gomega.Equal(corev1.ConditionTrue))
+}
diff --git a/e2e/fixtures/fixtures.go b/e2e/fixtures/fixtures.go
index ccc194ba..d90be2b7 100644
--- a/e2e/fixtures/fixtures.go
+++ b/e2e/fixtures/fixtures.go
@@ -70,12 +70,8 @@ func CheckInvariant(
 	threshold time.Duration,
 	f func() error,
 ) error {
-	err := f()
-	if err != nil {
-		return fmt.Errorf("invariant %s not true at beginning of test: %w", invariantName, err)
-	}
 	var waitGroup sync.WaitGroup
-	ticker := time.NewTicker(100 * time.Millisecond)
+	ticker := time.NewTicker(250 * time.Millisecond)
 	var last error
 	testFailed := false
 	quit := make(chan struct{})
diff --git a/e2e/fixtures/options.go b/e2e/fixtures/options.go
index bef9ca18..5d04c6b8 100644
--- a/e2e/fixtures/options.go
+++ b/e2e/fixtures/options.go
@@ -38,6 +38,7 @@ type FactoryOptions struct {
 	fdbImage                    string // TODO (johscheuer): Make this optional if we use the default
 	sidecarImage                string // TODO (johscheuer): Make this optional if we use the default
 	operatorImage               string
+	dataLoaderImage             string
 	registry                    string
 	fdbVersion                  string
 	username                    string
@@ -45,6 +46,7 @@ type FactoryOptions struct {
 	upgradeString               string
 	cloudProvider               string
 	enableChaosTests            bool
+	enableDataLoading           bool
 	cleanup                     bool
 	featureOperatorDNS          bool
 	featureOperatorLocalities   bool
@@ -133,17 +135,29 @@ func (options *FactoryOptions) BindFlags(fs *flag.FlagSet) {
 		"",
 		"defines the cloud provider used for the Kubernetes cluster, for defining cloud provider specific behaviour. Currently only Kind is support.",
 	)
+	fs.StringVar(
+		&options.dataLoaderImage,
+		"data-loader-image",
+		"foundationdb/fdb-data-loader:latest",
+		"defines the data loader image that should be used for testing",
+	)
 	fs.BoolVar(
 		&options.enableChaosTests,
 		"enable-chaos-tests",
 		true,
-		"defines if the chaos tests should be enabled in operator related tests cases",
+		"defines if the chaos tests should be enabled in operator related tests cases.",
+	)
+	fs.BoolVar(
+		&options.enableDataLoading,
+		"enable-data-load",
+		true,
+		"defines if the created FDB cluster should be loaded with some random data.",
 	)
 	fs.BoolVar(
 		&options.cleanup,
 		"cleanup",
 		true,
-		"defines if the test namespace should be removed after the test run",
+		"defines if the test namespace should be removed after the test run.",
 	)
 	fs.BoolVar(
 		&options.featureOperatorUnifiedImage,
diff --git a/e2e/fixtures/singleton.go b/e2e/fixtures/singleton.go
index d5680875..11db701e 100644
--- a/e2e/fixtures/singleton.go
+++ b/e2e/fixtures/singleton.go
@@ -21,13 +21,15 @@
 package fixtures
 
 import (
+	"math/rand"
+	"os/user"
+	"time"
+
 	chaosmesh "github.com/chaos-mesh/chaos-mesh/api/v1alpha1"
 	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes/scheme"
-	"math/rand"
-	"os/user"
-	"time"
 
 	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
 	corev1 "k8s.io/api/core/v1"
@@ -104,6 +106,10 @@ func getSingleton(options *FactoryOptions) (*singleton, error) {
 	if err != nil {
 		return nil, err
 	}
+	err = batchv1.AddToScheme(curScheme)
+	if err != nil {
+		return nil, err
+	}
 	err = chaosmesh.AddToScheme(curScheme)
 	if err != nil {
 		return nil, err
diff --git a/e2e/fixtures/status.go b/e2e/fixtures/status.go
index a54e0cf6..a0555676 100644
--- a/e2e/fixtures/status.go
+++ b/e2e/fixtures/status.go
@@ -263,6 +263,7 @@ func (fdbCluster FdbCluster) StatusInvariantChecker(
 // checkAvailability returns nil if the cluster is reachable. If the cluster is unreachable an error will be returned.
 func checkAvailability(status *fdbv1beta2.FoundationDBStatus) error {
 	if !status.Client.DatabaseStatus.Available {
+		log.Println("client messages", status.Client.Messages, "cluster messages", status.Cluster.Messages)
 		return fmt.Errorf("cluster is not available")
 	}
 
@@ -284,9 +285,9 @@ func (fdbCluster FdbCluster) InvariantClusterStatusAvailableWithThreshold(
 func (fdbCluster FdbCluster) InvariantClusterStatusAvailable() error {
 	return fdbCluster.StatusInvariantChecker(
 		"InvariantClusterStatusAvailable",
-		// Per default we allow 5 seconds unavailability. Otherwise we could get a few test failures when we do operations
+		// Per default we allow 15 seconds unavailability. Otherwise we could get a few test failures when we do operations
 		// like a replacement on a transaction system Pod and the recovery takes longer.
-		5*time.Second,
+		15*time.Second,
 		checkAvailability,
 	)
 }
diff --git a/e2e/fixtures/upgrade_test_configuration.go b/e2e/fixtures/upgrade_test_configuration.go
index 401895d1..cbf43cbd 100644
--- a/e2e/fixtures/upgrade_test_configuration.go
+++ b/e2e/fixtures/upgrade_test_configuration.go
@@ -29,7 +29,7 @@ import (
 )
 
 // UpgradeTestConfiguration represents the configuration for an upgrade test. This includes the initial FoundationDB version
-// and the target FoundationDB version to upgrade the cluster to.
+// and the target FoundationDB version to upgrade or downgrade the cluster to.
 type UpgradeTestConfiguration struct {
 	// InitialVersion represents the version before the upgrade.
 	InitialVersion fdbv1beta2.Version
@@ -37,7 +37,7 @@ type UpgradeTestConfiguration struct {
 	TargetVersion fdbv1beta2.Version
 }
 
-func parseUpgradeVersionPair(upgradeConfig string) UpgradeTestConfiguration {
+func parseUpgradeVersionPair(upgradeConfig string) *UpgradeTestConfiguration {
 	versions := strings.Split(upgradeConfig, ":")
 	if len(versions) != 2 {
 		log.Fatalf(
@@ -46,6 +46,11 @@ func parseUpgradeVersionPair(upgradeConfig string) UpgradeTestConfiguration {
 		)
 	}
 
+	// If both versions are the same ignore it.
+	if versions[0] == versions[1] {
+		return nil
+	}
+
 	initialVersion, err := fdbv1beta2.ParseFdbVersion(versions[0])
 	if err != nil {
 		log.Fatalf("\"%s\" is not a valid FDB version", versions[0])
@@ -56,11 +61,11 @@ func parseUpgradeVersionPair(upgradeConfig string) UpgradeTestConfiguration {
 		log.Fatalf("\"%s\" is not a valid FDB version", versions[1])
 	}
 
-	if initialVersion.IsAtLeast(targetVersion) {
-		log.Fatalf("downgrade from \"%s\" to \"%s\" is not supported", versions[0], versions[1])
+	if !initialVersion.SupportsVersionChange(targetVersion) {
+		log.Fatalf("version change from \"%s\" to \"%s\" is not supported", versions[0], versions[1])
 	}
 
-	return UpgradeTestConfiguration{
+	return &UpgradeTestConfiguration{
 		InitialVersion: initialVersion,
 		TargetVersion:  targetVersion,
 	}
@@ -69,7 +74,7 @@ func parseUpgradeVersionPair(upgradeConfig string) UpgradeTestConfiguration {
 // GetUpgradeVersions returns the upgrade versions as a string slice based on the command line flag. Each entry will be
 // a FoundationDB version. This slice can contain duplicate entries. For upgrade tests it's expected that two versions
 // form one test, e.g. where the odd number is the initial version and the even number is the
-func (factory *Factory) GetUpgradeVersions() []UpgradeTestConfiguration {
+func (factory *Factory) GetUpgradeVersions() []*UpgradeTestConfiguration {
 	return getUpgradeVersions(factory.options.upgradeString)
 }
 
@@ -77,15 +82,20 @@ func (factory *Factory) GetUpgradeVersions() []UpgradeTestConfiguration {
 // a FoundationDB version. This slice can contain duplicate entries. For upgrade tests it's expected that two versions
 // form one test, e.g. where the odd number is the initial version and the even number is the
 // This method is only internally used. Users that import this test suite should use the factory method.
-func getUpgradeVersions(upgradeString string) []UpgradeTestConfiguration {
+func getUpgradeVersions(upgradeString string) []*UpgradeTestConfiguration {
 	if upgradeString == "" {
 		return nil
 	}
 
 	upgradeVersionStrings := strings.Split(upgradeString, ",")
-	upgradeVersions := make([]UpgradeTestConfiguration, 0, len(upgradeVersionStrings))
+	upgradeVersions := make([]*UpgradeTestConfiguration, 0, len(upgradeVersionStrings))
 	for _, upgradeTest := range upgradeVersionStrings {
-		upgradeVersions = append(upgradeVersions, parseUpgradeVersionPair(upgradeTest))
+		upgradeTestConfig := parseUpgradeVersionPair(upgradeTest)
+		if upgradeTestConfig == nil {
+			continue
+		}
+
+		upgradeVersions = append(upgradeVersions, upgradeTestConfig)
 	}
 
 	return upgradeVersions
diff --git a/e2e/test_operator/operator_test.go b/e2e/test_operator/operator_test.go
index 087b51d8..d7f36092 100644
--- a/e2e/test_operator/operator_test.go
+++ b/e2e/test_operator/operator_test.go
@@ -76,6 +76,9 @@ var _ = BeforeSuite(func() {
 		factory.GetClusterOptions()...,
 	)
 
+	// Load some data async into the cluster. We will only block as long as the Job is created.
+	factory.CreateDataLoaderIfAbsent(fdbCluster)
+
 	// In order to test the robustness of the operator we try to kill the operator Pods every minute.
 	if factory.ChaosTestsEnabled() {
 		factory.ScheduleInjectPodKill(
@@ -104,6 +107,9 @@ var _ = Describe("Operator", Label("e2e", "pr"), func() {
 		}
 		Expect(fdbCluster.WaitForReconciliation()).ToNot(HaveOccurred())
 		factory.StopInvariantCheck()
+		// Make sure all data is present in the cluster
+		fdbCluster.EnsureTeamTrackersAreHealthy()
+		fdbCluster.EnsureTeamTrackersHaveMinReplicas()
 	})
 
 	JustBeforeEach(func() {
diff --git a/e2e/test_operator_ha_upgrades/operator_ha_upgrade_test.go b/e2e/test_operator_ha_upgrades/operator_ha_upgrade_test.go
index 7b5dba0e..ea2532b0 100644
--- a/e2e/test_operator_ha_upgrades/operator_ha_upgrade_test.go
+++ b/e2e/test_operator_ha_upgrades/operator_ha_upgrade_test.go
@@ -60,21 +60,34 @@ var _ = AfterSuite(func() {
 	}
 })
 
-func clusterSetupWithHealthCheckOption(beforeVersion string, enableOperatorPodChaos bool, enableHealthCheck bool) {
+type testConfig struct {
+	beforeVersion          string
+	enableOperatorPodChaos bool
+	enableHealthCheck      bool
+	loadData               bool
+}
+
+func clusterSetupWithTestConfig(config testConfig) {
 	// We set the before version here to overwrite the before version from the specific flag
 	// the specific flag will be removed in the future.
-	factory.SetBeforeVersion(beforeVersion)
+	factory.SetBeforeVersion(config.beforeVersion)
 	fdbCluster = factory.CreateFdbHaCluster(
 		fixtures.DefaultClusterConfigWithHaMode(fixtures.HaFourZoneSingleSat, false),
 		factory.GetClusterOptions(fixtures.UseVersionBeforeUpgrade)...,
 	)
-	if enableHealthCheck {
+
+	if config.enableHealthCheck {
 		Expect(
 			fdbCluster.GetPrimary().InvariantClusterStatusAvailableWithThreshold(15 * time.Second),
 		).ShouldNot(HaveOccurred())
 	}
 
-	if enableOperatorPodChaos && factory.ChaosTestsEnabled() {
+	if config.loadData {
+		// Load some data async into the cluster. We will only block as long as the Job is created.
+		factory.CreateDataLoaderIfAbsent(fdbCluster.GetPrimary())
+	}
+
+	if config.enableOperatorPodChaos && factory.ChaosTestsEnabled() {
 		for _, curCluster := range fdbCluster.GetAllClusters() {
 			factory.ScheduleInjectPodKill(
 				fixtures.GetOperatorSelector(curCluster.Namespace()),
@@ -86,7 +99,14 @@ func clusterSetupWithHealthCheckOption(beforeVersion string, enableOperatorPodCh
 }
 
 func clusterSetup(beforeVersion string, enableOperatorPodChaos bool) {
-	clusterSetupWithHealthCheckOption(beforeVersion, enableOperatorPodChaos, true)
+	clusterSetupWithTestConfig(
+		testConfig{
+			beforeVersion:          beforeVersion,
+			enableOperatorPodChaos: enableOperatorPodChaos,
+			enableHealthCheck:      true,
+			loadData:               false,
+		},
+	)
 }
 
 // Verify that bouncing is blocked in primary/remote/primary-satellite clusters.
@@ -148,7 +168,12 @@ var _ = Describe("Operator HA Upgrades", Label("e2e", "pr"), func() {
 	DescribeTable(
 		"with operator pod chaos and without foundationdb pod chaos",
 		func(beforeVersion string, targetVersion string) {
-			clusterSetup(beforeVersion, true /* = enableOperatorPodChaos */)
+			clusterSetupWithTestConfig(testConfig{
+				beforeVersion:          beforeVersion,
+				enableOperatorPodChaos: true,
+				enableHealthCheck:      true,
+				loadData:               true,
+			})
 			initialGeneration := fdbCluster.GetPrimary().GetStatus().Cluster.Generation
 			// Make use of a sync.Map here as we have to modify it concurrently.
 			var transactionSystemProcessGroups sync.Map
@@ -232,6 +257,9 @@ var _ = Describe("Operator HA Upgrades", Label("e2e", "pr"), func() {
 			// may get bounced at different times and also because of cluster controller
 			// change, which happens a number of times, during an upgrade).
 			Expect(finalGeneration).To(BeNumerically("<=", initialGeneration+80))
+			// Make sure the cluster has no data loss
+			fdbCluster.GetPrimary().EnsureTeamTrackersAreHealthy()
+			fdbCluster.GetPrimary().EnsureTeamTrackersHaveMinReplicas()
 		},
 		EntryDescription("Upgrade from %[1]s to %[2]s"),
 		fixtures.GenerateUpgradeTableEntries(testOptions),
@@ -338,6 +366,7 @@ var _ = Describe("Operator HA Upgrades", Label("e2e", "pr"), func() {
 			// Upgrade should make progress now - wait until all processes have upgraded
 			// to "targetVersion".
 			fdbCluster.VerifyVersion(targetVersion)
+			fdbCluster.GetPrimary().EnsureTeamTrackersHaveMinReplicas()
 		},
 		EntryDescription("Upgrade from %s to %s"),
 		fixtures.GenerateUpgradeTableEntries(testOptions),
@@ -400,6 +429,9 @@ var _ = Describe("Operator HA Upgrades", Label("e2e", "pr"), func() {
 				factory.DeletePod(randomPod)
 				return false
 			}).WithTimeout(30 * time.Minute).WithPolling(2 * time.Minute).Should(BeTrue())
+
+			// Make sure the cluster has no data loss
+			fdbCluster.GetPrimary().EnsureTeamTrackersHaveMinReplicas()
 		},
 		EntryDescription(
 			"Upgrade from %s to %s",
@@ -415,7 +447,12 @@ var _ = Describe("Operator HA Upgrades", Label("e2e", "pr"), func() {
 				Skip("chaos mesh is disabled")
 			}
 
-			clusterSetupWithHealthCheckOption(beforeVersion, false, false)
+			clusterSetupWithTestConfig(testConfig{
+				beforeVersion:          beforeVersion,
+				enableOperatorPodChaos: false,
+				enableHealthCheck:      false,
+				loadData:               false,
+			})
 
 			// 1. Introduce packet loss b/w pods.
 			log.Println("Injecting packet loss b/w pod")
@@ -442,6 +479,7 @@ var _ = Describe("Operator HA Upgrades", Label("e2e", "pr"), func() {
 			Expect(fdbCluster.UpgradeCluster(targetVersion, false)).NotTo(HaveOccurred())
 			// Verify that the upgrade proceeds
 			fdbCluster.VerifyVersion(targetVersion)
+			fdbCluster.GetPrimary().EnsureTeamTrackersHaveMinReplicas()
 		},
 		EntryDescription("Upgrade from %[1]s to %[2]s"),
 		fixtures.GenerateUpgradeTableEntries(testOptions),
@@ -484,7 +522,8 @@ var _ = Describe("Operator HA Upgrades", Label("e2e", "pr"), func() {
 			Expect(fdbCluster.UpgradeCluster(targetVersion, false)).NotTo(HaveOccurred())
 			// Verify that the upgrade proceeds
 			fdbCluster.VerifyVersion(targetVersion)
-
+			// Make sure the cluster has no data loss
+			fdbCluster.GetPrimary().EnsureTeamTrackersHaveMinReplicas()
 			// TODO add validation here processes are updated new version
 		},
 		EntryDescription("Upgrade from %[1]s to %[2]s"),
diff --git a/e2e/test_operator_upgrades/operator_upgrades_test.go b/e2e/test_operator_upgrades/operator_upgrades_test.go
index dda041ad..063b0628 100644
--- a/e2e/test_operator_upgrades/operator_upgrades_test.go
+++ b/e2e/test_operator_upgrades/operator_upgrades_test.go
@@ -158,6 +158,9 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 
 			// 5. Verify a final time that the cluster is reconciled, this should be quick.
 			Expect(fdbCluster.WaitForReconciliation()).NotTo(HaveOccurred())
+
+			// Make sure the cluster has no data loss.
+			fdbCluster.EnsureTeamTrackersHaveMinReplicas()
 		},
 
 		EntryDescription("Upgrade from %[1]s to %[2]s with pods deleted during rolling bounce"),
@@ -223,6 +226,9 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 
 			// Allow the operator to restart processes and the upgrade should continue and finish.
 			fdbCluster.SetKillProcesses(true)
+
+			// Make sure the cluster has no data loss.
+			fdbCluster.EnsureTeamTrackersHaveMinReplicas()
 		},
 
 		EntryDescription(
@@ -286,6 +292,8 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 
 			// 5. Upgrade should proceed after we stop killing the sidecar.
 			fdbCluster.VerifyVersion(targetVersion)
+			// Make sure the cluster has no data loss.
+			fdbCluster.EnsureTeamTrackersHaveMinReplicas()
 		},
 		EntryDescription("Upgrade from %[1]s to %[2]s with crash-looping sidecar"),
 		fixtures.GenerateUpgradeTableEntries(testOptions),
@@ -321,6 +329,9 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 			Eventually(func() []string {
 				return fdbCluster.GetStatus().Cluster.IncompatibleConnections
 			}).WithTimeout(5 * time.Minute).WithPolling(5 * time.Second).MustPassRepeatedly(5).Should(BeEmpty())
+
+			// Make sure the cluster has no data loss.
+			fdbCluster.EnsureTeamTrackersHaveMinReplicas()
 		},
 		EntryDescription("Upgrade from %[1]s to %[2]s with one coordinator not being restarted"),
 		fixtures.GenerateUpgradeTableEntries(testOptions),
@@ -374,6 +385,9 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 
 			// The cluster should still be able to upgrade.
 			Expect(fdbCluster.UpgradeCluster(targetVersion, true)).NotTo(HaveOccurred())
+
+			// Make sure the cluster has no data loss.
+			fdbCluster.EnsureTeamTrackersHaveMinReplicas()
 		},
 		EntryDescription("Upgrade from %[1]s to %[2]s and multiple processes are not restarted"),
 		fixtures.GenerateUpgradeTableEntries(testOptions),
@@ -504,6 +518,9 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 
 				return processesToUpdate
 			}).WithTimeout(30 * time.Minute).WithPolling(5 * time.Second).MustPassRepeatedly(5).Should(BeNumerically("==", 0))
+
+			// Make sure the cluster has no data loss.
+			fdbCluster.EnsureTeamTrackersHaveMinReplicas()
 		},
 		EntryDescription("Upgrade from %[1]s to %[2]s"),
 		fixtures.GenerateUpgradeTableEntries(testOptions),
@@ -554,6 +571,9 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 
 			// Make sure we remove the finalizer to not block the clean up.
 			Expect(factory.SetFinalizerForPod(&podMarkedForRemoval, []string{})).ShouldNot(HaveOccurred())
+
+			// Make sure the cluster has no data loss.
+			fdbCluster.EnsureTeamTrackersHaveMinReplicas()
 		},
 		EntryDescription("Upgrade from %[1]s to %[2]s"),
 		fixtures.GenerateUpgradeTableEntries(testOptions),
diff --git a/e2e/test_operator_upgrades_variations/operator_upgrades_variations_test.go b/e2e/test_operator_upgrades_variations/operator_upgrades_variations_test.go
index 433d8b9d..32f15f88 100644
--- a/e2e/test_operator_upgrades_variations/operator_upgrades_variations_test.go
+++ b/e2e/test_operator_upgrades_variations/operator_upgrades_variations_test.go
@@ -53,13 +53,25 @@ var _ = AfterSuite(func() {
 	}
 })
 
-func clusterSetupWithConfig(beforeVersion string, config *fixtures.ClusterConfig) *fixtures.FdbCluster {
-	factory.SetBeforeVersion(beforeVersion)
+type testConfig struct {
+	beforeVersion string
+	targetVersion string
+	clusterConfig *fixtures.ClusterConfig
+	loadData      bool
+}
+
+func clusterSetupWithConfig(config testConfig) *fixtures.FdbCluster {
+	factory.SetBeforeVersion(config.beforeVersion)
 	fdbCluster := factory.CreateFdbCluster(
-		config,
+		config.clusterConfig,
 		factory.GetClusterOptions(fixtures.UseVersionBeforeUpgrade)...,
 	)
 
+	if config.loadData {
+		// Load some data async into the cluster. We will only block as long as the Job is created.
+		factory.CreateDataLoaderIfAbsent(fdbCluster)
+	}
+
 	Expect(
 		fdbCluster.InvariantClusterStatusAvailableWithThreshold(15 * time.Second),
 	).ShouldNot(HaveOccurred())
@@ -67,13 +79,13 @@ func clusterSetupWithConfig(beforeVersion string, config *fixtures.ClusterConfig
 	return fdbCluster
 }
 
-func performUpgrade(beforeVersion string, targetVersion string, config *fixtures.ClusterConfig, validateFunc func(cluster *fixtures.FdbCluster)) {
-	cluster := clusterSetupWithConfig(beforeVersion, config)
+func performUpgrade(config testConfig, validateFunc func(cluster *fixtures.FdbCluster)) {
+	cluster := clusterSetupWithConfig(config)
 	startTime := time.Now()
-	Expect(cluster.UpgradeCluster(targetVersion, false)).NotTo(HaveOccurred())
+	Expect(cluster.UpgradeCluster(config.targetVersion, false)).NotTo(HaveOccurred())
 	validateFunc(cluster)
 
-	if !fixtures.VersionsAreProtocolCompatible(beforeVersion, targetVersion) {
+	if !fixtures.VersionsAreProtocolCompatible(config.beforeVersion, config.targetVersion) {
 		// Ensure that the operator is setting the IncorrectConfigMap and IncorrectCommandLine conditions during the upgrade
 		// process.
 		expectedConditions := map[fdbv1beta2.ProcessGroupConditionType]bool{
@@ -103,7 +115,7 @@ func performUpgrade(beforeVersion string, targetVersion string, config *fixtures
 		}
 
 		// Allow soft reconciliation and make sure the running version was updated
-		return cluster.Status.Generations.Reconciled == cluster.Generation && cluster.Status.RunningVersion == targetVersion
+		return cluster.Status.Generations.Reconciled == cluster.Generation && cluster.Status.RunningVersion == config.targetVersion
 	})).NotTo(HaveOccurred())
 
 	log.Println("Upgrade took:", time.Since(startTime).String())
@@ -116,6 +128,9 @@ func performUpgrade(beforeVersion string, targetVersion string, config *fixtures
 	// replacements during an upgrade.
 	expectedProcessCounts := (processCounts.Total()-processCounts.Storage)*2 + 5
 	Expect(len(transactionSystemProcessGroups)).To(BeNumerically("<=", expectedProcessCounts))
+	// Ensure we have not data loss.
+	cluster.EnsureTeamTrackersAreHealthy()
+	cluster.EnsureTeamTrackersHaveMinReplicas()
 }
 
 var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
@@ -133,8 +148,13 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 	DescribeTable(
 		"upgrading a cluster without chaos",
 		func(beforeVersion string, targetVersion string) {
-			performUpgrade(beforeVersion, targetVersion, &fixtures.ClusterConfig{
-				DebugSymbols: false,
+			performUpgrade(testConfig{
+				beforeVersion: beforeVersion,
+				targetVersion: targetVersion,
+				clusterConfig: &fixtures.ClusterConfig{
+					DebugSymbols: false,
+				},
+				loadData: true,
 			}, func(cluster *fixtures.FdbCluster) {})
 		},
 		EntryDescription("Upgrade from %[1]s to %[2]s"),
@@ -144,9 +164,14 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 	DescribeTable(
 		"with 2 storage servers per Pod",
 		func(beforeVersion string, targetVersion string) {
-			performUpgrade(beforeVersion, targetVersion, &fixtures.ClusterConfig{
-				DebugSymbols:        false,
-				StorageServerPerPod: 2,
+			performUpgrade(testConfig{
+				beforeVersion: beforeVersion,
+				targetVersion: targetVersion,
+				clusterConfig: &fixtures.ClusterConfig{
+					DebugSymbols:        false,
+					StorageServerPerPod: 2,
+				},
+				loadData: false,
 			}, func(cluster *fixtures.FdbCluster) {
 				Expect(cluster.GetCluster().Spec.StorageServersPerPod).To(Equal(2))
 			})
@@ -158,9 +183,14 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 	DescribeTable(
 		"with 2 log servers per Pod",
 		func(beforeVersion string, targetVersion string) {
-			performUpgrade(beforeVersion, targetVersion, &fixtures.ClusterConfig{
-				DebugSymbols:     false,
-				LogServersPerPod: 2,
+			performUpgrade(testConfig{
+				beforeVersion: beforeVersion,
+				targetVersion: targetVersion,
+				clusterConfig: &fixtures.ClusterConfig{
+					DebugSymbols:     false,
+					LogServersPerPod: 2,
+				},
+				loadData: false,
 			}, func(cluster *fixtures.FdbCluster) {
 				Expect(cluster.GetCluster().Spec.LogServersPerPod).To(Equal(2))
 			})
@@ -172,9 +202,14 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 	DescribeTable(
 		"with maintenance mode enabled",
 		func(beforeVersion string, targetVersion string) {
-			performUpgrade(beforeVersion, targetVersion, &fixtures.ClusterConfig{
-				DebugSymbols:       false,
-				UseMaintenanceMode: true,
+			performUpgrade(testConfig{
+				beforeVersion: beforeVersion,
+				targetVersion: targetVersion,
+				clusterConfig: &fixtures.ClusterConfig{
+					DebugSymbols:       false,
+					UseMaintenanceMode: true,
+				},
+				loadData: false,
 			}, func(cluster *fixtures.FdbCluster) {
 				Expect(cluster.GetCluster().UseMaintenaceMode()).To(BeTrue())
 			})
@@ -186,9 +221,14 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 	DescribeTable(
 		"with locality based exclusions",
 		func(beforeVersion string, targetVersion string) {
-			performUpgrade(beforeVersion, targetVersion, &fixtures.ClusterConfig{
-				DebugSymbols:               false,
-				UseLocalityBasedExclusions: true,
+			performUpgrade(testConfig{
+				beforeVersion: beforeVersion,
+				targetVersion: targetVersion,
+				clusterConfig: &fixtures.ClusterConfig{
+					DebugSymbols:               false,
+					UseLocalityBasedExclusions: true,
+				},
+				loadData: false,
 			}, func(cluster *fixtures.FdbCluster) {
 				Expect(cluster.GetCluster().UseLocalitiesForExclusion()).To(BeTrue())
 			})
@@ -206,9 +246,14 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 				Skip(fmt.Sprintf("FoundationDB version: %s, does not support the usage of DNS", beforeVersion))
 			}
 
-			performUpgrade(beforeVersion, targetVersion, &fixtures.ClusterConfig{
-				DebugSymbols: false,
-				UseDNS:       true,
+			performUpgrade(testConfig{
+				beforeVersion: beforeVersion,
+				targetVersion: targetVersion,
+				clusterConfig: &fixtures.ClusterConfig{
+					DebugSymbols: false,
+					UseDNS:       true,
+				},
+				loadData: false,
 			}, func(cluster *fixtures.FdbCluster) {
 				Expect(cluster.GetCluster().UseDNSInClusterFile()).To(BeTrue())
 			})
diff --git a/e2e/test_operator_velocity/operator_velocity_test.go b/e2e/test_operator_velocity/operator_velocity_test.go
index aa0e5ffa..219db0cf 100644
--- a/e2e/test_operator_velocity/operator_velocity_test.go
+++ b/e2e/test_operator_velocity/operator_velocity_test.go
@@ -233,7 +233,6 @@ var _ = Describe("Test Operator Velocity", Label("e2e", "nightly"), func() {
 			).NotTo(HaveOccurred())
 
 			cluster := fdbCluster.GetPrimary().GetCluster()
-
 			CheckKnobRollout(
 				fdbCluster,
 				newGeneralCustomParameters,
@@ -259,9 +258,7 @@ var _ = Describe("Test Operator Velocity", Label("e2e", "nightly"), func() {
 			// Partition a storage Pod from the rest of the cluster
 			pod := fixtures.ChooseRandomPod(fdbCluster.GetPrimary().GetStoragePods())
 			log.Printf("partition Pod: %s", pod.Name)
-			var err error
 			exp = factory.InjectPartition(fixtures.PodSelector(pod))
-			Expect(err).ShouldNot(HaveOccurred())
 		})
 
 		AfterEach(func() {
@@ -287,11 +284,16 @@ var _ = Describe("Test Operator Velocity", Label("e2e", "nightly"), func() {
 				),
 			).NotTo(HaveOccurred())
 
+			// We have to increase the timeout for this test as the maximum time this test might take, depends on the
+			// ordering of the operators taking the kill action. If the operator in the primary cluster is the last one
+			// the test will pass fairly fast. If the primary operator is not the last operator, this test take longer
+			// as the partitioned Pod will block the release of the lock and the next operator has to wait until the
+			// lock is expired.
 			CheckKnobRollout(
 				fdbCluster,
 				newGeneralCustomParameters,
 				newStorageCustomParameters,
-				normalKnobRolloutTimeoutSeconds)
+				normalKnobRolloutTimeoutSeconds+int(fdbCluster.GetPrimary().GetCluster().GetLockDuration().Seconds()))
 		})
 	})
 })
diff --git a/internal/locality/locality_test.go b/internal/locality/locality_test.go
index 31cf1379..32231e1c 100644
--- a/internal/locality/locality_test.go
+++ b/internal/locality/locality_test.go
@@ -22,12 +22,10 @@ package locality
 
 import (
 	"fmt"
+	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/podclient/mock"
 	"math"
 	"net"
 	"strings"
-	"time"
-
-	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/podclient/mock"
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
@@ -118,7 +116,7 @@ func generateDefaultStatus(tls bool) *fdbv1beta2.FoundationDBStatus {
 	}
 }
 
-var _ = Describe("Change coordinators", func() {
+var _ = Describe("Localities", func() {
 	var cluster *fdbv1beta2.FoundationDBCluster
 
 	BeforeEach(func() {
@@ -361,17 +359,87 @@ var _ = Describe("Change coordinators", func() {
 				fdbv1beta2.FDBLocalityZoneIDKey: 1,
 			},
 		),
-		Entry("default cluster with two usable regiosn",
+		Entry("default cluster with two usable regions and 4 DCs",
 			&fdbv1beta2.FoundationDBCluster{
 				Spec: fdbv1beta2.FoundationDBClusterSpec{
 					DatabaseConfiguration: fdbv1beta2.DatabaseConfiguration{
 						UsableRegions: 2,
+						Regions: []fdbv1beta2.Region{
+							{
+								DataCenters: []fdbv1beta2.DataCenter{
+									{
+										ID: "dc1",
+									},
+									{
+										ID: "dc2",
+									},
+								},
+							},
+							{
+								DataCenters: []fdbv1beta2.DataCenter{
+									{
+										ID: "dc3",
+									},
+									{
+										ID: "dc4",
+									},
+								},
+							},
+						},
 					},
 				},
 			},
 			map[string]int{
 				fdbv1beta2.FDBLocalityZoneIDKey: 1,
-				fdbv1beta2.FDBLocalityDCIDKey:   4,
+				fdbv1beta2.FDBLocalityDCIDKey:   3,
+			},
+		),
+		Entry("default cluster with two usable regions and 3 DCs",
+			&fdbv1beta2.FoundationDBCluster{
+				Spec: fdbv1beta2.FoundationDBClusterSpec{
+					DatabaseConfiguration: fdbv1beta2.DatabaseConfiguration{
+						UsableRegions: 2,
+						Regions: []fdbv1beta2.Region{
+							{
+								DataCenters: []fdbv1beta2.DataCenter{
+									{
+										ID: "dc1",
+									},
+									{
+										ID: "dc2",
+									},
+								},
+							},
+							{
+								DataCenters: []fdbv1beta2.DataCenter{
+									{
+										ID: "dc3",
+									},
+									{
+										ID: "dc2",
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			map[string]int{
+				fdbv1beta2.FDBLocalityZoneIDKey: 1,
+				fdbv1beta2.FDBLocalityDCIDKey:   3,
+			},
+		),
+		Entry("default cluster with one usable region and three data hall",
+			&fdbv1beta2.FoundationDBCluster{
+				Spec: fdbv1beta2.FoundationDBClusterSpec{
+					DatabaseConfiguration: fdbv1beta2.DatabaseConfiguration{
+						RedundancyMode: fdbv1beta2.RedundancyModeThreeDataHall,
+					},
+				},
+			},
+			map[string]int{
+				fdbv1beta2.FDBLocalityDataHallKey: 3,
+				fdbv1beta2.FDBLocalityZoneIDKey:   1,
 			},
 		),
 	)
@@ -651,38 +719,6 @@ var _ = Describe("Change coordinators", func() {
 			})
 		})
 
-		When("a process has a process group status with a pending Pod", func() {
-			BeforeEach(func() {
-				status.Cluster.Processes["4"] = fdbv1beta2.FoundationDBStatusProcessInfo{
-					Address: fdbv1beta2.ProcessAddress{
-						IPAddress: net.ParseIP("1.1.1.4"),
-						Port:      4501,
-					},
-					Locality: map[string]string{
-						fdbv1beta2.FDBLocalityInstanceIDKey: "test-4",
-					},
-				}
-
-				cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups,
-					&fdbv1beta2.ProcessGroupStatus{
-						ProcessGroupID: "test-4",
-						ProcessGroupConditions: []*fdbv1beta2.ProcessGroupCondition{
-							{
-								ProcessGroupConditionType: fdbv1beta2.PodPending,
-								Timestamp:                 time.Now().Add(-45 * time.Minute).Unix(),
-							},
-						},
-					})
-			})
-
-			It("should ignore the process", func() {
-				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
-				Expect(coordinatorsValid).To(BeTrue())
-				Expect(addressesValid).To(BeTrue())
-				Expect(err).To(BeNil())
-			})
-		})
-
 		When("a process misses the public-address flag", func() {
 			BeforeEach(func() {
 				process := status.Cluster.Processes["3"]
@@ -777,6 +813,40 @@ var _ = Describe("Change coordinators", func() {
 		Context("with multiple regions", func() {
 			BeforeEach(func() {
 				cluster.Spec.DatabaseConfiguration.UsableRegions = 2
+				cluster.Spec.DatabaseConfiguration.Regions = []fdbv1beta2.Region{
+					{
+						DataCenters: []fdbv1beta2.DataCenter{
+							{
+								ID: "dc1",
+							},
+							{
+								ID:        "dc2",
+								Satellite: 1,
+							},
+							{
+								ID:        "dc3",
+								Satellite: 1,
+								Priority:  1,
+							},
+						},
+					},
+					{
+						DataCenters: []fdbv1beta2.DataCenter{
+							{
+								ID: "dc3",
+							},
+							{
+								ID:        "dc2",
+								Satellite: 1,
+							},
+							{
+								ID:        "dc1",
+								Satellite: 1,
+								Priority:  1,
+							},
+						},
+					},
+				}
 
 				status.Cluster.Processes["4"] = generateDummyProcessInfo("test-4", "dc2", 4501, false)
 				status.Cluster.Processes["5"] = generateDummyProcessInfo("test-5", "dc2", 4501, false)
@@ -843,11 +913,12 @@ var _ = Describe("Change coordinators", func() {
 			Context("with coordinators divided across two DCs", func() {
 				BeforeEach(func() {
 					for _, process := range status.Cluster.Processes {
-						if process.Locality["dcid"] == "dc3" {
-							process.Locality["dcid"] = "dc1"
+						if process.Locality[fdbv1beta2.FDBLocalityDCIDlKey] == "dc3" {
+							process.Locality[fdbv1beta2.FDBLocalityDCIDlKey] = "dc1"
 						}
 					}
 				})
+
 				It("should report the coordinators as not valid", func() {
 					coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
 					Expect(coordinatorsValid).To(BeFalse())