Repository: FoundationDB/fdb-kubernetes-operator
Pull request #1881 (2023-11-09 22:18:03)

Add check for plugin version upgrades
# Description

Fixes: https://github.com/FoundationDB/fdb-kubernetes-operator/issues/1106

## Type of change

*Please select one of the options below.*

- New feature (non-breaking change which adds functionality)

## Discussion

*Are there any design details that you would like to discuss further?*
An alternative approach would be saving minimum-supported-plugin-version as a label in operator deployment(instead of GitHub api) and use that to check if client's plugin is outdated. Would like to hear your idea @johscheuer .  

## Testing

*Please describe the tests that you ran to verify your changes. Unit tests?
Manual testing?*
Added unit test and ran manual test. 

*Do we need to perform additional testing once this is merged, or perform in a larger testing environment?*
No 

## Documentation

*Did you update relevant documentation within this repository?*
No

*If this change is adding new functionality, do we need to describe it in our user manual?*
No

*If this change is adding or removing subreconcilers, have we updated the core technical design doc to reflect that?*
No

*If this change is adding new safety checks or new potential failure modes, have we documented and how to debug potential issues?*
No

## Follow-up

*Are there any follow-up issues that we should pursue in the future?*
No
*Does this introduce new defaults that we should re-evaluate in the future?*
No


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

* Commit ID: 0f459d776493eb11395c90296db5c34b8fbd124b
* Duration 2:47:10
* Result: :x: **FAILED**
* Error: `Error while executing command: if $fail_test; then exit 1; fi. Reason: exit status 1`
* [Build Log](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator/fdb-kubernetes-operator-pr/dc9197ff-a7e6-43ce-bded-e592b40dab37.gz) terminal output (available for 30 days)
* [Build Workspace](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator-pr/dc9197ff-a7e6-43ce-bded-e592b40dab37/workspace.zip) zip file of the working directory (available for 30 days)

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

