Repository: FoundationDB/fdb-kubernetes-operator
Pull request #1273 (2022-07-03 12:34:07)

Add changes for unified image support
# Description

Those changes add the required RBAC settings to run the unified image.

## Type of change

*Please select one of the options below.*

- Bug fix (non-breaking change which fixes an issue)

## Discussion

-

## Testing

Locally.

## Documentation

-

## Follow-up

-


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

* Commit ID: eb80ae80c12a2cc82ddde78035a77b1e977ecbe7
* Duration 0:53:17
* Result: :x: **FAILED**
* Error: `Error while executing command: make -C tests -kj prOperator. Reason: exit status 2`
* [Build Logs](https://d1e0l78xbkh3xa.cloudfront.net/build-logs/fdb-kubernetes-operator/fdb-kubernetes-operator-pr/4a131883-93c7-42cc-a426-c2c656c1c942.gz) (available for 30 days)


Patch:
diff --git a/.github/workflows/pull_request.yml b/.github/workflows/pull_request.yml
index d6f6ac68..a1437741 100644
--- a/.github/workflows/pull_request.yml
+++ b/.github/workflows/pull_request.yml
@@ -8,7 +8,6 @@ on:
 env:
   FDB_VER: "6.2.29"
 
-
 jobs:
   lint-go:
     name: Lint go code
@@ -73,10 +72,15 @@ jobs:
         chmod +x kind
         sudo mv kind /usr/local/bin/kind
         ./scripts/setup_kind_local_registry.sh ${{ matrix.kubever }}
+    #  https://github.com/goreleaser/goreleaser/issues/1311
+    - name: Get current semver tag
+      run: echo "::set-output name=CURRENT_TAG::$(git describe --tags --match "v*" --abbrev=0)"
+      id: current-tag
     - name: Check for uncommitted changes
       env:
         # Don't run any tests we run them in the next step
         SKIP_TEST: "1"
+        GORELEASER_CURRENT_TAG: ${{ steps.current-tag.outputs.CURRENT_TAG }}
       run: |
         make clean all
         git diff --exit-code
diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml
index 81b56406..a846242b 100644
--- a/.github/workflows/release.yaml
+++ b/.github/workflows/release.yaml
@@ -44,10 +44,15 @@ jobs:
         uses: actions/setup-go@v1
         with:
           go-version: 1.17.11
+      #  https://github.com/goreleaser/goreleaser/issues/1311
+      - name: Get current semver tag
+        run: echo "::set-output name=CURRENT_TAG::$(git describe --tags --match "v*" --abbrev=0)"
+        id: current-tag
       - name: Release binaries
         run: make release
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GORELEASER_CURRENT_TAG: ${{ steps.current-tag.outputs.CURRENT_TAG }}
   push_images:
     name: Push Docker images
     needs: create-release
diff --git a/api/v1beta2/foundationdb_database_configuration.go b/api/v1beta2/foundationdb_database_configuration.go
index 16336278..a5609dc8 100644
--- a/api/v1beta2/foundationdb_database_configuration.go
+++ b/api/v1beta2/foundationdb_database_configuration.go
@@ -737,6 +737,8 @@ const (
 	StorageEngineRocksDbExperimental StorageEngine = "ssd-rocksdb-experimental"
 	// StorageEngineRocksDbV1 defines the storage engine ssd-rocksdb-v1.
 	StorageEngineRocksDbV1 StorageEngine = "ssd-rocksdb-v1"
+	// StorageEngineShardedRocksDB defines the storage engine ssd-sharded-rocksdb.
+	StorageEngineShardedRocksDB StorageEngine = "ssd-sharded-rocksdb"
 )
 
 // RoleCounts represents the roles whose counts can be customized.
diff --git a/api/v1beta2/foundationdb_status.go b/api/v1beta2/foundationdb_status.go
index 07d3d1c6..a71da56b 100644
--- a/api/v1beta2/foundationdb_status.go
+++ b/api/v1beta2/foundationdb_status.go
@@ -89,6 +89,10 @@ type FoundationDBStatusClusterInfo struct {
 	// FaultTolerance provides information about the fault tolerance status
 	// of the cluster.
 	FaultTolerance FaultTolerance `json:"fault_tolerance,omitempty"`
+
+	// IncompatibleConnections provides information about processes that try to connect to the cluster with an
+	// incompatible version.
+	IncompatibleConnections []string `json:"incompatible_connections,omitempty"`
 }
 
 // FaultTolerance provides information about the fault tolerance status
diff --git a/api/v1beta2/foundationdb_version.go b/api/v1beta2/foundationdb_version.go
index 1ae8cf9e..1e11d464 100644
--- a/api/v1beta2/foundationdb_version.go
+++ b/api/v1beta2/foundationdb_version.go
@@ -186,6 +186,8 @@ func (version Version) IsStorageEngineSupported(storageEngine StorageEngine) boo
 		return version.IsAtLeast(Versions.SupportsRocksDBV1)
 	} else if storageEngine == StorageEngineRocksDbExperimental {
 		return !version.IsAtLeast(Versions.SupportsRocksDBV1)
+	} else if storageEngine == StorageEngineShardedRocksDB {
+		return version.IsAtLeast(Versions.SupportsShardedRocksDB)
 	}
 	return true
 }
@@ -208,12 +210,14 @@ var Versions = struct {
 	MinimumVersion,
 	SupportsRocksDBV1,
 	SupportsIsPresent,
+	SupportsShardedRocksDB,
 	Default Version
 }{
-	Default:           Version{Major: 6, Minor: 2, Patch: 20},
-	NextPatchVersion:  Version{Major: 6, Minor: 2, Patch: 21},
-	NextMajorVersion:  Version{Major: 7, Minor: 0, Patch: 0},
-	MinimumVersion:    Version{Major: 6, Minor: 2, Patch: 20},
-	SupportsRocksDBV1: Version{Major: 7, Minor: 1, Patch: 0, ReleaseCandidate: 4},
-	SupportsIsPresent: Version{Major: 7, Minor: 1, Patch: 4},
+	Default:                Version{Major: 6, Minor: 2, Patch: 20},
+	NextPatchVersion:       Version{Major: 6, Minor: 2, Patch: 21},
+	NextMajorVersion:       Version{Major: 7, Minor: 0, Patch: 0},
+	MinimumVersion:         Version{Major: 6, Minor: 2, Patch: 20},
+	SupportsRocksDBV1:      Version{Major: 7, Minor: 1, Patch: 0, ReleaseCandidate: 4},
+	SupportsIsPresent:      Version{Major: 7, Minor: 1, Patch: 4},
+	SupportsShardedRocksDB: Version{Major: 7, Minor: 2, Patch: 0},
 }
diff --git a/api/v1beta2/foundationdbcluster_types.go b/api/v1beta2/foundationdbcluster_types.go
index 92ca5c84..1052c49d 100644
--- a/api/v1beta2/foundationdbcluster_types.go
+++ b/api/v1beta2/foundationdbcluster_types.go
@@ -866,6 +866,10 @@ type FoundationDBClusterAutomationOptions struct {
 	// +kubebuilder:validation:Enum=Replace;ReplaceTransactionSystem;Delete
 	// +kubebuilder:default:=ReplaceTransactionSystem
 	PodUpdateStrategy PodUpdateStrategy `json:"podUpdateStrategy,omitempty"`
+
+	// UseManagementAPI defines if the operator should make use of the management API instead of
+	// using fdbcli to interact with the FoundationDB cluster.
+	UseManagementAPI *bool `json:"useManagementAPI,omitempty"`
 }
 
 // AutomaticReplacementOptions controls options for automatically replacing
@@ -1801,6 +1805,11 @@ func (cluster *FoundationDBCluster) GetMaxConcurrentReplacements() int {
 	return pointer.IntDeref(cluster.Spec.AutomationOptions.MaxConcurrentReplacements, math.MaxInt64)
 }
 
+// UseManagementAPI returns the value of UseManagementAPI or false if unset.
+func (cluster *FoundationDBCluster) UseManagementAPI() bool {
+	return pointer.BoolDeref(cluster.Spec.AutomationOptions.UseManagementAPI, false)
+}
+
 // PodUpdateMode defines the deletion mode for the cluster
 type PodUpdateMode string
 
@@ -1980,6 +1989,76 @@ func (cluster *FoundationDBCluster) AddProcessGroupsToRemovalList(processGroupID
 	}
 }
 
