diff --git a/.github/workflows/kindIntegTest.yml b/.github/workflows/kindIntegTest.yml
index e3efd98f..da2fd8f3 100644
--- a/.github/workflows/kindIntegTest.yml
+++ b/.github/workflows/kindIntegTest.yml
@@ -117,12 +117,12 @@ jobs:
     strategy:
       matrix:
         version:
-          - "6.8.49"
+          - "6.8.50"
         integration_test:
           - cdc_successful
         include:
-          - version: 6.8.49
-            serverImage: datastax/dse-mgmtapi-6_8:6.8.49-ubi8 # DSE 6.8.49
+          - version: 6.8.50
+            serverImage: datastax/dse-mgmtapi-6_8:6.8.50-ubi8 # DSE 6.8.50
             serverType: dse
             integration_test: "cdc_successful"
       fail-fast: true
@@ -155,7 +155,7 @@ jobs:
     strategy:
       matrix:
         version:
-          - "4.1.5"
+          - "4.1.6"
         integration_test:
           # Single worker tests:
           - additional_serviceoptions
@@ -202,12 +202,6 @@ jobs:
           #- config_fql # OSS only
           - decommission_dc
         # - stop_resume_scale_up # Odd insufficient CPU issues in kind+GHA
-        include:
-          - version: 4.1.5
-            serverImage: michaelburman290/cass-management-api:4.1.5-ubi8 # Modified version of cass-management-api
-            serverType: cassandra
-            integration_test: "smoke_test_read_only_fs"
-
         # let other tests continue to run
       # even if one fails
       fail-fast: false
@@ -239,9 +233,9 @@ jobs:
         version:
           - "3.11.17"
           - "4.0.13"
-          - "4.1.5"
+          - "4.1.6"
           - "6.8.50"
-          - "6.9.0"
+          - "6.9.2"
           - "1.0.0"
         integration_test:
           - test_all_the_things
@@ -250,8 +244,8 @@ jobs:
             serverImage: datastax/dse-mgmtapi-6_8:6.8.50-ubi8 # DSE 6.8.50
             serverType: dse
             integration_test: "test_all_the_things"
-          - version: 6.9.0
-            serverImage: datastax/dse-mgmtapi-6_8:6.9.0-ubi8 # DSE 6.9.0
+          - version: 6.9.2
+            serverImage: datastax/dse-mgmtapi-6_8:6.9.2-ubi8 # DSE 6.9.2
             serverType: dse
             integration_test: "test_all_the_things"
           - version: 1.0.0
@@ -288,7 +282,7 @@ jobs:
     strategy:
       matrix:
         version:
-          - "4.1.5"
+          - "4.1.6"
         integration_test:
           - pvc_expansion
       fail-fast: true
diff --git a/apis/cassandra/v1beta1/cassandradatacenter_types.go b/apis/cassandra/v1beta1/cassandradatacenter_types.go
index ec9aa20e..8a628e93 100644
--- a/apis/cassandra/v1beta1/cassandradatacenter_types.go
+++ b/apis/cassandra/v1beta1/cassandradatacenter_types.go
@@ -485,6 +485,9 @@ type CassandraDatacenterStatus struct {
 	// This field is used to perform validation checks preventing a user from changing the override
 	// +optional
 	DatacenterName *string `json:"datacenterName,omitempty"`
+
+	// +optional
+	MetadataVersion int64 `json:"metadataVersion,omitempty"`
 }
 
 // CassandraDatacenter is the Schema for the cassandradatacenters API