* Commit ID: 8338b1eb67f2a5f409e6629ac7ccfd93d551d41e
* Duration 2:46:00
* Result: :white_check_mark: **SUCCEEDED**
* Error: `N/A`
* [Build Log](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator/fdb-kubernetes-operator-pr/70b75b80-df3f-417b-aceb-917f81e24f79.gz) terminal output (available for 30 days)
* [Build Workspace](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator-pr/70b75b80-df3f-417b-aceb-917f81e24f79/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 a40dd1d8..60076d5a 100644
--- a/api/v1beta2/foundationdb_database_configuration.go
+++ b/api/v1beta2/foundationdb_database_configuration.go
@@ -234,12 +234,12 @@ func (configuration *DatabaseConfiguration) CountUniqueDataCenters() int {
 // This will fill in defaults of -1 for some fields that have a default
 // of 0, and will ensure that the region configuration is ordered
 // consistently.
-func (configuration DatabaseConfiguration) NormalizeConfigurationWithSeparatedProxies(version string) DatabaseConfiguration {
+func (configuration DatabaseConfiguration) NormalizeConfigurationWithSeparatedProxies(version string, areSeparatedProxiesConfigured bool) DatabaseConfiguration {
 	result := configuration.NormalizeConfiguration()
 
 	parsedVersion, _ := ParseFdbVersion(version)
 	if parsedVersion.HasSeparatedProxies() {
-		if !configuration.AreSeparatedProxiesConfigured() {
+		if !areSeparatedProxiesConfigured {
 			result.GrvProxies = 0
 			result.CommitProxies = 0
 		} else {
diff --git a/api/v1beta2/foundationdbcluster_types.go b/api/v1beta2/foundationdbcluster_types.go
index d0a2244c..b7dcd533 100644
--- a/api/v1beta2/foundationdbcluster_types.go
+++ b/api/v1beta2/foundationdbcluster_types.go
@@ -1313,6 +1313,7 @@ func (cluster *FoundationDBCluster) calculateProcessCountFromRole(count int, alt
 	if count < 0 {
 		return 0
 	}
+
 	return count
 }
 
@@ -1366,9 +1367,7 @@ func (cluster *FoundationDBCluster) GetProcessCountsWithDefaults() (ProcessCount
 	roleCounts := cluster.GetRoleCountsWithDefaults()
 	processCounts := cluster.Spec.ProcessCounts.DeepCopy()
 
-	isSatellite := false
-	isMain := false
-
+	var isSatellite, isMain bool
 	satelliteLogs := 0
 	for _, region := range cluster.Spec.DatabaseConfiguration.Regions {
 		for _, dataCenter := range region.DataCenters {
@@ -1396,12 +1395,14 @@ func (cluster *FoundationDBCluster) GetProcessCountsWithDefaults() (ProcessCount
 		processCounts.Storage = cluster.calculateProcessCount(false,
 			roleCounts.Storage)
 	}
+
 	if processCounts.Log == 0 {
 		processCounts.Log = cluster.calculateProcessCount(true,
 			cluster.calculateProcessCountFromRole(roleCounts.Logs+satelliteLogs, processCounts.Log),
 			cluster.calculateProcessCountFromRole(roleCounts.RemoteLogs, processCounts.Log),
 		)
 	}
+
 	if processCounts.Stateless == 0 {
 		primaryStatelessCount := cluster.calculateProcessCountFromRole(1, processCounts.Master) +
 			cluster.calculateProcessCountFromRole(1, processCounts.ClusterController) +
@@ -1815,7 +1816,7 @@ type ContainerOverrides struct {
 // DesiredDatabaseConfiguration builds the database configuration for the
 // cluster based on its spec.
 func (cluster *FoundationDBCluster) DesiredDatabaseConfiguration() DatabaseConfiguration {
-	configuration := cluster.Spec.DatabaseConfiguration.NormalizeConfigurationWithSeparatedProxies(cluster.GetRunningVersion())
+	configuration := cluster.Spec.DatabaseConfiguration.NormalizeConfigurationWithSeparatedProxies(cluster.GetRunningVersion(), cluster.Spec.DatabaseConfiguration.AreSeparatedProxiesConfigured())
 	configuration.RoleCounts = cluster.GetRoleCountsWithDefaults()
 	configuration.RoleCounts.Storage = 0
 
@@ -1916,7 +1917,8 @@ func (cluster *FoundationDBCluster) ShouldUseLocks() bool {
 		return !*disabled
 	}
 
-	return cluster.Spec.FaultDomain.ZoneCount > 1 || len(cluster.Spec.DatabaseConfiguration.Regions) > 1
+	return cluster.Spec.FaultDomain.ZoneCount > 1 || len(cluster.Spec.DatabaseConfiguration.Regions) > 1 ||
+		cluster.Spec.DatabaseConfiguration.RedundancyMode == RedundancyModeThreeDataHall
 }
 
 // GetLockPrefix gets the prefix for the keys where we store locking
diff --git a/controllers/cluster_controller.go b/controllers/cluster_controller.go
index 5ef48034..58d83098 100644
--- a/controllers/cluster_controller.go
+++ b/controllers/cluster_controller.go
@@ -476,9 +476,14 @@ func (r *FoundationDBClusterReconciler) getStatusFromClusterOrDummyStatus(logger
 	if err != nil {
 		return nil, err
 	}
-
 	defer adminClient.Close()
 
+	// If the cluster is not yet configured, we can reduce the timeout to make sure the initial reconcile steps
+	// are faster.
+	if !cluster.Status.Configured {
+		adminClient.SetTimeout(10 * time.Second)
+	}
+
 	status, err := adminClient.GetStatus()
 	if err == nil {
 		return status, nil
diff --git a/controllers/exclude_processes.go b/controllers/exclude_processes.go
index 4ea4074c..fff71be5 100644
--- a/controllers/exclude_processes.go
+++ b/controllers/exclude_processes.go
@@ -75,18 +75,19 @@ func (e excludeProcesses) reconcile(_ context.Context, r *FoundationDBClusterRec
 
 		desiredProcessesMap := desiredProcesses.Map()
 		for processClass := range fdbProcessesToExcludeByClass {
+			contextLogger := logger.WithValues("processClass", processClass)
 			ongoingExclusions := ongoingExclusionsByClass[processClass]
 			processesToExclude := fdbProcessesToExcludeByClass[processClass]
 
-			allowedExclusions, missingProcesses := canExcludeNewProcesses(logger, cluster, processClass, desiredProcessesMap[processClass], ongoingExclusions, r.InSimulation)
+			allowedExclusions, missingProcesses := getAllowedExclusionsAndMissingProcesses(contextLogger, cluster, processClass, desiredProcessesMap[processClass], ongoingExclusions, r.InSimulation)
 			if allowedExclusions <= 0 {
-				logger.Info("Waiting for missing processes before continuing with the exclusion", "processClass", processClass, "missingProcesses", missingProcesses, "addressesToExclude", processesToExclude, "allowedExclusions", allowedExclusions, "ongoingExclusions", ongoingExclusions)
+				contextLogger.Info("Waiting for missing processes before continuing with the exclusion", "missingProcesses", missingProcesses, "addressesToExclude", processesToExclude, "allowedExclusions", allowedExclusions, "ongoingExclusions", ongoingExclusions)
 				continue
 			}
 
 			// If we are not able to exclude all processes at once print a log message.
 			if len(processesToExclude) > allowedExclusions {
-				logger.Info("Some processes are still missing but continuing with the exclusion", "processClass", processClass, "missingProcesses", missingProcesses, "addressesToExclude", processesToExclude, "allowedExclusions", allowedExclusions, "ongoingExclusions", ongoingExclusions)
+				contextLogger.Info("Some processes are still missing but continuing with the exclusion", "missingProcesses", missingProcesses, "addressesToExclude", processesToExclude, "allowedExclusions", allowedExclusions, "ongoingExclusions", ongoingExclusions)
 			}
 
 			if len(processesToExclude) < allowedExclusions {
@@ -184,22 +185,23 @@ func getProcessesToExclude(exclusions []fdbv1beta2.ProcessAddress, cluster *fdbv
 	return fdbProcessesToExcludeByClass, ongoingExclusionsByClass
 }
 
-// canExcludeNewProcesses will check if new processes for the specified process class can be excluded. The calculation takes
+// getAllowedExclusionsAndMissingProcesses will check if new processes for the specified process class can be excluded. The calculation takes
 // the current ongoing exclusions into account and the desired process count. If there are process groups that have
 // the MissingProcesses condition this method will forbid exclusions until all process groups with this condition have
 // this condition for longer than ignoreMissingProcessDuration. The idea behind this is to try to exclude as many processes
 // at once e.g. to reduce the number of recoveries and data movement.
-func canExcludeNewProcesses(logger logr.Logger, cluster *fdbv1beta2.FoundationDBCluster, processClass fdbv1beta2.ProcessClass, desiredProcessCount int, ongoingExclusions int, inSimulation bool) (int, []fdbv1beta2.ProcessGroupID) {
+func getAllowedExclusionsAndMissingProcesses(logger logr.Logger, cluster *fdbv1beta2.FoundationDBCluster, processClass fdbv1beta2.ProcessClass, desiredProcessCount int, ongoingExclusions int, inSimulation bool) (int, []fdbv1beta2.ProcessGroupID) {
 	// Block excludes on missing processes not marked for removal unless they are missing for a long time and the process might be broken
 	// or the namespace quota was hit.
 	missingProcesses := make([]fdbv1beta2.ProcessGroupID, 0)
-	validProcesses := make([]fdbv1beta2.ProcessGroupID, 0)
+	var validProcesses int
 
 	exclusionsAllowed := true
 	for _, processGroup := range cluster.Status.ProcessGroups {
 		if processGroup.ProcessClass != processClass {
 			continue
 		}
+
 		// Those should already be filtered out by the previous method.
 		if processGroup.IsMarkedForRemoval() && processGroup.IsExcluded() {
 			continue
@@ -217,33 +219,27 @@ func canExcludeNewProcesses(logger logr.Logger, cluster *fdbv1beta2.FoundationDB
 			continue
 		}
 
-		validProcesses = append(validProcesses, processGroup.ProcessGroupID)
+		validProcesses++
 	}
 
 	if !exclusionsAllowed {
-		logger.Info("Found at least one missing process, that was not missing for a long time", "missingProcesses", missingProcesses)
+		logger.Info("Found at least one missing process, that was not missing for more than 5 minutes", "missingProcesses", missingProcesses)
 		return 0, missingProcesses
 	}
 
-	logger.V(1).Info("canExcludeNewProcesses", "validProcesses", len(validProcesses), "desiredProcessCount", desiredProcessCount, "ongoingExclusions", ongoingExclusions)
-
-	// The assumption here is that we will only exclude a process if there is a replacement ready for it. We could relax
-	// this requirement in the future and take the fault tolerance into account. We add the desired fault tolerance to
-	// have some buffer to prevent cases where the operator might need to exclude more processes but there are more missing
-	// processes.
-	allowedExclusions := cluster.DesiredFaultTolerance() + len(validProcesses) - desiredProcessCount - ongoingExclusions
-
-	// If automatic replacements are enabled and the allowed exclusions is less than or equal to 0, we have to check
-	// how many processes are missing and if more processes are missing than the automatic replacements is allowed to
-	// replace, we will allow exclusions for the count of automatic replacements removing the already ongoing exclusions.
-	// This code should make sure that the operator can automatically replace processes, even in the case where multiple
-	// processes are failing.
-	if cluster.GetEnableAutomaticReplacements() && allowedExclusions <= 0 {
-		automaticReplacements := cluster.GetMaxConcurrentAutomaticReplacements()
-		if len(missingProcesses) > automaticReplacements {
-			return automaticReplacements - ongoingExclusions, missingProcesses
-		}
+	return getAllowedExclusions(logger, validProcesses, desiredProcessCount, ongoingExclusions, cluster.DesiredFaultTolerance()), missingProcesses
+}
+
+// getAllowedExclusions will return the number of allowed exclusions. If no exclusions are allowed this method will return a 0.
+// The assumption here is that we will only exclude a process if there is a replacement ready for it. We add the desired fault
+// tolerance to have some buffer to prevent cases where the operator might need to exclude more processes but there are more
+// missing processes.
+func getAllowedExclusions(logger logr.Logger, validProcesses int, desiredProcessCount int, ongoingExclusions int, faultTolerance int) int {
+	logger.V(1).Info("getAllowedExclusions", "validProcesses", validProcesses, "desiredProcessCount", desiredProcessCount, "ongoingExclusions", ongoingExclusions, "faultTolerance", faultTolerance)
+	allowedExclusions := validProcesses + faultTolerance - desiredProcessCount - ongoingExclusions
+	if allowedExclusions < 0 {
+		return 0
 	}
 
-	return allowedExclusions, missingProcesses
+	return allowedExclusions
 }
diff --git a/controllers/update_database_configuration.go b/controllers/update_database_configuration.go
index e93c993d..e65e5d0b 100644
--- a/controllers/update_database_configuration.go
+++ b/controllers/update_database_configuration.go
@@ -65,7 +65,7 @@ func (u updateDatabaseConfiguration) reconcile(ctx context.Context, r *Foundatio
 
 	desiredConfiguration := cluster.DesiredDatabaseConfiguration()
 	desiredConfiguration.RoleCounts.Storage = 0
-	currentConfiguration := status.Cluster.DatabaseConfiguration.NormalizeConfigurationWithSeparatedProxies(cluster.Spec.Version)
+	currentConfiguration := status.Cluster.DatabaseConfiguration.NormalizeConfigurationWithSeparatedProxies(cluster.Spec.Version, cluster.Spec.DatabaseConfiguration.AreSeparatedProxiesConfigured())
 	// We have to reset the excluded servers here otherwise we will trigger a reconfiguration if one or more servers
 	// are excluded.
 	currentConfiguration.ExcludedServers = nil
diff --git a/controllers/update_status.go b/controllers/update_status.go
index 047d5191..9b0d2ed4 100644
--- a/controllers/update_status.go
+++ b/controllers/update_status.go
@@ -99,7 +99,7 @@ func (updateStatus) reconcile(ctx context.Context, r *FoundationDBClusterReconci
 	// about the current database configuration, leading to a wrong signal that the database configuration must be changed as
 	// the configuration will be overwritten with the default values.
 	if databaseStatus.Client.DatabaseStatus.Available {
-		clusterStatus.DatabaseConfiguration = databaseStatus.Cluster.DatabaseConfiguration.NormalizeConfigurationWithSeparatedProxies(cluster.Spec.Version)
+		clusterStatus.DatabaseConfiguration = databaseStatus.Cluster.DatabaseConfiguration.NormalizeConfigurationWithSeparatedProxies(cluster.Spec.Version, cluster.Spec.DatabaseConfiguration.AreSeparatedProxiesConfigured())
 		// Removing excluded servers as we don't want them during comparison.
 		clusterStatus.DatabaseConfiguration.ExcludedServers = nil
 		cluster.ClearMissingVersionFlags(&clusterStatus.DatabaseConfiguration)
@@ -318,6 +318,12 @@ func tryConnectionOptions(logger logr.Logger, cluster *fdbv1beta2.FoundationDBCl
 			return originalConnectionString, clientErr
 		}
 
+		// If the cluster is not yet configured, we can reduce the timeout to make sure the initial reconcile steps
+		// are faster.
+		if !cluster.Status.Configured {
+			adminClient.SetTimeout(10 * time.Second)
+		}
+
 		var activeConnectionString string
 		activeConnectionString, err = adminClient.GetConnectionString()
 
@@ -337,74 +343,98 @@ func tryConnectionOptions(logger logr.Logger, cluster *fdbv1beta2.FoundationDBCl
 }
 
 // checkAndSetProcessStatus checks the status of the Process and if missing or incorrect add it to the related status field
-func checkAndSetProcessStatus(logger logr.Logger, r *FoundationDBClusterReconciler, cluster *fdbv1beta2.FoundationDBCluster, pod *corev1.Pod, processMap map[fdbv1beta2.ProcessGroupID][]fdbv1beta2.FoundationDBStatusProcessInfo, processNumber int, processCount int, processGroupStatus *fdbv1beta2.ProcessGroupStatus) error {
-	processID := processGroupStatus.ProcessGroupID
-
-	if processCount > 1 {
-		processID = fdbv1beta2.ProcessGroupID(fmt.Sprintf("%s-%d", processID, processNumber))
-	}
-
-	processStatus := processMap[processID]
-	// Only set the MissingProcesses condition if the machine-readable status has at least one process. We can improve this check
+func checkAndSetProcessStatus(logger logr.Logger, r *FoundationDBClusterReconciler, cluster *fdbv1beta2.FoundationDBCluster, pod *corev1.Pod, processMap map[fdbv1beta2.ProcessGroupID][]fdbv1beta2.FoundationDBStatusProcessInfo, processCount int, processGroupStatus *fdbv1beta2.ProcessGroupStatus) error {
+	// Only perform any process specific validation if the machine-readable status has at least one process. We can improve this check
 	// later by validating additional messages in the machine-readable status.
-	processGroupStatus.UpdateCondition(fdbv1beta2.MissingProcesses, len(processStatus) == 0 && len(processMap) > 0)
-	if len(processStatus) == 0 {
+	if len(processMap) == 0 {
 		return nil
 	}
 
+	var excluded, hasIncorrectCommandLine, hasMissingProcesses, sidecarUnreachable bool
+
+	// Fetch the pod client and variables once per Pod.
 	podClient, message := r.getPodClient(cluster, pod)
 	if podClient == nil {
 		logger.Info("Unable to build pod client", "processGroupID", processGroupStatus.ProcessGroupID, "message", message)
 		return nil
 	}
 
-	var excluded, correct bool
+	substitutions, err := podClient.GetVariableSubstitutions()
+	if err != nil {
+		if internal.IsNetworkError(err) {
+			sidecarUnreachable = true
+		}
+	}
+
 	versionCompatibleUpgrade := cluster.VersionCompatibleUpgradeInProgress()
-	for _, process := range processStatus {
-		// Check if the process is reporting any messages, those will normally include error messages
-		if len(process.Messages) > 0 {
-			logger.Info("found error message(s) for the process", "processGroupID", processGroupStatus.ProcessGroupID, "messages", process.Messages)
+	for processNumber := 1; processNumber <= processCount; processNumber++ {
+		var processID fdbv1beta2.ProcessGroupID
+		if processCount > 1 {
+			processID = fdbv1beta2.ProcessGroupID(fmt.Sprintf("%s-%d", processGroupStatus.ProcessGroupID, processNumber))
+		} else {
+			processID = processGroupStatus.ProcessGroupID
 		}
 
-		if !excluded {
-			excluded = process.Excluded
+		processStatus := processMap[processID]
+		if !hasMissingProcesses {
+			hasMissingProcesses = len(processStatus) == 0
 		}
 
-		commandLine, err := internal.GetStartCommand(cluster, processGroupStatus.ProcessClass, podClient, processNumber, processCount)
-		if err != nil {
-			if internal.IsNetworkError(err) {
-				processGroupStatus.UpdateCondition(fdbv1beta2.SidecarUnreachable, true)
-				return nil
+		if len(processStatus) == 0 {
+			continue
+		}
+
+		for _, process := range processStatus {
+			// Check if the process is reporting any messages, those will normally include error messages.
+			if len(process.Messages) > 0 {
+				logger.Info("found error message(s) for the process", "processGroupID", processGroupStatus.ProcessGroupID, "messages", process.Messages)
 			}
 
-			return err
-		}
+			if !excluded {
+				excluded = process.Excluded
+			}
 
-		// If a version compatible upgrade is in progress, skip the version check since we will run a mixed set of versions
-		// until the cluster is fully reconciled.
-		versionMatch := true
-		if !versionCompatibleUpgrade {
-			versionMatch = process.Version == cluster.Spec.Version || process.Version == fmt.Sprintf("%s-PRERELEASE", cluster.Spec.Version)
-		}
+			if len(substitutions) == 0 {
+				continue
+			}
 
-		// If the `EmptyMonitorConf` is set, the commandline is by definition wrong since there should be no running processes.
-		correct = commandLine == process.CommandLine && versionMatch && !cluster.Spec.Buggify.EmptyMonitorConf
+			commandLine, err := internal.GetStartCommandWithSubstitutions(cluster, processGroupStatus.ProcessClass, substitutions, processNumber, processCount)
+			if err != nil {
+				return err
+			}
 
-		if !correct {
-			logger.Info("IncorrectProcess",
-				"expected", commandLine, "got", process.CommandLine,
-				"expectedVersion", cluster.Spec.Version,
-				"version", process.Version,
-				"processGroupID", processGroupStatus.ProcessGroupID,
-				"emptyMonitorConf", cluster.Spec.Buggify.EmptyMonitorConf)
+			// If a version compatible upgrade is in progress, skip the version check since we will run a mixed set of versions
+			// until the cluster is fully reconciled.
+			versionMatch := true
+			if !versionCompatibleUpgrade {
+				versionMatch = process.Version == cluster.Spec.Version || process.Version == fmt.Sprintf("%s-PRERELEASE", cluster.Spec.Version)
+			}
+
+			// If the `EmptyMonitorConf` is set, the commandline is by definition wrong since there should be no running processes.
+			if !(commandLine == process.CommandLine && versionMatch && !cluster.Spec.Buggify.EmptyMonitorConf) {
+				logger.Info("IncorrectProcess",
+					"expected", commandLine, "got", process.CommandLine,
+					"expectedVersion", cluster.Spec.Version,
+					"version", process.Version,
+					"processGroupID", processGroupStatus.ProcessGroupID,
+					"emptyMonitorConf", cluster.Spec.Buggify.EmptyMonitorConf)
+				hasIncorrectCommandLine = true
+			}
 		}
 	}
 
-	// If the process is excluded, update the exclude condition.
+	processGroupStatus.UpdateCondition(fdbv1beta2.MissingProcesses, hasMissingProcesses)
+	processGroupStatus.UpdateCondition(fdbv1beta2.SidecarUnreachable, sidecarUnreachable)
+	// If the processes are absent, we are not able to determine the state of the processes and therefore we won't change it.
+	if hasMissingProcesses {
+		return nil
+	}
+
 	processGroupStatus.UpdateCondition(fdbv1beta2.ProcessIsMarkedAsExcluded, excluded)
-	processGroupStatus.UpdateCondition(fdbv1beta2.IncorrectCommandLine, !correct)
-	// Reset status for sidecar unreachable, since we are here at this point we were able to reach the sidecar for the substitute variables.
-	processGroupStatus.UpdateCondition(fdbv1beta2.SidecarUnreachable, false)
+	if sidecarUnreachable {
+		return nil
+	}
+	processGroupStatus.UpdateCondition(fdbv1beta2.IncorrectCommandLine, hasIncorrectCommandLine)
 
 	return nil
 }
@@ -515,12 +545,9 @@ func validateProcessGroups(ctx context.Context, r *FoundationDBClusterReconciler
 			}
 		}
 
-		// In theory we could also support multiple processes per pod for different classes
-		for i := 1; i <= processCount; i++ {
-			err = checkAndSetProcessStatus(logger, r, cluster, pod, processMap, i, processCount, processGroup)
-			if err != nil {
-				return processGroups, err
-			}
+		err = checkAndSetProcessStatus(logger, r, cluster, pod, processMap, processCount, processGroup)
+		if err != nil {
+			return processGroups, err
 		}
 
 		configMapHash, err := internal.GetDynamicConfHash(configMap, processGroup.ProcessClass, imageType, processCount)
diff --git a/docs/manual/fault_domains.md b/docs/manual/fault_domains.md
index 7b39b84f..1db0d31c 100644
--- a/docs/manual/fault_domains.md
+++ b/docs/manual/fault_domains.md
@@ -253,7 +253,7 @@ 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.
+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, a `zoneCount` greater than 1 in its fault domain configuration, or `redundancyMode` equal to `three_data_hall`.
 
 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 01a84775..43778ef1 100644
--- a/fdbclient/admin_client.go
+++ b/fdbclient/admin_client.go
@@ -89,6 +89,9 @@ type cliAdminClient struct {
 	// fdbLibClient is an interface to interact with a FDB cluster over the FDB client libraries. In the real fdb lib client
 	// we will issue the actual requests against FDB. In the mock runner we will return predefined output.
 	fdbLibClient fdbLibClient
+
+	// timeout defines the timeout that should be used for interacting with FDB.
+	timeout time.Duration
 }
 
 // NewCliAdminClient generates an Admin client for a cluster
@@ -275,7 +278,7 @@ func (client *cliAdminClient) runCommand(command cliCommand) (string, error) {
 func (client *cliAdminClient) getStatusFromCli() (*fdbv1beta2.FoundationDBStatus, error) {
 	// Always use the max timeout here. Otherwise we will retry multiple times with an increasing timeout. As the
 	// timeout is only the upper bound using directly the max timeout reduces the calls to a single call.
-	output, err := client.runCommand(cliCommand{command: "status json", timeout: MaxCliTimeout})
+	output, err := client.runCommand(cliCommand{command: "status json", timeout: client.getTimeout()})
 	if err != nil {
 		return nil, err
 	}
@@ -313,8 +316,9 @@ func (client *cliAdminClient) GetStatus() (*fdbv1beta2.FoundationDBStatus, error
 	adminClientMutex.Lock()
 	defer adminClientMutex.Unlock()
 
+	startTime := time.Now()
 	// This will call directly the database and fetch the status information from the system key space.
-	status, err := getStatusFromDB(client.fdbLibClient, client.log, MaxCliTimeout)
+	status, err := getStatusFromDB(client.fdbLibClient, client.log, client.getTimeout())
 	// There is a limitation in the multi version client if the cluster is only partially upgraded e.g. because not
 	// all fdbserver processes are restarted, then the multi version client sometimes picks the wrong version
 	// to connect to the cluster. This will result in an empty status only reporting the unreachable coordinators.
@@ -326,7 +330,7 @@ func (client *cliAdminClient) GetStatus() (*fdbv1beta2.FoundationDBStatus, error
 		status, err = client.getStatus()
 	}
 
-	client.log.V(1).Info("Completed GetStatus() call", "error", err, "status", status)
+	client.log.V(1).Info("Completed GetStatus() call", "error", err, "status", status, "duration", time.Since(startTime).String())
 
 	return status, err
 }
@@ -348,7 +352,7 @@ func (client *cliAdminClient) ConfigureDatabase(configuration fdbv1beta2.Databas
 
 // GetMaintenanceZone gets current maintenance zone, if any. Returns empty string if maintenance mode is off
 func (client *cliAdminClient) GetMaintenanceZone() (string, error) {
-	mode, err := client.fdbLibClient.getValueFromDBUsingKey("\xff/maintenance", DefaultCLITimeout)
+	mode, err := client.fdbLibClient.getValueFromDBUsingKey("\xff/maintenance", client.getTimeout())
 	if err != nil {
 		return "", err
 	}
@@ -394,7 +398,7 @@ func (client *cliAdminClient) ExcludeProcesses(addresses []fdbv1beta2.ProcessAdd
 
 	excludeCommand.WriteString(fdbv1beta2.ProcessAddressesString(addresses, " "))
 
-	_, err = client.runCommand(cliCommand{command: excludeCommand.String(), timeout: MaxCliTimeout})
+	_, err = client.runCommand(cliCommand{command: excludeCommand.String(), timeout: client.getTimeout()})
 
 	return err
 }
@@ -445,7 +449,7 @@ func (client *cliAdminClient) KillProcesses(addresses []fdbv1beta2.ProcessAddres
 		fdbv1beta2.ProcessAddressesStringWithoutFlags(addresses, " "),
 	)
 	// Run the kill command once with the max timeout to reduce the risk of multiple recoveries happening.
-	_, err := client.runCommand(cliCommand{command: killCommand, timeout: MaxCliTimeout})
+	_, err := client.runCommand(cliCommand{command: killCommand, timeout: client.getTimeout()})
 
 	return err
 }
@@ -488,7 +492,7 @@ func cleanConnectionStringOutput(input string) string {
 func (client *cliAdminClient) GetConnectionString() (string, error) {
 	// This will call directly the database and fetch the connection string
 	// from the system key space.
-	outputBytes, err := getConnectionStringFromDB(client.fdbLibClient, MaxCliTimeout)
+	outputBytes, err := getConnectionStringFromDB(client.fdbLibClient, client.getTimeout())
 
 	if err != nil {
 		return "", err
@@ -694,3 +698,17 @@ func (client *cliAdminClient) WithValues(keysAndValues ...interface{}) {
 	}
 	cmdRunner.log = newLogger
 }
+
+// SetTimeout will overwrite the default timeout for interacting the FDB cluster.
+func (client *cliAdminClient) SetTimeout(timeout time.Duration) {
+	client.timeout = timeout
+}
+
+// getTimeout will return the timeout that is specified for the admin client or otherwise the MaxCliTimeout.
+func (client *cliAdminClient) getTimeout() time.Duration {
+	if client.timeout == 0 {
+		return MaxCliTimeout
+	}
+
+	return client.timeout
+}
diff --git a/fdbclient/common.go b/fdbclient/common.go
index 6ecec8b1..ddd1a906 100644
--- a/fdbclient/common.go
+++ b/fdbclient/common.go
@@ -40,7 +40,7 @@ import (
 var DefaultCLITimeout = 10 * time.Second
 
 // MaxCliTimeout is the maximum CLI timeout that will be used for requests that might be slower to respond.
-var MaxCliTimeout = 60 * time.Second
+var MaxCliTimeout = 40 * time.Second
 
 const (
 	defaultTransactionTimeout = 5 * time.Second
diff --git a/internal/monitor_conf.go b/internal/monitor_conf.go
index 660f4466..80104cff 100644
--- a/internal/monitor_conf.go
+++ b/internal/monitor_conf.go
@@ -39,11 +39,11 @@ func GetStartCommand(cluster *fdbv1beta2.FoundationDBCluster, processClass fdbv1
 		return "", err
 	}
 
-	return getStartCommandWithSubstitutions(cluster, processClass, substitutions, processNumber, processCount)
+	return GetStartCommandWithSubstitutions(cluster, processClass, substitutions, processNumber, processCount)
 }
 
-// getStartCommandWithSubstitutions will be used by GetStartCommand and for internal testing.
-func getStartCommandWithSubstitutions(cluster *fdbv1beta2.FoundationDBCluster, processClass fdbv1beta2.ProcessClass, substitutions map[string]string, processNumber int, processCount int) (string, error) {
+// GetStartCommandWithSubstitutions will be used by GetStartCommand and for internal testing.
+func GetStartCommandWithSubstitutions(cluster *fdbv1beta2.FoundationDBCluster, processClass fdbv1beta2.ProcessClass, substitutions map[string]string, processNumber int, processCount int) (string, error) {
 	if substitutions == nil {
 		return "", nil
 	}
diff --git a/kubectl-fdb/Readme.md b/kubectl-fdb/Readme.md
index 68cb50df..d358c76e 100644
--- a/kubectl-fdb/Readme.md
+++ b/kubectl-fdb/Readme.md
@@ -16,6 +16,9 @@ Install from release:
 
 ```bash
 pushd $TMPDIR
+if [ -f "$TMPDIR/kubectl-fdb-version.txt" ]; then
+   rm "$TMPDIR/kubectl-fdb-version.txt"
+fi
 OS=$(uname)
 ARCH=$(uname -m)
 VERSION="$(curl -s "https://api.github.com/repos/FoundationDB/fdb-kubernetes-operator/releases/latest" | jq -r '.tag_name')"
@@ -51,3 +54,11 @@ Run `kubectl fdb help` to get the latest help.
 We have a list of [planned operations](https://github.com/FoundationDB/fdb-kubernetes-operator/issues?q=is%3Aissue+is%3Aopen+label%3Aplugin)
 that we want to implement.
 Raise an issue if you miss a specific command to operate FDB on Kubernetes.
+
+## Plugin Version Check
+
+Per default the plugin will check if there is a newer version present.
+For this the plugin will contact the GitHub API and check for the latest release.
+If a newer version was found the command will exit with the request to update the plugin.
+The version check can be disabled by setting `--version-check=false` when issuing a command.
+The plugin will create a local cache in the temp directory to reduce the calls to the GitHub API, the file is located under `$TMPDIR/kubectl-fdb-version.txt`.
diff --git a/kubectl-fdb/cmd/plugin_version_checker.go b/kubectl-fdb/cmd/plugin_version_checker.go
new file mode 100644
index 00000000..6ab45e76
--- /dev/null
+++ b/kubectl-fdb/cmd/plugin_version_checker.go
@@ -0,0 +1,128 @@
+package cmd
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/hashicorp/go-retryablehttp"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+// VersionChecker interface to help us mock test the version checker
+type VersionChecker interface {
+	getLatestPluginVersion() (string, error)
+}
+
+// RealVersionChecker will do actual call to GitHub API to get version
+type RealVersionChecker struct{}
+
+// KubectlFbReleaseURL is the public GitHub URL we read the latest version from
+const KubectlFbReleaseURL = "https://api.github.com/repos/FoundationDB/fdb-kubernetes-operator/releases/latest"
+
+// LocalTempVersionFileName is where we cache plugin version for 24 hours, so we don't call GitHub for every single command(also rate-limit issue on GitHub api calls)
+const LocalTempVersionFileName = "kubectl-fdb-version.txt"
+
+// PluginVersionDetails Contains [partial] plugin version details from GitHub
+type PluginVersionDetails struct {
+	ID      int64  `json:"id"`
+	Version string `json:"tag_name"`
+	Name    string `json:"name"`
+}
+
+// It reads the latest fdb plugin version from local temp file,
+// if not exists, gets it from GitHub and store it locally and then returns it
+func (versionChecker *RealVersionChecker) getLatestPluginVersion() (string, error) {
+	fileName := filepath.Join(os.TempDir(), LocalTempVersionFileName)
+	_, err := os.Stat(fileName)
+	if os.IsNotExist(err) {
+		return updateLocalVersion(fileName)
+	}
+	if isVersionFileCreatedToday(fileName) {
+		return readVersionFromLocalFile(fileName)
+	}
+	return updateLocalVersion(fileName)
+}
+
+// readVersionFromLocalFile reads the version value from the local temp file
+func readVersionFromLocalFile(fileName string) (string, error) {
+	file, _ := os.OpenFile(fileName, os.O_RDONLY, 0644)
+	defer file.Close()
+	content, err := os.ReadFile(fileName)
+	if err != nil {
+		return "", err
+	}
+	return string(content), err
+}
+
+// Check if the local temp version file is created within the last 24 hours
+func isVersionFileCreatedToday(filename string) bool {
+	fileInfo, err := os.Stat(filename)
+	if err != nil {
+		return false
+	}
+	return time.Since(fileInfo.ModTime()) <= 24*time.Hour
+}
+
+// write version value in local temp file
+func writeVersionToLocalTempFile(fileName string, version string) (int, error) {
+	file, err := os.OpenFile(fileName, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0644)
+	if err != nil {
+		// Handle the error if the file cannot be opened.
+		panic(err)
+	}
+	defer file.Close()
+	return file.WriteString(version)
+}
+
+// read version from HitHub and update local version file
+func updateLocalVersion(fileName string) (string, error) {
+	latestVersion, err := readVersionFromGitHub()
+	if err != nil {
+		return "", err
+	}
+	_, err = writeVersionToLocalTempFile(fileName, latestVersion)
+	return latestVersion, err
+}
+
+// read the latest release version number from GitHub, due to GitHub api rate limit we don't do it for every command
+func readVersionFromGitHub() (string, error) {
+	retryClient := retryablehttp.NewClient()
+	retryClient.RetryMax = 2
+	retryClient.RetryWaitMax = 1 * time.Second
+	retryClient.Logger = nil
+	retryClient.CheckRetry = retryablehttp.ErrorPropagatedRetryPolicy
+
+	retryClient.HTTPClient.Timeout = 1 * time.Second
+	req, _ := retryablehttp.NewRequest(http.MethodGet, KubectlFbReleaseURL, nil)
+	resp, err := retryClient.Do(req)
+	if err != nil || resp == nil {
+		fmt.Println("Failed to fetch kubectl-fdb version from GitHub")
+		return "", err
+	}
+	if resp == nil {
+		return "", nil
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+
+	if err != nil {
+		fmt.Println("Error in reading version from GitHub")
+		return "", err
+	}
+	var result = PluginVersionDetails{}
+	if !strings.Contains(strings.ToUpper(resp.Status), "OK") {
+		fmt.Println("Failed to get parse version info from GitHub response")
+		return "", err
+	}
+	err = json.Unmarshal(body, &result)
+	if err != nil {
+		fmt.Println("Failed to read version from GitHub response\n", err)
+		return "", err
+	}
+	//removing 'v' from beginning of release version
+	return result.Version[1:], nil
+}
diff --git a/kubectl-fdb/cmd/root.go b/kubectl-fdb/cmd/root.go
index 2aebebb5..19e0ac6e 100644
--- a/kubectl-fdb/cmd/root.go
+++ b/kubectl-fdb/cmd/root.go
@@ -54,7 +54,7 @@ func newFDBOptions(streams genericclioptions.IOStreams) *fdbBOptions {
 }
 
 // NewRootCmd provides a cobra command wrapping FDB actions
-func NewRootCmd(streams genericclioptions.IOStreams) *cobra.Command {
+func NewRootCmd(streams genericclioptions.IOStreams, pluginVersionChecker VersionChecker) *cobra.Command {
 	rand.Seed(time.Now().Unix())
 
 	o := newFDBOptions(streams)
@@ -67,6 +67,13 @@ func NewRootCmd(streams genericclioptions.IOStreams) *cobra.Command {
 		RunE: func(cmd *cobra.Command, args []string) error {
 			return cmd.Help()
 		},
+		PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
+			versionCheck, _ := cmd.Flags().GetBool("version-check")
+			if versionCheck {
+				return usingLatestPluginVersion(cmd, pluginVersionChecker)
+			}
+			return nil
+		},
 	}
 	cmd.SetOut(o.Out)
 	cmd.SetErr(o.ErrOut)
@@ -74,6 +81,7 @@ func NewRootCmd(streams genericclioptions.IOStreams) *cobra.Command {
 
 	viper.SetDefault("license", "apache 2")
 	cmd.PersistentFlags().StringP("operator-name", "o", "fdb-kubernetes-operator-controller-manager", "Name of the Deployment for the operator.")
+	cmd.PersistentFlags().Bool("version-check", true, "If the plugin should check if a newer release of the plugin exists. If so the command will not be executed.")
 	cmd.PersistentFlags().BoolP("wait", "w", true, "If the plugin should wait for confirmation before executing any action")
 	cmd.PersistentFlags().Uint16P("sleep", "z", 0, "The plugin should sleep between sequential operations for the defined time in seconds (default 0)")
 	o.configFlags.AddFlags(cmd.Flags())
@@ -145,3 +153,25 @@ func printStatement(cmd *cobra.Command, line string, mesType messageType) {
 	cmd.Printf("✔ %s\n", line)
 	color.Unset()
 }
+
+// will check plugin version and won't let any interaction happen with cluster, if it's not latest release version
+func usingLatestPluginVersion(cmd *cobra.Command, pluginVersionChecker VersionChecker) error {
+	// If the user has a self build plugin we are not performing any checks.
+	if strings.ToLower(pluginVersion) == "latest" {
+		return nil
+	}
+	latestPluginVersion, err := pluginVersionChecker.getLatestPluginVersion()
+	if err != nil {
+		return err
+	}
+
+	if pluginVersion != latestPluginVersion {
+		versionMessage := "kubectl-fdb plugin is not up-to-date, please install the latest version and try again!\n" +
+			"Your version:[" + pluginVersion + "], latest release version:[" + latestPluginVersion + "].\n" +
+			"Installation instructions can be found here: https://github.com/FoundationDB/fdb-kubernetes-operator/blob/main/kubectl-fdb/Readme.md"
+		cmd.Println(versionMessage)
+		return fmt.Errorf("outdated plugin version")
+	}
+
+	return nil
+}
diff --git a/kubectl-fdb/main.go b/kubectl-fdb/main.go
index e4f6f338..fffa904c 100644
--- a/kubectl-fdb/main.go
+++ b/kubectl-fdb/main.go
@@ -31,8 +31,7 @@ import (
 func main() {
 	flags := pflag.NewFlagSet("kubectl-fdb", pflag.ExitOnError)
 	pflag.CommandLine = flags
-
-	root := cmd.NewRootCmd(genericclioptions.IOStreams{In: os.Stdin, Out: os.Stdout, ErrOut: os.Stderr})
+	root := cmd.NewRootCmd(genericclioptions.IOStreams{In: os.Stdin, Out: os.Stdout, ErrOut: os.Stderr}, &cmd.RealVersionChecker{})
 	if err := root.Execute(); err != nil {
 		os.Exit(1)
 	}
diff --git a/mock-kubernetes-client/client/client.go b/mock-kubernetes-client/client/client.go
index 7d428fe5..175b8e95 100644
--- a/mock-kubernetes-client/client/client.go
+++ b/mock-kubernetes-client/client/client.go
@@ -170,8 +170,11 @@ func (client *MockClient) generateIP() string {
 
 // Create creates a new object
 func (client *MockClient) Create(ctx context.Context, object ctrlClient.Object, options ...ctrlClient.CreateOption) error {
-	// Ensure the default values are set.
-	object.SetCreationTimestamp(metav1.Time{Time: time.Now()})
+	// Ensure the default values are set if not present.
+	if object.GetCreationTimestamp().Time.IsZero() {
+		object.SetCreationTimestamp(metav1.Time{Time: time.Now()})
+	}
+
 	object.SetGeneration(object.GetGeneration() + 1)
 	object.SetUID(uuid.NewUUID())
 
diff --git a/pkg/fdbadminclient/admin_client.go b/pkg/fdbadminclient/admin_client.go
index 8d92b116..5e22701e 100644
--- a/pkg/fdbadminclient/admin_client.go
+++ b/pkg/fdbadminclient/admin_client.go
@@ -22,6 +22,7 @@ package fdbadminclient
 
 import (
 	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
+	"time"
 )
 
 // AdminClient describes an interface for running administrative commands on a
@@ -115,4 +116,7 @@ type AdminClient interface {
 	// WithValues will update the logger used by the current AdminClient to contain the provided key value pairs. The provided
 	// arguments must be even.
 	WithValues(keysAndValues ...interface{})
+
+	// SetTimeout will overwrite the default timeout for interacting the FDB cluster.
+	SetTimeout(timeout time.Duration)
 }
diff --git a/pkg/fdbadminclient/mock/admin_client_mock.go b/pkg/fdbadminclient/mock/admin_client_mock.go
index 53eab762..b2bdcfa6 100644
--- a/pkg/fdbadminclient/mock/admin_client_mock.go
+++ b/pkg/fdbadminclient/mock/admin_client_mock.go
@@ -1039,3 +1039,6 @@ func (client *AdminClient) GetWorstDurabilityLag() (fdbv1beta2.FoundationDBStatu
 // WithValues will update the logger used by the current AdminClient to contain the provided key value pairs. The provided
 // arguments must be even.
 func (client *AdminClient) WithValues(_ ...interface{}) {}
+
+// SetTimeout will overwrite the default timeout for interacting the FDB cluster.
+func (client *AdminClient) SetTimeout(_ time.Duration) {}
diff --git a/setup/setup.go b/setup/setup.go
index 7abedd99..4963dc1d 100644
--- a/setup/setup.go
+++ b/setup/setup.go
@@ -91,7 +91,7 @@ func (o *Options) BindFlags(fs *flag.FlagSet) {
 	)
 	fs.StringVar(&o.LogFile, "log-file", "", "The path to a file to write logs to.")
 	fs.IntVar(&o.CliTimeout, "cli-timeout", 10, "The timeout to use for CLI commands in seconds.")
-	fs.IntVar(&o.MaxCliTimeout, "max-cli-timeout", 60, "The maximum timeout to use for CLI commands in seconds. This timeout is used for CLI requests that are known to be potentially slow like get status or exclude.")
+	fs.IntVar(&o.MaxCliTimeout, "max-cli-timeout", 40, "The maximum timeout to use for CLI commands in seconds. This timeout is used for CLI requests that are known to be potentially slow like get status or exclude.")
 	fs.IntVar(&o.MaxConcurrentReconciles, "max-concurrent-reconciles", 1, "Defines the maximum number of concurrent reconciles for all controllers.")
 	fs.BoolVar(&o.CleanUpOldLogFile, "cleanup-old-cli-logs", true, "Defines if the operator should delete old fdbcli log files.")
 	fs.DurationVar(&o.LogFileMinAge, "log-file-min-age", 5*time.Minute, "Defines the minimum age of fdbcli log files before removing when \"--cleanup-old-cli-logs\" is set.")


Test patch:
diff --git a/api/v1beta2/foundationdbcluster_types_test.go b/api/v1beta2/foundationdbcluster_types_test.go
index bca0a5d5..aa0cac2f 100644
--- a/api/v1beta2/foundationdbcluster_types_test.go
+++ b/api/v1beta2/foundationdbcluster_types_test.go
@@ -2534,7 +2534,7 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			It("should set the correct value (-1) for log routers", func() {
 				spec := DatabaseConfiguration{}
 				spec.RemoteLogs = 9
-				normalized := spec.NormalizeConfigurationWithSeparatedProxies(version)
+				normalized := spec.NormalizeConfigurationWithSeparatedProxies(version, false)
 				Expect(normalized.LogRouters).To(Equal(-1))
 				Expect(normalized.RemoteLogs).To(Equal(9))
 			})
@@ -2582,7 +2582,7 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 						},
 					},
 				}
-				normalized := spec.NormalizeConfigurationWithSeparatedProxies(version)
+				normalized := spec.NormalizeConfigurationWithSeparatedProxies(version, false)
 				Expect(normalized.Regions).To(Equal([]Region{
 					{
 						DataCenters: []DataCenter{
@@ -2666,7 +2666,7 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 						},
 					},
 				}
-				normalized := spec.NormalizeConfigurationWithSeparatedProxies(version)
+				normalized := spec.NormalizeConfigurationWithSeparatedProxies(version, false)
 				Expect(normalized.Regions).To(Equal([]Region{
 					{
 						DataCenters: []DataCenter{
@@ -3499,6 +3499,14 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			Expect(cluster.ShouldUseLocks()).To(BeTrue())
 
 			cluster.Spec.FaultDomain.ZoneCount = 0
+			Expect(cluster.ShouldUseLocks()).To(BeFalse())
+
+			cluster.Spec.DatabaseConfiguration.RedundancyMode = RedundancyModeThreeDataHall
+			Expect(cluster.ShouldUseLocks()).To(BeTrue())
+
+			cluster.Spec.DatabaseConfiguration.RedundancyMode = RedundancyModeDouble
+			Expect(cluster.ShouldUseLocks()).To(BeFalse())
+
 			cluster.Spec.DatabaseConfiguration.Regions = []Region{
 				{},
 				{},
diff --git a/controllers/exclude_processes_test.go b/controllers/exclude_processes_test.go
index a9da13f2..96627fb2 100644
--- a/controllers/exclude_processes_test.go
+++ b/controllers/exclude_processes_test.go
@@ -38,6 +38,7 @@ var _ = Describe("exclude_processes", func() {
 	var allowedExclusions int
 	var ongoingExclusions int
 	var missingProcesses []fdbv1beta2.ProcessGroupID
+	var processClass fdbv1beta2.ProcessClass
 
 	When("validating if processes can be excluded", func() {
 		BeforeEach(func() {
@@ -62,220 +63,211 @@ var _ = Describe("exclude_processes", func() {
 		JustBeforeEach(func() {
 			processCounts, err := cluster.GetProcessCountsWithDefaults()
 			Expect(err).NotTo(HaveOccurred())
-			allowedExclusions, missingProcesses = canExcludeNewProcesses(globalControllerLogger, cluster, fdbv1beta2.ProcessClassStorage, processCounts.Storage, ongoingExclusions, false)
+			allowedExclusions, missingProcesses = getAllowedExclusionsAndMissingProcesses(globalControllerLogger, cluster, processClass, processCounts.Map()[processClass], ongoingExclusions, false)
 		})
 
 		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() {
-						Expect(allowedExclusions).To(BeNumerically("==", 0))
-						Expect(missingProcesses).To(BeEmpty())
-					})
+			When("using the storage process class", func() {
+				BeforeEach(func() {
+					processClass = fdbv1beta2.ProcessClassStorage
 				})
 
-				When("no additional processes are running", func() {
-					When("the redundancy mode is single", func() {
+				When("all processes are healthy", func() {
+					When("no additional processes are running", func() {
 						It("should not allow the exclusion", func() {
 							Expect(allowedExclusions).To(BeNumerically("==", 0))
 							Expect(missingProcesses).To(BeEmpty())
 						})
 					})
 
-					When("the redundancy mode is double", func() {
-						BeforeEach(func() {
-							cluster.Spec.DatabaseConfiguration.RedundancyMode = fdbv1beta2.RedundancyModeDouble
+					When("no additional processes are running", func() {
+						When("the redundancy mode is single", func() {
+							It("should not allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 0))
+								Expect(missingProcesses).To(BeEmpty())
+							})
 						})
 
-						It("should not allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", cluster.DesiredFaultTolerance()))
-							Expect(missingProcesses).To(BeEmpty())
-						})
-					})
+						When("the redundancy mode is double", func() {
+							BeforeEach(func() {
+								cluster.Spec.DatabaseConfiguration.RedundancyMode = fdbv1beta2.RedundancyModeDouble
+							})
 
-					When("the redundancy mode is triple", func() {
-						BeforeEach(func() {
-							cluster.Spec.DatabaseConfiguration.RedundancyMode = fdbv1beta2.RedundancyModeTriple
-						})
+							It("should not allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", cluster.DesiredFaultTolerance()))
+								Expect(missingProcesses).To(BeEmpty())
+							})
 
-						It("should not allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", cluster.DesiredFaultTolerance()))
-							Expect(missingProcesses).To(BeEmpty())
-						})
-					})
-				})
+							When("there are failed processes", func() {
+								BeforeEach(func() {
 
-				When("one additional process is running", func() {
-					BeforeEach(func() {
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-					})
+									for idx, processGroup := range cluster.Status.ProcessGroups {
+										if processGroup.ProcessClass != processClass {
+											continue
+										}
 
-					It("should allow the exclusion", func() {
-						Expect(allowedExclusions).To(BeNumerically("==", 1))
-						Expect(missingProcesses).To(BeEmpty())
-					})
+										cluster.Status.ProcessGroups[idx].UpdateCondition(fdbv1beta2.MissingProcesses, true)
+										cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
+										break
+									}
+								})
 
-					When("there is one ongoing exclusion", func() {
-						BeforeEach(func() {
-							ongoingExclusions = 1
+								It("should not allow the exclusion", func() {
+									Expect(allowedExclusions).To(BeZero())
+									Expect(missingProcesses).To(ConsistOf(fdbv1beta2.ProcessGroupID("storage-1")))
+								})
+							})
 						})
 
-						It("should not allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", 0))
-							Expect(missingProcesses).To(BeEmpty())
+						When("the redundancy mode is triple", func() {
+							BeforeEach(func() {
+								cluster.Spec.DatabaseConfiguration.RedundancyMode = fdbv1beta2.RedundancyModeTriple
+							})
+
+							It("should allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", cluster.DesiredFaultTolerance()))
+								Expect(missingProcesses).To(BeEmpty())
+							})
+
+							When("there are failed processes", func() {
+								BeforeEach(func() {
+
+									for idx, processGroup := range cluster.Status.ProcessGroups {
+										if processGroup.ProcessClass != processClass {
+											continue
+										}
+
+										cluster.Status.ProcessGroups[idx].UpdateCondition(fdbv1beta2.MissingProcesses, true)
+										cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
+										break
+									}
+								})
+
+								It("should allow the exclusion", func() {
+									Expect(allowedExclusions).To(BeNumerically("==", cluster.DesiredFaultTolerance()-1))
+									Expect(missingProcesses).To(ConsistOf(fdbv1beta2.ProcessGroupID("storage-1")))
+								})
+							})
 						})
 					})
 
-					When("the additional process is marked for removal", func() {
+					When("one additional process is running", func() {
 						BeforeEach(func() {
-							// In this case the process group is marked for removal but is counted against the valid
-							// processes as the process is not yet excluded.
-							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].MarkForRemoval()
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
 						})
 
 						It("should allow the exclusion", func() {
 							Expect(allowedExclusions).To(BeNumerically("==", 1))
 							Expect(missingProcesses).To(BeEmpty())
 						})
-					})
 
-					When("the additional process is marked for removal and is excluded", func() {
-						BeforeEach(func() {
-							// In this case we have no additional processes running, as the valid processes is the
-							// same as the desired count of processes.
-							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].MarkForRemoval()
-							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].SetExclude()
+						When("there is one ongoing exclusion", func() {
+							BeforeEach(func() {
+								ongoingExclusions = 1
+							})
+
+							It("should not allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 0))
+								Expect(missingProcesses).To(BeEmpty())
+							})
 						})
 
-						It("should not allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", 0))
-							Expect(missingProcesses).To(BeEmpty())
+						When("the additional process is marked for removal", func() {
+							BeforeEach(func() {
+								// In this case the process group is marked for removal but is counted against the valid
+								// processes as the process is not yet excluded.
+								cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].MarkForRemoval()
+							})
+
+							It("should allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 1))
+								Expect(missingProcesses).To(BeEmpty())
+							})
 						})
-					})
-				})
-			})
 
-			When("one process group is missing", func() {
-				BeforeEach(func() {
-					createMissingProcesses(cluster, 1, fdbv1beta2.ProcessClassStorage)
-				})
+						When("the additional process is marked for removal and is excluded", func() {
+							BeforeEach(func() {
+								// In this case we have no additional processes running, as the valid processes is the
+								// same as the desired count of processes.
+								cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].MarkForRemoval()
+								cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].SetExclude()
+							})
 
-				When("no additional processes are running", func() {
-					It("should not allow the exclusion", func() {
-						Expect(allowedExclusions).To(BeNumerically("==", 0))
-						Expect(missingProcesses).To(HaveLen(1))
+							It("should not allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 0))
+								Expect(missingProcesses).To(BeEmpty())
+							})
+						})
 					})
 				})
 
-				When("additional processes are running", func() {
+				When("one process group is missing", func() {
 					BeforeEach(func() {
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-						// Add two process groups
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1338", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-					})
-
-					It("should not allow the exclusion", func() {
-						Expect(allowedExclusions).To(BeNumerically("==", 0))
-						Expect(missingProcesses).To(HaveLen(1))
+						createMissingProcesses(cluster, 1, processClass)
 					})
 
-					When("there is one ongoing exclusion", func() {
-						BeforeEach(func() {
-							ongoingExclusions = 1
-						})
-
+					When("no additional processes are running", func() {
 						It("should not allow the exclusion", func() {
 							Expect(allowedExclusions).To(BeNumerically("==", 0))
 							Expect(missingProcesses).To(HaveLen(1))
 						})
 					})
 
-					When("the missing timestamp is older than 5 minutes", func() {
+					When("additional processes are running", func() {
 						BeforeEach(func() {
-							for idx, processGroup := range cluster.Status.ProcessGroups {
-								timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
-								if timestamp == nil {
-									continue
-								}
-
-								cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
-							}
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							// Add two process groups
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1338", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
 						})
 
-						It("should allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", 1))
+						It("should not allow the exclusion", func() {
+							Expect(allowedExclusions).To(BeNumerically("==", 0))
 							Expect(missingProcesses).To(HaveLen(1))
 						})
-					})
-				})
-			})
-
-			When("more process groups are missing than the operator is allowed to automatically replace", func() {
-				BeforeEach(func() {
-					createMissingProcesses(cluster, cluster.GetMaxConcurrentAutomaticReplacements()+1, fdbv1beta2.ProcessClassStorage)
-				})
 
-				It("should not allow the exclusion", func() {
-					Expect(allowedExclusions).To(BeNumerically("==", 0))
-					Expect(missingProcesses).To(Equal([]fdbv1beta2.ProcessGroupID{"storage-1", "storage-2"}))
-				})
-
-				When("the missing timestamp is older than 5 minutes", func() {
-					BeforeEach(func() {
-						for idx, processGroup := range cluster.Status.ProcessGroups {
-							timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
-							if timestamp == nil {
-								continue
-							}
-
-							cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
-						}
-					})
-
-					When("automatic replacements enabled", func() {
-						When("no exclusions are ongoing", func() {
-							It("should allow the exclusion to replace as many processes as automatic replacements are allowed", func() {
-								Expect(allowedExclusions).To(BeNumerically("==", cluster.GetMaxConcurrentAutomaticReplacements()))
-								Expect(missingProcesses).To(HaveLen(cluster.GetMaxConcurrentAutomaticReplacements() + 1))
-							})
-						})
-
-						When("one exclusion is ongoing", func() {
+						When("there is one ongoing exclusion", func() {
 							BeforeEach(func() {
 								ongoingExclusions = 1
 							})
 
-							It("should allow the exclusion to replace as many processes as automatic replacements are allowed", func() {
-								Expect(allowedExclusions).To(BeNumerically("==", cluster.GetMaxConcurrentAutomaticReplacements()-ongoingExclusions))
-								Expect(missingProcesses).To(HaveLen(cluster.GetMaxConcurrentAutomaticReplacements() + 1))
+							It("should not allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 0))
+								Expect(missingProcesses).To(HaveLen(1))
 							})
 						})
-					})
 
-					When("automatic replacements disabled", func() {
-						BeforeEach(func() {
-							cluster.Spec.AutomationOptions.Replacements.Enabled = pointer.Bool(false)
-						})
+						When("the missing timestamp is older than 5 minutes", func() {
+							BeforeEach(func() {
+								for idx, processGroup := range cluster.Status.ProcessGroups {
+									timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
+									if timestamp == nil {
+										continue
+									}
 
-						It("should not allow the exclusions", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", -1*len(missingProcesses)))
-							Expect(missingProcesses).To(HaveLen(cluster.GetMaxConcurrentAutomaticReplacements() + 1))
+									cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
+								}
+							})
+
+							It("should allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 1))
+								Expect(missingProcesses).To(HaveLen(1))
+							})
 						})
 					})
 				})
-			})
 
-			When("two process groups of a different type are missing", func() {
-				BeforeEach(func() {
-					createMissingProcesses(cluster, 2, fdbv1beta2.ProcessClassLog)
-				})
+				When("two process groups of a different type are missing", func() {
+					BeforeEach(func() {
+						createMissingProcesses(cluster, 2, fdbv1beta2.ProcessClassLog)
+					})
 
-				It("should allow the exclusion", func() {
-					Expect(allowedExclusions).To(BeNumerically("==", 0))
-					Expect(missingProcesses).To(BeEmpty())
+					It("should allow the exclusion", func() {
+						Expect(allowedExclusions).To(BeNumerically("==", 0))
+						Expect(missingProcesses).To(BeEmpty())
+					})
 				})
 			})
 		})
@@ -293,63 +285,124 @@ var _ = Describe("exclude_processes", func() {
 				Expect(err).NotTo(HaveOccurred())
 			})
 
-			When("two process groups are missing", func() {
+			When("the storage class is used", func() {
 				BeforeEach(func() {
-					createMissingProcesses(cluster, 2, fdbv1beta2.ProcessClassStorage)
+					processClass = fdbv1beta2.ProcessClassStorage
 				})
 
-				When("no additional processes are running", func() {
-					It("should not allow the exclusion", func() {
-						Expect(allowedExclusions).To(BeNumerically("==", 0))
-						Expect(missingProcesses).To(HaveLen(2))
-					})
-				})
-
-				When("additional processes are running", func() {
+				When("two process groups are missing", func() {
 					BeforeEach(func() {
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1338", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1339", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1340", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+						createMissingProcesses(cluster, 2, processClass)
 					})
 
-					It("should not allow the exclusion", func() {
-						Expect(allowedExclusions).To(BeNumerically("==", 0))
-						Expect(missingProcesses).To(HaveLen(2))
+					When("no additional processes are running", func() {
+						It("should not allow the exclusion", func() {
+							Expect(allowedExclusions).To(BeNumerically("==", 0))
+							Expect(missingProcesses).To(HaveLen(2))
+						})
 					})
 
-					When("the missing timestamps are older than 5 minutes", func() {
+					When("additional processes are running", func() {
 						BeforeEach(func() {
-							for idx, processGroup := range cluster.Status.ProcessGroups {
-								timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
-								if timestamp == nil {
-									continue
-								}
-
-								cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
-							}
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1338", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1339", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1340", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
 						})
 
-						It("should allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", 2))
+						It("should not allow the exclusion", func() {
+							Expect(allowedExclusions).To(BeNumerically("==", 0))
 							Expect(missingProcesses).To(HaveLen(2))
 						})
+
+						When("the missing timestamps are older than 5 minutes", func() {
+							BeforeEach(func() {
+								for idx, processGroup := range cluster.Status.ProcessGroups {
+									timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
+									if timestamp == nil {
+										continue
+									}
+
+									cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
+								}
+							})
+
+							It("should allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 2))
+								Expect(missingProcesses).To(HaveLen(2))
+							})
+						})
+					})
+				})
+
+				When("five process groups are missing", func() {
+					BeforeEach(func() {
+						createMissingProcesses(cluster, 5, processClass)
+					})
+
+					It("should not allow the exclusion", func() {
+						Expect(allowedExclusions).To(BeNumerically("==", 0))
+						Expect(missingProcesses).To(Equal([]fdbv1beta2.ProcessGroupID{"storage-1", "storage-10", "storage-11", "storage-12", "storage-13"}))
 					})
 				})
 			})
 
-			When("five process groups are missing", func() {
+			When("the log class is used", func() {
 				BeforeEach(func() {
-					createMissingProcesses(cluster, 5, fdbv1beta2.ProcessClassStorage)
+					processClass = fdbv1beta2.ProcessClassLog
 				})
 
-				It("should not allow the exclusion", func() {
-					Expect(allowedExclusions).To(BeNumerically("==", 0))
-					Expect(missingProcesses).To(Equal([]fdbv1beta2.ProcessGroupID{"storage-1", "storage-10", "storage-11", "storage-12", "storage-13"}))
+				When("two process groups are missing", func() {
+					BeforeEach(func() {
+						createMissingProcesses(cluster, 2, processClass)
+					})
+
+					When("no additional processes are running", func() {
+						It("should not allow the exclusion", func() {
+							Expect(allowedExclusions).To(BeNumerically("==", 0))
+							Expect(missingProcesses).To(HaveLen(2))
+						})
+					})
+
+					When("additional processes are running", func() {
+						BeforeEach(func() {
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1338", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1339", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1340", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+						})
+
+						It("should not allow the exclusion", func() {
+							Expect(allowedExclusions).To(BeNumerically("==", 0))
+							Expect(missingProcesses).To(HaveLen(2))
+						})
+
+						When("the missing timestamps are older than 5 minutes", func() {
+							BeforeEach(func() {
+								for idx, processGroup := range cluster.Status.ProcessGroups {
+									timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
+									if timestamp == nil {
+										continue
+									}
+
+									cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
+								}
+							})
+
+							It("should allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 2))
+								Expect(missingProcesses).To(HaveLen(2))
+							})
+						})
+					})
 				})
 			})
 		})
@@ -652,6 +705,53 @@ var _ = Describe("exclude_processes", func() {
 			})
 		})
 	})
+
+	DescribeTable("when getting the allowed exclusions", func(validProcesses int, desiredProcessCount int, ongoingExclusions int, faultTolerance int, expected int) {
+		Expect(getAllowedExclusions(GinkgoLogr, validProcesses, desiredProcessCount, ongoingExclusions, faultTolerance)).To(BeNumerically("==", expected))
+	},
+		Entry("when as many valid processes are running as desired",
+			10,
+			10,
+			0,
+			2,
+			2),
+		Entry("when as many valid processes are running as desired but one exclusion is ongoing",
+			10,
+			10,
+			1,
+			2,
+			1),
+		Entry("when as many valid processes are running as desired but two exclusions are ongoing",
+			10,
+			10,
+			2,
+			2,
+			0),
+		Entry("when more valid processes are running than desired",
+			20,
+			10,
+			0,
+			2,
+			12),
+		Entry("when more valid processes are running than desired but there are 10 exclusions ongoing",
+			20,
+			10,
+			10,
+			2,
+			2),
+		Entry("when only 8 valid processes are running",
+			8,
+			10,
+			0,
+			2,
+			0),
+		Entry("when no valid processes are running",
+			0,
+			10,
+			0,
+			2,
+			0),
+	)
 })
 
 func createMissingProcesses(cluster *fdbv1beta2.FoundationDBCluster, count int, processClass fdbv1beta2.ProcessClass) {
diff --git a/e2e/fixtures/factory.go b/e2e/fixtures/factory.go
index 0c268202..6a08e415 100644
--- a/e2e/fixtures/factory.go
+++ b/e2e/fixtures/factory.go
@@ -123,7 +123,7 @@ func (factory *Factory) getClient() *kubernetes.Clientset {
 
 // DeletePod deletes the provided Pod
 func (factory *Factory) DeletePod(pod *corev1.Pod) {
-	gomega.Expect(factory.GetControllerRuntimeClient().Delete(ctx.TODO(), pod)).NotTo(gomega.HaveOccurred())
+	factory.Delete(pod)
 }
 
 // GetPod returns the Pod matching the namespace and name
@@ -385,7 +385,7 @@ func (factory *Factory) startFDBFromClusterSpec(
 	factory.logClusterInfo(spec)
 
 	fdbCluster, err := factory.ensureFdbClusterExists(spec, config)
-	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "cluster was not created in the expected time")
 	factory.singleton.namespaces = append(
 		factory.singleton.namespaces,
 		fdbCluster.cluster.Namespace,
@@ -818,6 +818,16 @@ func (factory *Factory) CreateIfAbsent(object client.Object) error {
 	return nil
 }
 
+// Delete will delete the provided resource if it exists.
+func (factory *Factory) Delete(object client.Object) {
+	err := factory.GetControllerRuntimeClient().Delete(ctx.TODO(), object)
+	if err == nil || k8serrors.IsNotFound(err) {
+		return
+	}
+
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+}
+
 // GetOperatorImage returns the operator image provided via command line. If a registry was defined the registry will be
 // prepended.
 func (factory *Factory) GetOperatorImage() string {
diff --git a/e2e/fixtures/fdb_cluster.go b/e2e/fixtures/fdb_cluster.go
index 8624b970..a14c06bb 100644
--- a/e2e/fixtures/fdb_cluster.go
+++ b/e2e/fixtures/fdb_cluster.go
@@ -633,7 +633,7 @@ func (fdbCluster *FdbCluster) ReplacePod(pod corev1.Pod, waitForReconcile bool)
 }
 
 // ReplacePods replaces the provided Pods in the current FoundationDBCluster.
-func (fdbCluster *FdbCluster) ReplacePods(pods []corev1.Pod) {
+func (fdbCluster *FdbCluster) ReplacePods(pods []corev1.Pod, waitForReconcile bool) {
 	for _, pod := range pods {
 		fdbCluster.cluster.Spec.ProcessGroupsToRemove = append(
 			fdbCluster.cluster.Spec.ProcessGroupsToRemove,
@@ -642,6 +642,10 @@ func (fdbCluster *FdbCluster) ReplacePods(pods []corev1.Pod) {
 	}
 	fdbCluster.UpdateClusterSpec()
 
+	if !waitForReconcile {
+		return
+	}
+
 	gomega.Expect(fdbCluster.WaitForReconciliation()).NotTo(gomega.HaveOccurred())
 }
 
diff --git a/e2e/fixtures/fdb_operator_fixtures.go b/e2e/fixtures/fdb_operator_fixtures.go
index 3d372787..6cad5afb 100644
--- a/e2e/fixtures/fdb_operator_fixtures.go
+++ b/e2e/fixtures/fdb_operator_fixtures.go
@@ -56,7 +56,6 @@ func (factory *Factory) ensureFdbClusterExists(
 	fdbCluster.WaitUntilExists()
 	// Wait until cluster is reconciled -- otherwise, the operator may not have
 	// assigned pods, etc.
-
 	err = fdbCluster.WaitForReconciliation(CreationTrackerLoggerOption(config.CreationTracker))
 	if err != nil {
 		return nil, err
diff --git a/e2e/fixtures/kubernetes_fixtures.go b/e2e/fixtures/kubernetes_fixtures.go
index 9be51da5..12c9bd73 100644
--- a/e2e/fixtures/kubernetes_fixtures.go
+++ b/e2e/fixtures/kubernetes_fixtures.go
@@ -53,7 +53,7 @@ func (factory *Factory) MultipleNamespaces(dcIDs []string) []string {
 	res := make([]string, len(dcIDs))
 	for idx, dcID := range dcIDs {
 		namespace := factory.createNamespace(dcID)
-		log.Println("Create namespace" + namespace)
+		log.Println("Create namespace", namespace)
 		res[idx] = namespace
 	}
 
diff --git a/e2e/test_operator/operator_test.go b/e2e/test_operator/operator_test.go
index 4f658a42..6310a352 100644
--- a/e2e/test_operator/operator_test.go
+++ b/e2e/test_operator/operator_test.go
@@ -374,7 +374,7 @@ var _ = Describe("Operator", Label("e2e", "pr"), func() {
 		})
 
 		BeforeEach(func() {
-			useLocalitiesForExclusion = pointer.BoolDeref(fdbCluster.GetCluster().Spec.AutomationOptions.UseLocalitiesForExclusion, false)
+			useLocalitiesForExclusion = fdbCluster.GetCluster().UseLocalitiesForExclusion()
 		})
 
 		AfterEach(func() {
@@ -383,6 +383,7 @@ var _ = Describe("Operator", Label("e2e", "pr"), func() {
 			spec := fdbCluster.GetCluster().Spec.DeepCopy()
 			spec.AutomationOptions.UseLocalitiesForExclusion = pointer.Bool(useLocalitiesForExclusion)
 			fdbCluster.UpdateClusterSpecWithSpec(spec)
+			Expect(fdbCluster.GetCluster().UseLocalitiesForExclusion()).To(Equal(useLocalitiesForExclusion))
 		})
 
 		When("IP addresses are used for exclusion", func() {
@@ -399,9 +400,19 @@ var _ = Describe("Operator", Label("e2e", "pr"), func() {
 
 		When("localities are used for exclusion", func() {
 			BeforeEach(func() {
-				spec := fdbCluster.GetCluster().Spec.DeepCopy()
+				cluster := fdbCluster.GetCluster()
+
+				fdbVersion, err := fdbv1beta2.ParseFdbVersion(cluster.GetRunningVersion())
+				Expect(err).NotTo(HaveOccurred())
+
+				if !fdbVersion.SupportsLocalityBasedExclusions() {
+					Skip("provided FDB version: " + cluster.GetRunningVersion() + " doesn't support locality based exclusions")
+				}
+
+				spec := cluster.Spec.DeepCopy()
 				spec.AutomationOptions.UseLocalitiesForExclusion = pointer.Bool(true)
 				fdbCluster.UpdateClusterSpecWithSpec(spec)
+				Expect(fdbCluster.GetCluster().UseLocalitiesForExclusion()).To(BeTrue())
 			})
 
 			It("should remove the targeted Pod", func() {
@@ -902,7 +913,7 @@ var _ = Describe("Operator", Label("e2e", "pr"), func() {
 		var replacedPods []corev1.Pod
 
 		BeforeEach(func() {
-			fdbCluster.ReplacePods(fixtures.RandomPickPod(fdbCluster.GetStatelessPods().Items, 4))
+			fdbCluster.ReplacePods(fixtures.RandomPickPod(fdbCluster.GetStatelessPods().Items, 4), true)
 		})
 
 		AfterEach(func() {
@@ -1533,9 +1544,7 @@ var _ = Describe("Operator", Label("e2e", "pr"), func() {
 			pod = fixtures.ChooseRandomPod(fdbCluster.GetStatelessPods())
 			log.Printf("exclude Pod: %s", pod.Name)
 			Expect(pod.Status.PodIP).NotTo(BeEmpty())
-			_, _, err := fdbCluster.RunFdbCliCommandInOperatorWithoutRetry(fmt.Sprintf("exclude %s", pod.Status.PodIP), false, 30)
-			Expect(err).NotTo(HaveOccurred())
-
+			_, _ = fdbCluster.RunFdbCliCommandInOperator(fmt.Sprintf("exclude %s", pod.Status.PodIP), false, 30)
 			// Make sure we trigger a reconciliation to speed up the exclusion detection.
 			fdbCluster.ForceReconcile()
 		})
@@ -1691,4 +1700,48 @@ var _ = Describe("Operator", Label("e2e", "pr"), func() {
 			}).WithTimeout(10 * time.Minute).WithPolling(5 * time.Second).Should(BeNumerically("==", 1))
 		})
 	})
+
+	When("using proxies instead of grv and commit proxies", func() {
+		var originalRoleCounts fdbv1beta2.RoleCounts
+
+		BeforeEach(func() {
+			spec := fdbCluster.GetCluster().Spec.DeepCopy()
+			originalRoleCounts = spec.DatabaseConfiguration.RoleCounts
+			spec.DatabaseConfiguration.RoleCounts.Proxies = spec.DatabaseConfiguration.RoleCounts.GrvProxies + spec.DatabaseConfiguration.RoleCounts.CommitProxies
+			// Reset the more specific GRV and Commit proxy count.
+			spec.DatabaseConfiguration.RoleCounts.GrvProxies = 0
+			spec.DatabaseConfiguration.RoleCounts.CommitProxies = 0
+
+			fmt.Println("Original:", fixtures.ToJSON(originalRoleCounts), "new roleCounts:", fixtures.ToJSON(spec.DatabaseConfiguration.RoleCounts))
+			fdbCluster.UpdateClusterSpecWithSpec(spec)
+			Expect(fdbCluster.WaitForReconciliation()).NotTo(HaveOccurred())
+		})
+
+		AfterEach(func() {
+			spec := fdbCluster.GetCluster().Spec.DeepCopy()
+			spec.DatabaseConfiguration.RoleCounts = originalRoleCounts
+			fdbCluster.UpdateClusterSpecWithSpec(spec)
+			Expect(fdbCluster.WaitForReconciliation()).NotTo(HaveOccurred())
+		})
+
+		It("should configure the database to run with GRV and commit proxies but keep the proxies in the status field of the FoundationDB resource", func() {
+			// Make sure the FoundationDB configured GRV and commit proxies.
+			status := fdbCluster.GetStatus()
+			Expect(status.Cluster.DatabaseConfiguration.RoleCounts.CommitProxies).To(BeNumerically(">", 0))
+			Expect(status.Cluster.DatabaseConfiguration.RoleCounts.GrvProxies).To(BeNumerically(">", 0))
+			Expect(status.Cluster.DatabaseConfiguration.RoleCounts.Proxies).NotTo(BeZero())
+
+			fmt.Println("Status configuration:", fixtures.ToJSON(status.Cluster.DatabaseConfiguration))
+			// Verify the FoundationDB Cluster resource
+			cluster := fdbCluster.GetCluster()
+			// Make sure that the spec of the FoundationDB resource only has proxies defined.
+			Expect(cluster.Spec.DatabaseConfiguration.RoleCounts.GrvProxies).To(BeZero())
+			Expect(cluster.Spec.DatabaseConfiguration.RoleCounts.CommitProxies).To(BeZero())
+			Expect(cluster.Spec.DatabaseConfiguration.RoleCounts.Proxies).NotTo(BeZero())
+			// Make sure that the status of the FoundationDB resource only has proxies defined.
+			Expect(cluster.Status.DatabaseConfiguration.RoleCounts.GrvProxies).To(BeZero())
+			Expect(cluster.Status.DatabaseConfiguration.RoleCounts.CommitProxies).To(BeZero())
+			Expect(cluster.Status.DatabaseConfiguration.RoleCounts.Proxies).NotTo(BeZero())
+		})
+	})
 })
diff --git a/e2e/test_operator/suite_test.go b/e2e/test_operator/suite_test.go
index 82b59f4a..2a5649e2 100644
--- a/e2e/test_operator/suite_test.go
+++ b/e2e/test_operator/suite_test.go
@@ -28,7 +28,7 @@ import (
 	"github.com/onsi/gomega"
 )
 
-func TestOperatorCreationVelocity(t *testing.T) {
+func TestOperator(t *testing.T) {
 	gomega.SetDefaultEventuallyTimeout(10 * time.Second)
 	fixtures.RunGinkgoTests(t, "Operator test suite")
 }
diff --git a/e2e/test_operator_ha/operator_ha_test.go b/e2e/test_operator_ha/operator_ha_test.go
new file mode 100644
index 00000000..a4b56833
--- /dev/null
+++ b/e2e/test_operator_ha/operator_ha_test.go
@@ -0,0 +1,162 @@
+/*
+ * operator_ha_test.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 operatorha
+
+/*
+This test suite includes functional tests to ensure normal operational tasks are working fine.
+Those tests include replacements of healthy or fault Pods and setting different configurations.
+
+The assumption is that every test case reverts the changes that were done on the cluster.
+In order to improve the test speed we only create one FoundationDB HA cluster initially.
+This cluster will be used for all tests.
+*/
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"log"
+	"strconv"
+	"time"
+
+	"github.com/FoundationDB/fdb-kubernetes-operator/e2e/fixtures"
+	chaosmesh "github.com/chaos-mesh/chaos-mesh/api/v1alpha1"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var (
+	factory     *fixtures.Factory
+	fdbCluster  *fixtures.HaFdbCluster
+	testOptions *fixtures.FactoryOptions
+)
+
+func init() {
+	testOptions = fixtures.InitFlags()
+}
+
+var _ = BeforeSuite(func() {
+	factory = fixtures.CreateFactory(testOptions)
+	fdbCluster = factory.CreateFdbHaCluster(
+		fixtures.DefaultClusterConfigWithHaMode(fixtures.HaFourZoneSingleSat, false),
+		factory.GetClusterOptions()...,
+	)
+
+	// Load some data into the cluster.
+	factory.CreateDataLoaderIfAbsent(fdbCluster.GetPrimary())
+
+	// In order to test the robustness of the operator we try to kill the operator Pods every minute.
+	if factory.ChaosTestsEnabled() {
+		for _, cluster := range fdbCluster.GetAllClusters() {
+			factory.ScheduleInjectPodKill(
+				fixtures.GetOperatorSelector(cluster.Namespace()),
+				"*/2 * * * *",
+				chaosmesh.OneMode,
+			)
+		}
+	}
+})
+
+var _ = AfterSuite(func() {
+	if CurrentSpecReport().Failed() {
+		log.Printf("failed due to %s", CurrentSpecReport().FailureMessage())
+	}
+	factory.Shutdown()
+})
+
+var _ = Describe("Operator HA tests", Label("e2e", "pr"), func() {
+	var availabilityCheck bool
+
+	AfterEach(func() {
+		// Reset availabilityCheck if a test case removes this check.
+		availabilityCheck = true
+		if CurrentSpecReport().Failed() {
+			factory.DumpStateHaCluster(fdbCluster)
+		}
+		Expect(fdbCluster.WaitForReconciliation()).ToNot(HaveOccurred())
+		factory.StopInvariantCheck()
+		// Make sure all data is present in the cluster
+		fdbCluster.GetPrimary().EnsureTeamTrackersAreHealthy()
+		fdbCluster.GetPrimary().EnsureTeamTrackersHaveMinReplicas()
+	})
+
+	JustBeforeEach(func() {
+		if availabilityCheck {
+			Expect(fdbCluster.GetPrimary().InvariantClusterStatusAvailable()).NotTo(HaveOccurred())
+		}
+	})
+
+	When("replacing satellite Pods and the new Pods are stuck in pending", func() {
+		var desiredRunningPods int
+		var quota *corev1.ResourceQuota
+
+		BeforeEach(func() {
+			satellite := fdbCluster.GetPrimarySatellite()
+			satelliteCluster := satellite.GetCluster()
+
+			processCounts, err := satelliteCluster.GetProcessCountsWithDefaults()
+			Expect(err).NotTo(HaveOccurred())
+
+			// Create Quota to limit the PVCs that can be created to 0. This will mean no new PVCs can be created.
+			quota = &corev1.ResourceQuota{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "testing-quota",
+					Namespace: satellite.Namespace(),
+				},
+				Spec: corev1.ResourceQuotaSpec{
+					Hard: corev1.ResourceList{
+						"persistentvolumeclaims": resource.MustParse(strconv.Itoa(0)),
+					},
+				},
+			}
+			Expect(factory.CreateIfAbsent(quota)).NotTo(HaveOccurred())
+
+			desiredRunningPods = processCounts.Log - satelliteCluster.DesiredFaultTolerance()
+
+			// Replace all Pods for this cluster.
+			satellite.ReplacePods(satellite.GetAllPods().Items, false)
+		})
+
+		AfterEach(func() {
+			// Make sure that the quota is deleted and new PVCs can be created.
+			factory.Delete(quota)
+			Expect(fdbCluster.GetPrimarySatellite().WaitForReconciliation()).NotTo(HaveOccurred())
+		})
+
+		It("should not replace too many Pods and bring down the satellite", func() {
+			satellite := fdbCluster.GetPrimarySatellite()
+
+			Consistently(func() int {
+				var runningPods int
+				for _, pod := range satellite.GetAllPods().Items {
+					if pod.Status.Phase != corev1.PodRunning {
+						continue
+					}
+
+					runningPods++
+				}
+
+				// We should add here another check that the cluster stays in the primary.
+				return runningPods
+			}).WithTimeout(10 * time.Minute).WithPolling(2 * time.Second).Should(BeNumerically(">=", desiredRunningPods))
+		})
+	})
+})
diff --git a/e2e/test_operator_ha/suite_test.go b/e2e/test_operator_ha/suite_test.go
new file mode 100644
index 00000000..ece141dc
--- /dev/null
+++ b/e2e/test_operator_ha/suite_test.go
@@ -0,0 +1,34 @@
+/*
+ * suite_test.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 operatorha
+
+import (
+	"testing"
+	"time"
+
+	"github.com/FoundationDB/fdb-kubernetes-operator/e2e/fixtures"
+	"github.com/onsi/gomega"
+)
+
+func TestOperatorHA(t *testing.T) {
+	gomega.SetDefaultEventuallyTimeout(10 * time.Second)
+	fixtures.RunGinkgoTests(t, "Operator HA test suite")
+}
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 0abf9d66..91973ccb 100644
--- a/e2e/test_operator_ha_upgrades/operator_ha_upgrade_test.go
+++ b/e2e/test_operator_ha_upgrades/operator_ha_upgrade_test.go
@@ -542,6 +542,13 @@ var _ = Describe("Operator HA Upgrades", Label("e2e", "pr"), func() {
 	DescribeTable(
 		"when locality based exclusions are used and the resources are limited in a satellite namespace",
 		func(beforeVersion string, targetVersion string) {
+			fdbVersion, err := fdbv1beta2.ParseFdbVersion(beforeVersion)
+			Expect(err).NotTo(HaveOccurred())
+
+			if !fdbVersion.SupportsLocalityBasedExclusions() {
+				Skip("provided FDB version: " + beforeVersion + " doesn't support locality based exclusions")
+			}
+
 			clusterConfig := fixtures.DefaultClusterConfigWithHaMode(fixtures.HaFourZoneSingleSat, false)
 			clusterConfig.UseLocalityBasedExclusions = true
 
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 a9717d70..ff80e60b 100644
--- a/e2e/test_operator_upgrades_variations/operator_upgrades_variations_test.go
+++ b/e2e/test_operator_upgrades_variations/operator_upgrades_variations_test.go
@@ -233,6 +233,13 @@ var _ = Describe("Operator Upgrades", Label("e2e", "pr"), func() {
 	DescribeTable(
 		"with locality based exclusions",
 		func(beforeVersion string, targetVersion string) {
+			fdbVersion, err := fdbv1beta2.ParseFdbVersion(beforeVersion)
+			Expect(err).NotTo(HaveOccurred())
+
+			if !fdbVersion.SupportsLocalityBasedExclusions() {
+				Skip("provided FDB version: " + beforeVersion + " doesn't support locality based exclusions")
+			}
+
 			performUpgrade(testConfig{
 				beforeVersion: beforeVersion,
 				targetVersion: targetVersion,
diff --git a/internal/monitor_conf_test.go b/internal/monitor_conf_test.go
index da46b8c9..9ef3a373 100644
--- a/internal/monitor_conf_test.go
+++ b/internal/monitor_conf_test.go
@@ -446,7 +446,7 @@ var _ = Describe("monitor_conf", func() {
 			It("should substitute the variables in the start command", func() {
 				substitutions, err := GetSubstitutionsFromClusterAndPod(logr.Discard(), cluster, pod)
 				Expect(err).NotTo(HaveOccurred())
-				command, err = getStartCommandWithSubstitutions(cluster, processClass, substitutions, 1, 1)
+				command, err = GetStartCommandWithSubstitutions(cluster, processClass, substitutions, 1, 1)
 				Expect(err).NotTo(HaveOccurred())
 
 				Expect(command).To(Equal(strings.Join([]string{
@@ -472,7 +472,7 @@ var _ = Describe("monitor_conf", func() {
 
 					substitutions, err := GetSubstitutionsFromClusterAndPod(logr.Discard(), cluster, pod)
 					Expect(err).NotTo(HaveOccurred())
-					command, err = getStartCommandWithSubstitutions(cluster, processClass, substitutions, 1, 1)
+					command, err = GetStartCommandWithSubstitutions(cluster, processClass, substitutions, 1, 1)
 					Expect(err).NotTo(HaveOccurred())
 
 					Expect(command).To(Equal(strings.Join([]string{
@@ -501,7 +501,7 @@ var _ = Describe("monitor_conf", func() {
 			It("should generate the unsorted command-line", func() {
 				substitutions, err := GetSubstitutionsFromClusterAndPod(logr.Discard(), cluster, pod)
 				Expect(err).NotTo(HaveOccurred())
-				command, err = getStartCommandWithSubstitutions(cluster, processClass, substitutions, 1, 1)
+				command, err = GetStartCommandWithSubstitutions(cluster, processClass, substitutions, 1, 1)
 				Expect(err).NotTo(HaveOccurred())
 
 				Expect(command).To(Equal(strings.Join([]string{
@@ -523,7 +523,7 @@ var _ = Describe("monitor_conf", func() {
 				It("should fill in the process number", func() {
 					substitutions, err := GetSubstitutionsFromClusterAndPod(logr.Discard(), cluster, pod)
 					Expect(err).NotTo(HaveOccurred())
-					command, err = getStartCommandWithSubstitutions(cluster, processClass, substitutions, 2, 3)
+					command, err = GetStartCommandWithSubstitutions(cluster, processClass, substitutions, 2, 3)
 					Expect(err).NotTo(HaveOccurred())
 
 					Expect(command).To(Equal(strings.Join([]string{
@@ -552,7 +552,7 @@ var _ = Describe("monitor_conf", func() {
 			It("should generate the sorted command-line", func() {
 				substitutions, err := GetSubstitutionsFromClusterAndPod(logr.Discard(), cluster, pod)
 				Expect(err).NotTo(HaveOccurred())
-				command, err = getStartCommandWithSubstitutions(cluster, processClass, substitutions, 1, 1)
+				command, err = GetStartCommandWithSubstitutions(cluster, processClass, substitutions, 1, 1)
 				Expect(err).NotTo(HaveOccurred())
 
 				Expect(command).To(Equal(strings.Join([]string{
@@ -575,7 +575,7 @@ var _ = Describe("monitor_conf", func() {
 			It("should substitute the variables in the start command", func() {
 				substitutions, err := GetSubstitutionsFromClusterAndPod(logr.Discard(), cluster, pod)
 				Expect(err).NotTo(HaveOccurred())
-				command, err = getStartCommandWithSubstitutions(cluster, processClass, substitutions, 1, 2)
+				command, err = GetStartCommandWithSubstitutions(cluster, processClass, substitutions, 1, 2)
 				Expect(err).NotTo(HaveOccurred())
 
 				id := "storage-1"
@@ -594,7 +594,7 @@ var _ = Describe("monitor_conf", func() {
 					"--seed_cluster_file=/var/dynamic-conf/fdb.cluster",
 				}, " ")))
 
-				command, err = getStartCommandWithSubstitutions(cluster, processClass, substitutions, 2, 2)
+				command, err = GetStartCommandWithSubstitutions(cluster, processClass, substitutions, 2, 2)
 				Expect(err).NotTo(HaveOccurred())
 				Expect(command).To(Equal(strings.Join([]string{
 					"/usr/bin/fdbserver",
@@ -620,7 +620,7 @@ var _ = Describe("monitor_conf", func() {
 
 				substitutions, err := GetSubstitutionsFromClusterAndPod(logr.Discard(), cluster, pod)
 				Expect(err).NotTo(HaveOccurred())
-				command, err = getStartCommandWithSubstitutions(cluster, fdbv1beta2.ProcessClassStorage, substitutions, 1, 1)
+				command, err = GetStartCommandWithSubstitutions(cluster, fdbv1beta2.ProcessClassStorage, substitutions, 1, 1)
 				Expect(err).NotTo(HaveOccurred())
 			})
 
@@ -652,7 +652,7 @@ var _ = Describe("monitor_conf", func() {
 
 				substitutions, err := GetSubstitutionsFromClusterAndPod(logr.Discard(), cluster, pod)
 				Expect(err).NotTo(HaveOccurred())
-				command, err = getStartCommandWithSubstitutions(cluster, fdbv1beta2.ProcessClassStorage, substitutions, 1, 1)
+				command, err = GetStartCommandWithSubstitutions(cluster, fdbv1beta2.ProcessClassStorage, substitutions, 1, 1)
 				Expect(err).NotTo(HaveOccurred())
 			})
 
@@ -679,7 +679,7 @@ var _ = Describe("monitor_conf", func() {
 				cluster.Status.RunningVersion = fdbv1beta2.Versions.Default.String()
 				substitutions, err := GetSubstitutionsFromClusterAndPod(logr.Discard(), cluster, pod)
 				Expect(err).NotTo(HaveOccurred())
-				command, err = getStartCommandWithSubstitutions(cluster, fdbv1beta2.ProcessClassStorage, substitutions, 1, 1)
+				command, err = GetStartCommandWithSubstitutions(cluster, fdbv1beta2.ProcessClassStorage, substitutions, 1, 1)
 				Expect(err).NotTo(HaveOccurred())
 			})
 
diff --git a/kubectl-fdb/cmd/analyze_test.go b/kubectl-fdb/cmd/analyze_test.go
index 7d509342..b09c2b74 100644
--- a/kubectl-fdb/cmd/analyze_test.go
+++ b/kubectl-fdb/cmd/analyze_test.go
@@ -486,7 +486,7 @@ var _ = Describe("[plugin] analyze cluster", func() {
 			errBuffer := bytes.Buffer{}
 			inBuffer := bytes.Buffer{}
 
-			rootCmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer})
+			rootCmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{})
 			rootCmd.SetArgs([]string{"analyze"})
 
 			err := rootCmd.Execute()
diff --git a/kubectl-fdb/cmd/deprecation_test.go b/kubectl-fdb/cmd/deprecation_test.go
index 5c673e77..2c1f8d8c 100644
--- a/kubectl-fdb/cmd/deprecation_test.go
+++ b/kubectl-fdb/cmd/deprecation_test.go
@@ -209,7 +209,7 @@ var _ = Describe("[plugin] deprecation command", func() {
 			errBuffer := bytes.Buffer{}
 			inBuffer := bytes.Buffer{}
 
-			cmd = NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer})
+			cmd = NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{})
 		})
 
 		It("should print out the help information", func() {
diff --git a/kubectl-fdb/cmd/restart_test.go b/kubectl-fdb/cmd/restart_test.go
index 89f37994..86105f30 100644
--- a/kubectl-fdb/cmd/restart_test.go
+++ b/kubectl-fdb/cmd/restart_test.go
@@ -42,7 +42,7 @@ var _ = Describe("[plugin] root command", func() {
 		})
 
 		It("should not throw an error", func() {
-			cmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer})
+			cmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{})
 
 			args := []string{"restart", "-c", "sample"}
 			cmd.SetArgs(args)
diff --git a/kubectl-fdb/cmd/root_test.go b/kubectl-fdb/cmd/root_test.go
index d34b0145..63b58293 100644
--- a/kubectl-fdb/cmd/root_test.go
+++ b/kubectl-fdb/cmd/root_test.go
@@ -42,7 +42,7 @@ var _ = Describe("[plugin] root command", func() {
 		})
 
 		It("should not throw an error", func() {
-			cmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer})
+			cmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{})
 			cmd.SetArgs([]string{})
 			Expect(cmd.Execute()).NotTo(HaveOccurred())
 		})
diff --git a/kubectl-fdb/cmd/suite_test.go b/kubectl-fdb/cmd/suite_test.go
index ccc2cc5c..56ba5f29 100644
--- a/kubectl-fdb/cmd/suite_test.go
+++ b/kubectl-fdb/cmd/suite_test.go
@@ -19,6 +19,19 @@ var cluster *fdbv1beta2.FoundationDBCluster
 var clusterName = "test"
 var namespace = "test"
 
+// MockVersionChecker will help to mock the calls to GitHub API to get 'latest' version in tests
+type MockVersionChecker struct {
+	MockedVersion string
+}
+
+func (versionChecker *MockVersionChecker) getLatestPluginVersion() (string, error) {
+	if versionChecker.MockedVersion == "" {
+		return "latest", nil
+	}
+
+	return versionChecker.MockedVersion, nil
+}
+
 func TestCmd(t *testing.T) {
 	RegisterFailHandler(Fail)
 	RunSpecs(t, "FDB plugin")
diff --git a/kubectl-fdb/cmd/version_test.go b/kubectl-fdb/cmd/version_test.go
index 40a2b79a..97b960d7 100644
--- a/kubectl-fdb/cmd/version_test.go
+++ b/kubectl-fdb/cmd/version_test.go
@@ -46,7 +46,7 @@ var _ = Describe("[plugin] version command", func() {
 			errBuffer = bytes.Buffer{}
 			inBuffer = bytes.Buffer{}
 
-			rootCmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer})
+			rootCmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{})
 
 			args := []string{"version", "--client-only"}
 			rootCmd.SetArgs(args)
@@ -162,4 +162,33 @@ var _ = Describe("[plugin] version command", func() {
 				}),
 		)
 	})
+	When("running the version command with old version", func() {
+		var outBuffer bytes.Buffer
+		var errBuffer bytes.Buffer
+		var inBuffer bytes.Buffer
+
+		AfterEach(func() {
+			pluginVersion = "latest"
+		})
+		BeforeEach(func() {
+			pluginVersion = "1.0.0"
+			// We use these buffers to check the input/output
+			outBuffer = bytes.Buffer{}
+			errBuffer = bytes.Buffer{}
+			inBuffer = bytes.Buffer{}
+
+			rootCmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{MockedVersion: "2.0.0"})
+
+			args := []string{"version", "--client-only"}
+			rootCmd.SetArgs(args)
+
+			err := rootCmd.Execute()
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("should print out the client version", func() {
+			Expect(outBuffer.String()).To(ContainSubstring(
+				"kubectl-fdb plugin is not up-to-date, please install the latest version and try again!"))
+		})
+	})
 })
diff --git a/mock-kubernetes-client/client/client_test.go b/mock-kubernetes-client/client/client_test.go
index 7a0d274e..9c8b2add 100644
--- a/mock-kubernetes-client/client/client_test.go
+++ b/mock-kubernetes-client/client/client_test.go
@@ -63,11 +63,8 @@ var _ = Describe("[mock client]", func() {
 	var mockClient *MockClient
 
 	BeforeEach(func() {
-		err := scheme.AddToScheme(scheme.Scheme)
-		Expect(err).NotTo(HaveOccurred())
-		err = fdbv1beta2.AddToScheme(scheme.Scheme)
-		Expect(err).NotTo(HaveOccurred())
-
+		Expect(scheme.AddToScheme(scheme.Scheme)).NotTo(HaveOccurred())
+		Expect(fdbv1beta2.AddToScheme(scheme.Scheme)).NotTo(HaveOccurred())
 		mockClient = NewMockClient(scheme.Scheme)
 	})
 
@@ -88,6 +85,24 @@ var _ = Describe("[mock client]", func() {
 		})
 	})
 
+	When("creating and getting an object with a custom creationTimestamp", func() {
+		It("should create and get the object", func() {
+			pod := createDummyPod()
+			expectedCreationTime := time.Now().Add(-15 * time.Minute)
+			pod.CreationTimestamp.Time = expectedCreationTime
+			Expect(mockClient.Create(context.TODO(), pod)).NotTo(HaveOccurred())
+			Expect(pod.ObjectMeta.Generation).To(Equal(int64(1)))
+
+			podCopy := &corev1.Pod{}
+			Expect(mockClient.Get(context.TODO(), types.NamespacedName{Namespace: "default", Name: "pod1"}, podCopy)).NotTo(HaveOccurred())
+			Expect(podCopy.Name).To(Equal("pod1"))
+			Expect(len(podCopy.Spec.Containers)).To(Equal(1))
+			Expect(podCopy.Spec.Containers[0].Name).To(Equal("test-container"))
+			Expect(podCopy.ObjectMeta.Generation).To(Equal(int64(1)))
+			Expect(podCopy.ObjectMeta.CreationTimestamp.Time.Minute()).To(Equal(expectedCreationTime.Minute()))
+		})
+	})
+
 	When("creating and getting a Pod", func() {
 		var initialPod *corev1.Pod