diff --git a/pkg/daemon/ceph/client/mon.go b/pkg/daemon/ceph/client/mon.go
index dec00f07b..08cfc92a8 100644
--- a/pkg/daemon/ceph/client/mon.go
+++ b/pkg/daemon/ceph/client/mon.go
@@ -152,3 +152,14 @@ func SetMonStretchTiebreaker(context *clusterd.Context, clusterInfo *ClusterInfo
 	logger.Infof("successfully set mon tiebreaker %q in failure domain %q", monName, bucketType)
 	return nil
 }
+
+// SetNewTiebreaker sets the new tiebreaker mon in the stretch cluster during a failover
+func SetNewTiebreaker(context *clusterd.Context, clusterInfo *ClusterInfo, monName string) error {
+	logger.Infof("setting new mon tiebreaker %q in arbiter zone", monName)
+	args := []string{"mon", "set_new_tiebreaker", monName}
+	if _, err := NewCephCommand(context, clusterInfo, args).Run(); err != nil {
+		return errors.Wrapf(err, "failed to set new mon tiebreaker %q", monName)
+	}
+	logger.Infof("successfully set new mon tiebreaker %q in arbiter zone", monName)
+	return nil
+}
diff --git a/pkg/daemon/ceph/client/mon_test.go b/pkg/daemon/ceph/client/mon_test.go
index 83ef7fe54..9c3a8c441 100644
--- a/pkg/daemon/ceph/client/mon_test.go
+++ b/pkg/daemon/ceph/client/mon_test.go
@@ -88,15 +88,22 @@ func TestStretchElectionStrategy(t *testing.T) {
 func TestStretchClusterMonTiebreaker(t *testing.T) {
 	monName := "a"
 	failureDomain := "rack"
+	setTiebreaker := false
+	enabledStretch := false
 	executor := &exectest.MockExecutor{}
 	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
 		logger.Infof("Command: %s %v", command, args)
 		switch {
 		case args[0] == "mon" && args[1] == "enable_stretch_mode":
+			enabledStretch = true
 			assert.Equal(t, monName, args[2])
 			assert.Equal(t, defaultStretchCrushRuleName, args[3])
 			assert.Equal(t, failureDomain, args[4])
 			return "", nil
+		case args[0] == "mon" && args[1] == "set_new_tiebreaker":
+			setTiebreaker = true
+			assert.Equal(t, monName, args[2])
+			return "", nil
 		}
 		return "", errors.Errorf("unexpected ceph command %q", args)
 	}
@@ -105,6 +112,14 @@ func TestStretchClusterMonTiebreaker(t *testing.T) {
 
 	err := SetMonStretchTiebreaker(context, clusterInfo, monName, failureDomain)
 	assert.NoError(t, err)
+	assert.True(t, enabledStretch)
+	assert.False(t, setTiebreaker)
+	enabledStretch = false
+
+	err = SetNewTiebreaker(context, clusterInfo, monName)
+	assert.NoError(t, err)
+	assert.True(t, setTiebreaker)
+	assert.False(t, enabledStretch)
 }
 
 func TestMonDump(t *testing.T) {
diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go
index c14751d3c..c6482b19f 100755
--- a/pkg/operator/ceph/cluster/cluster.go
+++ b/pkg/operator/ceph/cluster/cluster.go
@@ -137,7 +137,8 @@ func (c *cluster) reconcileCephDaemons(rookImage string, cephVersion cephver.Cep
 
 	// If a stretch cluster, enable the arbiter after the OSDs are created with the CRUSH map
 	if c.Spec.IsStretchCluster() {
-		if err := c.mons.ConfigureArbiter(); err != nil {
+		failingOver := false
+		if err := c.mons.ConfigureArbiter(failingOver); err != nil {
 			return errors.Wrap(err, "failed to configure stretch arbiter")
 		}
 	}
diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go
index 4cbb3510e..2cf00189c 100644
--- a/pkg/operator/ceph/cluster/mon/health.go
+++ b/pkg/operator/ceph/cluster/mon/health.go
@@ -28,6 +28,7 @@ import (
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	cephutil "github.com/rook/rook/pkg/daemon/ceph/util"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
+	"github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -44,6 +45,8 @@ var (
 	timeZero                       = time.Duration(0)
 	// Check whether mons are on the same node once per operator restart since it's a rare scheduling condition
 	needToCheckMonsOnSameNode = true
+	// Version of Ceph where the arbiter failover is supported
+	arbiterFailoverSupportedCephVersion = version.CephVersion{Major: 16, Minor: 2, Extra: 7}
 )
 
 // HealthChecker aggregates the mon/cluster info needed to check the health of the monitors
@@ -322,34 +325,49 @@ func (c *Cluster) failMon(monCount, desiredMonCount int, name string) bool {
 		if err := c.removeMon(name); err != nil {
 			logger.Errorf("failed to remove mon %q. %v", name, err)
 		}
-	} else {
-		if c.spec.IsStretchCluster() && name == c.arbiterMon {
-			// Ceph does not currently support updating the arbiter mon
-			// or else the mons in the two datacenters will not be aware anymore
-			// of the arbiter mon. Thus, disabling failover until the arbiter
-			// mon can be updated in ceph.
-			logger.Warningf("refusing to failover arbiter mon %q on a stretched cluster", name)
-			return false
-		}
+		return true
+	}
 
-		// prevent any voluntary mon drain while failing over
-		if err := c.blockMonDrain(types.NamespacedName{Name: monPDBName, Namespace: c.Namespace}); err != nil {
-			logger.Errorf("failed to block mon drain. %v", err)
-		}
+	if err := c.allowFailover(name); err != nil {
+		logger.Warningf("aborting mon %q failover. %v", name, err)
+		return false
+	}
 
-		// bring up a new mon to replace the unhealthy mon
-		if err := c.failoverMon(name); err != nil {
-			logger.Errorf("failed to failover mon %q. %v", name, err)
-		}
+	// prevent any voluntary mon drain while failing over
+	if err := c.blockMonDrain(types.NamespacedName{Name: monPDBName, Namespace: c.Namespace}); err != nil {
+		logger.Errorf("failed to block mon drain. %v", err)
+	}
 
-		// allow any voluntary mon drain after failover
-		if err := c.allowMonDrain(types.NamespacedName{Name: monPDBName, Namespace: c.Namespace}); err != nil {
-			logger.Errorf("failed to allow mon drain. %v", err)
-		}
+	// bring up a new mon to replace the unhealthy mon
+	if err := c.failoverMon(name); err != nil {
+		logger.Errorf("failed to failover mon %q. %v", name, err)
+	}
+
+	// allow any voluntary mon drain after failover
+	if err := c.allowMonDrain(types.NamespacedName{Name: monPDBName, Namespace: c.Namespace}); err != nil {
+		logger.Errorf("failed to allow mon drain. %v", err)
 	}
 	return true
 }
 
+func (c *Cluster) allowFailover(name string) error {
+	if !c.spec.IsStretchCluster() {
+		// always failover if not a stretch cluster
+		return nil
+	}
+	if name != c.arbiterMon {
+		// failover if it's a non-arbiter
+		return nil
+	}
+	if c.ClusterInfo.CephVersion.IsAtLeast(arbiterFailoverSupportedCephVersion) {
+		// failover the arbiter if at least v16.2.7
+		return nil
+	}
+
+	// Ceph does not support updating the arbiter mon in older versions
+	return errors.Errorf("refusing to failover arbiter mon %q on a stretched cluster until upgrading to ceph version %s", name, arbiterFailoverSupportedCephVersion.String())
+}
+
 func (c *Cluster) removeOrphanMonResources() {
 	if c.spec.Mon.VolumeClaimTemplate == nil {
 		logger.Debug("skipping check for orphaned mon pvcs since using the host path")
@@ -435,6 +453,9 @@ func (c *Cluster) failoverMon(name string) error {
 
 	// remove the failed mon from a local list of the existing mons for finding a stretch zone
 	existingMons := c.clusterInfoToMonConfig(name)
+	// Cache the name of the current arbiter in case it is updated during the failover
+	// This allows a simple check for updating the arbiter later in this method
+	currentArbiter := c.arbiterMon
 	zone, err := c.findAvailableZoneIfStretched(existingMons)
 	if err != nil {
 		return errors.Wrap(err, "failed to find available stretch zone")
@@ -473,12 +494,11 @@ func (c *Cluster) failoverMon(name string) error {
 	}
 
 	// Assign to a zone if a stretch cluster
-	if c.spec.IsStretchCluster() {
-		if name == c.arbiterMon {
-			// Update the arbiter mon for the stretch cluster if it changed
-			if err := c.ConfigureArbiter(); err != nil {
-				return errors.Wrap(err, "failed to configure stretch arbiter")
-			}
+	if c.spec.IsStretchCluster() && name == currentArbiter {
+		// Update the arbiter mon for the stretch cluster if it changed
+		failingOver := true
+		if err := c.ConfigureArbiter(failingOver); err != nil {
+			return errors.Wrap(err, "failed to configure stretch arbiter")
 		}
 	}
 
diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go
index bbec6a522..f01d06c1b 100644
--- a/pkg/operator/ceph/cluster/mon/health_test.go
+++ b/pkg/operator/ceph/cluster/mon/health_test.go
@@ -31,6 +31,7 @@ import (
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	clienttest "github.com/rook/rook/pkg/daemon/ceph/client/test"
 	"github.com/rook/rook/pkg/operator/ceph/config"
+	"github.com/rook/rook/pkg/operator/ceph/version"
 	testopk8s "github.com/rook/rook/pkg/operator/k8sutil/test"
 	"github.com/rook/rook/pkg/operator/test"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
@@ -153,6 +154,40 @@ func TestCheckHealth(t *testing.T) {
 	}
 }
 
+func TestSkipMonFailover(t *testing.T) {
+	c := New(&clusterd.Context{}, "ns", cephv1.ClusterSpec{}, nil, nil)
+	c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
+	monName := "arb"
+
+	t.Run("don't skip failover for non-stretch", func(t *testing.T) {
+		assert.NoError(t, c.allowFailover(monName))
+	})
+
+	t.Run("don't skip failover for non-arbiter", func(t *testing.T) {
+		c.spec.Mon.Count = 5
+		c.spec.Mon.StretchCluster = &cephv1.StretchClusterSpec{
+			Zones: []cephv1.StretchClusterZoneSpec{
+				{Name: "a"},
+				{Name: "b"},
+				{Name: "c", Arbiter: true},
+			},
+		}
+
+		assert.NoError(t, c.allowFailover(monName))
+	})
+
+	t.Run("skip failover for arbiter if an older version of ceph", func(t *testing.T) {
+		c.arbiterMon = monName
+		c.ClusterInfo.CephVersion = version.CephVersion{Major: 16, Minor: 2, Extra: 6}
+		assert.Error(t, c.allowFailover(monName))
+	})
+
+	t.Run("don't skip failover for arbiter if a newer version of ceph", func(t *testing.T) {
+		c.ClusterInfo.CephVersion = version.CephVersion{Major: 16, Minor: 2, Extra: 7}
+		assert.NoError(t, c.allowFailover(monName))
+	})
+}
+
 func TestEvictMonOnSameNode(t *testing.T) {
 	ctx := context.TODO()
 	clientset := test.New(t, 1)
diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go
index d987ef7fb..a5e793404 100644
--- a/pkg/operator/ceph/cluster/mon/mon.go
+++ b/pkg/operator/ceph/cluster/mon/mon.go
@@ -339,11 +339,20 @@ func (c *Cluster) isArbiterZone(zone string) bool {
 	return c.getArbiterZone() == zone
 }
 
-func (c *Cluster) ConfigureArbiter() error {
+func (c *Cluster) ConfigureArbiter(failingOver bool) error {
 	if c.arbiterMon == "" {
 		return errors.New("arbiter not specified for the stretch cluster")
 	}
 
+	failureDomain := c.stretchFailureDomainName()
+	if failingOver {
+		// Set the new mon tiebreaker
+		if err := cephclient.SetNewTiebreaker(c.context, c.ClusterInfo, c.arbiterMon); err != nil {
+			return errors.Wrap(err, "failed to set new mon tiebreaker")
+		}
+		return nil
+	}
+
 	monDump, err := cephclient.GetMonDump(c.context, c.ClusterInfo)
 	if err != nil {
 		logger.Warningf("attempting to enable arbiter after failed to detect if already enabled. %v", err)
@@ -355,7 +364,6 @@ func (c *Cluster) ConfigureArbiter() error {
 	// Wait for the CRUSH map to have at least two zones
 	// The timeout is relatively short since the operator will requeue the reconcile
 	// and try again at a higher level if not yet found
-	failureDomain := c.stretchFailureDomainName()
 	logger.Infof("enabling stretch mode... waiting for two failure domains of type %q to be found in the CRUSH map after OSD initialization", failureDomain)
 	pollInterval := 5 * time.Second
 	totalWaitTime := 2 * time.Minute