@@ -596,7 +599,7 @@ func (dc *CassandraDatacenter) SetCondition(condition DatacenterCondition) {
 // GetDatacenterLabels ...
 func (dc *CassandraDatacenter) GetDatacenterLabels() map[string]string {
 	labels := dc.GetClusterLabels()
-	labels[DatacenterLabel] = CleanLabelValue(dc.DatacenterName())
+	labels[DatacenterLabel] = CleanLabelValue(dc.Name)
 	return labels
 }
 
@@ -661,19 +664,19 @@ func (dc *CassandraDatacenter) GetSeedServiceName() string {
 }
 
 func (dc *CassandraDatacenter) GetAdditionalSeedsServiceName() string {
-	return CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.SanitizedName() + "-additional-seed-service"
+	return CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.LabelResourceName() + "-additional-seed-service"
 }
 
 func (dc *CassandraDatacenter) GetAllPodsServiceName() string {
-	return CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.SanitizedName() + "-all-pods-service"
+	return CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.LabelResourceName() + "-all-pods-service"
 }
 
 func (dc *CassandraDatacenter) GetDatacenterServiceName() string {
-	return CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.SanitizedName() + "-service"
+	return CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.LabelResourceName() + "-service"
 }
 
 func (dc *CassandraDatacenter) GetNodePortServiceName() string {
-	return CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.SanitizedName() + "-node-port-service"
+	return CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.LabelResourceName() + "-node-port-service"
 }
 
 func (dc *CassandraDatacenter) ShouldGenerateSuperuserSecret() bool {
@@ -970,9 +973,17 @@ func SplitRacks(nodeCount, rackCount int) []int {
 	return topology
 }
 
-// SanitizedName returns a sanitized version of the name returned by DatacenterName()
-func (dc *CassandraDatacenter) SanitizedName() string {
-	return CleanupForKubernetes(dc.DatacenterName())
+func (dc *CassandraDatacenter) DatacenterNameStatus() bool {
+	return dc.Status.DatacenterName != nil
+}
+
+// LabelResourceName returns a sanitized version of the name returned by DatacenterName()
+func (dc *CassandraDatacenter) LabelResourceName() string {
+	// If existing cluster, return dc.DatacenterName() else return dc.Name
+	if dc.DatacenterNameStatus() {
+		return CleanupForKubernetes(*dc.Status.DatacenterName)
+	}
+	return CleanupForKubernetes(dc.Name)
 }
 
 // DatacenterName returns the Cassandra DC name override if it exists,
diff --git a/apis/cassandra/v1beta1/cassandradatacenter_types_test.go b/apis/cassandra/v1beta1/cassandradatacenter_types_test.go
index 266897c2..7c8a3ccb 100644
--- a/apis/cassandra/v1beta1/cassandradatacenter_types_test.go
+++ b/apis/cassandra/v1beta1/cassandradatacenter_types_test.go
@@ -4,7 +4,9 @@ import (
 	"encoding/json"
 	"testing"
 
+	"github.com/Jeffail/gabs/v2"
 	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 var internodeEnabledAll = `
@@ -125,3 +127,50 @@ func TestUseClientImage(t *testing.T) {
 		}
 	}
 }
+
+func TestDatacenterNoOverrideConfig(t *testing.T) {
+	assert := assert.New(t)
+	dc := CassandraDatacenter{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "dc1",
+		},
+		Spec: CassandraDatacenterSpec{
+			ClusterName: "cluster1",
+		},
+	}
+
+	config, err := dc.GetConfigAsJSON(dc.Spec.Config)
+	assert.NoError(err)
+
+	container, err := gabs.ParseJSON([]byte(config))
+	assert.NoError(err)
+
+	dataCenterInfo := container.ChildrenMap()["datacenter-info"]
+	assert.NotEmpty(dataCenterInfo)
+	assert.Equal(dc.Name, dataCenterInfo.ChildrenMap()["name"].Data().(string))
+	assert.Equal(dc.DatacenterName(), dc.Name)
+}
+
+func TestDatacenterOverrideInConfig(t *testing.T) {
+	assert := assert.New(t)
+	dc := CassandraDatacenter{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "dc1",
+		},
+		Spec: CassandraDatacenterSpec{
+			ClusterName:    "cluster1",
+			DatacenterName: "Home_Dc",
+		},
+	}
+
+	config, err := dc.GetConfigAsJSON(dc.Spec.Config)
+	assert.NoError(err)
+
+	container, err := gabs.ParseJSON([]byte(config))
+	assert.NoError(err)
+
+	dataCenterInfo := container.ChildrenMap()["datacenter-info"]
+	assert.NotEmpty(dataCenterInfo)
+	assert.Equal(dc.Spec.DatacenterName, dataCenterInfo.ChildrenMap()["name"].Data().(string))
+	assert.Equal(dc.DatacenterName(), dc.Spec.DatacenterName)
+}
diff --git a/config/crd/bases/cassandra.datastax.com_cassandradatacenters.yaml b/config/crd/bases/cassandra.datastax.com_cassandradatacenters.yaml
index 2d7a9493..83d5ad0e 100644
--- a/config/crd/bases/cassandra.datastax.com_cassandradatacenters.yaml
+++ b/config/crd/bases/cassandra.datastax.com_cassandradatacenters.yaml
@@ -11221,6 +11221,9 @@ spec:
                   with the management API
                 format: date-time
                 type: string
+              metadataVersion:
+                format: int64
+                type: integer
               nodeReplacements:
                 items:
                   type: string
diff --git a/config/manager/image_config.yaml b/config/manager/image_config.yaml
index 04aed7f0..2e85d348 100644
--- a/config/manager/image_config.yaml
+++ b/config/manager/image_config.yaml
@@ -18,7 +18,7 @@ defaults:
   # Note, postfix is ignored if repository is not set
   cassandra:
     repository: "k8ssandra/cass-management-api"
-    suffix: "-ubi8"
+    suffix: "-ubi"
   dse:
     repository: "datastax/dse-mgmtapi-6_8"
     suffix: "-ubi8"
diff --git a/internal/controllers/cassandra/cassandradatacenter_controller_test.go b/internal/controllers/cassandra/cassandradatacenter_controller_test.go
index 2b0fe192..f172a7c2 100644
--- a/internal/controllers/cassandra/cassandradatacenter_controller_test.go
+++ b/internal/controllers/cassandra/cassandradatacenter_controller_test.go
@@ -155,8 +155,9 @@ var _ = Describe("CassandraDatacenter tests", func() {
 				refreshDatacenter(ctx, &dc)
 
 				By("Updating the size to 3")
+				patch := client.MergeFrom(dc.DeepCopy())
 				dc.Spec.Size = 3
-				Expect(k8sClient.Update(ctx, &dc)).To(Succeed())
+				Expect(k8sClient.Patch(ctx, &dc, patch)).To(Succeed())
 
 				waitForDatacenterCondition(ctx, dcName, cassdcapi.DatacenterScalingUp, corev1.ConditionTrue).Should(Succeed())
 				waitForDatacenterProgress(ctx, dcName, cassdcapi.ProgressUpdating).Should(Succeed())
diff --git a/internal/controllers/control/cassandratask_controller.go b/internal/controllers/control/cassandratask_controller.go
index 45bd78fd..4d76c7d8 100644
--- a/internal/controllers/control/cassandratask_controller.go
+++ b/internal/controllers/control/cassandratask_controller.go
@@ -201,7 +201,7 @@ func (r *CassandraTaskReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 		return ctrl.Result{}, errors.Wrapf(err, "unable to fetch target CassandraDatacenter: %s", cassTask.Spec.Datacenter)
 	}
 
-	logger = log.FromContext(ctx, "datacenterName", dc.SanitizedName(), "clusterName", dc.Spec.ClusterName)
+	logger = log.FromContext(ctx, "datacenterName", dc.LabelResourceName(), "clusterName", dc.Spec.ClusterName)
 	log.IntoContext(ctx, logger)
 
 	// If we're active, we can proceed - otherwise verify if we're allowed to run
diff --git a/pkg/images/images_test.go b/pkg/images/images_test.go
index 13582e19..07eb9a69 100644
--- a/pkg/images/images_test.go
+++ b/pkg/images/images_test.go
@@ -94,7 +94,7 @@ func TestDefaultImageConfigParsing(t *testing.T) {
 
 	path, err = GetCassandraImage("cassandra", "4.1.4")
 	assert.NoError(err)
-	assert.Equal("k8ssandra/cass-management-api:4.1.4-ubi8", path)
+	assert.Equal("k8ssandra/cass-management-api:4.1.4-ubi", path)
 }
 
 func TestImageConfigParsing(t *testing.T) {
diff --git a/pkg/reconciliation/construct_podtemplatespec.go b/pkg/reconciliation/construct_podtemplatespec.go
index 2dfc83e7..24006730 100644
--- a/pkg/reconciliation/construct_podtemplatespec.go
+++ b/pkg/reconciliation/construct_podtemplatespec.go
@@ -351,7 +351,7 @@ func addVolumes(dc *api.CassandraDatacenter, baseTemplate *corev1.PodTemplateSpe
 			Name: "encryption-cred-storage",
 			VolumeSource: corev1.VolumeSource{
 				Secret: &corev1.SecretVolumeSource{
-					SecretName: fmt.Sprintf("%s-keystore", dc.SanitizedName()),
+					SecretName: fmt.Sprintf("%s-keystore", dc.LabelResourceName()),
 				},
 			},
 		}
@@ -558,7 +558,7 @@ func getConfigDataEnVars(dc *api.CassandraDatacenter) ([]corev1.EnvVar, error) {
 			return envVars, nil
 		}
 
-		return nil, fmt.Errorf("datacenter %s is missing %s annotation", dc.SanitizedName(), api.ConfigHashAnnotation)
+		return nil, fmt.Errorf("datacenter %s is missing %s annotation", dc.LabelResourceName(), api.ConfigHashAnnotation)
 	}
 
 	configData, err := dc.GetConfigAsJSON(dc.Spec.Config)
diff --git a/pkg/reconciliation/construct_podtemplatespec_test.go b/pkg/reconciliation/construct_podtemplatespec_test.go
index c116a73e..6e9a5a0d 100644
--- a/pkg/reconciliation/construct_podtemplatespec_test.go
+++ b/pkg/reconciliation/construct_podtemplatespec_test.go
@@ -1548,7 +1548,7 @@ func Test_makeImage(t *testing.T) {
 				serverType:    "cassandra",
 				serverVersion: "3.11.10",
 			},
-			want:      "localhost:5000/k8ssandra/cass-management-api:3.11.10-ubi8",
+			want:      "localhost:5000/k8ssandra/cass-management-api:3.11.10-ubi",
 			errString: "",
 		},
 		{
diff --git a/pkg/reconciliation/construct_statefulset.go b/pkg/reconciliation/construct_statefulset.go
index f9227710..a8ed4fbc 100644
--- a/pkg/reconciliation/construct_statefulset.go
+++ b/pkg/reconciliation/construct_statefulset.go
@@ -26,7 +26,7 @@ func newNamespacedNameForStatefulSet(
 	dc *api.CassandraDatacenter,
 	rackName string) types.NamespacedName {
 
-	name := api.CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.SanitizedName() + "-" + api.CleanupSubdomain(rackName) + "-sts"
+	name := api.CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.LabelResourceName() + "-" + api.CleanupSubdomain(rackName) + "-sts"
 	ns := dc.Namespace
 
 	return types.NamespacedName{
diff --git a/pkg/reconciliation/construct_statefulset_test.go b/pkg/reconciliation/construct_statefulset_test.go
index a0530b9e..f53983c1 100644
--- a/pkg/reconciliation/construct_statefulset_test.go
+++ b/pkg/reconciliation/construct_statefulset_test.go
@@ -641,7 +641,7 @@ func Test_newStatefulSetForCassandraDatacenter_dcNameOverride(t *testing.T) {
 		oplabels.NameLabel:      oplabels.NameLabelValue,
 		oplabels.CreatedByLabel: oplabels.CreatedByLabelValue,
 		oplabels.VersionLabel:   "4.0.1",
-		api.DatacenterLabel:     "MySuperDC",
+		api.DatacenterLabel:     "dc1",
 		api.ClusterLabel:        "piclem",
 		api.RackLabel:           dc.Spec.Racks[0].Name,
 	}
@@ -652,7 +652,7 @@ func Test_newStatefulSetForCassandraDatacenter_dcNameOverride(t *testing.T) {
 		oplabels.NameLabel:      oplabels.NameLabelValue,
 		oplabels.CreatedByLabel: oplabels.CreatedByLabelValue,
 		oplabels.VersionLabel:   "4.0.1",
-		api.DatacenterLabel:     "MySuperDC",
+		api.DatacenterLabel:     "dc1",
 		api.ClusterLabel:        "piclem",
 		api.RackLabel:           dc.Spec.Racks[0].Name,
 		api.CassNodeState:       stateReadyToStart,
diff --git a/pkg/reconciliation/constructor.go b/pkg/reconciliation/constructor.go
index edb29ab2..dd06171b 100644
--- a/pkg/reconciliation/constructor.go
+++ b/pkg/reconciliation/constructor.go
@@ -31,7 +31,7 @@ func newPodDisruptionBudgetForDatacenter(dc *api.CassandraDatacenter) *policyv1.
 
 	pdb := &policyv1.PodDisruptionBudget{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:        dc.SanitizedName() + "-pdb",
+			Name:        dc.LabelResourceName() + "-pdb",
 			Namespace:   dc.Namespace,
 			Labels:      labels,
 			Annotations: anns,
@@ -62,8 +62,11 @@ func setOperatorProgressStatus(rc *ReconciliationContext, newState api.ProgressS
 	rc.Datacenter.Status.CassandraOperatorProgress = newState
 
 	if newState == api.ProgressReady {
+		if rc.Datacenter.Status.MetadataVersion < 1 {
+			rc.Datacenter.Status.MetadataVersion = 1
+		}
 		if rc.Datacenter.Status.DatacenterName == nil {
-			rc.Datacenter.Status.DatacenterName = &rc.Datacenter.Spec.DatacenterName
+			rc.Datacenter.Status.DatacenterName = &rc.Datacenter.Name
 		}
 	}
 	if err := rc.Client.Status().Patch(rc.Ctx, rc.Datacenter, patch); err != nil {
@@ -73,17 +76,6 @@ func setOperatorProgressStatus(rc *ReconciliationContext, newState api.ProgressS
 
 	monitoring.UpdateOperatorDatacenterProgressStatusMetric(rc.Datacenter, newState)
 
-	// The allow-upgrade=once annotation is temporary and should be removed after first successful reconcile
-	if metav1.HasAnnotation(rc.Datacenter.ObjectMeta, api.UpdateAllowedAnnotation) && rc.Datacenter.Annotations[api.UpdateAllowedAnnotation] == string(api.AllowUpdateOnce) {
-		// remove the annotation
-		patch = client.MergeFrom(rc.Datacenter.DeepCopy())
-		delete(rc.Datacenter.ObjectMeta.Annotations, api.UpdateAllowedAnnotation)
-		if err := rc.Client.Patch(rc.Ctx, rc.Datacenter, patch); err != nil {
-			rc.ReqLogger.Error(err, "error removing the allow-upgrade=once annotation")
-			return err
-		}
-	}
-
 	return nil
 }
 
@@ -101,5 +93,16 @@ func setDatacenterStatus(rc *ReconciliationContext) error {
 		return err
 	}
 
+	// The allow-upgrade=once annotation is temporary and should be removed after first successful reconcile
+	if metav1.HasAnnotation(rc.Datacenter.ObjectMeta, api.UpdateAllowedAnnotation) && rc.Datacenter.Annotations[api.UpdateAllowedAnnotation] == string(api.AllowUpdateOnce) {
+		// remove the annotation
+		patch := client.MergeFrom(rc.Datacenter.DeepCopy())
+		delete(rc.Datacenter.ObjectMeta.Annotations, api.UpdateAllowedAnnotation)
+		if err := rc.Client.Patch(rc.Ctx, rc.Datacenter, patch); err != nil {
+			rc.ReqLogger.Error(err, "error removing the allow-upgrade=once annotation")
+			return err
+		}
+	}
+
 	return nil
 }
diff --git a/pkg/reconciliation/context.go b/pkg/reconciliation/context.go
index c3e7dbb0..1069103b 100644
--- a/pkg/reconciliation/context.go
+++ b/pkg/reconciliation/context.go
@@ -92,7 +92,7 @@ func CreateReconciliationContext(
 	}
 
 	rc.ReqLogger = rc.ReqLogger.
-		WithValues("datacenterName", dc.SanitizedName()).
+		WithValues("datacenterName", dc.LabelResourceName()).
 		WithValues("clusterName", dc.Spec.ClusterName)
 
 	log.IntoContext(ctx, rc.ReqLogger)
@@ -146,8 +146,8 @@ func (rc *ReconciliationContext) validateDatacenterNameConflicts() []error {
 		errs = append(errs, fmt.Errorf("failed to list CassandraDatacenters in namespace %s: %w", dc.Namespace, err))
 	} else {
 		for _, existingDc := range cassandraDatacenters.Items {
-			if existingDc.SanitizedName() == dc.SanitizedName() && existingDc.Name != dc.Name {
-				errs = append(errs, fmt.Errorf("datacenter name/override %s/%s is already in use by CassandraDatacenter %s/%s", dc.Name, dc.SanitizedName(), existingDc.Name, existingDc.SanitizedName()))
+			if existingDc.LabelResourceName() == dc.LabelResourceName() && existingDc.Name != dc.Name {
+				errs = append(errs, fmt.Errorf("datacenter name/override %s/%s is already in use by CassandraDatacenter %s/%s", dc.Name, dc.LabelResourceName(), existingDc.Name, existingDc.LabelResourceName()))
 			}
 		}
 	}
@@ -164,7 +164,7 @@ func (rc *ReconciliationContext) validateDatacenterNameOverride() []error {
 		return errs
 	} else {
 		if *dc.Status.DatacenterName != dc.Spec.DatacenterName {
-			errs = append(errs, fmt.Errorf("datacenter %s name override '%s' cannot be changed after creation to '%s'.", dc.Name, dc.Spec.DatacenterName, *dc.Status.DatacenterName))
+			errs = append(errs, fmt.Errorf("datacenter %s name override '%s' cannot be changed after creation to '%s'", dc.Name, dc.Spec.DatacenterName, *dc.Status.DatacenterName))
 		}
 	}
 
diff --git a/pkg/reconciliation/handler_reconcile_test.go b/pkg/reconciliation/handler_reconcile_test.go
index 5d995235..1b66984d 100644
--- a/pkg/reconciliation/handler_reconcile_test.go
+++ b/pkg/reconciliation/handler_reconcile_test.go
@@ -19,6 +19,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -26,7 +27,7 @@ import (
 
 func TestReconcile(t *testing.T) {
 	var (
-		name            = "cluster-example-cluster"
+		name            = "dc1-example"
 		namespace       = "default"
 		size      int32 = 2
 	)
@@ -74,6 +75,7 @@ func TestReconcile(t *testing.T) {
 		Client:   fakeClient,
 		Scheme:   s,
 		Recorder: record.NewFakeRecorder(100),
+		Log:      ctrl.Log.WithName("controllers").WithName("CassandraDatacenter"),
 	}
 
 	request := reconcile.Request{
@@ -88,8 +90,8 @@ func TestReconcile(t *testing.T) {
 		t.Fatalf("Reconciliation Failure: (%v)", err)
 	}
 
-	if result != (reconcile.Result{Requeue: true, RequeueAfter: 2 * time.Second}) {
-		t.Error("Reconcile did not return a correct result.")
+	if result != (reconcile.Result{Requeue: true, RequeueAfter: 10 * time.Second}) {
+		t.Errorf("Reconcile did not return a correct result. (%v)", result)
 	}
 }
 
diff --git a/pkg/reconciliation/handler_test.go b/pkg/reconciliation/handler_test.go
index d32a2f3b..f8bcb60c 100644
--- a/pkg/reconciliation/handler_test.go
+++ b/pkg/reconciliation/handler_test.go
@@ -220,7 +220,11 @@ func TestConflictingDcNameOverride(t *testing.T) {
 				Spec: api.CassandraDatacenterSpec{
 					ClusterName:    "cluster1",
 					DatacenterName: "CassandraDatacenter_example",
-				}}}
+				},
+				Status: api.CassandraDatacenterStatus{
+					DatacenterName: ptr.To[string]("CassandraDatacenter_example"),
+				},
+			}}
 		})
 
 	errs := rc.validateDatacenterNameConflicts()
diff --git a/pkg/reconciliation/reconcile_configsecret.go b/pkg/reconciliation/reconcile_configsecret.go
index 038c3b76..947040d9 100644
--- a/pkg/reconciliation/reconcile_configsecret.go
+++ b/pkg/reconciliation/reconcile_configsecret.go
@@ -131,7 +131,7 @@ func getConfigFromConfigSecret(dc *api.CassandraDatacenter, secret *corev1.Secre
 
 // getDatacenterConfigSecretName The format is clusterName-dcName-config
 func getDatacenterConfigSecretName(dc *api.CassandraDatacenter) string {
-	return api.CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.SanitizedName() + "-config"
+	return api.CleanupForKubernetes(dc.Spec.ClusterName) + "-" + dc.LabelResourceName() + "-config"
 }
 
 // getDatacenterConfigSecret Fetches the secret from the api server or creates a new secret
diff --git a/pkg/reconciliation/reconcile_datacenter.go b/pkg/reconciliation/reconcile_datacenter.go
index f55f67ae..ec8c11e8 100644
--- a/pkg/reconciliation/reconcile_datacenter.go
+++ b/pkg/reconciliation/reconcile_datacenter.go
@@ -50,13 +50,12 @@ func (rc *ReconciliationContext) ProcessDeletion() result.ReconcileResult {
 	}
 
 	if _, found := rc.Datacenter.Annotations[api.DecommissionOnDeleteAnnotation]; found {
-		podList, err := rc.listPods(rc.Datacenter.GetDatacenterLabels())
+		dcPods, err := rc.listPods(rc.Datacenter.GetDatacenterLabels())
 		if err != nil {
 			rc.ReqLogger.Error(err, "Failed to list pods, unable to proceed with deletion")
 			return result.Error(err)
 		}
-		dcPods := PodPtrsFromPodList(podList)
-		if len(podList.Items) > 0 {
+		if len(dcPods) > 0 {
 			rc.ReqLogger.V(1).Info("Deletion is being processed by the decommission check")
 			dcs, err := rc.getClusterDatacenters(dcPods)
 			if err != nil {
diff --git a/pkg/reconciliation/reconcile_fql.go b/pkg/reconciliation/reconcile_fql.go
index ee3a1143..f6e20ad1 100644
--- a/pkg/reconciliation/reconcile_fql.go
+++ b/pkg/reconciliation/reconcile_fql.go
@@ -25,7 +25,7 @@ func (rc *ReconciliationContext) CheckFullQueryLogging() result.ReconcileResult
 		rc.ReqLogger.Error(err, "error listing all pods in the cluster to progress full query logging reconciliation")
 		return result.RequeueSoon(2)
 	}
-	for _, podPtr := range PodPtrsFromPodList(podList) {
+	for _, podPtr := range podList {
 		features, err := rc.NodeMgmtClient.FeatureSet(podPtr)
 		if err != nil {
 			rc.ReqLogger.Error(err, "failed to verify featureset for FQL support")
diff --git a/pkg/reconciliation/reconcile_racks.go b/pkg/reconciliation/reconcile_racks.go
index 2d79cb1e..d40012b2 100644
--- a/pkg/reconciliation/reconcile_racks.go
+++ b/pkg/reconciliation/reconcile_racks.go
@@ -34,10 +34,9 @@ import (
 )
 
 var (
-	ResultShouldNotRequeue     reconcile.Result = reconcile.Result{Requeue: false}
-	ResultShouldRequeueNow     reconcile.Result = reconcile.Result{Requeue: true}
-	ResultShouldRequeueSoon    reconcile.Result = reconcile.Result{Requeue: true, RequeueAfter: 2 * time.Second}
-	ResultShouldRequeueTenSecs reconcile.Result = reconcile.Result{Requeue: true, RequeueAfter: 10 * time.Second}
+	ResultShouldNotRequeue  reconcile.Result = reconcile.Result{Requeue: false}
+	ResultShouldRequeueNow  reconcile.Result = reconcile.Result{Requeue: true}
+	ResultShouldRequeueSoon reconcile.Result = reconcile.Result{Requeue: true, RequeueAfter: 2 * time.Second}
 
 	QuietDurationFunc func(int) time.Duration = func(secs int) time.Duration { return time.Duration(secs) * time.Second }
 )
@@ -618,7 +617,7 @@ func (rc *ReconciliationContext) CheckRackStoppedState() result.ReconcileResult
 				emittedStoppingEvent = true
 			}
 
-			rackPods := FilterPodListByLabels(rc.dcPods, rc.Datacenter.GetRackLabels(rackInfo.RackName))
+			rackPods := rc.rackPods(rackInfo.RackName)
 
 			nodesDrained := 0
 			nodeDrainErrors := 0
@@ -751,7 +750,7 @@ func (rc *ReconciliationContext) CheckPodsReady(endpointData httphelper.CassMeta
 		return result.Error(err)
 	}
 	if atLeastOneFirstNodeNotReady {
-		return result.RequeueSoon(2)
+		return result.RequeueSoon(10)
 	}
 
 	// step 3 - if the cluster isn't healthy, that's ok, but go back to step 1
@@ -1432,8 +1431,8 @@ func (rc *ReconciliationContext) isClusterHealthy() bool {
 func (rc *ReconciliationContext) labelSeedPods(rackInfo *RackInformation) (int, error) {
 	logger := rc.ReqLogger.WithName("labelSeedPods")
 
-	rackLabels := rc.Datacenter.GetRackLabels(rackInfo.RackName)
-	rackPods := FilterPodListByLabels(rc.dcPods, rackLabels)
+	rackPods := rc.rackPods(rackInfo.RackName)
+
 	sort.SliceStable(rackPods, func(i, j int) bool {
 		return rackPods[i].Name < rackPods[j].Name
 	})
@@ -2150,7 +2149,7 @@ func (rc *ReconciliationContext) refreshSeeds() error {
 	return nil
 }
 
-func (rc *ReconciliationContext) listPods(selector map[string]string) (*corev1.PodList, error) {
+func (rc *ReconciliationContext) listPods(selector map[string]string) ([]*corev1.Pod, error) {
 	rc.ReqLogger.Info("reconcile_racks::listPods")
 
 	listOptions := &client.ListOptions{
@@ -2165,7 +2164,11 @@ func (rc *ReconciliationContext) listPods(selector map[string]string) (*corev1.P
 		},
 	}
 
-	return podList, rc.Client.List(rc.Ctx, podList, listOptions)
+	if err := rc.Client.List(rc.Ctx, podList, listOptions); err != nil {
+		return nil, err
+	}
+
+	return PodPtrsFromPodList(podList), nil
 }
 
 func (rc *ReconciliationContext) CheckRollingRestart() result.ReconcileResult {
@@ -2425,21 +2428,40 @@ func (rc *ReconciliationContext) fixMissingPVC() (bool, error) {
 	return false, nil
 }
 
+func (rc *ReconciliationContext) datacenterPods() []*corev1.Pod {
+	if rc.dcPods != nil {
+		return rc.dcPods
+	}
+
+	dcSelector := rc.Datacenter.GetDatacenterLabels()
+	dcPods := FilterPodListByLabels(rc.clusterPods, dcSelector)
+
+	if rc.Datacenter.Status.MetadataVersion < 1 && rc.Datacenter.Status.DatacenterName != nil && *rc.Datacenter.Status.DatacenterName == rc.Datacenter.Spec.DatacenterName {
+		rc.ReqLogger.Info("Fetching datacenter pods with the old metadata version labels")
+		dcSelector[api.DatacenterLabel] = api.CleanLabelValue(rc.Datacenter.Spec.DatacenterName)
+		dcPods = append(dcPods, FilterPodListByLabels(rc.clusterPods, dcSelector)...)
+	}
+
+	return dcPods
+}
+
+func (rc *ReconciliationContext) rackPods(rackName string) []*corev1.Pod {
+	return FilterPodListByLabels(rc.datacenterPods(), map[string]string{api.RackLabel: rackName})
+}
+
 // ReconcileAllRacks determines if a rack needs to be reconciled.
 func (rc *ReconciliationContext) ReconcileAllRacks() (reconcile.Result, error) {
 	rc.ReqLogger.Info("reconciliationContext::reconcileAllRacks")
 
 	logger := rc.ReqLogger
 
-	podList, err := rc.listPods(rc.Datacenter.GetClusterLabels())
+	pods, err := rc.listPods(rc.Datacenter.GetClusterLabels())
 	if err != nil {
 		logger.Error(err, "error listing all pods in the cluster")
 	}
 
-	rc.clusterPods = PodPtrsFromPodList(podList)
-
-	dcSelector := rc.Datacenter.GetDatacenterLabels()
-	rc.dcPods = FilterPodListByLabels(rc.clusterPods, dcSelector)
+	rc.clusterPods = pods
+	rc.dcPods = rc.datacenterPods()
 
 	endpointData := rc.getCassMetadataEndpoints()
 
diff --git a/pkg/reconciliation/reconcile_racks_test.go b/pkg/reconciliation/reconcile_racks_test.go
index bb53bd67..89cc2eee 100644
--- a/pkg/reconciliation/reconcile_racks_test.go
+++ b/pkg/reconciliation/reconcile_racks_test.go
@@ -6,7 +6,6 @@ package reconciliation
 import (
 	"context"
 	"fmt"
-	"github.com/pkg/errors"
 	"io"
 	"net/http"
 	"reflect"
@@ -16,6 +15,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pkg/errors"
+
 	"k8s.io/utils/ptr"
 
 	api "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
@@ -2780,3 +2781,96 @@ func TestDatacenterStatus(t *testing.T) {
 	assert.NoError(err)
 	assert.Equal(float64(0), val)
 }
+
+func TestDatacenterPods(t *testing.T) {
+	rc, _, cleanupMockScr := setupTest()
+	defer cleanupMockScr()
+	assert := assert.New(t)
+
+	desiredStatefulSet, err := newStatefulSetForCassandraDatacenter(
+		nil,
+		"default",
+		rc.Datacenter,
+		3)
+	assert.NoErrorf(err, "error occurred creating statefulset")
+
+	desiredStatefulSet.Status.ReadyReplicas = *desiredStatefulSet.Spec.Replicas
+
+	trackObjects := []runtime.Object{
+		desiredStatefulSet,
+		rc.Datacenter,
+	}
+
+	mockPods := mockReadyPodsForStatefulSet(desiredStatefulSet, rc.Datacenter.Spec.ClusterName, rc.Datacenter.Name)
+	for idx := range mockPods {
+		mp := mockPods[idx]
+		trackObjects = append(trackObjects, mp)
+	}
+
+	rc.Client = fake.NewClientBuilder().WithStatusSubresource(rc.Datacenter).WithRuntimeObjects(trackObjects...).Build()
+
+	nextRack := &RackInformation{}
+	nextRack.RackName = "default"
+	nextRack.NodeCount = 1
+	nextRack.SeedCount = 1
+
+	rackInfo := []*RackInformation{nextRack}
+
+	rc.desiredRackInformation = rackInfo
+	rc.statefulSets = make([]*appsv1.StatefulSet, len(rackInfo))
+
+	rc.clusterPods = mockPods
+	assert.Equal(int(*desiredStatefulSet.Spec.Replicas), len(rc.datacenterPods()))
+}
+
+func TestDatacenterPodsOldLabels(t *testing.T) {
+	rc, _, cleanupMockScr := setupTest()
+	defer cleanupMockScr()
+	assert := assert.New(t)
+
+	// We fake the process a bit to get old style naming and labels
+	rc.Datacenter.Name = "overrideMe"
+
+	desiredStatefulSet, err := newStatefulSetForCassandraDatacenter(
+		nil,
+		"default",
+		rc.Datacenter,
+		3)
+	assert.NoErrorf(err, "error occurred creating statefulset")
+
+	desiredStatefulSet.Status.ReadyReplicas = *desiredStatefulSet.Spec.Replicas
+
+	trackObjects := []runtime.Object{
+		desiredStatefulSet,
+		rc.Datacenter,
+	}
+
+	mockPods := mockReadyPodsForStatefulSet(desiredStatefulSet, rc.Datacenter.Spec.ClusterName, rc.Datacenter.Name)
+	for idx := range mockPods {
+		mp := mockPods[idx]
+		trackObjects = append(trackObjects, mp)
+	}
+
+	rc.Client = fake.NewClientBuilder().WithStatusSubresource(rc.Datacenter).WithRuntimeObjects(trackObjects...).Build()
+
+	nextRack := &RackInformation{}
+	nextRack.RackName = "default"
+	nextRack.NodeCount = 1
+	nextRack.SeedCount = 1
+
+	rackInfo := []*RackInformation{nextRack}
+
+	rc.desiredRackInformation = rackInfo
+	rc.statefulSets = make([]*appsv1.StatefulSet, len(rackInfo))
+
+	rc.clusterPods = mockPods
+
+	// Lets modify the Datacenter names and set the status like it used to be in some older versions
+	rc.Datacenter.Spec.DatacenterName = "overrideMe"
+	rc.Datacenter.Name = "dc1"
+	rc.Datacenter.Status.DatacenterName = ptr.To[string]("overrideMe")
+	rc.Datacenter.Status.MetadataVersion = 0
+
+	// We should still find the pods
+	assert.Equal(int(*desiredStatefulSet.Spec.Replicas), len(rc.datacenterPods()))
+}
diff --git a/scripts/release-helm-chart.sh b/scripts/release-helm-chart.sh
index 4a37b775..d3db2030 100755
--- a/scripts/release-helm-chart.sh
+++ b/scripts/release-helm-chart.sh
@@ -6,7 +6,7 @@ if [[ ! $0 == scripts/* ]]; then
 fi
 
 # This script assumes k8ssandra is checked out at ../k8ssandra and is checked out at main
-if [ "$#" -le 1 ]; then
+if [ "$#" -lt 1 ]; then
     echo "Usage: scripts/release-helm-chart.sh version legacy"
     echo "Script assumes you are in the correct branch / tag and that k8ssandra repository"
     echo "has been checked out to ../k8ssandra/. If legacy is set, the script will generate"
diff --git a/tests/decommission_dc/decommission_dc_suite_test.go b/tests/decommission_dc/decommission_dc_suite_test.go
index 7aa3c746..39f9a491 100644
--- a/tests/decommission_dc/decommission_dc_suite_test.go
+++ b/tests/decommission_dc/decommission_dc_suite_test.go
@@ -12,22 +12,19 @@ import (
 	"github.com/k8ssandra/cass-operator/tests/kustomize"
 	ginkgo_util "github.com/k8ssandra/cass-operator/tests/util/ginkgo"
 	"github.com/k8ssandra/cass-operator/tests/util/kubectl"
-
-	api "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
 )
 
 var (
-	testName        = "Delete DC and verify it is correctly decommissioned in multi-dc cluster"
-	namespace       = "test-decommission-dc"
-	dc1Name         = "dc1"
-	dc1OverrideName = "My_Super_Dc"
-	dc2Name         = "dc2"
-	dc1Yaml         = "../testdata/default-two-rack-two-node-dc.yaml"
-	dc2Yaml         = "../testdata/default-two-rack-two-node-dc2.yaml"
-	dc1Label        = fmt.Sprintf("cassandra.datastax.com/datacenter=%s", api.CleanLabelValue(dc1OverrideName))
-	dc2Label        = fmt.Sprintf("cassandra.datastax.com/datacenter=%s", dc2Name)
-	seedLabel       = "cassandra.datastax.com/seed-node=true"
-	taskYaml        = "../testdata/tasks/rebuild_task.yaml"
+	testName  = "Delete DC and verify it is correctly decommissioned in multi-dc cluster"
+	namespace = "test-decommission-dc"
+	dc1Name   = "dc1"
+	dc2Name   = "dc2"
+	dc1Yaml   = "../testdata/default-two-rack-two-node-dc.yaml"
+	dc2Yaml   = "../testdata/default-two-rack-two-node-dc2.yaml"
+	dc1Label  = fmt.Sprintf("cassandra.datastax.com/datacenter=%s", dc1Name)
+	dc2Label  = fmt.Sprintf("cassandra.datastax.com/datacenter=%s", dc2Name)
+	seedLabel = "cassandra.datastax.com/seed-node=true"
+	taskYaml  = "../testdata/tasks/rebuild_task.yaml"
 	// dcLabel   = fmt.Sprintf("cassandra.datastax.com/datacenter=%s", dcName)
 	ns = ginkgo_util.NewWrapper(testName, namespace)
 )
@@ -137,7 +134,7 @@ var _ = Describe(testName, func() {
 			// Wait for the task to be completed
 			ns.WaitForCompleteTask("rebuild-dc")
 
-			podNames := ns.GetDatacenterReadyPodNames(dc1OverrideName)
+			podNames := ns.GetDatacenterReadyPodNames(dc1Name)
 			Expect(len(podNames)).To(Equal(2))
 			dcs := findDatacenters(podNames[0])
 
@@ -160,7 +157,7 @@ var _ = Describe(testName, func() {
 			ns.WaitForOutputAndLog(step, k, "[]", 300)
 
 			// Verify nodetool status has only a single Datacenter
-			podNames = ns.GetDatacenterReadyPodNames(dc1OverrideName)
+			podNames = ns.GetDatacenterReadyPodNames(dc1Name)
 
 			if len(podNames) != 2 {
 				// This is to catch why the test sometimes fails on the check (string parsing? or real issue?)
diff --git a/tests/rolling_restart_with_override/rolling_restart_suite_with_override_test.go b/tests/rolling_restart_with_override/rolling_restart_suite_with_override_test.go
index 99b73826..c0d49cd1 100644
--- a/tests/rolling_restart_with_override/rolling_restart_suite_with_override_test.go
+++ b/tests/rolling_restart_with_override/rolling_restart_suite_with_override_test.go
@@ -20,14 +20,13 @@ import (
 )
 
 var (
-	testName       = "DC override Rolling Restart"
-	namespace      = "test-override-with-rolling-restart"
-	dcName         = "dc1"
-	dcNameOverride = "My_Super_Dc"
-	dcYaml         = "../testdata/default-two-rack-two-node-dc.yaml"
-	taskYaml       = "../testdata/tasks/rolling_restart_override.yaml"
-	dcResource     = fmt.Sprintf("CassandraDatacenter/%s", dcName)
-	ns             = ginkgo_util.NewWrapper(testName, namespace)
+	testName   = "DC override Rolling Restart"
+	namespace  = "test-override-with-rolling-restart"
+	dcName     = "dc1"
+	dcYaml     = "../testdata/default-two-rack-two-node-dc.yaml"
+	taskYaml   = "../testdata/tasks/rolling_restart_override.yaml"
+	dcResource = fmt.Sprintf("CassandraDatacenter/%s", dcName)
+	ns         = ginkgo_util.NewWrapper(testName, namespace)
 )
 
 func TestLifecycle(t *testing.T) {
@@ -86,7 +85,7 @@ var _ = Describe(testName, func() {
 			step = "get ready pods"
 			json = "jsonpath={.items[*].status.containerStatuses[0].ready}"
 			k = kubectl.Get("pods").
-				WithLabel(fmt.Sprintf("cassandra.datastax.com/datacenter=%s", api.CleanLabelValue(dcNameOverride))).
+				WithLabel(fmt.Sprintf("cassandra.datastax.com/datacenter=%s", api.CleanLabelValue(dcName))).
 				WithFlag("field-selector", "status.phase=Running").
 				FormatOutput(json)
 
@@ -105,7 +104,7 @@ var _ = Describe(testName, func() {
 			// Verify each pod does have the annotation..
 			json := `jsonpath={.items[0].metadata.annotations.control\.k8ssandra\.io/restartedAt}`
 			k = kubectl.Get("pods").
-				WithLabel(fmt.Sprintf("cassandra.datastax.com/datacenter=%s", api.CleanLabelValue(dcNameOverride))).
+				WithLabel(fmt.Sprintf("cassandra.datastax.com/datacenter=%s", api.CleanLabelValue(dcName))).
 				WithFlag("field-selector", "status.phase=Running").
 				FormatOutput(json)
 			ns.WaitForOutputPatternAndLog(step, k, `^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$`, 360)
diff --git a/tests/test_all_the_things/test_all_the_things_suite_test.go b/tests/test_all_the_things/test_all_the_things_suite_test.go
index ef0ccac4..98c3243a 100644
--- a/tests/test_all_the_things/test_all_the_things_suite_test.go
+++ b/tests/test_all_the_things/test_all_the_things_suite_test.go
@@ -22,14 +22,13 @@ import (
 )
 
 var (
-	testName       = "Test all the things"
-	namespace      = "test-test-all-the-things"
-	dcName         = "dc1"
-	dcNameOverride = "My_Super_Dc"
-	dcYaml         = "../testdata/default-two-rack-two-node-dc.yaml"
-	dcResource     = fmt.Sprintf("CassandraDatacenter/%s", dcName)
-	dcLabel        = fmt.Sprintf("cassandra.datastax.com/datacenter=%s", api.CleanLabelValue(dcNameOverride))
-	ns             = ginkgo_util.NewWrapper(testName, namespace)
+	testName   = "Test all the things"
+	namespace  = "test-test-all-the-things"
+	dcName     = "dc1"
+	dcYaml     = "../testdata/default-two-rack-two-node-dc.yaml"
+	dcResource = fmt.Sprintf("CassandraDatacenter/%s", dcName)
+	dcLabel    = fmt.Sprintf("cassandra.datastax.com/datacenter=%s", api.CleanLabelValue(dcName))
+	ns         = ginkgo_util.NewWrapper(testName, namespace)
 )
 
 func TestLifecycle(t *testing.T) {
@@ -89,12 +88,12 @@ var _ = Describe(testName, func() {
 			ns.WaitForDatacenterOperatorProgress(dcName, "Updating", 60)
 			ns.WaitForDatacenterConditionWithTimeout(dcName, "ScalingUp", string(corev1.ConditionFalse), 1200)
 			// Ensure that when 'ScaleUp' becomes 'false' that our pods are in fact up and running
-			Expect(len(ns.GetDatacenterReadyPodNames(api.CleanLabelValue(dcNameOverride)))).To(Equal(4))
+			Expect(len(ns.GetDatacenterReadyPodNames(api.CleanLabelValue(dcName)))).To(Equal(4))
 
 			ns.ExpectDoneReconciling(dcName)
 			ns.WaitForDatacenterReady(dcName)
 
-			ns.ExpectDatacenterNameStatusUpdated(dcName, dcNameOverride)
+			ns.ExpectDatacenterNameStatusUpdated(dcName, dcName)
 
 			// Ensure we have a single CassandraTask created which is a cleanup (and it succeeded)
 			ns.WaitForCompletedCassandraTasks(dcName, "cleanup", 1)
@@ -114,7 +113,7 @@ var _ = Describe(testName, func() {
 				FormatOutput(json)
 			ns.WaitForOutputAndLog(step, k, "4", 20)
 
-			ns.WaitForDatacenterToHaveNoPods(api.CleanLabelValue(dcNameOverride))
+			ns.WaitForDatacenterToHaveNoPods(api.CleanLabelValue(dcName))
 
 			step = "resume the dc"
 			json = "{\"spec\": {\"stopped\": false}}"
@@ -133,7 +132,7 @@ var _ = Describe(testName, func() {
 			wg.Add(1)
 			go func() {
 				k = kubectl.Logs("-f").
-					WithLabel(fmt.Sprintf("statefulset.kubernetes.io/pod-name=cluster1-%s-r1-sts-0", api.CleanupForKubernetes(dcNameOverride))).
+					WithLabel(fmt.Sprintf("statefulset.kubernetes.io/pod-name=cluster1-%s-r1-sts-0", dcName)).
 					WithFlag("container", "cassandra")
 				output, err := ns.Output(k)
 				Expect(err).ToNot(HaveOccurred())
@@ -148,7 +147,7 @@ var _ = Describe(testName, func() {
 
 			found, err := regexp.MatchString("node/drain status=200 OK", logOutput)
 			if err == nil && !found {
-				ns.Log(fmt.Sprintf("logOutput, pod: statefulset.kubernetes.io/pod-name=cluster1-%s-r1-sts-0 => %s", api.CleanLabelValue(dcNameOverride), logOutput))
+				ns.Log(fmt.Sprintf("logOutput, pod: statefulset.kubernetes.io/pod-name=cluster1-%s-r1-sts-0 => %s", dcName, logOutput))
 			}
 			if err != nil {
 				ns.Log(fmt.Sprintf("Regexp parsing failed: %v", err))
diff --git a/tests/testdata/default-single-rack-single-node-dc-with-readonly-fs.yaml b/tests/testdata/default-single-rack-single-node-dc-with-readonly-fs.yaml
index 8a734a74..64633004 100644
--- a/tests/testdata/default-single-rack-single-node-dc-with-readonly-fs.yaml
+++ b/tests/testdata/default-single-rack-single-node-dc-with-readonly-fs.yaml
@@ -6,7 +6,7 @@ spec:
   clusterName: cluster1
   serverType: cassandra
   serverVersion: "4.1.5"
-  serverImage: michaelburman290/cass-management-api:4.1.5-ubi8
+  serverImage: michaelburman290/cass-management-api:4.1.6-ubi
   managementApiAuth:
     insecure: {}
   readOnlyRootFilesystem: true
diff --git a/tests/testdata/default-three-rack-three-node-dc-4x.yaml b/tests/testdata/default-three-rack-three-node-dc-4x.yaml
index 68e0ec79..bada916b 100644
--- a/tests/testdata/default-three-rack-three-node-dc-4x.yaml
+++ b/tests/testdata/default-three-rack-three-node-dc-4x.yaml
@@ -5,7 +5,8 @@ metadata:
 spec:
   clusterName: cluster1
   serverType: cassandra
-  serverVersion: 4.1.4
+  datacenterName: My_Super_Dc
+  serverVersion: 4.1.6
   managementApiAuth:
     insecure: {}
   size: 3
diff --git a/tests/testdata/image_config_parsing.yaml b/tests/testdata/image_config_parsing.yaml
index 1f8caf26..5f6a382f 100644
--- a/tests/testdata/image_config_parsing.yaml
+++ b/tests/testdata/image_config_parsing.yaml
@@ -18,7 +18,7 @@ defaults:
   # Note, postfix is ignored if repository is not set
   cassandra:
     repository: "k8ssandra/cass-management-api"
-    suffix: "-ubi8"
+    suffix: "-ubi"
   dse:
     repository: "datastax/dse-mgmtapi-6_8"
     suffix: "-ubi8"
diff --git a/tests/upgrade_operator/upgrade_operator_suite_test.go b/tests/upgrade_operator/upgrade_operator_suite_test.go
index 6a8eafd4..126ded2a 100644
--- a/tests/upgrade_operator/upgrade_operator_suite_test.go
+++ b/tests/upgrade_operator/upgrade_operator_suite_test.go
@@ -20,6 +20,7 @@ var (
 	testName   = "Upgrade Operator"
 	namespace  = "test-upgrade-operator"
 	dcName     = "dc1"
+	podId      = "pod/cluster1-my-super-dc-r1-sts-0"
 	dcYaml     = "../testdata/default-three-rack-three-node-dc-4x.yaml"
 	dcResource = fmt.Sprintf("CassandraDatacenter/%s", dcName)
 	dcLabel    = fmt.Sprintf("cassandra.datastax.com/datacenter=%s", dcName)
@@ -79,11 +80,11 @@ var _ = Describe(testName, func() {
 			k := kubectl.ApplyFiles(dcYaml)
 			ns.ExecAndLog(step, k)
 
-			ns.WaitForDatacenterReady(dcName)
+			ns.WaitForDatacenterOperatorProgress(dcName, "Ready", 1800)
 
 			// Get UID of the cluster pod
 			step = "get Cassandra pods UID"
-			k = kubectl.Get("pod/cluster1-dc1-r1-sts-0").FormatOutput("jsonpath={.metadata.uid}")
+			k = kubectl.Get(podId).FormatOutput("jsonpath={.metadata.uid}")
 			createdPodUID := ns.OutputAndLog(step, k)
 
 			step = "get name of 1.19.1 operator pod"
@@ -106,7 +107,7 @@ var _ = Describe(testName, func() {
 
 			// Verify Pod hasn't restarted
 			step = "get Cassandra pods UID"
-			k = kubectl.Get("pod/cluster1-dc1-r1-sts-0").FormatOutput("jsonpath={.metadata.uid}")
+			k = kubectl.Get(podId).FormatOutput("jsonpath={.metadata.uid}")
 			postUpgradeCassPodUID := ns.OutputAndLog(step, k)
 
 			Expect(createdPodUID).To(Equal(postUpgradeCassPodUID))
@@ -120,7 +121,7 @@ var _ = Describe(testName, func() {
 			// Get current system-logger image
 			// Verify the Pod now has updated system-logger container image
 			step = "get Cassandra pod system-logger"
-			k = kubectl.Get("pod/cluster1-dc1-r1-sts-0").FormatOutput("jsonpath={.spec.containers[?(@.name == 'server-system-logger')].image}")
+			k = kubectl.Get(podId).FormatOutput("jsonpath={.spec.containers[?(@.name == 'server-system-logger')].image}")
 			loggerImage := ns.OutputAndLog(step, k)
 			Expect(loggerImage).To(Equal("cr.k8ssandra.io/k8ssandra/system-logger:v1.19.1"))
 
@@ -137,14 +138,14 @@ var _ = Describe(testName, func() {
 
 			// Verify pod has been restarted
 			step = "get Cassandra pods UID"
-			k = kubectl.Get("pod/cluster1-dc1-r1-sts-0").FormatOutput("jsonpath={.metadata.uid}")
+			k = kubectl.Get(podId).FormatOutput("jsonpath={.metadata.uid}")
 			postAllowUpgradeUID := ns.OutputAndLog(step, k)
 
 			Expect(postUpgradeCassPodUID).ToNot(Equal(postAllowUpgradeUID))
 
 			// Verify the Pod now has updated system-logger container image
 			step = "get Cassandra pod system-logger"
-			k = kubectl.Get("pod/cluster1-dc1-r1-sts-0").FormatOutput("jsonpath={.spec.containers[?(@.name == 'server-system-logger')].image}")
+			k = kubectl.Get(podId).FormatOutput("jsonpath={.spec.containers[?(@.name == 'server-system-logger')].image}")
 			loggerImageNew := ns.OutputAndLog(step, k)
 			Expect(loggerImage).To(Not(Equal(loggerImageNew)))
 
diff --git a/tests/util/ginkgo/lib.go b/tests/util/ginkgo/lib.go
index a05e5d86..aed205d9 100644
--- a/tests/util/ginkgo/lib.go
+++ b/tests/util/ginkgo/lib.go
@@ -6,7 +6,6 @@ package ginkgo_util
 import (
 	"encoding/base64"
 	"fmt"
-	"github.com/pkg/errors"
 	"os"
 	"path/filepath"
 	"regexp"
@@ -15,6 +14,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pkg/errors"
+
 	ginkgo "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"gopkg.in/yaml.v2"
@@ -424,16 +425,6 @@ func (ns *NsWrapper) Log(step string) {
 	ginkgo.By(step)
 }
 
-func (ns *NsWrapper) getDcNameWithOverride(dcName string) string {
-	json := "jsonpath={.spec.datacenterName}"
-	k := kubectl.Get("CassandraDatacenter", dcName).FormatOutput(json)
-	dcNameOverride := ns.OutputPanic(k)
-	if dcNameOverride == "" {
-		return dcName
-	}
-	return dcNameOverride
-}
-
 func (ns *NsWrapper) WaitForDatacenterReadyWithTimeouts(dcName string, podCountTimeout int, dcReadyTimeout int) {
 	json := "jsonpath={.spec.size}"
 	k := kubectl.Get("CassandraDatacenter", dcName).FormatOutput(json)
@@ -441,7 +432,7 @@ func (ns *NsWrapper) WaitForDatacenterReadyWithTimeouts(dcName string, podCountT
 	size, err := strconv.Atoi(sizeString)
 	Expect(err).ToNot(HaveOccurred())
 
-	ns.WaitForDatacenterReadyPodCountWithTimeout(ns.getDcNameWithOverride(dcName), size, podCountTimeout)
+	ns.WaitForDatacenterReadyPodCountWithTimeout(dcName, size, podCountTimeout)
 	ns.WaitForDatacenterOperatorProgress(dcName, "Ready", dcReadyTimeout)
 }