+// AddProcessGroupsToNoScheduleList adds the provided process group IDs to the no-schedule list.
+// If a process group ID is already present on that list it won't be added a second time.
+func (cluster *FoundationDBCluster) AddProcessGroupsToNoScheduleList(processGroupIDs []string) {
+	noScheduleProcesses := map[string]None{}
+
+	for _, id := range cluster.Spec.Buggify.NoSchedule {
+		noScheduleProcesses[id] = None{}
+	}
+
+	for _, processGroupID := range processGroupIDs {
+		if _, ok := noScheduleProcesses[processGroupID]; ok {
+			continue
+		}
+
+		cluster.Spec.Buggify.NoSchedule = append(cluster.Spec.Buggify.NoSchedule, processGroupID)
+	}
+}
+
+// RemoveProcessGroupsFromNoScheduleList removes the provided process group IDs from the no-schedule list.
+func (cluster *FoundationDBCluster) RemoveProcessGroupsFromNoScheduleList(processGroupIDs []string) {
+	processGroupIDsToRemove := make(map[string]None)
+	for _, processGroupID := range processGroupIDs {
+		processGroupIDsToRemove[processGroupID] = None{}
+	}
+
+	idx := 0
+	for _, processGroupID := range cluster.Spec.Buggify.NoSchedule {
+		if _, ok := processGroupIDsToRemove[processGroupID]; ok {
+			continue
+		}
+		cluster.Spec.Buggify.NoSchedule[idx] = processGroupID
+		idx++
+	}
+
+	cluster.Spec.Buggify.NoSchedule = cluster.Spec.Buggify.NoSchedule[:idx]
+}
+
+// AddProcessGroupsToCrashLoopList adds the provided process group IDs to the crash-loop list.
+// If a process group ID is already present on that list or all the processes are set into crash-loop
+// it won't be added a second time.
+func (cluster *FoundationDBCluster) AddProcessGroupsToCrashLoopList(processGroupIDs []string) {
+	crashLoop, _ := cluster.GetCrashLoopProcessGroups()
+
+	for _, processGroupID := range processGroupIDs {
+		if _, ok := crashLoop[processGroupID]; ok {
+			continue
+		}
+
+		cluster.Spec.Buggify.CrashLoop = append(cluster.Spec.Buggify.CrashLoop, processGroupID)
+	}
+}
+
+// RemoveProcessGroupsFromCrashLoopList removes the provided process group IDs from the crash-loop list.
+func (cluster *FoundationDBCluster) RemoveProcessGroupsFromCrashLoopList(processGroupIDs []string) {
+	processGroupIDsToRemove := make(map[string]None)
+	for _, processGroupID := range processGroupIDs {
+		processGroupIDsToRemove[processGroupID] = None{}
+	}
+
+	idx := 0
+	for _, processGroupID := range cluster.Spec.Buggify.CrashLoop {
+		if _, ok := processGroupIDsToRemove[processGroupID]; ok {
+			continue
+		}
+		cluster.Spec.Buggify.CrashLoop[idx] = processGroupID
+		idx++
+	}
+	cluster.Spec.Buggify.CrashLoop = cluster.Spec.Buggify.CrashLoop[:idx]
+}
+
 // AddProcessGroupsToRemovalWithoutExclusionList adds the provided process group IDs to the remove without exclusion list.
 // If a process group ID is already present on that list it won't be added a second time.
 func (cluster *FoundationDBCluster) AddProcessGroupsToRemovalWithoutExclusionList(processGroupIDs []string) {
@@ -2015,14 +2094,15 @@ func (cluster *FoundationDBCluster) GetRunningVersion() string {
 // if all process group IDs in a cluster should be crash looping.
 func (cluster *FoundationDBCluster) GetCrashLoopProcessGroups() (map[string]None, bool) {
 	crashLoopPods := make(map[string]None, len(cluster.Spec.Buggify.CrashLoop))
+	crashLoopAll := false
 	for _, processGroupID := range cluster.Spec.Buggify.CrashLoop {
 		if processGroupID == "*" {
-			return nil, true
+			crashLoopAll = true
 		}
 		crashLoopPods[processGroupID] = None{}
 	}
 
-	return crashLoopPods, false
+	return crashLoopPods, crashLoopAll
 }
 
 // Validate checks if all settings in the cluster are valid, if not and error will be returned. If multiple issues are
diff --git a/api/v1beta2/zz_generated.deepcopy.go b/api/v1beta2/zz_generated.deepcopy.go
index 9e213845..ebccc41d 100644
--- a/api/v1beta2/zz_generated.deepcopy.go
+++ b/api/v1beta2/zz_generated.deepcopy.go
@@ -500,6 +500,11 @@ func (in *FoundationDBClusterAutomationOptions) DeepCopyInto(out *FoundationDBCl
 		*out = new(int)
 		**out = **in
 	}
+	if in.UseManagementAPI != nil {
+		in, out := &in.UseManagementAPI, &out.UseManagementAPI
+		*out = new(bool)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FoundationDBClusterAutomationOptions.
@@ -953,6 +958,11 @@ func (in *FoundationDBStatusClusterInfo) DeepCopyInto(out *FoundationDBStatusClu
 	in.Clients.DeepCopyInto(&out.Clients)
 	in.Layers.DeepCopyInto(&out.Layers)
 	out.FaultTolerance = in.FaultTolerance
+	if in.IncompatibleConnections != nil {
+		in, out := &in.IncompatibleConnections, &out.IncompatibleConnections
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FoundationDBStatusClusterInfo.
diff --git a/charts/fdb-operator/Chart.yaml b/charts/fdb-operator/Chart.yaml
index 15073fe0..ebcdccf9 100644
--- a/charts/fdb-operator/Chart.yaml
+++ b/charts/fdb-operator/Chart.yaml
@@ -21,7 +21,7 @@ version: 0.2.0
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application.
-appVersion: v1.7.0
+appVersion: v1.8.0
 
 maintainers:
   - name: "foundationdb-ci"
diff --git a/charts/fdb-operator/values.yaml b/charts/fdb-operator/values.yaml
index b9d9dac4..4bf3380e 100644
--- a/charts/fdb-operator/values.yaml
+++ b/charts/fdb-operator/values.yaml
@@ -1,7 +1,7 @@
 ---
 image:
   repository: foundationdb/fdb-kubernetes-operator
-  tag: v1.7.0
+  tag: v1.8.0
   pullPolicy: IfNotPresent
 
 initContainers:
@@ -18,7 +18,7 @@ initContainers:
   7.1:
     image:
       repository: foundationdb/foundationdb-kubernetes-sidecar
-      tag: 7.1.5-1
+      tag: 7.1.15-1
       pullPolicy: IfNotPresent
 
 globalMode:
diff --git a/config/crd/bases/apps.foundationdb.org_foundationdbclusters.yaml b/config/crd/bases/apps.foundationdb.org_foundationdbclusters.yaml
index 44b1ffca..d7347e6c 100644
--- a/config/crd/bases/apps.foundationdb.org_foundationdbclusters.yaml
+++ b/config/crd/bases/apps.foundationdb.org_foundationdbclusters.yaml
@@ -9355,6 +9355,8 @@ spec:
                     type: object
                   useLocalitiesForExclusion:
                     type: boolean
+                  useManagementAPI:
+                    type: boolean
                   useNonBlockingExcludes:
                     type: boolean
                   waitBetweenRemovalsSeconds:
diff --git a/config/deployment/manager.yaml b/config/deployment/manager.yaml
index b6eacbe7..4c108457 100644
--- a/config/deployment/manager.yaml
+++ b/config/deployment/manager.yaml
@@ -58,13 +58,13 @@ spec:
             - "--copy-binary"
             - "fdbrestore"
             - "--output-dir"
-            - "/var/output-files/6.3.10"
+            - "/var/output-files/6.3.24"
             - "--init-mode"
           volumeMounts:
             - name: fdb-binaries
               mountPath: /var/output-files
         - name: foundationdb-kubernetes-init-7-1
-          image: foundationdb/foundationdb-kubernetes-sidecar:7.1.11-1
+          image: foundationdb/foundationdb-kubernetes-sidecar:7.1.15-1
           args:
             - "--copy-library"
             - "7.1"
@@ -75,7 +75,7 @@ spec:
             - "--copy-binary"
             - "fdbrestore"
             - "--output-dir"
-            - "/var/output-files/7.1.5"
+            - "/var/output-files/7.1.15"
             - "--init-mode"
           volumeMounts:
             - name: fdb-binaries
@@ -83,7 +83,7 @@ spec:
       containers:
       - command:
         - /manager
-        image: foundationdb/fdb-kubernetes-operator:v1.7.0
+        image: foundationdb/fdb-kubernetes-operator:v1.8.0
         name: manager
         env:
           - name: WATCH_NAMESPACE
diff --git a/config/samples/deployment.yaml b/config/samples/deployment.yaml
index 5e8ce243..9d52dc69 100644
--- a/config/samples/deployment.yaml
+++ b/config/samples/deployment.yaml
@@ -150,7 +150,7 @@ spec:
           valueFrom:
             fieldRef:
               fieldPath: metadata.namespace
-        image: foundationdb/fdb-kubernetes-operator:v1.7.0
+        image: foundationdb/fdb-kubernetes-operator:v1.8.0
         name: manager
         ports:
         - containerPort: 8080
@@ -201,7 +201,7 @@ spec:
         - --copy-binary
         - fdbrestore
         - --output-dir
-        - /var/output-files/6.3.10
+        - /var/output-files/6.3.24
         - --init-mode
         image: foundationdb/foundationdb-kubernetes-sidecar:6.3.24-1
         name: foundationdb-kubernetes-init-6-3
@@ -218,9 +218,9 @@ spec:
         - --copy-binary
         - fdbrestore
         - --output-dir
-        - /var/output-files/7.1.5
+        - /var/output-files/7.1.15
         - --init-mode
-        image: foundationdb/foundationdb-kubernetes-sidecar:7.1.11-1
+        image: foundationdb/foundationdb-kubernetes-sidecar:7.1.15-1
         name: foundationdb-kubernetes-init-7-1
         volumeMounts:
         - mountPath: /var/output-files
diff --git a/controllers/admin_client_mock.go b/controllers/admin_client_mock.go
index ffa5d052..1803c8e9 100644
--- a/controllers/admin_client_mock.go
+++ b/controllers/admin_client_mock.go
@@ -463,6 +463,16 @@ func (client *mockAdminClient) KillProcesses(addresses []fdbv1beta2.ProcessAddre
 	}
 	adminClientMutex.Unlock()
 
+	if client.Cluster.Status.RunningVersion != client.Cluster.Spec.Version {
+		// We have to do this in the mock client, in the real world the tryConnectionOptions in update_status,
+		// will update the version.
+		client.Cluster.Status.RunningVersion = client.Cluster.Spec.Version
+		err := client.KubeClient.Status().Update(context.TODO(), client.Cluster)
+		if err != nil {
+			return err
+		}
+	}
+
 	client.UnfreezeStatus()
 	return nil
 }
diff --git a/controllers/bounce_processes.go b/controllers/bounce_processes.go
index a81976b0..62d62e35 100644
--- a/controllers/bounce_processes.go
+++ b/controllers/bounce_processes.go
@@ -172,7 +172,7 @@ func (bounceProcesses) reconcile(ctx context.Context, r *FoundationDBClusterReco
 
 		if useLocks && upgrading {
 			var req *requeue
-			addresses, req = getAddressesForUpgrade(r, adminClient, lockClient, cluster, version)
+			addresses, req = getAddressesForUpgrade(r, status, lockClient, cluster, version)
 			if req != nil {
 				return req
 			}
@@ -187,13 +187,13 @@ func (bounceProcesses) reconcile(ctx context.Context, r *FoundationDBClusterReco
 		if err != nil {
 			return &requeue{curError: err}
 		}
-	}
 
-	if upgrading {
-		cluster.Status.RunningVersion = cluster.Spec.Version
-		err = r.Status().Update(ctx, cluster)
-		if err != nil {
-			return &requeue{curError: err}
+		// If the cluster was upgraded we will requeue and let the update_status command set the correct version.
+		// Updating the version in this method has the drawback that we upgrade the version independent of the success
+		// of the kill command. The kill command is not reliable, which means that some kill request might not be
+		// delivered and the return value will still not contain any error.
+		if upgrading {
+			return &requeue{message: "fetch latest status after upgrade"}
 		}
 	}
 
@@ -202,20 +202,14 @@ func (bounceProcesses) reconcile(ctx context.Context, r *FoundationDBClusterReco
 
 // getAddressesForUpgrade checks that all processes in a cluster are ready to be
 // upgraded and returns the full list of addresses.
-func getAddressesForUpgrade(r *FoundationDBClusterReconciler, adminClient fdbadminclient.AdminClient, lockClient fdbadminclient.LockClient, cluster *fdbv1beta2.FoundationDBCluster, version fdbv1beta2.Version) ([]fdbv1beta2.ProcessAddress, *requeue) {
+func getAddressesForUpgrade(r *FoundationDBClusterReconciler, databaseStatus *fdbv1beta2.FoundationDBStatus, lockClient fdbadminclient.LockClient, cluster *fdbv1beta2.FoundationDBCluster, version fdbv1beta2.Version) ([]fdbv1beta2.ProcessAddress, *requeue) {
 	logger := log.WithValues("namespace", cluster.Namespace, "cluster", cluster.Name, "reconciler", "bounceProcesses")
 	pendingUpgrades, err := lockClient.GetPendingUpgrades(version)
 	if err != nil {
 		return nil, &requeue{curError: err}
 	}
 
-	databaseStatus, err := adminClient.GetStatus()
-	if err != nil {
-		return nil, &requeue{curError: err}
-	}
-
 	if !databaseStatus.Client.DatabaseStatus.Available {
-		logger.Info("Deferring upgrade until database is available")
 		r.Recorder.Event(cluster, corev1.EventTypeNormal, "UpgradeRequeued", "Database is unavailable")
 		return nil, &requeue{message: "Deferring upgrade until database is available"}
 	}
@@ -223,7 +217,7 @@ func getAddressesForUpgrade(r *FoundationDBClusterReconciler, adminClient fdbadm
 	notReadyProcesses := make([]string, 0)
 	addresses := make([]fdbv1beta2.ProcessAddress, 0, len(databaseStatus.Cluster.Processes))
 	for _, process := range databaseStatus.Cluster.Processes {
-		processID := process.Locality["instance_id"]
+		processID := process.Locality[fdbv1beta2.FDBLocalityInstanceIDKey]
 		if process.Version == version.String() {
 			continue
 		}
diff --git a/controllers/cluster_controller.go b/controllers/cluster_controller.go
index 6fe79f6f..a8bb5412 100644
--- a/controllers/cluster_controller.go
+++ b/controllers/cluster_controller.go
@@ -30,10 +30,9 @@ import (
 	"sort"
 	"time"
 
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
 	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/fdbadminclient"
 	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/podmanager"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 
@@ -53,15 +52,16 @@ import (
 // FoundationDBClusterReconciler reconciles a FoundationDBCluster object
 type FoundationDBClusterReconciler struct {
 	client.Client
-	Recorder               record.EventRecorder
-	Log                    logr.Logger
-	InSimulation           bool
-	PodLifecycleManager    podmanager.PodLifecycleManager
-	PodClientProvider      func(*fdbv1beta2.FoundationDBCluster, *corev1.Pod) (podclient.FdbPodClient, error)
-	DatabaseClientProvider DatabaseClientProvider
-	DeprecationOptions     internal.DeprecationOptions
-	GetTimeout             time.Duration
-	PostTimeout            time.Duration
+	Recorder                           record.EventRecorder
+	Log                                logr.Logger
+	InSimulation                       bool
+	EnableRestartIncompatibleProcesses bool
+	PodLifecycleManager                podmanager.PodLifecycleManager
+	PodClientProvider                  func(*fdbv1beta2.FoundationDBCluster, *corev1.Pod) (podclient.FdbPodClient, error)
+	DatabaseClientProvider             DatabaseClientProvider
+	DeprecationOptions                 internal.DeprecationOptions
+	GetTimeout                         time.Duration
+	PostTimeout                        time.Duration
 }
 
 // NewFoundationDBClusterReconciler creates a new FoundationDBClusterReconciler with defaults.
@@ -139,6 +139,7 @@ func (r *FoundationDBClusterReconciler) Reconcile(ctx context.Context, request c
 		addPVCs{},
 		addPods{},
 		generateInitialClusterFile{},
+		removeIncompatibleProcesses{},
 		updateSidecarVersions{},
 		updatePodConfig{},
 		updateLabels{},
diff --git a/controllers/remove_incompatible_processes.go b/controllers/remove_incompatible_processes.go
new file mode 100644
index 00000000..2a6f2042
--- /dev/null
+++ b/controllers/remove_incompatible_processes.go
@@ -0,0 +1,119 @@
+/*
+ * remove_incompatible_processes.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 controllers
+
+import (
+	"context"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/go-logr/logr"
+
+	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
+	"github.com/FoundationDB/fdb-kubernetes-operator/internal"
+)
+
+// removeIncompatibleProcesses is a reconciler that will restart incompatible fdbserver processes, this can happen
+// during an upgrade when the kill command doesn't reach all processes, see: https://github.com/FoundationDB/fdb-kubernetes-operator/issues/1281
+type removeIncompatibleProcesses struct{}
+
+// reconcile runs the reconciler's work.
+func (removeIncompatibleProcesses) reconcile(ctx context.Context, r *FoundationDBClusterReconciler, cluster *fdbv1beta2.FoundationDBCluster) *requeue {
+	logger := log.WithValues("namespace", cluster.Namespace, "cluster", cluster.Name, "reconciler", "removeIncompatibleProcesses")
+	err := processIncompatibleProcesses(ctx, r, logger, cluster)
+
+	if err != nil {
+		return &requeue{curError: err, delay: 15 * time.Second}
+	}
+
+	return nil
+}
+
+func processIncompatibleProcesses(ctx context.Context, r *FoundationDBClusterReconciler, logger logr.Logger, cluster *fdbv1beta2.FoundationDBCluster) error {
+	if !r.EnableRestartIncompatibleProcesses {
+		logger.Info("skipping disabled subreconciler")
+		return nil
+	}
+
+	pods, err := r.PodLifecycleManager.GetPods(ctx, r, cluster, internal.GetPodListOptions(cluster, "", "")...)
+	if err != nil {
+		return err
+	}
+
+	podMap := internal.CreatePodMap(cluster, pods)
+
+	adminClient, err := r.getDatabaseClientProvider().GetAdminClient(cluster, r.Client)
+	if err != nil {
+		return err
+	}
+	defer adminClient.Close()
+
+	status, err := adminClient.GetStatus()
+	if err != nil {
+		return err
+	}
+
+	if len(status.Cluster.IncompatibleConnections) == 0 {
+		return nil
+	}
+
+	logger.Info("incompatible connections", "incompatibleConnections", status.Cluster.IncompatibleConnections)
+	incompatibleConnections := map[string]fdbv1beta2.None{}
+	for _, incompatibleAddress := range status.Cluster.IncompatibleConnections {
+		address, err := fdbv1beta2.ParseProcessAddress(incompatibleAddress)
+		if err != nil {
+			logger.Error(err, "could not parse address in incompatible connections", "address", incompatibleAddress)
+			continue
+		}
+
+		incompatibleConnections[address.IPAddress.String()] = fdbv1beta2.None{}
+	}
+
+	incompatiblePods := make([]*corev1.Pod, 0, len(incompatibleConnections))
+	for _, processGroup := range cluster.Status.ProcessGroups {
+		pod, ok := podMap[processGroup.ProcessGroupID]
+		if !ok || pod == nil {
+			logger.V(1).Info("Could not find Pod for process group ID",
+				"processGroupID", processGroup.ProcessGroupID)
+			continue
+		}
+
+		if isIncompatible(incompatibleConnections, processGroup) {
+			logger.Info("recreate Pod for process group with incompatible version", "processGroupID", processGroup.ProcessGroupID)
+			incompatiblePods = append(incompatiblePods, pod)
+		}
+	}
+
+	// Do an unsafe update of the Pods since they are not reachable anyway
+	return r.PodLifecycleManager.UpdatePods(ctx, r, cluster, incompatiblePods, true)
+}
+
+// isIncompatible checks if the process group is in the list of incompatible connections.
+func isIncompatible(incompatibleConnections map[string]fdbv1beta2.None, processGroup *fdbv1beta2.ProcessGroupStatus) bool {
+	for _, address := range processGroup.Addresses {
+		if _, ok := incompatibleConnections[address]; ok {
+			return true
+		}
+	}
+
+	return false
+}
diff --git a/controllers/update_database_configuration.go b/controllers/update_database_configuration.go
index 4368e636..2bca030d 100644
--- a/controllers/update_database_configuration.go
+++ b/controllers/update_database_configuration.go
@@ -23,9 +23,10 @@ package controllers
 import (
 	"context"
 	"fmt"
-	"reflect"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
+	"k8s.io/utils/pointer"
 
 	fdbtypes "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
 )
@@ -46,7 +47,6 @@ func (u updateDatabaseConfiguration) reconcile(ctx context.Context, r *Foundatio
 
 	desiredConfiguration := cluster.DesiredDatabaseConfiguration()
 	desiredConfiguration.RoleCounts.Storage = 0
-	needsChange := false
 	var currentConfiguration fdbtypes.DatabaseConfiguration
 
 	status, err := adminClient.GetStatus()
@@ -55,21 +55,20 @@ func (u updateDatabaseConfiguration) reconcile(ctx context.Context, r *Foundatio
 	}
 
 	initialConfig := !cluster.Status.Configured
-
-	available := initialConfig || status.Client.DatabaseStatus.Available
 	dataState := status.Cluster.Data.State
-	dataHealthy := initialConfig || dataState.Healthy
 
-	if !available {
+	if !(initialConfig || status.Client.DatabaseStatus.Available) {
 		logger.Info("Skipping database configuration change because database is unavailable")
 		return nil
 	}
 
 	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
 	cluster.ClearMissingVersionFlags(&currentConfiguration)
-	needsChange = initialConfig || !reflect.DeepEqual(desiredConfiguration, currentConfiguration)
 
-	if needsChange {
+	if initialConfig || !equality.Semantic.DeepEqual(desiredConfiguration, currentConfiguration) {
 		var nextConfiguration fdbtypes.DatabaseConfiguration
 		if initialConfig {
 			nextConfiguration = desiredConfiguration
@@ -77,16 +76,15 @@ func (u updateDatabaseConfiguration) reconcile(ctx context.Context, r *Foundatio
 			nextConfiguration = currentConfiguration.GetNextConfigurationChange(desiredConfiguration)
 		}
 		configurationString, _ := nextConfiguration.GetConfigurationString(cluster.Spec.Version)
-		var enabled = cluster.Spec.AutomationOptions.ConfigureDatabase
 
-		if !dataHealthy {
+		if !(initialConfig || dataState.Healthy) {
 			logger.Info("Waiting for data distribution to be healthy", "stateName", dataState.Name, "stateDescription", dataState.Description)
 			r.Recorder.Event(cluster, corev1.EventTypeNormal, "NeedsConfigurationChange",
 				fmt.Sprintf("Spec require configuration change to `%s`, but data distribution is not fully healthy: %s (%s)", configurationString, dataState.Name, dataState.Description))
 			return nil
 		}
 
-		if enabled != nil && !*enabled {
+		if !pointer.BoolDeref(cluster.Spec.AutomationOptions.ConfigureDatabase, true) {
 			r.Recorder.Event(cluster, corev1.EventTypeNormal, "NeedsConfigurationChange",
 				fmt.Sprintf("Spec require configuration change to `%s`, but configuration changes are disabled", configurationString))
 			cluster.Status.Generations.NeedsConfigurationChange = cluster.ObjectMeta.Generation
@@ -105,7 +103,7 @@ func (u updateDatabaseConfiguration) reconcile(ctx context.Context, r *Foundatio
 			}
 		}
 
-		logger.Info("Configuring database")
+		logger.Info("Configuring database", "current configuration", currentConfiguration, "desired configuration", desiredConfiguration)
 		r.Recorder.Event(cluster, corev1.EventTypeNormal, "ConfiguringDatabase",
 			fmt.Sprintf("Setting database configuration to `%s`", configurationString),
 		)
@@ -123,8 +121,7 @@ func (u updateDatabaseConfiguration) reconcile(ctx context.Context, r *Foundatio
 		}
 		logger.Info("Configured database")
 
-		if !reflect.DeepEqual(nextConfiguration, desiredConfiguration) {
-			logger.Info("Requeuing for next stage of database configuration change")
+		if !equality.Semantic.DeepEqual(nextConfiguration, desiredConfiguration) {
 			return &requeue{message: "Requeuing for next stage of database configuration change"}
 		}
 	}
diff --git a/controllers/update_pods.go b/controllers/update_pods.go
index 0ccf95f2..cdbbaac1 100644
--- a/controllers/update_pods.go
+++ b/controllers/update_pods.go
@@ -73,6 +73,7 @@ func (updatePods) reconcile(ctx context.Context, r *FoundationDBClusterReconcile
 			logger.V(1).Info("Could not find Pod for process group ID",
 				"processGroupID", processGroup.ProcessGroupID)
 			continue
+			// TODO should not be continue but rather be a requeue?
 		}
 
 		if shouldRequeueDueToTerminatingPod(pod, cluster, processGroup.ProcessGroupID) {
@@ -145,16 +146,16 @@ func (updatePods) reconcile(ctx context.Context, r *FoundationDBClusterReconcile
 		}
 	}
 
+	if len(updates) == 0 {
+		return nil
+	}
+
 	adminClient, err := r.getDatabaseClientProvider().GetAdminClient(cluster, r.Client)
 	if err != nil {
 		return &requeue{curError: err}
 	}
 	defer adminClient.Close()
 
-	if len(updates) == 0 {
-		return nil
-	}
-
 	return deletePodsForUpdates(ctx, r, cluster, adminClient, updates, logger)
 }
 
@@ -218,7 +219,7 @@ func deletePodsForUpdates(ctx context.Context, r *FoundationDBClusterReconciler,
 	}
 
 	// Only lock the cluster if we are not running in the delete "All" mode.
-	// Otherwise we want to delete all Pods and don't require a lock to sync with other clusters.
+	// Otherwise, we want to delete all Pods and don't require a lock to sync with other clusters.
 	if deletionMode != fdbv1beta2.PodUpdateModeAll {
 		hasLock, err := r.takeLock(cluster, "updating pods")
 		if !hasLock {
@@ -226,7 +227,7 @@ func deletePodsForUpdates(ctx context.Context, r *FoundationDBClusterReconciler,
 		}
 	}
 
-	logger.Info("Deleting pods", "zone", zone, "count", len(deletions), "deletionMode", string(cluster.Spec.AutomationOptions.DeletionMode))
+	logger.Info("Deleting pods", "zone", zone, "count", len(deletions), "deletionMode", string(deletionMode))
 	r.Recorder.Event(cluster, corev1.EventTypeNormal, "UpdatingPods", fmt.Sprintf("Recreating pods in zone %s", zone))
 
 	err = r.PodLifecycleManager.UpdatePods(logr.NewContext(ctx, logger), r, cluster, deletions, false)
diff --git a/controllers/update_status.go b/controllers/update_status.go
index d69c36b3..c220671c 100644
--- a/controllers/update_status.go
+++ b/controllers/update_status.go
@@ -23,7 +23,6 @@ package controllers
 import (
 	"context"
 	"fmt"
-	"reflect"
 	"sort"
 	"time"
 
@@ -95,6 +94,7 @@ func (updateStatus) reconcile(ctx context.Context, r *FoundationDBClusterReconci
 		}
 	}
 
+	versionMap := map[string]int{}
 	for _, process := range databaseStatus.Cluster.Processes {
 		processID, ok := process.Locality["process_id"]
 		// if the processID is not set we fall back to the instanceID
@@ -102,8 +102,16 @@ func (updateStatus) reconcile(ctx context.Context, r *FoundationDBClusterReconci
 			processID = process.Locality["instance_id"]
 		}
 		processMap[processID] = append(processMap[processID], process)
+		versionMap[process.Version]++
 	}
 
+	// Update the running version based on the reported version of the FDB processes
+	version, err := getRunningVersion(versionMap, cluster.Status.RunningVersion)
+	if err != nil {
+		return &requeue{curError: err}
+	}
+	cluster.Status.RunningVersion = version
+
 	status.HasListenIPsForAllPods = cluster.NeedsExplicitListenAddress()
 	status.DatabaseConfiguration = databaseStatus.Cluster.DatabaseConfiguration.NormalizeConfigurationWithSeparatedProxies(cluster.Spec.Version, cluster.Spec.DatabaseConfiguration.AreSeparatedProxiesConfigured())
 	// Removing excluded servers as we don't want them during comparison.
@@ -192,7 +200,7 @@ func (updateStatus) reconcile(ctx context.Context, r *FoundationDBClusterReconci
 		status.ConnectionString = cluster.Spec.SeedConnectionString
 	}
 
-	status.HasIncorrectConfigMap = status.HasIncorrectConfigMap || !reflect.DeepEqual(existingConfigMap.Data, configMap.Data) || !metadataMatches(existingConfigMap.ObjectMeta, configMap.ObjectMeta)
+	status.HasIncorrectConfigMap = status.HasIncorrectConfigMap || !equality.Semantic.DeepEqual(existingConfigMap.Data, configMap.Data) || !metadataMatches(existingConfigMap.ObjectMeta, configMap.ObjectMeta)
 
 	service := internal.GetHeadlessService(cluster)
 	existingService := &corev1.Service{}
@@ -337,6 +345,7 @@ func tryConnectionOptions(cluster *fdbv1beta2.FoundationDBCluster, r *Foundation
 				"version", version, "connectionString", connectionString)
 		}
 	}
+
 	return originalVersion, originalConnectionString, nil
 }
 
@@ -686,3 +695,35 @@ func refreshProcessGroupStatus(ctx context.Context, r *FoundationDBClusterReconc
 
 	return nil
 }
+
+func getRunningVersion(versionMap map[string]int, fallback string) (string, error) {
+	if len(versionMap) == 0 {
+		return fallback, nil
+	}
+
+	var currentCandidate fdbv1beta2.Version
+	var currentMaxCount int
+
+	for version, count := range versionMap {
+		if count < currentMaxCount {
+			continue
+		}
+
+		parsedVersion, err := fdbv1beta2.ParseFdbVersion(version)
+		if err != nil {
+			return fallback, err
+		}
+		// In this case we want to ensure we always pick the newer version to have a stable return value. Otherwise,
+		// it could happen that the version will be flapping between two versions.
+		if count == currentMaxCount {
+			if currentCandidate.IsAtLeast(parsedVersion) {
+				continue
+			}
+		}
+
+		currentCandidate = parsedVersion
+		currentMaxCount = count
+	}
+
+	return currentCandidate.String(), nil
+}
diff --git a/docs/changelog/v1.8.0.md b/docs/changelog/v1.8.0.md
new file mode 100644
index 00000000..6f363963
--- /dev/null
+++ b/docs/changelog/v1.8.0.md
@@ -0,0 +1,15 @@
+# v1.8.0
+
+## Changes
+
+### Operator
+
+- Fix bug where config update is triggered to often [#1280](https://github.com/FoundationDB/fdb-kubernetes-operator/pull/1280)
+- Use DB key for connection string [#1262](https://github.com/FoundationDB/fdb-kubernetes-operator/pull/1262)
+- Retry flaky test [#1267](https://github.com/FoundationDB/fdb-kubernetes-operator/pull/1267)
+
+### Tooling
+
+- Update customization.md [#1290](https://github.com/FoundationDB/fdb-kubernetes-operator/pull/1290)
+- Fix setup of deployment [#1282](https://github.com/FoundationDB/fdb-kubernetes-operator/pull/1282)
+- Fix how goreleaser fetches the current tag [#1272](https://github.com/FoundationDB/fdb-kubernetes-operator/pull/1272/)
diff --git a/docs/cluster_spec.md b/docs/cluster_spec.md
index 1d2affa1..eb6d2112 100644
--- a/docs/cluster_spec.md
+++ b/docs/cluster_spec.md
@@ -163,6 +163,7 @@ FoundationDBClusterAutomationOptions provides flags for enabling or disabling op
 | removalMode | RemovalMode defines the removal mode for this cluster. This can be PodUpdateModeNone, PodUpdateModeAll, PodUpdateModeZone or PodUpdateModeProcessGroup. The RemovalMode defines how process groups are deleted in order when they are marked for removal. | [PodUpdateMode](#podupdatemode) | false |
 | waitBetweenRemovalsSeconds | WaitBetweenRemovalsSeconds defines how long to wait between the last removal and the next removal. This is only an upper limit if the process group and the according resources are deleted faster than the provided duration the operator will move on with the next removal. The idea is to prevent a race condition were the operator deletes a resource but the Kubernetes API is slower to trigger the actual deletion, and we are running into a situation where the fault tolerance check still includes the already deleted processes. Defaults to 60. | *int | false |
 | podUpdateStrategy | PodUpdateStrategy defines how Pod spec changes are rolled out either by replacing Pods or by deleting Pods. The default for this is ReplaceTransactionSystem. | [PodUpdateStrategy](#podupdatestrategy) | false |
+| useManagementAPI | UseManagementAPI defines if the operator should make use of the management API instead of using fdbcli to interact with the FoundationDB cluster. | *bool | false |
 
 [Back to TOC](#table-of-contents)
 
diff --git a/docs/manual/customization.md b/docs/manual/customization.md
index bfc49baa..d9a540be 100644
--- a/docs/manual/customization.md
+++ b/docs/manual/customization.md
@@ -19,7 +19,7 @@ spec:
     storageServersPerPod: 2
 ```
 
-A change to the `storageServersPerPod` will replace all of the storage pods. For more information about this feature read the [multiple storage servers per pod](/docs/design/multiple_storage_per_disk.md) design doc.
+A change to the `storageServersPerPod` will replace all of the storage pods. For more information about this feature read the [multiple storage servers per pod](/docs/design/implemented/multiple_storage_per_disk.md) design doc.
 
 ## Customizing the Volumes
 
diff --git a/docs/manual/replacements_and_deletions.md b/docs/manual/replacements_and_deletions.md
index 766695e8..3576ac36 100644
--- a/docs/manual/replacements_and_deletions.md
+++ b/docs/manual/replacements_and_deletions.md
@@ -58,7 +58,7 @@ The default deletion mode is `Zone`.
 
 * `All` will delete all pods at once.
 * `Zone` deletes all Pods in fault domain at once.
-* `ProcessGroup` delete one Pod ar a time.
+* `ProcessGroup` delete one Pod at a time.
 
 Depending on your requirements and the underlying Kubernetes cluster you might choose a different deletion mode than the default.
 
diff --git a/docs/manual/technical_design.md b/docs/manual/technical_design.md
index 5830b352..ea46cc94 100644
--- a/docs/manual/technical_design.md
+++ b/docs/manual/technical_design.md
@@ -88,30 +88,31 @@ See the [LockOptions](../cluster_spec.md#LockOptions) documentation for more opt
 
 The cluster reconciler runs the following subreconcilers:
 
-1. UpdateStatus
-1. UpdateLockConfiguration
-1. UpdateConfigMap
-1. CheckClientCompatibility
-1. ReplaceMisconfiguredProcessGroups
-1. ReplaceFailedProcessGroups
-1. DeletePodsForBuggification
-1. AddProcessGroups
-1. AddServices
-1. AddPVCs
-1. AddPods
-1. GenerateInitialClusterFile
-1. UpdateSidecarVersions
-1. UpdatePodConfig
-1. UpdateLabels
-1. UpdateDatabaseConfiguration
-1. ChooseRemovals
-1. ExcludeProcesses
-1. ChangeCoordinators
-1. BounceProcesses
-1. UpdatePods
-1. RemoveServices
-1. RemoveProcessGroups
-1. UpdateStatus (again)
+1. [UpdateStatus](#updatestatus)
+1. [UpdateLockConfiguration](#updatelockconfiguration)
+1. [UpdateConfigMap](#updateconfigmap)
+1. [CheckClientCompatibility](#checkclientcompatibility)
+1. [DeletePodsForBuggification](#deletepodsforbuggification)
+1. [ReplaceMisconfiguredProcessGroups](#replacemisconfiguredprocessgroups)
+1. [ReplaceFailedProcessGroups](#replacefailedprocessGroups)
+1. [AddProcessGroups](#addprocessgroups)
+1. [AddServices](#addservices)
+1. [AddPVCs](#addpvcs)
+1. [AddPods](#addpods)
+1. [GenerateInitialClusterFile](#generateinitialclusterFile)
+1. [RemoveIncompatibleProcesses](#removeincompatibleprocesses)
+1. [UpdateSidecarVersions](#updatesidecarversions)
+1. [UpdatePodConfig](#updatepodconfig)
+1. [UpdateLabels](#updatelabels)
+1. [UpdateDatabaseConfiguration](#updatedatabaseconfiguration)
+1. [ChooseRemovals](#chooseremovals)
+1. [ExcludeProcesses](#excludeprocesses)
+1. [ChangeCoordinators](#changecoordinators)
+1. [BounceProcesses](#bounceprocesses)
+1. [UpdatePods](#updatepods)
+1. [RemoveProcessGroups](#removeprocessgroups)
+1. [RemoveServices](#removeservices)
+1. [UpdateStatus (again)](#updatestatus)
 
 ### Tracking Reconciliation Stages
 
@@ -139,6 +140,12 @@ The `CheckClientCompatibility` subreconciler is used during upgrades to ensure t
 
 You can skip this check by setting the `ignoreUpgradabilityChecks` flag in the cluster spec.
 
+### DeletePodsForBuggification
+
+The `DeletePodsForBuggification` subreconciler deletes pods that need to be recreated in order to set buggification options. These options are set through the `buggify` section in the cluster spec.
+
+When pods are deleted for buggification, we apply fewer safety checks, and buggification will often put the cluster in an unhealthy state.
+
 ### ReplaceMisconfiguredProcessGroups
 
 The `ReplaceMisconfiguredProcessGroups` subreconciler checks for process groups that need to be replaced in order to safely bring them up on a new configuration. The core action this subreconciler takes is setting the `removalTimestamp` field on the `ProcessGroup` in the cluster status. Later subreconcilers will do the work for handling the replacement, whether processes are marked for replacement through this subreconciler or another mechanism.
@@ -151,12 +158,6 @@ The `ReplaceFailedProcessGroups` subreconciler checks for process groups that ne
 
 See the [Replacements and Deletions](replacements_and_deletions.md) document for more details on when we do these replacements.
 
-### DeletePodsForBuggification
-
-The `DeletePodsForBuggification` subreconciler deletes pods that need to be recreated in order to set buggification options. These options are set through the `buggify` section in the cluster spec.
-
-When pods are deleted for buggification, we apply fewer safety checks, and buggification will often put the cluster in an unhealthy state.
-
 ### AddProcessGroups
 
 The `AddProcessGroups` subreconciler compares the desired process counts, calculated from the cluster spec, with the number of process groups in the cluster status. If the spec requires any additional process groups, this step will add them to the status. It will not create resources, and will mark the new process groups with conditions that indicate they are missing resources.
@@ -177,6 +178,12 @@ The `AddPods` subreconciler creates any pods that are required for the cluster.
 
 The `GenerateInitialClusterFile` creates the cluster file for the cluster. If the cluster already has a cluster file, this will take no action. The cluster file is the service discovery mechanism for the cluster. It includes addresses for coordinator processes, which are chosen statically. The coordinators are used to elect the cluster controller and inform servers and clients about which process is serving as cluster controller. The cluster file is stored in the `connectionString` field in the cluster status. You can manually specify the cluster file in the `seedConnectionString` field in the cluster spec. If both of these are blank, the operator will choose coordinators that satisfy the cluster's fault tolerance requirements. Coordinators cannot be chosen until the pods have been created and the processes have been assigned IP addresses, which by default comes from the pod's IP. Once the initial cluster file has been generated, we store it in the cluster status and requeue reconciliation so we can update the config map with the new cluster file.
 
+### RemoveIncompatibleProcesses
+
+The `RemoveIncompatibleProcesses` subreconciler will check the FoundationDB cluster status for incompatible connections.
+If the cluster has some incompatible connections the subreconciler will match those IP addresses with the process groups.
+For matching process groups the subrecociler will delete the associated Pod and let it recreate with the new image.
+
 ### UpdateSidecarVersions
 
 The `UpdateSidecarVersions` subreconciler updates the image for the `foundationdb-kubernetes-sidecar` container in each pod to match the `version` in the cluster spec.
diff --git a/fdbclient/admin_client.go b/fdbclient/admin_client.go
index fa52864c..b58b7191 100644
--- a/fdbclient/admin_client.go
+++ b/fdbclient/admin_client.go
@@ -44,6 +44,14 @@ const (
 	fdbcliStr = "fdbcli"
 )
 
+type timeoutError struct {
+	err error
+}
+
+func (timeoutErr timeoutError) Error() string {
+	return fmt.Sprintf("timeout: %s", timeoutErr.err.Error())
+}
+
 var adminClientMutex sync.Mutex
 
 var maxCommandOutput = parseMaxCommandOutput()
@@ -117,6 +125,9 @@ type cliCommand struct {
 
 	// args provides alternative arguments in place of the exec command.
 	args []string
+
+	// timeout provides a way to overwrite the default cli timeout.
+	timeout time.Duration
 }
 
 // hasTimeoutArg determines whether a command accepts a timeout argument.
@@ -140,6 +151,15 @@ func (command cliCommand) getClusterFileFlag() string {
 	return "-C"
 }
 
+// getTimeout returns the timeout for the command
+func (command cliCommand) getTimeout() int {
+	if command.timeout != 0 {
+		return int(command.timeout.Seconds())
+	}
+
+	return DefaultCLITimeout
+}
+
 // getBinaryPath generates the path to an FDB binary.
 func getBinaryPath(binaryName string, version string) string {
 	parsed, _ := fdbv1beta2.ParseFdbVersion(version)
@@ -162,7 +182,7 @@ func (client *cliAdminClient) runCommand(command cliCommand) (string, error) {
 	}
 
 	binary := getBinaryPath(binaryName, version)
-	hardTimeout := DefaultCLITimeout
+	hardTimeout := command.getTimeout()
 	args := make([]string, 0, 9)
 	args = append(args, command.args...)
 	if len(args) == 0 {
@@ -195,8 +215,8 @@ func (client *cliAdminClient) runCommand(command cliCommand) (string, error) {
 	}
 
 	if command.hasTimeoutArg() {
-		args = append(args, "--timeout", strconv.Itoa(DefaultCLITimeout))
-		hardTimeout += DefaultCLITimeout
+		args = append(args, "--timeout", strconv.Itoa(command.getTimeout()))
+		hardTimeout += command.getTimeout()
 	}
 	timeoutContext, cancelFunction := context.WithTimeout(context.Background(), time.Second*time.Duration(hardTimeout))
 	defer cancelFunction()
@@ -210,6 +230,12 @@ func (client *cliAdminClient) runCommand(command cliCommand) (string, error) {
 		if canCast {
 			client.log.Error(exitError, "Error from FDB command", "namespace", client.Cluster.Namespace, "cluster", client.Cluster.Name, "code", exitError.ProcessState.ExitCode(), "stdout", string(output), "stderr", string(exitError.Stderr))
 		}
+
+		// If we hit a timeout report it as a timeout error
+		if strings.Contains(string(output), "Specified timeout reached") {
+			return "", timeoutError{err: err}
+		}
+
 		return "", err
 	}
 
@@ -226,30 +252,88 @@ func (client *cliAdminClient) runCommand(command cliCommand) (string, error) {
 	return outputString, nil
 }
 
-// GetStatus gets the database's status
-func (client *cliAdminClient) GetStatus() (*fdbv1beta2.FoundationDBStatus, error) {
-	adminClientMutex.Lock()
-	defer adminClientMutex.Unlock()
+// runCommandWithBackoff is a wrapper around runCommand which allows retrying commands if they hit a timeout.
+func (client *cliAdminClient) runCommandWithBackoff(command string) (string, error) {
+	maxTimeoutInSeconds := 40
+	currentTimeoutInSeconds := DefaultCLITimeout
+
+	var rawResult string
+	var err error
 
-	if client.useClientLibrary {
+	// This method will be retrying to get the status if a timeout is seen. The timeout will be doubled everytime we try
+	// it with the default timeout of 10 we will try it 3 times with the following timeouts: 10s - 20s - 40s. We have
+	// seen that during upgrades of version incompatible version, when not all coordinators are properly restarted that
+	// the response time will be increased.
+	for currentTimeoutInSeconds <= maxTimeoutInSeconds {
+		rawResult, err = client.runCommand(cliCommand{command: command, timeout: time.Duration(currentTimeoutInSeconds) * time.Second})
+		if err == nil {
+			break
+		}
+
+		if _, ok := err.(timeoutError); ok {
+			client.log.Info("timeout issue will retry with higher timeout")
+			currentTimeoutInSeconds *= 2
+			continue
+		}
+
+		// If any error other than a timeout happens return this error and don't retry.
+		return "", err
+	}
+
+	return rawResult, err
+}
+
+func (client *cliAdminClient) getStatus(useClientLibrary bool) (*fdbv1beta2.FoundationDBStatus, error) {
+	var contents []byte
+	var err error
+
+	if useClientLibrary {
 		// This will call directly the database and fetch the status information
 		// from the system key space.
-		return getStatusFromDB(client.Cluster, client.log)
+		contents, err = getStatusFromDB(client.Cluster, client.log)
+	} else {
+		var rawResult string
+		rawResult, err = client.runCommandWithBackoff("status json")
+		if err != nil {
+			return nil, err
+		}
+
+		contents, err = internal.RemoveWarningsInJSON(rawResult)
 	}
-	contents, err := client.runCommand(cliCommand{command: "status json"})
+
 	if err != nil {
 		return nil, err
 	}
-	client.log.V(1).Info("Fetched status JSON", "contents", contents)
-	contents, err = internal.RemoveWarningsInJSON(contents)
+
+	status := &fdbv1beta2.FoundationDBStatus{}
+	err = json.Unmarshal(contents, status)
 	if err != nil {
 		return nil, err
 	}
-	status := &fdbv1beta2.FoundationDBStatus{}
-	err = json.Unmarshal([]byte(contents), status)
+	client.log.V(1).Info("Fetched status JSON", "status", status)
+
+	return status, nil
+}
+
+// GetStatus gets the database's status
+func (client *cliAdminClient) GetStatus() (*fdbv1beta2.FoundationDBStatus, error) {
+	adminClientMutex.Lock()
+	defer adminClientMutex.Unlock()
+
+	status, err := client.getStatus(client.useClientLibrary)
 	if err != nil {
 		return nil, err
 	}
+
+	// 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.
+	// In this case we want to fall back to use fdbcli which is version specific and will work.
+	if len(status.Cluster.Processes) == 0 && client.useClientLibrary && client.Cluster.Status.Configured {
+		client.log.Info("retry fetching status with fdbcli instead of using the client library")
+		return client.getStatus(false)
+	}
+
 	return status, nil
 }
 
@@ -498,8 +582,9 @@ func (client *cliAdminClient) KillProcesses(addresses []fdbv1beta2.ProcessAddres
 	if len(addresses) == 0 {
 		return nil
 	}
+
 	_, err := client.runCommand(cliCommand{command: fmt.Sprintf(
-		"kill; kill %s; status",
+		"kill; kill %[1]s; sleep 1; kill %[1]s; sleep 1; kill %[1]s",
 		fdbv1beta2.ProcessAddressesStringWithoutFlags(addresses, " "),
 	)})
 	return err
@@ -529,24 +614,35 @@ func (client *cliAdminClient) ChangeCoordinators(addresses []fdbv1beta2.ProcessA
 
 // GetConnectionString fetches the latest connection string.
 func (client *cliAdminClient) GetConnectionString() (string, error) {
-	output, err := client.runCommand(cliCommand{command: "status minimal"})
-	if err != nil {
-		return "", err
-	}
+	var connectionStringBytes []byte
+	var err error
 
-	if !strings.Contains(output, "The database is available") {
-		return "", fmt.Errorf("unable to fetch connection string: %s", output)
-	}
+	if client.Cluster.UseManagementAPI() {
+		// This will call directly the database and fetch the connection string
+		// from the system key space.
+		connectionStringBytes, err = getConnectionStringFromDB(client.Cluster)
+	} else {
+		var output string
+		output, err = client.runCommandWithBackoff("status minimal")
+		if err != nil {
+			return "", err
+		}
 
-	connectionStringBytes, err := os.ReadFile(client.clusterFilePath)
+		if !strings.Contains(output, "The database is available") {
+			return "", fmt.Errorf("unable to fetch connection string: %s", output)
+		}
+
+		connectionStringBytes, err = os.ReadFile(client.clusterFilePath)
+	}
 	if err != nil {
 		return "", err
 	}
-
-	connectionString, err := fdbv1beta2.ParseConnectionString(string(connectionStringBytes))
+	var connectionString fdbv1beta2.ConnectionString
+	connectionString, err = fdbv1beta2.ParseConnectionString(string(connectionStringBytes))
 	if err != nil {
 		return "", err
 	}
+
 	return connectionString.String(), nil
 }
 
@@ -667,12 +763,12 @@ func (client *cliAdminClient) GetBackupStatus() (*fdbv1beta2.FoundationDBLiveBac
 	}
 
 	status := &fdbv1beta2.FoundationDBLiveBackupStatus{}
-	statusString, err = internal.RemoveWarningsInJSON(statusString)
+	statusBytes, err := internal.RemoveWarningsInJSON(statusString)
 	if err != nil {
 		return nil, err
 	}
 
-	err = json.Unmarshal([]byte(statusString), &status)
+	err = json.Unmarshal(statusBytes, &status)
 	if err != nil {
 		return nil, err
 	}
diff --git a/fdbclient/common.go b/fdbclient/common.go
index 630c6915..5e147c8e 100644
--- a/fdbclient/common.go
+++ b/fdbclient/common.go
@@ -21,7 +21,6 @@
 package fdbclient
 
 import (
-	"encoding/json"
 	"fmt"
 	"os"
 
@@ -73,11 +72,7 @@ func getFDBDatabase(cluster *fdbv1beta2.FoundationDBCluster) (fdb.Database, erro
 	return database, nil
 }
 
-// getStatusFromDB gets the database's status directly from the system key
-func getStatusFromDB(cluster *fdbv1beta2.FoundationDBCluster, log logr.Logger) (*fdbv1beta2.FoundationDBStatus, error) {
-	log.Info("Fetch status from FDB", "namespace", cluster.Namespace, "cluster", cluster.Name)
-	statusKey := "\xff\xff/status/json"
-
+func getValueFromDBUsingKey(cluster *fdbv1beta2.FoundationDBCluster, fdbKey string, extraTimeout int64) ([]byte, error) {
 	database, err := getFDBDatabase(cluster)
 	if err != nil {
 		return nil, err
@@ -88,37 +83,39 @@ func getStatusFromDB(cluster *fdbv1beta2.FoundationDBCluster, log logr.Logger) (
 		if err != nil {
 			return nil, err
 		}
-		// Wait default timeout seconds to receive status for larger clusters.
-		err = transaction.Options().SetTimeout(int64(DefaultCLITimeout * 1000))
+		err = transaction.Options().SetTimeout(int64(DefaultCLITimeout) * extraTimeout)
 		if err != nil {
 			return nil, err
 		}
 
-		statusBytes := transaction.Get(fdb.Key(statusKey)).MustGet()
-		if len(statusBytes) == 0 {
+		rawResult := transaction.Get(fdb.Key(fdbKey)).MustGet()
+		if len(rawResult) == 0 {
 			return nil, err
 		}
 
-		return statusBytes, err
+		return rawResult, err
 	})
 
 	if err != nil {
 		return nil, err
 	}
 
-	statusBytes, ok := result.([]byte)
+	byteResult, ok := result.([]byte)
 	if !ok {
 		return nil, fmt.Errorf("could not cast result into byte slice")
 	}
+	return byteResult, nil
+}
 
-	status := &fdbv1beta2.FoundationDBStatus{}
-	err = json.Unmarshal(statusBytes, &status)
-	if err == nil {
-		log.V(1).Info("Retrieved JSON status", "raw", string(statusBytes), "parsed", status)
-		log.Info("Successfully fetched status from FDB", "namespace", cluster.Namespace, "cluster", cluster.Name)
-	}
+// getConnectionStringFromDB gets the database's connection string directly from the system key
+func getConnectionStringFromDB(cluster *fdbv1beta2.FoundationDBCluster) ([]byte, error) {
+	return getValueFromDBUsingKey(cluster, "\xff\xff/connection_string", 1)
+}
 
-	return status, err
+// getStatusFromDB gets the database's status directly from the system key
+func getStatusFromDB(cluster *fdbv1beta2.FoundationDBCluster, log logr.Logger) ([]byte, error) {
+	log.Info("Fetch status from FDB", "namespace", cluster.Namespace, "cluster", cluster.Name)
+	return getValueFromDBUsingKey(cluster, "\xff\xff/status/json", 1000)
 }
 
 type realDatabaseClientProvider struct {
diff --git a/internal/fdb_status_helper.go b/internal/fdb_status_helper.go
index 9cacc8c6..02f72231 100644
--- a/internal/fdb_status_helper.go
+++ b/internal/fdb_status_helper.go
@@ -27,11 +27,11 @@ import (
 
 // RemoveWarningsInJSON removes any warning messages that might appear in the status output from the fdbcli and returns
 // the JSON output without the warning message.
-func RemoveWarningsInJSON(jsonString string) (string, error) {
+func RemoveWarningsInJSON(jsonString string) ([]byte, error) {
 	idx := strings.Index(jsonString, "{")
 	if idx == -1 {
-		return "", fmt.Errorf("the JSON string doesn't contain a starting '{'")
+		return nil, fmt.Errorf("the JSON string doesn't contain a starting '{'")
 	}
 
-	return strings.TrimSpace(jsonString[idx:]), nil
+	return []byte(strings.TrimSpace(jsonString[idx:])), nil
 }
diff --git a/internal/pod_models.go b/internal/pod_models.go
index 6e97d65c..185076af 100644
--- a/internal/pod_models.go
+++ b/internal/pod_models.go
@@ -37,6 +37,17 @@ import (
 
 var processClassSanitizationPattern = regexp.MustCompile("[^a-z0-9-]")
 
+// GetProcessGroupIDFromPodName returns the process group ID for a given Pod name.
+func GetProcessGroupIDFromPodName(cluster *fdbv1beta2.FoundationDBCluster, podName string) string {
+	tmpName := strings.ReplaceAll(podName, cluster.Name, "")[1:]
+
+	if cluster.Spec.ProcessGroupIDPrefix != "" {
+		return fmt.Sprintf("%s-%s", cluster.Spec.ProcessGroupIDPrefix, tmpName)
+	}
+
+	return tmpName
+}
+
 // GetProcessGroupID generates an ID for a process group.
 //
 // This will return the pod name and the processGroupID ID.
diff --git a/kubectl-fdb/cmd/analyze.go b/kubectl-fdb/cmd/analyze.go
index 3848ec19..bd5fd4ac 100644
--- a/kubectl-fdb/cmd/analyze.go
+++ b/kubectl-fdb/cmd/analyze.go
@@ -143,11 +143,11 @@ kubectl fdb -n test-namespace analyze sample-cluster-1
 kubectl fdb analyze --auto-fix sample-cluster-1
 
 # Analyze the cluster "sample-cluster-1" in the current namespace and ignore the IncorrectCommandLine and IncorrectPodSpec condition
-kubectl fdb analyze cluster --ignore-condition=IncorrectCommandLine --ignore-condition=IncorrectPodSpec sample-cluster-1
+kubectl fdb analyze --ignore-condition=IncorrectCommandLine --ignore-condition=IncorrectPodSpec sample-cluster-1
 
 # Per default the plugin will print out how many process groups are marked for removal instead of printing out each process group.
 # This can be disabled by using the ignore-removals flag to print out the details about process groups that are marked for removal.
-kubectl fdb analyze cluster --ignore-removals=false sample-cluster-1
+kubectl fdb analyze --ignore-removals=false sample-cluster-1
 `,
 	}
 	cmd.SetOut(o.Out)
@@ -399,7 +399,7 @@ func analyzeCluster(cmd *cobra.Command, kubeClient client.Client, clusterName st
 		confirmed := false
 
 		if len(failedProcessGroups) > 0 {
-			err := replaceProcessGroups(kubeClient, cluster.Name, failedProcessGroups, namespace, true, false, wait, false)
+			err := replaceProcessGroups(kubeClient, cluster.Name, failedProcessGroups, namespace, true, wait, false, true)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/buggify.go b/kubectl-fdb/cmd/buggify.go
new file mode 100644
index 00000000..ccebc4f9
--- /dev/null
+++ b/kubectl-fdb/cmd/buggify.go
@@ -0,0 +1,80 @@
+/*
+ * buggify.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 cmd
+
+import (
+	"k8s.io/cli-runtime/pkg/genericclioptions"
+
+	"github.com/spf13/cobra"
+)
+
+func newBuggifyCmd(streams genericclioptions.IOStreams) *cobra.Command {
+	o := newFDBOptions(streams)
+
+	cmd := &cobra.Command{
+		Use:   "buggify",
+		Short: "Subcommand to add process groups to buggify list for a given cluster",
+		Long: "Subcommand to add process groups to buggify list for a given cluster. " +
+			"Supported options: crash-loop, no-schedule, empty-monitor-conf.",
+		RunE: func(c *cobra.Command, args []string) error {
+			return c.Help()
+		},
+		Example: `
+kubectl fdb -n <namespace> buggify <option> -c <cluster> pod-1 pod-2
+
+# Add process groups into crash loop state for a cluster in the current namespace
+kubectl fdb buggify crash-loop -c cluster pod-1 pod-2
+
+# Remove process groups from crash loop state from a cluster in the current namespace
+kubectl fdb buggify crash-loop --clear -c cluster pod-1 pod-2
+
+# Clean crash loop list of a cluster in the current namespace
+kubectl fdb buggify crash-loop --clean -c cluster
+
+# Add process groups into no-schedule state for a cluster in the current namespace
+kubectl fdb buggify no-schedule -c cluster pod-1 pod-2
+
+# Remove process groups from no-schedule state from a cluster in the current namespace
+kubectl fdb buggify no-schedule --clear -c cluster pod-1 pod-2
+
+# Clean no-schedule list of a cluster in the current namespace
+kubectl fdb buggify no-schedule  --clean -c cluster
+
+# Setting empty-monitor-conf to true
+kubectl fdb buggify empty-monitor-conf -c cluster
+
+# Setting empty-monitor-conf to false
+kubectl fdb buggify empty-monitor-conf --unset -c cluster
+`,
+	}
+	cmd.SetOut(o.Out)
+	cmd.SetErr(o.ErrOut)
+	cmd.SetIn(o.In)
+
+	cmd.AddCommand(
+		newBuggifyCrashLoop(streams),
+		newBuggifyNoSchedule(streams),
+		newBuggifyEmptyMonitorConf(streams),
+	)
+	o.configFlags.AddFlags(cmd.Flags())
+
+	return cmd
+}
diff --git a/kubectl-fdb/cmd/buggify_crash_loop.go b/kubectl-fdb/cmd/buggify_crash_loop.go
new file mode 100644
index 00000000..87df8251
--- /dev/null
+++ b/kubectl-fdb/cmd/buggify_crash_loop.go
@@ -0,0 +1,150 @@
+/*
+ * buggify_crash_loop.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 cmd
+
+import (
+	ctx "context"
+	"fmt"
+	"log"
+
+	"github.com/spf13/cobra"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/cli-runtime/pkg/genericclioptions"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func newBuggifyCrashLoop(streams genericclioptions.IOStreams) *cobra.Command {
+	o := newFDBOptions(streams)
+
+	cmd := &cobra.Command{
+		Use:   "crash-loop",
+		Short: "Updates the crash-loop list of the given cluster",
+		Long:  "Updates the crash-loop list of the given cluster",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			wait, err := cmd.Root().Flags().GetBool("wait")
+			if err != nil {
+				return err
+			}
+			clear, err := cmd.Flags().GetBool("clear")
+			if err != nil {
+				return err
+			}
+			clean, err := cmd.Flags().GetBool("clean")
+			if err != nil {
+				return err
+			}
+			cluster, err := cmd.Flags().GetString("fdb-cluster")
+			if err != nil {
+				return err
+			}
+
+			kubeClient, err := getKubeClient(o)
+			if err != nil {
+				return err
+			}
+
+			namespace, err := getNamespace(*o.configFlags.Namespace)
+			if err != nil {
+				return err
+			}
+
+			return updateCrashLoopList(kubeClient, cluster, args, namespace, wait, clear, clean)
+		},
+		Example: `
+# Add process groups into crash loop state for a cluster in the current namespace
+kubectl fdb buggify crash-loop -c cluster pod-1 pod-2
+
+# Remove process groups from crash loop state from a cluster in the current namespace
+kubectl fdb buggify crash-loop --clear -c cluster pod-1 pod-2
+
+# Clean crash loop list of a cluster in the current namespace
+kubectl fdb buggify crash-loop --clean -c cluster
+
+# Add process groups into crash loop state for a cluster in the namespace default
+kubectl fdb -n default buggify crash-loop -c cluster pod-1 pod-2
+`,
+	}
+	cmd.Flags().StringP("fdb-cluster", "c", "", "updates the crash-loop list in the provided cluster.")
+	cmd.Flags().Bool("clear", false, "removes the process groups from the crash-loop list.")
+	cmd.Flags().Bool("clean", false, "removes all process groups from the crash-loop list.")
+	err := cmd.MarkFlagRequired("fdb-cluster")
+	if err != nil {
+		log.Fatal(err)
+	}
+	cmd.SetOut(o.Out)
+	cmd.SetErr(o.ErrOut)
+	cmd.SetIn(o.In)
+
+	o.configFlags.AddFlags(cmd.Flags())
+
+	return cmd
+}
+
+// updateCrashLoopList updates the crash-loop list of the cluster
+func updateCrashLoopList(kubeClient client.Client, clusterName string, pods []string, namespace string, wait bool, clear bool, clean bool) error {
+	cluster, err := loadCluster(kubeClient, namespace, clusterName)
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			return fmt.Errorf("could not get cluster: %s/%s", namespace, clusterName)
+		}
+		return err
+	}
+
+	processGroupIDs, err := getProcessGroupIDsFromPodName(cluster, pods)
+	if err != nil {
+		return err
+	}
+
+	patch := client.MergeFrom(cluster.DeepCopy())
+	if clean {
+		if wait {
+			if !confirmAction(fmt.Sprintf("Clearing crash-loop list from cluster %s/%s", namespace, clusterName)) {
+				return fmt.Errorf("user aborted the removal")
+			}
+		}
+		cluster.Spec.Buggify.CrashLoop = nil
+		return kubeClient.Patch(ctx.TODO(), cluster, patch)
+	}
+
+	if len(processGroupIDs) == 0 {
+		return fmt.Errorf("please provide atleast one pod")
+	}
+
+	if wait {
+		if clear {
+			if !confirmAction(fmt.Sprintf("Removing %v to crash-loop from cluster %s/%s", processGroupIDs, namespace, clusterName)) {
+				return fmt.Errorf("user aborted the removal")
+			}
+		} else {
+			if !confirmAction(fmt.Sprintf("Adding %v to crash-loop from cluster %s/%s", processGroupIDs, namespace, clusterName)) {
+				return fmt.Errorf("user aborted the removal")
+			}
+		}
+	}
+
+	if clear {
+		cluster.RemoveProcessGroupsFromCrashLoopList(processGroupIDs)
+	} else {
+		cluster.AddProcessGroupsToCrashLoopList(processGroupIDs)
+	}
+
+	return kubeClient.Patch(ctx.TODO(), cluster, patch)
+}
diff --git a/kubectl-fdb/cmd/buggify_empty_monitor_conf.go b/kubectl-fdb/cmd/buggify_empty_monitor_conf.go
new file mode 100644
index 00000000..9ae6baf8
--- /dev/null
+++ b/kubectl-fdb/cmd/buggify_empty_monitor_conf.go
@@ -0,0 +1,115 @@
+/*
+ * buggify_empty_monitor_conf.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 cmd
+
+import (
+	ctx "context"
+	"fmt"
+	"log"
+
+	"github.com/spf13/cobra"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/cli-runtime/pkg/genericclioptions"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func newBuggifyEmptyMonitorConf(streams genericclioptions.IOStreams) *cobra.Command {
+	o := newFDBOptions(streams)
+
+	cmd := &cobra.Command{
+		Use:   "empty-monitor-conf",
+		Short: "Instructs the operator to update all of the fdbmonitor.conf files to have zero fdbserver processes configured",
+		Long:  "Instructs the operator to update all of the fdbmonitor.conf files to have zero fdbserver processes configured",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			wait, err := cmd.Root().Flags().GetBool("wait")
+			if err != nil {
+				return err
+			}
+			set, err := cmd.Flags().GetBool("set-value")
+			if err != nil {
+				return err
+			}
+			cluster, err := cmd.Flags().GetString("fdb-cluster")
+			if err != nil {
+				return err
+			}
+
+			kubeClient, err := getKubeClient(o)
+			if err != nil {
+				return err
+			}
+
+			namespace, err := getNamespace(*o.configFlags.Namespace)
+			if err != nil {
+				return err
+			}
+
+			return updateMonitorConf(kubeClient, cluster, namespace, wait, set)
+		},
+		Example: `
+# Setting empty-monitor-conf to true
+kubectl fdb buggify empty-monitor-conf -c cluster
+
+# Setting empty-monitor-conf to false
+kubectl fdb buggify empty-monitor-conf --set-value=false -c cluster
+`,
+	}
+
+	cmd.Flags().StringP("fdb-cluster", "c", "", "updates the empty-monitor-conf for the cluster.")
+	cmd.Flags().Bool("set-value", true, "set the empty-monitor-conf to true.")
+	err := cmd.MarkFlagRequired("fdb-cluster")
+	if err != nil {
+		log.Fatal(err)
+	}
+	cmd.SetOut(o.Out)
+	cmd.SetErr(o.ErrOut)
+	cmd.SetIn(o.In)
+
+	o.configFlags.AddFlags(cmd.Flags())
+
+	return cmd
+}
+
+// updateMonitorConf updates the monitor conf of the cluster
+func updateMonitorConf(kubeClient client.Client, clusterName string, namespace string, wait bool, set bool) error {
+	cluster, err := loadCluster(kubeClient, namespace, clusterName)
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			return fmt.Errorf("could not get cluster: %s/%s", namespace, clusterName)
+		}
+		return err
+	}
+
+	if cluster.Spec.Buggify.EmptyMonitorConf == set {
+		return fmt.Errorf("empty-monitor-conf is already set to %v", set)
+	}
+
+	patch := client.MergeFrom(cluster.DeepCopy())
+
+	if wait {
+		if !confirmAction(fmt.Sprintf("Setting empty-monitor-conf to %v for cluster %s/%s", set, namespace, clusterName)) {
+			return fmt.Errorf("user aborted the removal")
+		}
+	}
+
+	cluster.Spec.Buggify.EmptyMonitorConf = set
+	return kubeClient.Patch(ctx.TODO(), cluster, patch)
+}
diff --git a/kubectl-fdb/cmd/buggify_no_schedule.go b/kubectl-fdb/cmd/buggify_no_schedule.go
new file mode 100644
index 00000000..48656521
--- /dev/null
+++ b/kubectl-fdb/cmd/buggify_no_schedule.go
@@ -0,0 +1,151 @@
+/*
+ * buggify_no_schedule.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 cmd
+
+import (
+	ctx "context"
+	"fmt"
+	"log"
+
+	"github.com/spf13/cobra"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/cli-runtime/pkg/genericclioptions"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func newBuggifyNoSchedule(streams genericclioptions.IOStreams) *cobra.Command {
+	o := newFDBOptions(streams)
+
+	cmd := &cobra.Command{
+		Use:   "no-schedule",
+		Short: "Updates the no-schedule list of the given cluster",
+		Long:  "Updates the no-schedule list of the given cluster\"",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			wait, err := cmd.Root().Flags().GetBool("wait")
+			if err != nil {
+				return err
+			}
+			clear, err := cmd.Flags().GetBool("clear")
+			if err != nil {
+				return err
+			}
+			clean, err := cmd.Flags().GetBool("clean")
+			if err != nil {
+				return err
+			}
+			cluster, err := cmd.Flags().GetString("fdb-cluster")
+			if err != nil {
+				return err
+			}
+
+			kubeClient, err := getKubeClient(o)
+			if err != nil {
+				return err
+			}
+
+			namespace, err := getNamespace(*o.configFlags.Namespace)
+			if err != nil {
+				return err
+			}
+
+			return updateNoScheduleList(kubeClient, cluster, args, namespace, wait, clear, clean)
+		},
+		Example: `
+# Add process groups into no-schedule state for a cluster in the current namespace
+kubectl fdb buggify no-schedule -c cluster pod-1 pod-2
+
+# Remove process groups from no-schedule state from a cluster in the current namespace
+kubectl fdb buggify no-schedule --clear -c cluster pod-1 pod-2
+
+# Clean no-schedule list of a cluster in the current namespace
+kubectl fdb buggify no-schedule  --clean -c cluster
+
+# Add process groups into no-schedule state for a cluster in the namespace default
+kubectl fdb -n default buggify no-schedule  -c cluster pod-1 pod-2
+`,
+	}
+
+	cmd.Flags().StringP("fdb-cluster", "c", "", "updates the no-schedule list in the provided cluster.")
+	cmd.Flags().Bool("clear", false, "removes the process groups from the no-schedule list.")
+	cmd.Flags().Bool("clean", false, "removes all process groups from the no-schedule list.")
+	err := cmd.MarkFlagRequired("fdb-cluster")
+	if err != nil {
+		log.Fatal(err)
+	}
+	cmd.SetOut(o.Out)
+	cmd.SetErr(o.ErrOut)
+	cmd.SetIn(o.In)
+
+	o.configFlags.AddFlags(cmd.Flags())
+
+	return cmd
+}
+
+// updateNoScheduleList updates the removal list of the cluster
+func updateNoScheduleList(kubeClient client.Client, clusterName string, pods []string, namespace string, wait bool, clear bool, clean bool) error {
+	cluster, err := loadCluster(kubeClient, namespace, clusterName)
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			return fmt.Errorf("could not get cluster: %s/%s", namespace, clusterName)
+		}
+		return err
+	}
+
+	processGroupIDs, err := getProcessGroupIDsFromPodName(cluster, pods)
+	if err != nil {
+		return err
+	}
+
+	patch := client.MergeFrom(cluster.DeepCopy())
+	if clean {
+		if wait {
+			if !confirmAction(fmt.Sprintf("Clearing no-schedule list from cluster %s/%s", namespace, clusterName)) {
+				return fmt.Errorf("user aborted the removal")
+			}
+		}
+		cluster.Spec.Buggify.NoSchedule = nil
+		return kubeClient.Patch(ctx.TODO(), cluster, patch)
+	}
+
+	if len(processGroupIDs) == 0 {
+		return fmt.Errorf("please provide atleast one pod")
+	}
+
+	if wait {
+		if clear {
+			if !confirmAction(fmt.Sprintf("Removing %v from no-schedule from cluster %s/%s", processGroupIDs, namespace, clusterName)) {
+				return fmt.Errorf("user aborted the removal")
+			}
+		} else {
+			if !confirmAction(fmt.Sprintf("Adding %v from no-schedule from cluster %s/%s", processGroupIDs, namespace, clusterName)) {
+				return fmt.Errorf("user aborted the removal")
+			}
+		}
+	}
+
+	if clear {
+		cluster.RemoveProcessGroupsFromNoScheduleList(processGroupIDs)
+	} else {
+		cluster.AddProcessGroupsToNoScheduleList(processGroupIDs)
+	}
+
+	return kubeClient.Patch(ctx.TODO(), cluster, patch)
+}
diff --git a/kubectl-fdb/cmd/cordon.go b/kubectl-fdb/cmd/cordon.go
index 2dd4ea53..4b7f761e 100644
--- a/kubectl-fdb/cmd/cordon.go
+++ b/kubectl-fdb/cmd/cordon.go
@@ -145,7 +145,7 @@ func cordonNode(kubeClient client.Client, cluster *fdbv1beta2.FoundationDBCluste
 		}
 
 		for _, pod := range pods.Items {
-			// With the field selector above this shouldn't be required but it's good to
+			// With the field selector above this shouldn't be required, but it's good to
 			// have a second check.
 			if pod.Spec.NodeName != node {
 				fmt.Printf("Pod: %s is not running on node %s will be ignored\n", pod.Name, node)
@@ -161,5 +161,5 @@ func cordonNode(kubeClient client.Client, cluster *fdbv1beta2.FoundationDBCluste
 		}
 	}
 
-	return replaceProcessGroups(kubeClient, cluster.Name, processGroups, namespace, withExclusion, false, wait, false)
+	return replaceProcessGroups(kubeClient, cluster.Name, processGroups, namespace, withExclusion, wait, false, true)
 }
diff --git a/kubectl-fdb/cmd/exclusion_status.go b/kubectl-fdb/cmd/exclusion_status.go
index 07e1907b..aeacf466 100644
--- a/kubectl-fdb/cmd/exclusion_status.go
+++ b/kubectl-fdb/cmd/exclusion_status.go
@@ -155,7 +155,7 @@ func getExclusionStatus(cmd *cobra.Command, restConfig *rest.Config, kubeClient
 		}
 
 		status := &fdbv1beta2.FoundationDBStatus{}
-		err = json.Unmarshal([]byte(res), status)
+		err = json.Unmarshal(res, status)
 		if err != nil {
 			// If an error occurs retry
 			cmd.PrintErrln(err)
diff --git a/kubectl-fdb/cmd/k8s_client.go b/kubectl-fdb/cmd/k8s_client.go
index f9771a80..35b874b9 100644
--- a/kubectl-fdb/cmd/k8s_client.go
+++ b/kubectl-fdb/cmd/k8s_client.go
@@ -23,6 +23,8 @@ package cmd
 import (
 	"bytes"
 	ctx "context"
+	"fmt"
+	"strings"
 
 	fdbv1beta1 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta1"
 	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
@@ -198,3 +200,23 @@ func getAllPodsFromClusterWithCondition(kubeClient client.Client, clusterName st
 
 	return processes, nil
 }
+
+// getProcessGroupIDsFromPodName returns the process group IDs based on the cluster configuration.
+func getProcessGroupIDsFromPodName(cluster *fdbv1beta2.FoundationDBCluster, podNames []string) ([]string, error) {
+	processGroupIDs := make([]string, 0, len(podNames))
+
+	// TODO(johscheuer): We could validate if the provided process group is actually part of the cluster
+	for _, podName := range podNames {
+		if podName == "" {
+			continue
+		}
+
+		if !strings.HasPrefix(podName, cluster.Name) {
+			return nil, fmt.Errorf("cluster name %s is not set as prefix for Pod name %s, please ensure the specified Pod is part of the cluster", cluster.Name, podName)
+		}
+
+		processGroupIDs = append(processGroupIDs, internal.GetProcessGroupIDFromPodName(cluster, podName))
+	}
+
+	return processGroupIDs, nil
+}
diff --git a/kubectl-fdb/cmd/remove_process_group.go b/kubectl-fdb/cmd/remove_process_group.go
index 1511799a..41ec570c 100644
--- a/kubectl-fdb/cmd/remove_process_group.go
+++ b/kubectl-fdb/cmd/remove_process_group.go
@@ -26,9 +26,7 @@ import (
 	"log"
 
 	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
-	"github.com/FoundationDB/fdb-kubernetes-operator/internal"
 	"github.com/spf13/cobra"
-	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/cli-runtime/pkg/genericclioptions"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -54,10 +52,6 @@ func newRemoveProcessGroupCmd(streams genericclioptions.IOStreams) *cobra.Comman
 			if err != nil {
 				return err
 			}
-			withShrink, err := cmd.Flags().GetBool("shrink")
-			if err != nil {
-				return err
-			}
 			useProcessGroupID, err := cmd.Flags().GetBool("use-process-group-id")
 			if err != nil {
 				return err
@@ -77,15 +71,7 @@ func newRemoveProcessGroupCmd(streams genericclioptions.IOStreams) *cobra.Comman
 				return err
 			}
 
-			processGroups := args
-			if !useProcessGroupID {
-				processGroups, err = getProcessGroupIDsFromPod(kubeClient, cluster, processGroups, namespace)
-				if err != nil {
-					return err
-				}
-			}
-
-			return replaceProcessGroups(kubeClient, cluster, processGroups, namespace, withExclusion, withShrink, wait, removeAllFailed)
+			return replaceProcessGroups(kubeClient, cluster, args, namespace, withExclusion, wait, removeAllFailed, useProcessGroupID)
 		},
 		Example: `
 # Remove process groups for a cluster in the current namespace
@@ -105,7 +91,6 @@ kubectl fdb -n default remove process-group -c cluster --remove-all-failed
 
 	cmd.Flags().StringP("fdb-cluster", "c", "", "remove process groupss from the provided cluster.")
 	cmd.Flags().BoolP("exclusion", "e", true, "define if the process groups should be removed with exclusion.")
-	cmd.Flags().Bool("shrink", false, "define if the removed process groups should not be replaced.")
 	cmd.Flags().Bool("remove-all-failed", false, "define if all failed processes should be replaced.")
 	cmd.Flags().Bool("use-process-group-id", false, "define if the process-group should be used instead of the Pod name.")
 	err := cmd.MarkFlagRequired("fdb-cluster")
@@ -121,36 +106,12 @@ kubectl fdb -n default remove process-group -c cluster --remove-all-failed
 	return cmd
 }
 
-func getProcessGroupIDsFromPod(kubeClient client.Client, clusterName string, podNames []string, namespace string) ([]string, error) {
-	processGroups := make([]string, 0, len(podNames))
-	// Build a map to filter faster
-	podNameMap := map[string]struct{}{}
-	for _, name := range podNames {
-		podNameMap[name] = struct{}{}
-	}
-
-	cluster, err := loadCluster(kubeClient, namespace, clusterName)
-	if err != nil {
-		return processGroups, err
-	}
-	pods, err := getPodsForCluster(kubeClient, cluster, namespace)
-	if err != nil {
-		return processGroups, err
-	}
-
-	for _, pod := range pods.Items {
-		if _, ok := podNameMap[pod.Name]; !ok {
-			continue
-		}
-
-		processGroups = append(processGroups, pod.Labels[cluster.GetProcessGroupIDLabel()])
+// replaceProcessGroups adds process groups to the removal list of the cluster
+func replaceProcessGroups(kubeClient client.Client, clusterName string, processGroupIDs []string, namespace string, withExclusion bool, wait bool, removeAllFailed bool, useProcessGroupID bool) error {
+	if len(processGroupIDs) == 0 && !removeAllFailed {
+		return nil
 	}
 
-	return processGroups, nil
-}
-
-// replaceProcessGroups adds process groups to the removal list of the cluster
-func replaceProcessGroups(kubeClient client.Client, clusterName string, processGroups []string, namespace string, withExclusion bool, withShrink bool, wait bool, removeAllFailed bool) error {
 	cluster, err := loadCluster(kubeClient, namespace, clusterName)
 
 	if err != nil {
@@ -160,32 +121,18 @@ func replaceProcessGroups(kubeClient client.Client, clusterName string, processG
 		return err
 	}
 
-	if len(processGroups) == 0 && !removeAllFailed {
-		return nil
-	}
-
-	shrinkMap := make(map[fdbv1beta2.ProcessClass]int)
-
-	if withShrink {
-		var pods corev1.PodList
-		err := kubeClient.List(ctx.Background(), &pods,
-			client.InNamespace(namespace),
-			client.MatchingLabels(cluster.GetMatchLabels()),
-		)
+	// In this case the user has Pod name specified
+	if !useProcessGroupID {
+		processGroupIDs, err = getProcessGroupIDsFromPodName(cluster, processGroupIDs)
 		if err != nil {
 			return err
 		}
-
-		for _, pod := range pods.Items {
-			class := internal.GetProcessClassFromMeta(cluster, pod.ObjectMeta)
-			shrinkMap[class]++
-		}
 	}
 
 	patch := client.MergeFrom(cluster.DeepCopy())
 
 	processGroupSet := map[string]fdbv1beta2.None{}
-	for _, processGroup := range processGroups {
+	for _, processGroup := range processGroupIDs {
 		processGroupSet[processGroup] = fdbv1beta2.None{}
 	}
 
@@ -198,26 +145,21 @@ func replaceProcessGroups(kubeClient client.Client, clusterName string, processG
 
 			needsReplacement, _ := processGroupStatus.NeedsReplacement(0)
 			if needsReplacement {
-				processGroups = append(processGroups, processGroupStatus.ProcessGroupID)
+				processGroupIDs = append(processGroupIDs, processGroupStatus.ProcessGroupID)
 			}
 		}
 	}
 
-	for class, amount := range shrinkMap {
-		cluster.Spec.ProcessCounts.DecreaseCount(class, amount)
-	}
-
 	if wait {
-		confirmed := confirmAction(fmt.Sprintf("Remove %v from cluster %s/%s with exclude: %t and shrink: %t", processGroups, namespace, clusterName, withExclusion, withShrink))
-		if !confirmed {
+		if !confirmAction(fmt.Sprintf("Remove %v from cluster %s/%s with exclude: %t", processGroupIDs, namespace, clusterName, withExclusion)) {
 			return fmt.Errorf("user aborted the removal")
 		}
 	}
 
 	if withExclusion {
-		cluster.AddProcessGroupsToRemovalList(processGroups)
+		cluster.AddProcessGroupsToRemovalList(processGroupIDs)
 	} else {
-		cluster.AddProcessGroupsToRemovalWithoutExclusionList(processGroups)
+		cluster.AddProcessGroupsToRemovalWithoutExclusionList(processGroupIDs)
 	}
 
 	return kubeClient.Patch(ctx.TODO(), cluster, patch)
diff --git a/kubectl-fdb/cmd/root.go b/kubectl-fdb/cmd/root.go
index fb044801..adbd9543 100644
--- a/kubectl-fdb/cmd/root.go
+++ b/kubectl-fdb/cmd/root.go
@@ -25,6 +25,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+
 	"strings"
 
 	"github.com/spf13/viper"
@@ -80,6 +81,7 @@ func NewRootCmd(streams genericclioptions.IOStreams) *cobra.Command {
 		newDeprecationCmd(streams),
 		newFixCoordinatorIPsCmd(streams),
 		newGetCmd(streams),
+		newBuggifyCmd(streams),
 	)
 
 	return cmd
diff --git a/mock-kubernetes-client/client/client.go b/mock-kubernetes-client/client/client.go
index c9a521e9..a7276771 100644
--- a/mock-kubernetes-client/client/client.go
+++ b/mock-kubernetes-client/client/client.go
@@ -65,7 +65,7 @@ func (client *MockClient) Clear() {
 
 // Scheme returns the runtime Scheme
 func (client *MockClient) Scheme() *runtime.Scheme {
-	return nil
+	return runtime.NewScheme()
 }
 
 // RESTMapper returns the RESTMapper
diff --git a/setup/setup.go b/setup/setup.go
index 45d25559..7998cddc 100644
--- a/setup/setup.go
+++ b/setup/setup.go
@@ -50,24 +50,25 @@ var operatorVersion = "latest"
 
 // Options provides all configuration Options for the operator
 type Options struct {
-	MetricsAddr             string
-	EnableLeaderElection    bool
-	LeaderElectionID        string
-	LogFile                 string
-	CliTimeout              int
-	DeprecationOptions      internal.DeprecationOptions
-	MaxConcurrentReconciles int
-	CleanUpOldLogFile       bool
-	LogFileMinAge           time.Duration
-	LogFileMaxSize          int
-	LogFileMaxAge           int
-	MaxNumberOfOldLogFiles  int
-	CompressOldFiles        bool
-	PrintVersion            bool
-	LabelSelector           string
-	WatchNamespace          string
-	GetTimeout              time.Duration
-	PostTimeout             time.Duration
+	EnableLeaderElection               bool
+	CleanUpOldLogFile                  bool
+	CompressOldFiles                   bool
+	PrintVersion                       bool
+	EnableRestartIncompatibleProcesses bool
+	MetricsAddr                        string
+	LeaderElectionID                   string
+	LogFile                            string
+	LabelSelector                      string
+	WatchNamespace                     string
+	CliTimeout                         int
+	MaxConcurrentReconciles            int
+	LogFileMaxSize                     int
+	LogFileMaxAge                      int
+	MaxNumberOfOldLogFiles             int
+	LogFileMinAge                      time.Duration
+	GetTimeout                         time.Duration
+	PostTimeout                        time.Duration
+	DeprecationOptions                 internal.DeprecationOptions
 }
 
 // BindFlags will parse the given flagset for the operator option flags
@@ -94,6 +95,7 @@ func (o *Options) BindFlags(fs *flag.FlagSet) {
 	fs.StringVar(&o.WatchNamespace, "watch-namespace", os.Getenv("WATCH_NAMESPACE"), "Defines which namespace the operator should watch")
 	fs.DurationVar(&o.GetTimeout, "get-timeout", 5*time.Second, "http timeout for get requests to the FDB sidecar")
 	fs.DurationVar(&o.PostTimeout, "post-timeout", 10*time.Second, "http timeout for post requests to the FDB sidecar")
+	fs.BoolVar(&o.EnableRestartIncompatibleProcesses, "enable-restart-incompatible-processes", true, "This flag enables/disables in the operator to restart incompatible fdbserver processes.")
 }
 
 // StartManager will start the FoundationDB operator manager.
@@ -180,6 +182,7 @@ func StartManager(
 		clusterReconciler.GetTimeout = operatorOpts.GetTimeout
 		clusterReconciler.PostTimeout = operatorOpts.PostTimeout
 		clusterReconciler.Log = logr.WithName("controllers").WithName("FoundationDBCluster")
+		clusterReconciler.EnableRestartIncompatibleProcesses = operatorOpts.EnableRestartIncompatibleProcesses
 
 		if err := clusterReconciler.SetupWithManager(mgr, operatorOpts.MaxConcurrentReconciles, *labelSelector, watchedObjects...); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "FoundationDBCluster")


Test patch:
diff --git a/api/v1beta2/foundationdb_status_test.go b/api/v1beta2/foundationdb_status_test.go
index 9b4598ed..cfb97b79 100644
--- a/api/v1beta2/foundationdb_status_test.go
+++ b/api/v1beta2/foundationdb_status_test.go
@@ -70,6 +70,7 @@ var _ = Describe("FoundationDBStatus", func() {
 					DatabaseStatus: FoundationDBStatusClientDBStatus{Available: true, Healthy: true},
 				},
 				Cluster: FoundationDBStatusClusterInfo{
+					IncompatibleConnections: []string{},
 					FaultTolerance: FaultTolerance{
 						MaxZoneFailuresWithoutLosingAvailability: 1,
 						MaxZoneFailuresWithoutLosingData:         1,
@@ -459,6 +460,7 @@ var _ = Describe("FoundationDBStatus", func() {
 
 	When("parsing the status json with a 7.1.0-rc1 cluster", func() {
 		status := FoundationDBStatusClusterInfo{
+			IncompatibleConnections: []string{},
 			DatabaseConfiguration: DatabaseConfiguration{
 				RedundancyMode:  "double",
 				StorageEngine:   StorageEngineSSD2,
diff --git a/api/v1beta2/foundationdbcluster_types_test.go b/api/v1beta2/foundationdbcluster_types_test.go
index 92d299ef..b851f87f 100644
--- a/api/v1beta2/foundationdbcluster_types_test.go
+++ b/api/v1beta2/foundationdbcluster_types_test.go
@@ -4626,6 +4626,315 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 				},
 				fmt.Errorf("storage engine ssd-rocksdb-v1 is not supported on version 6.1.3, stateless is not a valid process class for coordinators"),
 			),
+			Entry("using invalid version for sharded rocksdb",
+				&FoundationDBCluster{
+					Spec: FoundationDBClusterSpec{
+						Version: "7.1.4",
+						DatabaseConfiguration: DatabaseConfiguration{
+							StorageEngine: StorageEngineShardedRocksDB,
+						},
+					},
+				},
+				fmt.Errorf("storage engine ssd-sharded-rocksdb is not supported on version 7.1.4"),
+			),
+			Entry("using valid version for sharded rocksdb",
+				&FoundationDBCluster{
+					Spec: FoundationDBClusterSpec{
+						Version: "7.2.0",
+						DatabaseConfiguration: DatabaseConfiguration{
+							StorageEngine: StorageEngineShardedRocksDB,
+						},
+					},
+				},
+				nil,
+			),
 		)
 	})
+
+	When("adding processes to the no-schedule list", func() {
+		var cluster *FoundationDBCluster
+
+		BeforeEach(func() {
+			cluster = &FoundationDBCluster{}
+		})
+
+		When("the no-schedule list is empty", func() {
+			type testCase struct {
+				Instances                     []string
+				ExpectedInstancesInNoSchedule []string
+			}
+
+			DescribeTable("should add all targeted processes to the no-schedule list",
+				func(tc testCase) {
+					cluster.AddProcessGroupsToNoScheduleList(tc.Instances)
+					Expect(cluster.Spec.Buggify.NoSchedule).To(ContainElements(tc.ExpectedInstancesInNoSchedule))
+					Expect(len(cluster.Spec.Buggify.NoSchedule)).To(Equal(len(tc.ExpectedInstancesInNoSchedule)))
+				},
+				Entry("Adding single instance",
+					testCase{
+						Instances:                     []string{"instance-1"},
+						ExpectedInstancesInNoSchedule: []string{"instance-1"},
+					}),
+				Entry("Adding multiple instances",
+					testCase{
+						Instances:                     []string{"instance-1", "instance-2"},
+						ExpectedInstancesInNoSchedule: []string{"instance-1", "instance-2"},
+					}),
+			)
+		})
+
+		When("the no-schedule list is not empty", func() {
+			BeforeEach(func() {
+				cluster.Spec.Buggify.NoSchedule = []string{"instance-1"}
+			})
+
+			type testCase struct {
+				Instances                     []string
+				ExpectedInstancesInNoSchedule []string
+			}
+
+			DescribeTable("should add all targeted processes to no-schedule list",
+				func(tc testCase) {
+					cluster.AddProcessGroupsToNoScheduleList(tc.Instances)
+					Expect(cluster.Spec.Buggify.NoSchedule).To(ContainElements(tc.ExpectedInstancesInNoSchedule))
+					Expect(len(cluster.Spec.Buggify.NoSchedule)).To(Equal(len(tc.ExpectedInstancesInNoSchedule)))
+				},
+				Entry("Adding single instance",
+					testCase{
+						Instances:                     []string{"instance-2"},
+						ExpectedInstancesInNoSchedule: []string{"instance-1", "instance-2"},
+					}),
+				Entry("Adding multiple instances",
+					testCase{
+						Instances:                     []string{"instance-2", "instance-3"},
+						ExpectedInstancesInNoSchedule: []string{"instance-1", "instance-2", "instance-3"},
+					}),
+			)
+		})
+
+	})
+
+	When("removing processes from the no-schedule list", func() {
+		var cluster *FoundationDBCluster
+
+		BeforeEach(func() {
+			cluster = &FoundationDBCluster{
+				Spec: FoundationDBClusterSpec{
+					Buggify: BuggifyConfig{
+						NoSchedule: []string{"instance-1", "instance-2", "instance-3"},
+					},
+				},
+			}
+		})
+
+		type testCase struct {
+			Instances                         []string
+			ExpectedInstancesInNoScheduleList []string
+		}
+
+		DescribeTable("should remove all targeted processes from the no-schedule list",
+			func(tc testCase) {
+				cluster.RemoveProcessGroupsFromNoScheduleList(tc.Instances)
+				Expect(cluster.Spec.Buggify.NoSchedule).To(ContainElements(tc.ExpectedInstancesInNoScheduleList))
+				Expect(len(cluster.Spec.Buggify.NoSchedule)).To(Equal(len(tc.ExpectedInstancesInNoScheduleList)))
+			},
+			Entry("Removing single instance",
+				testCase{
+					Instances:                         []string{"instance-1"},
+					ExpectedInstancesInNoScheduleList: []string{"instance-2", "instance-3"},
+				}),
+			Entry("Removing multiple instances",
+				testCase{
+					Instances:                         []string{"instance-2", "instance-3"},
+					ExpectedInstancesInNoScheduleList: []string{"instance-1"},
+				}),
+		)
+
+	})
+
+	When("adding processes to the crash-loop list", func() {
+		var cluster *FoundationDBCluster
+
+		BeforeEach(func() {
+			cluster = &FoundationDBCluster{}
+		})
+
+		When("the crash-loop list is empty", func() {
+			type testCase struct {
+				Instances                    []string
+				ExpectedInstancesInCrashLoop []string
+			}
+
+			DescribeTable("should add all targeted processes to the crash-loop list",
+				func(tc testCase) {
+					cluster.AddProcessGroupsToCrashLoopList(tc.Instances)
+					Expect(cluster.Spec.Buggify.CrashLoop).To(ContainElements(tc.ExpectedInstancesInCrashLoop))
+					Expect(len(cluster.Spec.Buggify.CrashLoop)).To(Equal(len(tc.ExpectedInstancesInCrashLoop)))
+				},
+				Entry("Adding single instance",
+					testCase{
+						Instances:                    []string{"instance-1"},
+						ExpectedInstancesInCrashLoop: []string{"instance-1"},
+					}),
+				Entry("Adding multiple instances",
+					testCase{
+						Instances:                    []string{"instance-1", "instance-2"},
+						ExpectedInstancesInCrashLoop: []string{"instance-1", "instance-2"},
+					}),
+				Entry("Adding all instances",
+					testCase{
+						Instances:                    []string{"*"},
+						ExpectedInstancesInCrashLoop: []string{"*"},
+					}),
+			)
+		})
+
+		When("the crash-loop list is not empty", func() {
+			BeforeEach(func() {
+				cluster.Spec.Buggify.CrashLoop = []string{"instance-1"}
+			})
+
+			type testCase struct {
+				Instances                    []string
+				ExpectedInstancesInCrashLoop []string
+			}
+
+			DescribeTable("should add all targeted processes to crash-loop list",
+				func(tc testCase) {
+					cluster.AddProcessGroupsToCrashLoopList(tc.Instances)
+					Expect(cluster.Spec.Buggify.CrashLoop).To(ContainElements(tc.ExpectedInstancesInCrashLoop))
+					Expect(len(cluster.Spec.Buggify.CrashLoop)).To(Equal(len(tc.ExpectedInstancesInCrashLoop)))
+				},
+				Entry("Adding single instance",
+					testCase{
+						Instances:                    []string{"instance-2"},
+						ExpectedInstancesInCrashLoop: []string{"instance-1", "instance-2"},
+					}),
+				Entry("Adding multiple instances",
+					testCase{
+						Instances:                    []string{"instance-2", "instance-3"},
+						ExpectedInstancesInCrashLoop: []string{"instance-1", "instance-2", "instance-3"},
+					}),
+				Entry("Adding all instances",
+					testCase{
+						Instances:                    []string{"*"},
+						ExpectedInstancesInCrashLoop: []string{"instance-1", "*"},
+					}),
+			)
+		})
+
+		When("the crash-loop list contains *", func() {
+			BeforeEach(func() {
+				cluster.Spec.Buggify.CrashLoop = []string{"*"}
+			})
+
+			type testCase struct {
+				Instances                    []string
+				ExpectedInstancesInCrashLoop []string
+			}
+
+			DescribeTable("should add all targeted processes to crash-loop list",
+				func(tc testCase) {
+					cluster.AddProcessGroupsToCrashLoopList(tc.Instances)
+					Expect(cluster.Spec.Buggify.CrashLoop).To(ContainElements(tc.ExpectedInstancesInCrashLoop))
+					Expect(len(cluster.Spec.Buggify.CrashLoop)).To(Equal(len(tc.ExpectedInstancesInCrashLoop)))
+				},
+				Entry("Adding single instance",
+					testCase{
+						Instances:                    []string{"instance-1"},
+						ExpectedInstancesInCrashLoop: []string{"instance-1", "*"},
+					}),
+				Entry("Adding multiple instances",
+					testCase{
+						Instances:                    []string{"instance-2", "instance-3"},
+						ExpectedInstancesInCrashLoop: []string{"*", "instance-2", "instance-3"},
+					}),
+				Entry("Adding all instances",
+					testCase{
+						Instances:                    []string{"*"},
+						ExpectedInstancesInCrashLoop: []string{"*"},
+					}),
+			)
+		})
+
+	})
+
+	When("removing processes from the crash-loop list", func() {
+		var cluster *FoundationDBCluster
+
+		When("the crash-loop list does not contain *", func() {
+			BeforeEach(func() {
+				cluster = &FoundationDBCluster{
+					Spec: FoundationDBClusterSpec{
+						Buggify: BuggifyConfig{
+							CrashLoop: []string{"instance-1", "instance-2", "instance-3"},
+						},
+					},
+				}
+			})
+
+			type testCase struct {
+				Instances                    []string
+				ExpectedInstancesInCrashLoop []string
+			}
+
+			DescribeTable("should remove all targeted processes from the crash-loop list",
+				func(tc testCase) {
+					cluster.RemoveProcessGroupsFromCrashLoopList(tc.Instances)
+					Expect(cluster.Spec.Buggify.CrashLoop).To(ContainElements(tc.ExpectedInstancesInCrashLoop))
+					Expect(len(cluster.Spec.Buggify.CrashLoop)).To(Equal(len(tc.ExpectedInstancesInCrashLoop)))
+				},
+				Entry("Removing single instance",
+					testCase{
+						Instances:                    []string{"instance-1"},
+						ExpectedInstancesInCrashLoop: []string{"instance-2", "instance-3"},
+					}),
+				Entry("Removing multiple instances",
+					testCase{
+						Instances:                    []string{"instance-2", "instance-3"},
+						ExpectedInstancesInCrashLoop: []string{"instance-1"},
+					}),
+			)
+		})
+
+		When("the crash-loop list contain *", func() {
+			BeforeEach(func() {
+				cluster = &FoundationDBCluster{
+					Spec: FoundationDBClusterSpec{
+						Buggify: BuggifyConfig{
+							CrashLoop: []string{"*", "instance-1", "instance-2", "instance-3"},
+						},
+					},
+				}
+			})
+
+			type testCase struct {
+				Instances                    []string
+				ExpectedInstancesInCrashLoop []string
+			}
+
+			DescribeTable("should remove all targeted processes from the crash-loop list",
+				func(tc testCase) {
+					cluster.RemoveProcessGroupsFromCrashLoopList(tc.Instances)
+					Expect(cluster.Spec.Buggify.CrashLoop).To(ContainElements(tc.ExpectedInstancesInCrashLoop))
+					Expect(len(cluster.Spec.Buggify.CrashLoop)).To(Equal(len(tc.ExpectedInstancesInCrashLoop)))
+				},
+				Entry("Removing single instance",
+					testCase{
+						Instances:                    []string{"instance-1"},
+						ExpectedInstancesInCrashLoop: []string{"*", "instance-2", "instance-3"},
+					}),
+				Entry("Removing multiple instances",
+					testCase{
+						Instances:                    []string{"instance-2", "instance-3"},
+						ExpectedInstancesInCrashLoop: []string{"*", "instance-1"},
+					}),
+				Entry("Removing *",
+					testCase{
+						Instances:                    []string{"*"},
+						ExpectedInstancesInCrashLoop: []string{"instance-1", "instance-2", "instance-3"},
+					}),
+			)
+		})
+	})
 })
diff --git a/config/tests/unified_image/images.yaml b/config/tests/unified_image/images.yaml
index b95065bb..0d9e06ff 100644
--- a/config/tests/unified_image/images.yaml
+++ b/config/tests/unified_image/images.yaml
@@ -8,3 +8,6 @@
       - tagSuffix: "-local"
 - op: remove
   path: "/spec/processes/general/podTemplate/spec/initContainers/0"
+- op: add
+  path: "/spec/processes/general/podTemplate/spec/serviceAccount"
+  value: "fdb-kubernetes"
diff --git a/config/tests/unified_image/kustomization.yaml b/config/tests/unified_image/kustomization.yaml
index f144c0a3..aacafbd5 100644
--- a/config/tests/unified_image/kustomization.yaml
+++ b/config/tests/unified_image/kustomization.yaml
@@ -4,6 +4,7 @@ apiVersion: kustomize.config.k8s.io/v1beta1
 kind: Kustomization
 resources:
 - "../base"
+- "unified_image_role.yaml"
 patchesJson6902:
 - path: images.yaml
   target:
diff --git a/config/tests/unified_image/unified_image_role.yaml b/config/tests/unified_image/unified_image_role.yaml
new file mode 100644
index 00000000..d66f3309
--- /dev/null
+++ b/config/tests/unified_image/unified_image_role.yaml
@@ -0,0 +1,31 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: fdb-kubernetes
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: fdb-kubernetes
+rules:
+  - apiGroups:
+      - ""
+    resources:
+      - "pods"
+    verbs:
+      - "get"
+      - "watch"
+      - "update"
+      - "patch"
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: fdb-kubernetes
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: fdb-kubernetes
+subjects:
+  - kind: ServiceAccount
+    name: fdb-kubernetes
diff --git a/controllers/bounce_processes_test.go b/controllers/bounce_processes_test.go
index a00afaf8..68669b6b 100644
--- a/controllers/bounce_processes_test.go
+++ b/controllers/bounce_processes_test.go
@@ -166,8 +166,8 @@ var _ = Describe("bounceProcesses", func() {
 			}
 		})
 
-		It("should not requeue", func() {
-			Expect(requeue).To(BeNil())
+		It("should requeue", func() {
+			Expect(requeue).NotTo(BeNil())
 		})
 
 		It("should kill all the processes", func() {
@@ -233,8 +233,8 @@ var _ = Describe("bounceProcesses", func() {
 					Expect(err).NotTo(HaveOccurred())
 				})
 
-				It("should not requeue", func() {
-					Expect(requeue).To(BeNil())
+				It("should requeue", func() {
+					Expect(requeue).NotTo(BeNil())
 				})
 
 				It("should kill all the processes", func() {
@@ -272,8 +272,8 @@ var _ = Describe("bounceProcesses", func() {
 					cluster.Spec.LockOptions.DisableLocks = &disabled
 				})
 
-				It("should not requeue", func() {
-					Expect(requeue).To(BeNil())
+				It("should requeue", func() {
+					Expect(requeue).NotTo(BeNil())
 				})
 
 				It("should kill all the processes", func() {
diff --git a/controllers/cluster_controller_test.go b/controllers/cluster_controller_test.go
index 6ae56417..3bb38be7 100644
--- a/controllers/cluster_controller_test.go
+++ b/controllers/cluster_controller_test.go
@@ -2649,6 +2649,50 @@ var _ = Describe("cluster_controller", func() {
 					})
 				})
 			})
+			When("using ssd-sharded-rocksdb", func() {
+				When("using 7.2.0", func() {
+					BeforeEach(func() {
+						cluster.Spec.DatabaseConfiguration.StorageEngine = fdbv1beta2.StorageEngineShardedRocksDB
+						cluster.Spec.Version = "7.2.0"
+						err := k8sClient.Update(context.TODO(), cluster)
+						Expect(err).NotTo(HaveOccurred())
+					})
+					It("generations are matching", func() {
+						generations, err := reloadClusterGenerations(cluster)
+						Expect(err).NotTo(HaveOccurred())
+						Expect(generations.Reconciled).To(Equal(cluster.ObjectMeta.Generation))
+					})
+				})
+				When("using 7.1.0", func() {
+					BeforeEach(func() {
+						cluster.Spec.DatabaseConfiguration.StorageEngine = fdbv1beta2.StorageEngineShardedRocksDB
+						cluster.Spec.Version = "7.1.0"
+						err := k8sClient.Update(context.TODO(), cluster)
+						Expect(err).NotTo(HaveOccurred())
+						shouldCompleteReconciliation = false
+					})
+					It("generations are not matching", func() {
+						generations, err := reloadClusterGenerations(cluster)
+						Expect(err).NotTo(HaveOccurred())
+						Expect(generations.Reconciled).ToNot(Equal(cluster.ObjectMeta.Generation))
+					})
+				})
+				When("using 6.3.24", func() {
+					BeforeEach(func() {
+						cluster.Spec.DatabaseConfiguration.StorageEngine = fdbv1beta2.StorageEngineShardedRocksDB
+						cluster.Spec.Version = "6.3.24"
+						err := k8sClient.Update(context.TODO(), cluster)
+						Expect(err).NotTo(HaveOccurred())
+						shouldCompleteReconciliation = false
+					})
+					It("generations are not matching", func() {
+						generations, err := reloadClusterGenerations(cluster)
+						Expect(err).NotTo(HaveOccurred())
+						Expect(generations.Reconciled).ToNot(Equal(cluster.ObjectMeta.Generation))
+					})
+				})
+			})
+
 		})
 
 		When("When a process have an incorrect commandline", func() {
diff --git a/controllers/remove_incompatible_processes_test.go b/controllers/remove_incompatible_processes_test.go
new file mode 100644
index 00000000..8443c9b9
--- /dev/null
+++ b/controllers/remove_incompatible_processes_test.go
@@ -0,0 +1,183 @@
+/*
+ * remove_incompatible_processes_tests.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 controllers
+
+import (
+	"context"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+
+	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
+	"github.com/FoundationDB/fdb-kubernetes-operator/internal"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("restart_incompatible_pods", func() {
+	DescribeTable("", func(incompatibleConnections map[string]fdbv1beta2.None, processGroup *fdbv1beta2.ProcessGroupStatus, expected bool) {
+		Expect(isIncompatible(incompatibleConnections, processGroup)).To(Equal(expected))
+	},
+		Entry("empty incompatible map",
+			map[string]fdbv1beta2.None{},
+			&fdbv1beta2.ProcessGroupStatus{
+				Addresses: []string{"1.1.1.1"},
+			},
+			false),
+		Entry("nil incompatible map",
+			nil,
+			&fdbv1beta2.ProcessGroupStatus{
+				Addresses: []string{"1.1.1.1"},
+			},
+			false),
+		Entry("incompatible map contains another address",
+			map[string]fdbv1beta2.None{
+				"1.1.1.2": {},
+			},
+			&fdbv1beta2.ProcessGroupStatus{
+				Addresses: []string{"1.1.1.1"},
+			},
+			false),
+		Entry("incompatible map contains matching address",
+			map[string]fdbv1beta2.None{
+				"1.1.1.1": {},
+			},
+			&fdbv1beta2.ProcessGroupStatus{
+				Addresses: []string{"1.1.1.1"},
+			},
+			true),
+	)
+
+	When("running a reconcile for the restart incompatible process reconciler", func() {
+		var cluster *fdbv1beta2.FoundationDBCluster
+		var initialCount int
+
+		BeforeEach(func() {
+			cluster = internal.CreateDefaultCluster()
+			err := k8sClient.Create(context.TODO(), cluster)
+			Expect(err).NotTo(HaveOccurred())
+
+			result, err := reconcileCluster(cluster)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(result.Requeue).To(BeFalse())
+
+			generation, err := reloadCluster(cluster)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(generation).To(Equal(int64(1)))
+
+			pods := &corev1.PodList{}
+			err = k8sClient.List(context.TODO(), pods, getListOptions(cluster)...)
+			Expect(err).NotTo(HaveOccurred())
+			initialCount = len(pods.Items)
+		})
+
+		JustBeforeEach(func() {
+			err := processIncompatibleProcesses(context.TODO(), clusterReconciler, logr.Discard(), cluster)
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		When("no incompatible processes are reported", func() {
+			BeforeEach(func() {
+				clusterReconciler.EnableRestartIncompatibleProcesses = true
+				adminClient, err := newMockAdminClientUncast(cluster, k8sClient)
+				Expect(err).NotTo(HaveOccurred())
+				adminClient.frozenStatus = &fdbv1beta2.FoundationDBStatus{
+					Cluster: fdbv1beta2.FoundationDBStatusClusterInfo{
+						IncompatibleConnections: []string{},
+					},
+				}
+			})
+
+			It("should have no deletions", func() {
+				pods := &corev1.PodList{}
+				err := k8sClient.List(context.TODO(), pods, getListOptions(cluster)...)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(len(pods.Items)).To(BeNumerically("==", initialCount))
+			})
+		})
+
+		When("no matching incompatible processes are reported", func() {
+			BeforeEach(func() {
+				clusterReconciler.EnableRestartIncompatibleProcesses = true
+				adminClient, err := newMockAdminClientUncast(cluster, k8sClient)
+				Expect(err).NotTo(HaveOccurred())
+				adminClient.frozenStatus = &fdbv1beta2.FoundationDBStatus{
+					Cluster: fdbv1beta2.FoundationDBStatusClusterInfo{
+						IncompatibleConnections: []string{
+							"192.192.192.192",
+						},
+					},
+				}
+			})
+
+			It("should have no deletions", func() {
+				pods := &corev1.PodList{}
+				err := k8sClient.List(context.TODO(), pods, getListOptions(cluster)...)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(len(pods.Items)).To(BeNumerically("==", initialCount))
+			})
+		})
+
+		When("matching incompatible processes are reported and the subreconciler is enabled", func() {
+			BeforeEach(func() {
+				clusterReconciler.EnableRestartIncompatibleProcesses = true
+				adminClient, err := newMockAdminClientUncast(cluster, k8sClient)
+				Expect(err).NotTo(HaveOccurred())
+				adminClient.frozenStatus = &fdbv1beta2.FoundationDBStatus{
+					Cluster: fdbv1beta2.FoundationDBStatusClusterInfo{
+						IncompatibleConnections: []string{
+							cluster.Status.ProcessGroups[0].Addresses[0],
+						},
+					},
+				}
+			})
+
+			It("should have one deletion", func() {
+				pods := &corev1.PodList{}
+				err := k8sClient.List(context.TODO(), pods, getListOptions(cluster)...)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(len(pods.Items)).To(BeNumerically("==", initialCount-1))
+			})
+		})
+
+		When("matching incompatible processes are reported and the subreconciler is disabled", func() {
+			BeforeEach(func() {
+				clusterReconciler.EnableRestartIncompatibleProcesses = false
+				adminClient, err := newMockAdminClientUncast(cluster, k8sClient)
+				Expect(err).NotTo(HaveOccurred())
+				adminClient.frozenStatus = &fdbv1beta2.FoundationDBStatus{
+					Cluster: fdbv1beta2.FoundationDBStatusClusterInfo{
+						IncompatibleConnections: []string{
+							cluster.Status.ProcessGroups[0].Addresses[0],
+						},
+					},
+				}
+			})
+
+			It("should have no deletions", func() {
+				pods := &corev1.PodList{}
+				err := k8sClient.List(context.TODO(), pods, getListOptions(cluster)...)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(len(pods.Items)).To(BeNumerically("==", initialCount))
+			})
+		})
+	})
+})
diff --git a/controllers/update_status_test.go b/controllers/update_status_test.go
index 3e3e28f0..01cad27f 100644
--- a/controllers/update_status_test.go
+++ b/controllers/update_status_test.go
@@ -465,4 +465,20 @@ var _ = Describe("update_status", func() {
 			})
 		})
 	})
+
+	DescribeTable("when getting the running version from the running processes", func(versionMap map[string]int, fallback string, expected string) {
+		Expect(getRunningVersion(versionMap, fallback)).To(Equal(expected))
+	},
+		Entry("when nearly all processes running on the new version", map[string]int{
+			"7.1.11": 1,
+			"7.1.15": 99,
+		}, "0", "7.1.15"),
+		Entry("when all processes running on the same version", map[string]int{
+			"7.1.15": 100,
+		}, "0", "7.1.15"),
+		Entry("when half of the processes running the old/new version", map[string]int{
+			"7.1.11": 50,
+			"7.1.15": 50,
+		}, "0", "7.1.15"),
+		Entry("when the versionMap is empty", map[string]int{}, "7.1.15", "7.1.15"))
 })
diff --git a/internal/fdb_status_helper_test.go b/internal/fdb_status_helper_test.go
index 8b195b30..889fef01 100644
--- a/internal/fdb_status_helper_test.go
+++ b/internal/fdb_status_helper_test.go
@@ -31,7 +31,7 @@ var _ = Describe("fdb_status_helper_test", func() {
 	When("Removing warnings in JSON", func() {
 		type testCase struct {
 			input       string
-			expected    string
+			expected    []byte
 			expectedErr error
 		}
 
@@ -53,7 +53,7 @@ var _ = Describe("fdb_status_helper_test", func() {
 			Entry("Valid JSON without warning",
 				testCase{
 					input:       "{}",
-					expected:    "{}",
+					expected:    []byte("{}"),
 					expectedErr: nil,
 				},
 			),
@@ -63,14 +63,14 @@ var _ = Describe("fdb_status_helper_test", func() {
  # Warning Slow response
 
  {}`,
-					expected:    "{}",
+					expected:    []byte("{}"),
 					expectedErr: nil,
 				},
 			),
 			Entry("Invalid JSON",
 				testCase{
 					input:       "}",
-					expected:    "",
+					expected:    nil,
 					expectedErr: fmt.Errorf("the JSON string doesn't contain a starting '{'"),
 				},
 			),
diff --git a/internal/pod_models_test.go b/internal/pod_models_test.go
index ce331797..566a7a07 100644
--- a/internal/pod_models_test.go
+++ b/internal/pod_models_test.go
@@ -3181,4 +3181,21 @@ var _ = Describe("pod_models", func() {
 			Expect(GetPodDNSName(cluster, "operator-test-storage-1")).To(Equal("operator-test-storage-1.operator-test-1.my-ns.svc.cluster.example"))
 		})
 	})
+
+	DescribeTable("getting the process group ID from the Pod name", func(cluster *fdbv1beta2.FoundationDBCluster, podName string, expected string) {
+		Expect(GetProcessGroupIDFromPodName(cluster, podName)).To(Equal(expected))
+	},
+		Entry("cluster without prefix", &fdbv1beta2.FoundationDBCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test",
+			},
+		}, "test-storage-1", "storage-1"),
+		Entry("cluster with prefix", &fdbv1beta2.FoundationDBCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test",
+			},
+			Spec: fdbv1beta2.FoundationDBClusterSpec{
+				ProcessGroupIDPrefix: "prefix",
+			},
+		}, "test-storage-1", "prefix-storage-1"))
 })
diff --git a/kubectl-fdb/cmd/buggify_crash_loop_test.go b/kubectl-fdb/cmd/buggify_crash_loop_test.go
new file mode 100644
index 00000000..bc020eee
--- /dev/null
+++ b/kubectl-fdb/cmd/buggify_crash_loop_test.go
@@ -0,0 +1,213 @@
+/*
+ * buggify_crash_loop_test.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 cmd
+
+import (
+	ctx "context"
+
+	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("[plugin] buggify crash-loop instances command", func() {
+	clusterName := "test"
+	namespace := "test"
+
+	var cluster *fdbv1beta2.FoundationDBCluster
+
+	BeforeEach(func() {
+		cluster = &fdbv1beta2.FoundationDBCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      clusterName,
+				Namespace: namespace,
+			},
+			Spec: fdbv1beta2.FoundationDBClusterSpec{
+				ProcessCounts: fdbv1beta2.ProcessCounts{
+					Storage: 1,
+				},
+			},
+		}
+	})
+
+	When("running buggify crash-loop instances command", func() {
+		When("adding instances to crash-loop list from a cluster", func() {
+			type testCase struct {
+				Instances                    []string
+				ExpectedInstancesInCrashLoop []string
+			}
+
+			DescribeTable("should add all targeted processes to crash-loop list",
+				func(tc testCase) {
+					scheme := runtime.NewScheme()
+					_ = clientgoscheme.AddToScheme(scheme)
+					_ = fdbv1beta2.AddToScheme(scheme)
+					kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
+
+					err := updateCrashLoopList(kubeClient, clusterName, tc.Instances, namespace, false, false, false)
+					Expect(err).NotTo(HaveOccurred())
+
+					var resCluster fdbv1beta2.FoundationDBCluster
+					err = kubeClient.Get(ctx.Background(), client.ObjectKey{
+						Namespace: namespace,
+						Name:      clusterName,
+					}, &resCluster)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(tc.ExpectedInstancesInCrashLoop).To(ContainElements(resCluster.Spec.Buggify.CrashLoop))
+					Expect(len(tc.ExpectedInstancesInCrashLoop)).To(BeNumerically("==", len(resCluster.Spec.Buggify.CrashLoop)))
+				},
+				Entry("Adding single instance.",
+					testCase{
+						Instances:                    []string{"test-storage-1"},
+						ExpectedInstancesInCrashLoop: []string{"storage-1"},
+					}),
+				Entry("Adding multiple instances.",
+					testCase{
+						Instances:                    []string{"test-storage-1", "test-storage-2"},
+						ExpectedInstancesInCrashLoop: []string{"storage-1", "storage-2"},
+					}),
+			)
+
+			When("a process group was already in crash-loop", func() {
+				var kubeClient client.Client
+
+				BeforeEach(func() {
+					cluster.Spec.Buggify.CrashLoop = []string{"storage-1"}
+					scheme := runtime.NewScheme()
+					_ = clientgoscheme.AddToScheme(scheme)
+					_ = fdbv1beta2.AddToScheme(scheme)
+					kubeClient = fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
+				})
+
+				type testCase struct {
+					Instances                    []string
+					ExpectedInstancesInCrashLoop []string
+				}
+
+				DescribeTable("should add all targeted processes to crash-loop list",
+					func(tc testCase) {
+						err := updateCrashLoopList(kubeClient, clusterName, tc.Instances, namespace, false, false, false)
+						Expect(err).NotTo(HaveOccurred())
+
+						var resCluster fdbv1beta2.FoundationDBCluster
+						err = kubeClient.Get(ctx.Background(), client.ObjectKey{
+							Namespace: namespace,
+							Name:      clusterName,
+						}, &resCluster)
+						Expect(err).NotTo(HaveOccurred())
+						Expect(tc.ExpectedInstancesInCrashLoop).To(ContainElements(resCluster.Spec.Buggify.CrashLoop))
+						Expect(len(tc.ExpectedInstancesInCrashLoop)).To(BeNumerically("==", len(resCluster.Spec.Buggify.CrashLoop)))
+					},
+					Entry("Adding the same instance.",
+						testCase{
+							Instances:                    []string{"test-storage-1"},
+							ExpectedInstancesInCrashLoop: []string{"storage-1"},
+						}),
+					Entry("Adding different instance.",
+						testCase{
+							Instances:                    []string{"test-storage-2"},
+							ExpectedInstancesInCrashLoop: []string{"storage-1", "storage-2"},
+						}),
+					Entry("Adding multiple instances.",
+						testCase{
+							Instances:                    []string{"test-storage-2", "test-storage-3"},
+							ExpectedInstancesInCrashLoop: []string{"storage-1", "storage-2", "storage-3"},
+						}),
+				)
+			})
+		})
+
+		When("removing instances from crash-loop list from a cluster", func() {
+			var kubeClient client.Client
+
+			BeforeEach(func() {
+				cluster.Spec.Buggify.CrashLoop = []string{"storage-1", "storage-2", "storage-3"}
+				scheme := runtime.NewScheme()
+				_ = clientgoscheme.AddToScheme(scheme)
+				_ = fdbv1beta2.AddToScheme(scheme)
+				kubeClient = fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
+			})
+
+			type testCase struct {
+				Instances                    []string
+				ExpectedInstancesInCrashLoop []string
+			}
+
+			DescribeTable("should remove all targeted processes from the crash-loop list",
+				func(tc testCase) {
+					err := updateCrashLoopList(kubeClient, clusterName, tc.Instances, namespace, false, true, false)
+					Expect(err).NotTo(HaveOccurred())
+
+					var resCluster fdbv1beta2.FoundationDBCluster
+					err = kubeClient.Get(ctx.Background(), client.ObjectKey{
+						Namespace: namespace,
+						Name:      clusterName,
+					}, &resCluster)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(tc.ExpectedInstancesInCrashLoop).To(ContainElements(resCluster.Spec.Buggify.CrashLoop))
+					Expect(len(tc.ExpectedInstancesInCrashLoop)).To(BeNumerically("==", len(resCluster.Spec.Buggify.CrashLoop)))
+				},
+				Entry("Removing single instance.",
+					testCase{
+						Instances:                    []string{"test-storage-1"},
+						ExpectedInstancesInCrashLoop: []string{"storage-2", "storage-3"},
+					}),
+				Entry("Removing multiple instances.",
+					testCase{
+						Instances:                    []string{"test-storage-2", "test-storage-3"},
+						ExpectedInstancesInCrashLoop: []string{"storage-1"},
+					}),
+			)
+
+		})
+
+		When("clearing crash-loop list", func() {
+			var kubeClient client.Client
+
+			BeforeEach(func() {
+				cluster.Spec.Buggify.CrashLoop = []string{"storage-1", "storage-2", "storage-3"}
+				scheme := runtime.NewScheme()
+				_ = clientgoscheme.AddToScheme(scheme)
+				_ = fdbv1beta2.AddToScheme(scheme)
+				kubeClient = fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
+			})
+
+			It("should clear the crash-loop list", func() {
+				err := updateCrashLoopList(kubeClient, clusterName, []string{}, namespace, false, false, true)
+				Expect(err).NotTo(HaveOccurred())
+
+				var resCluster fdbv1beta2.FoundationDBCluster
+				err = kubeClient.Get(ctx.Background(), client.ObjectKey{
+					Namespace: namespace,
+					Name:      clusterName,
+				}, &resCluster)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(len(resCluster.Spec.Buggify.CrashLoop)).To(Equal(0))
+			})
+		})
+	})
+})
diff --git a/kubectl-fdb/cmd/buggify_empty-monitor_conf_test.go b/kubectl-fdb/cmd/buggify_empty-monitor_conf_test.go
new file mode 100644
index 00000000..703a0d05
--- /dev/null
+++ b/kubectl-fdb/cmd/buggify_empty-monitor_conf_test.go
@@ -0,0 +1,106 @@
+/*
+ * buggify_empty_monitor_conf_test.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 cmd
+
+import (
+	ctx "context"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+var _ = Describe("[plugin] buggify empty-monitor-conf instances command", func() {
+	clusterName := "test"
+	namespace := "test"
+
+	var cluster *fdbv1beta2.FoundationDBCluster
+
+	BeforeEach(func() {
+		cluster = &fdbv1beta2.FoundationDBCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      clusterName,
+				Namespace: namespace,
+			},
+			Spec: fdbv1beta2.FoundationDBClusterSpec{
+				ProcessCounts: fdbv1beta2.ProcessCounts{
+					Storage: 1,
+				},
+			},
+		}
+	})
+
+	When("running buggify empty-monitor-conf instances command", func() {
+		var kubeClient client.Client
+
+		When("should update empty-monitor-conf", func() {
+			BeforeEach(func() {
+				cluster.Spec.Buggify.EmptyMonitorConf = false
+				scheme := runtime.NewScheme()
+				_ = clientgoscheme.AddToScheme(scheme)
+				_ = fdbv1beta2.AddToScheme(scheme)
+				kubeClient = fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
+			})
+
+			It("setting monitor config to true", func() {
+				err := updateMonitorConf(kubeClient, clusterName, namespace, false, true)
+				Expect(err).NotTo(HaveOccurred())
+
+				var resCluster fdbv1beta2.FoundationDBCluster
+				err = kubeClient.Get(ctx.Background(), client.ObjectKey{
+					Namespace: namespace,
+					Name:      clusterName,
+				}, &resCluster)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(resCluster.Spec.Buggify.EmptyMonitorConf).To(BeTrue())
+			})
+		})
+
+		When("should update empty-monitor-conf", func() {
+			BeforeEach(func() {
+				cluster.Spec.Buggify.EmptyMonitorConf = true
+				scheme := runtime.NewScheme()
+				_ = clientgoscheme.AddToScheme(scheme)
+				_ = fdbv1beta2.AddToScheme(scheme)
+				kubeClient = fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
+			})
+
+			It("setting monitor config to false", func() {
+				err := updateMonitorConf(kubeClient, clusterName, namespace, false, false)
+				Expect(err).NotTo(HaveOccurred())
+
+				var resCluster fdbv1beta2.FoundationDBCluster
+				err = kubeClient.Get(ctx.Background(), client.ObjectKey{
+					Namespace: namespace,
+					Name:      clusterName,
+				}, &resCluster)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(resCluster.Spec.Buggify.EmptyMonitorConf).To(BeFalse())
+			})
+		})
+	})
+})
diff --git a/kubectl-fdb/cmd/buggify_no_schedule_test.go b/kubectl-fdb/cmd/buggify_no_schedule_test.go
new file mode 100644
index 00000000..e1788323
--- /dev/null
+++ b/kubectl-fdb/cmd/buggify_no_schedule_test.go
@@ -0,0 +1,213 @@
+/*
+ * buggify_no_schedule_test.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 cmd
+
+import (
+	ctx "context"
+
+	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("[plugin] buggify no-schedule instances command", func() {
+	clusterName := "test"
+	namespace := "test"
+
+	var cluster *fdbv1beta2.FoundationDBCluster
+
+	BeforeEach(func() {
+		cluster = &fdbv1beta2.FoundationDBCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      clusterName,
+				Namespace: namespace,
+			},
+			Spec: fdbv1beta2.FoundationDBClusterSpec{
+				ProcessCounts: fdbv1beta2.ProcessCounts{
+					Storage: 1,
+				},
+			},
+		}
+	})
+
+	When("running buggify no-schedule instances command", func() {
+		When("adding instances to no-schedule list from a cluster", func() {
+			type testCase struct {
+				Instances                     []string
+				ExpectedInstancesInNoSchedule []string
+			}
+
+			DescribeTable("should add all targeted processes to no-schedule list",
+				func(tc testCase) {
+					scheme := runtime.NewScheme()
+					_ = clientgoscheme.AddToScheme(scheme)
+					_ = fdbv1beta2.AddToScheme(scheme)
+					kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
+
+					err := updateNoScheduleList(kubeClient, clusterName, tc.Instances, namespace, false, false, false)
+					Expect(err).NotTo(HaveOccurred())
+
+					var resCluster fdbv1beta2.FoundationDBCluster
+					err = kubeClient.Get(ctx.Background(), client.ObjectKey{
+						Namespace: namespace,
+						Name:      clusterName,
+					}, &resCluster)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(tc.ExpectedInstancesInNoSchedule).To(ContainElements(resCluster.Spec.Buggify.NoSchedule))
+					Expect(len(tc.ExpectedInstancesInNoSchedule)).To(BeNumerically("==", len(resCluster.Spec.Buggify.NoSchedule)))
+				},
+				Entry("Adding single instance.",
+					testCase{
+						Instances:                     []string{"test-storage-1"},
+						ExpectedInstancesInNoSchedule: []string{"storage-1"},
+					}),
+				Entry("Adding multiple instances.",
+					testCase{
+						Instances:                     []string{"test-storage-1", "test-storage-2"},
+						ExpectedInstancesInNoSchedule: []string{"storage-1", "storage-2"},
+					}),
+			)
+
+			When("a process group is already in no-schedule", func() {
+				var kubeClient client.Client
+
+				BeforeEach(func() {
+					cluster.Spec.Buggify.NoSchedule = []string{"storage-1"}
+					scheme := runtime.NewScheme()
+					_ = clientgoscheme.AddToScheme(scheme)
+					_ = fdbv1beta2.AddToScheme(scheme)
+					kubeClient = fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
+				})
+
+				type testCase struct {
+					Instances                     []string
+					ExpectedInstancesInNoSchedule []string
+				}
+
+				DescribeTable("should add all targeted processes to no-schedule list",
+					func(tc testCase) {
+						err := updateNoScheduleList(kubeClient, clusterName, tc.Instances, namespace, false, false, false)
+						Expect(err).NotTo(HaveOccurred())
+
+						var resCluster fdbv1beta2.FoundationDBCluster
+						err = kubeClient.Get(ctx.Background(), client.ObjectKey{
+							Namespace: namespace,
+							Name:      clusterName,
+						}, &resCluster)
+						Expect(err).NotTo(HaveOccurred())
+						Expect(tc.ExpectedInstancesInNoSchedule).To(ContainElements(resCluster.Spec.Buggify.NoSchedule))
+						Expect(len(tc.ExpectedInstancesInNoSchedule)).To(BeNumerically("==", len(resCluster.Spec.Buggify.NoSchedule)))
+					},
+					Entry("Adding the same instance.",
+						testCase{
+							Instances:                     []string{"test-storage-1"},
+							ExpectedInstancesInNoSchedule: []string{"storage-1"},
+						}),
+					Entry("Adding different instance.",
+						testCase{
+							Instances:                     []string{"test-storage-2"},
+							ExpectedInstancesInNoSchedule: []string{"storage-1", "storage-2"},
+						}),
+					Entry("Adding multiple instances.",
+						testCase{
+							Instances:                     []string{"test-storage-2", "test-storage-3"},
+							ExpectedInstancesInNoSchedule: []string{"storage-1", "storage-2", "storage-3"},
+						}),
+				)
+			})
+		})
+
+		When("removing process group from no-schedule list from a cluster", func() {
+			var kubeClient client.Client
+
+			BeforeEach(func() {
+				cluster.Spec.Buggify.NoSchedule = []string{"storage-1", "storage-2", "storage-3"}
+				scheme := runtime.NewScheme()
+				_ = clientgoscheme.AddToScheme(scheme)
+				_ = fdbv1beta2.AddToScheme(scheme)
+				kubeClient = fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
+			})
+
+			type testCase struct {
+				Instances                     []string
+				ExpectedInstancesInNoSchedule []string
+			}
+
+			DescribeTable("should remove all targeted processes from the no-schedule list",
+				func(tc testCase) {
+					err := updateNoScheduleList(kubeClient, clusterName, tc.Instances, namespace, false, true, false)
+					Expect(err).NotTo(HaveOccurred())
+
+					var resCluster fdbv1beta2.FoundationDBCluster
+					err = kubeClient.Get(ctx.Background(), client.ObjectKey{
+						Namespace: namespace,
+						Name:      clusterName,
+					}, &resCluster)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(tc.ExpectedInstancesInNoSchedule).To(Equal(resCluster.Spec.Buggify.NoSchedule))
+					Expect(len(tc.ExpectedInstancesInNoSchedule)).To(BeNumerically("==", len(resCluster.Spec.Buggify.NoSchedule)))
+				},
+				Entry("Removing single instance.",
+					testCase{
+						Instances:                     []string{"test-storage-1"},
+						ExpectedInstancesInNoSchedule: []string{"storage-2", "storage-3"},
+					}),
+				Entry("Removing multiple instances.",
+					testCase{
+						Instances:                     []string{"test-storage-2", "test-storage-3"},
+						ExpectedInstancesInNoSchedule: []string{"storage-1"},
+					}),
+			)
+
+		})
+
+		When("clearing no-schedule list", func() {
+			var kubeClient client.Client
+
+			BeforeEach(func() {
+				cluster.Spec.Buggify.NoSchedule = []string{"storage-1", "storage-2", "storage-3"}
+				scheme := runtime.NewScheme()
+				_ = clientgoscheme.AddToScheme(scheme)
+				_ = fdbv1beta2.AddToScheme(scheme)
+				kubeClient = fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
+			})
+
+			It("should clear the no-schedule list", func() {
+				err := updateNoScheduleList(kubeClient, clusterName, nil, namespace, false, false, true)
+				Expect(err).NotTo(HaveOccurred())
+
+				var resCluster fdbv1beta2.FoundationDBCluster
+				err = kubeClient.Get(ctx.Background(), client.ObjectKey{
+					Namespace: namespace,
+					Name:      clusterName,
+				}, &resCluster)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(len(resCluster.Spec.Buggify.NoSchedule)).To(Equal(0))
+			})
+		})
+	})
+})
diff --git a/kubectl-fdb/cmd/k8s_client_test.go b/kubectl-fdb/cmd/k8s_client_test.go
index 0033870b..8114ba36 100644
--- a/kubectl-fdb/cmd/k8s_client_test.go
+++ b/kubectl-fdb/cmd/k8s_client_test.go
@@ -182,4 +182,85 @@ var _ = Describe("[plugin] using the Kubernetes client", func() {
 				}),
 		)
 	})
+
+	When("getting the process groups IDs from Pods", func() {
+		clusterName := "test"
+		namespace := "test"
+		var cluster fdbv1beta2.FoundationDBCluster
+
+		When("the cluster doesn't have a prefix", func() {
+			BeforeEach(func() {
+				cluster = fdbv1beta2.FoundationDBCluster{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      clusterName,
+						Namespace: namespace,
+					},
+					Spec: fdbv1beta2.FoundationDBClusterSpec{
+						ProcessCounts: fdbv1beta2.ProcessCounts{
+							Storage: 1,
+						},
+					},
+				}
+			})
+
+			DescribeTable("should get all process groups IDs",
+				func(podNames []string, expected []string) {
+					instances, err := getProcessGroupIDsFromPodName(&cluster, podNames)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(instances).To(ContainElements(expected))
+					Expect(len(instances)).To(BeNumerically("==", len(expected)))
+				},
+				Entry("Filter one instance",
+					[]string{"test-storage-1"},
+					[]string{"storage-1"},
+				),
+				Entry("Filter two instances",
+					[]string{"test-storage-1", "test-storage-2"},
+					[]string{"storage-1", "storage-2"},
+				),
+				Entry("Filter no instance",
+					[]string{""},
+					[]string{},
+				),
+			)
+		})
+
+		When("the cluster has a prefix", func() {
+			BeforeEach(func() {
+				cluster = fdbv1beta2.FoundationDBCluster{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      clusterName,
+						Namespace: namespace,
+					},
+					Spec: fdbv1beta2.FoundationDBClusterSpec{
+						ProcessGroupIDPrefix: "banana",
+						ProcessCounts: fdbv1beta2.ProcessCounts{
+							Storage: 1,
+						},
+					},
+				}
+			})
+
+			DescribeTable("should get all process groups IDs",
+				func(podNames []string, expected []string) {
+					instances, err := getProcessGroupIDsFromPodName(&cluster, podNames)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(instances).To(ContainElements(expected))
+					Expect(len(instances)).To(BeNumerically("==", len(expected)))
+				},
+				Entry("Filter one instance",
+					[]string{"test-storage-1"},
+					[]string{"banana-storage-1"},
+				),
+				Entry("Filter two instances",
+					[]string{"test-storage-1", "test-storage-2"},
+					[]string{"banana-storage-1", "banana-storage-2"},
+				),
+				Entry("Filter no instance",
+					[]string{""},
+					[]string{},
+				),
+			)
+		})
+	})
 })
diff --git a/kubectl-fdb/cmd/remove_process_group_test.go b/kubectl-fdb/cmd/remove_process_group_test.go
index b38054cf..0444d6a0 100644
--- a/kubectl-fdb/cmd/remove_process_group_test.go
+++ b/kubectl-fdb/cmd/remove_process_group_test.go
@@ -24,7 +24,6 @@ import (
 	ctx "context"
 
 	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
-	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
@@ -35,7 +34,7 @@ import (
 	. "github.com/onsi/gomega"
 )
 
-var _ = Describe("[plugin] remove instances command", func() {
+var _ = Describe("[plugin] remove process groups command", func() {
 	clusterName := "test"
 	namespace := "test"
 
@@ -55,99 +54,20 @@ var _ = Describe("[plugin] remove instances command", func() {
 		}
 	})
 
-	When("running remove instances command", func() {
-		When("getting the instance IDs from Pods", func() {
-			var podList corev1.PodList
-
-			BeforeEach(func() {
-				podList = corev1.PodList{
-					Items: []corev1.Pod{
-						{
-							ObjectMeta: metav1.ObjectMeta{
-								Name:      "instance-1",
-								Namespace: namespace,
-								Labels: map[string]string{
-									fdbv1beta2.FDBProcessClassLabel:   string(fdbv1beta2.ProcessClassStorage),
-									fdbv1beta2.FDBClusterLabel:        clusterName,
-									fdbv1beta2.FDBProcessGroupIDLabel: "storage-1",
-								},
-							},
-						},
-						{
-							ObjectMeta: metav1.ObjectMeta{
-								Name:      "instance-2",
-								Namespace: namespace,
-								Labels: map[string]string{
-									fdbv1beta2.FDBProcessClassLabel:   string(fdbv1beta2.ProcessClassStorage),
-									fdbv1beta2.FDBClusterLabel:        clusterName,
-									fdbv1beta2.FDBProcessGroupIDLabel: "storage-2",
-								},
-							},
-						},
-					},
-				}
-			})
-
-			type testCase struct {
-				Instances         []string
-				ExpectedInstances []string
-			}
-
-			DescribeTable("should get all instance IDs",
-				func(input testCase) {
-					scheme := runtime.NewScheme()
-					_ = clientgoscheme.AddToScheme(scheme)
-					_ = fdbv1beta2.AddToScheme(scheme)
-					kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster, &podList).Build()
-
-					instances, err := getProcessGroupIDsFromPod(kubeClient, clusterName, input.Instances, namespace)
-					Expect(err).NotTo(HaveOccurred())
-					Expect(input.ExpectedInstances).To(Equal(instances))
-				},
-				Entry("Filter one instance",
-					testCase{
-						Instances:         []string{"instance-1"},
-						ExpectedInstances: []string{"storage-1"},
-					}),
-				Entry("Filter two instances",
-					testCase{
-						Instances:         []string{"instance-1", "instance-2"},
-						ExpectedInstances: []string{"storage-1", "storage-2"},
-					}),
-				Entry("Filter no instance",
-					testCase{
-						Instances:         []string{""},
-						ExpectedInstances: []string{},
-					}),
-			)
-		})
-
-		When("removing instances from a cluster", func() {
-			var podList corev1.PodList
-
+	When("running remove process groups command", func() {
+		When("removing process groups from a cluster", func() {
 			BeforeEach(func() {
 				cluster.Status = fdbv1beta2.FoundationDBClusterStatus{
 					ProcessGroups: []*fdbv1beta2.ProcessGroupStatus{
 						{
-							ProcessGroupID: "failed",
+							ProcessGroupID: "storage-42",
 							Addresses:      []string{"1.2.3.4"},
 							ProcessGroupConditions: []*fdbv1beta2.ProcessGroupCondition{
 								fdbv1beta2.NewProcessGroupCondition(fdbv1beta2.MissingProcesses),
 							},
 						},
-					},
-				}
-				podList = corev1.PodList{
-					Items: []corev1.Pod{
 						{
-							ObjectMeta: metav1.ObjectMeta{
-								Name:      "instance-1",
-								Namespace: namespace,
-								Labels: map[string]string{
-									fdbv1beta2.FDBProcessClassLabel: string(fdbv1beta2.ProcessClassStorage),
-									fdbv1beta2.FDBClusterLabel:      clusterName,
-								},
-							},
+							ProcessGroupID: "storage-1",
 						},
 					},
 				}
@@ -156,7 +76,6 @@ var _ = Describe("[plugin] remove instances command", func() {
 			type testCase struct {
 				Instances                                 []string
 				WithExclusion                             bool
-				WithShrink                                bool
 				ExpectedInstancesToRemove                 []string
 				ExpectedInstancesToRemoveWithoutExclusion []string
 				ExpectedProcessCounts                     fdbv1beta2.ProcessCounts
@@ -168,9 +87,9 @@ var _ = Describe("[plugin] remove instances command", func() {
 					scheme := runtime.NewScheme()
 					_ = clientgoscheme.AddToScheme(scheme)
 					_ = fdbv1beta2.AddToScheme(scheme)
-					kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster, &podList).Build()
+					kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
 
-					err := replaceProcessGroups(kubeClient, clusterName, tc.Instances, namespace, tc.WithExclusion, tc.WithShrink, false, tc.RemoveAllFailed)
+					err := replaceProcessGroups(kubeClient, clusterName, tc.Instances, namespace, tc.WithExclusion, false, tc.RemoveAllFailed, false)
 					Expect(err).NotTo(HaveOccurred())
 
 					var resCluster fdbv1beta2.FoundationDBCluster
@@ -185,60 +104,32 @@ var _ = Describe("[plugin] remove instances command", func() {
 					Expect(len(tc.ExpectedInstancesToRemoveWithoutExclusion)).To(BeNumerically("==", len(resCluster.Spec.ProcessGroupsToRemoveWithoutExclusion)))
 					Expect(tc.ExpectedProcessCounts.Storage).To(Equal(resCluster.Spec.ProcessCounts.Storage))
 				},
-				Entry("Remove instance with exclusion",
+				Entry("Remove process group with exclusion",
 					testCase{
-						Instances:                 []string{"instance-1"},
+						Instances:                 []string{"test-storage-1"},
 						WithExclusion:             true,
-						WithShrink:                false,
-						ExpectedInstancesToRemove: []string{"instance-1"},
+						ExpectedInstancesToRemove: []string{"storage-1"},
 						ExpectedInstancesToRemoveWithoutExclusion: []string{},
 						ExpectedProcessCounts: fdbv1beta2.ProcessCounts{
 							Storage: 1,
 						},
 						RemoveAllFailed: false,
 					}),
-				Entry("Remove instance without exclusion",
+				Entry("Remove process group without exclusion",
 					testCase{
-						Instances:                 []string{"instance-1"},
+						Instances:                 []string{"test-storage-1"},
 						WithExclusion:             false,
-						WithShrink:                false,
 						ExpectedInstancesToRemove: []string{},
-						ExpectedInstancesToRemoveWithoutExclusion: []string{"instance-1"},
+						ExpectedInstancesToRemoveWithoutExclusion: []string{"storage-1"},
 						ExpectedProcessCounts: fdbv1beta2.ProcessCounts{
 							Storage: 1,
 						},
 					}),
-				Entry("Remove instance with exclusion and shrink",
-					testCase{
-						Instances:                 []string{"instance-1"},
-						WithExclusion:             true,
-						WithShrink:                true,
-						ExpectedInstancesToRemove: []string{"instance-1"},
-						ExpectedInstancesToRemoveWithoutExclusion: []string{},
-						ExpectedProcessCounts: fdbv1beta2.ProcessCounts{
-							Storage: 0,
-						},
-						RemoveAllFailed: false,
-					}),
-
-				Entry("Remove instance without exclusion and shrink",
-					testCase{
-						Instances:                 []string{"instance-1"},
-						WithExclusion:             false,
-						WithShrink:                true,
-						ExpectedInstancesToRemove: []string{},
-						ExpectedInstancesToRemoveWithoutExclusion: []string{"instance-1"},
-						ExpectedProcessCounts: fdbv1beta2.ProcessCounts{
-							Storage: 0,
-						},
-						RemoveAllFailed: false,
-					}),
 				Entry("Remove all failed instances",
 					testCase{
-						Instances:                 []string{"failed"},
+						Instances:                 []string{"test-storage-42"},
 						WithExclusion:             true,
-						WithShrink:                false,
-						ExpectedInstancesToRemove: []string{"failed"},
+						ExpectedInstancesToRemove: []string{"storage-42"},
 						ExpectedInstancesToRemoveWithoutExclusion: []string{},
 						ExpectedProcessCounts: fdbv1beta2.ProcessCounts{
 							Storage: 1,
@@ -247,20 +138,20 @@ var _ = Describe("[plugin] remove instances command", func() {
 					}),
 			)
 
-			When("a procress group was already marked for removal", func() {
+			When("a process group was already marked for removal", func() {
 				var kubeClient client.Client
 
 				BeforeEach(func() {
-					cluster.Spec.ProcessGroupsToRemove = []string{"instance-1"}
+					cluster.Spec.ProcessGroupsToRemove = []string{"storage-1"}
 					scheme := runtime.NewScheme()
 					_ = clientgoscheme.AddToScheme(scheme)
 					_ = fdbv1beta2.AddToScheme(scheme)
-					kubeClient = fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster, &podList).Build()
+					kubeClient = fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(cluster).Build()
 				})
 
 				When("adding the same process group to the removal list without exclusion", func() {
 					It("should add the process group to the removal without exclusion list", func() {
-						removals := []string{"instance-1"}
+						removals := []string{"test-storage-1"}
 						err := replaceProcessGroups(kubeClient, clusterName, removals, namespace, false, false, false, false)
 						Expect(err).NotTo(HaveOccurred())
 
@@ -271,16 +162,16 @@ var _ = Describe("[plugin] remove instances command", func() {
 						}, &resCluster)
 
 						Expect(err).NotTo(HaveOccurred())
-						Expect(resCluster.Spec.ProcessGroupsToRemove).To(ContainElements(removals))
+						Expect(resCluster.Spec.ProcessGroupsToRemove).To(ContainElements("storage-1"))
 						Expect(len(resCluster.Spec.ProcessGroupsToRemove)).To(BeNumerically("==", len(removals)))
-						Expect(resCluster.Spec.ProcessGroupsToRemoveWithoutExclusion).To(ContainElements(removals))
+						Expect(resCluster.Spec.ProcessGroupsToRemoveWithoutExclusion).To(ContainElements("storage-1"))
 						Expect(len(resCluster.Spec.ProcessGroupsToRemoveWithoutExclusion)).To(BeNumerically("==", len(removals)))
 					})
 				})
 
 				When("adding the same process group to the removal list", func() {
 					It("should add the process group to the removal without exclusion list", func() {
-						removals := []string{"instance-1"}
+						removals := []string{"test-storage-1"}
 						err := replaceProcessGroups(kubeClient, clusterName, removals, namespace, true, false, false, false)
 						Expect(err).NotTo(HaveOccurred())
 
@@ -291,7 +182,7 @@ var _ = Describe("[plugin] remove instances command", func() {
 						}, &resCluster)
 
 						Expect(err).NotTo(HaveOccurred())
-						Expect(resCluster.Spec.ProcessGroupsToRemove).To(ContainElements(removals))
+						Expect(resCluster.Spec.ProcessGroupsToRemove).To(ContainElements("storage-1"))
 						Expect(len(resCluster.Spec.ProcessGroupsToRemove)).To(BeNumerically("==", len(removals)))
 						Expect(len(resCluster.Spec.ProcessGroupsToRemoveWithoutExclusion)).To(BeNumerically("==", 0))
 					})