diff --git a/pkg/controller/vitesscell/reconcile_vtgate.go b/pkg/controller/vitesscell/reconcile_vtgate.go
index 70559fcc..136e5055 100644
--- a/pkg/controller/vitesscell/reconcile_vtgate.go
+++ b/pkg/controller/vitesscell/reconcile_vtgate.go
@@ -71,7 +71,7 @@ func (m *secretCellsMapper) Map(ctx context.Context, obj client.Object) []reconc
 	return requests
 }
 
-func (r *ReconcileVitessCell) reconcileVtgate(ctx context.Context, vtc *planetscalev2.VitessCell) (reconcile.Result, error) {
+func (r *ReconcileVitessCell) reconcileVtgate(ctx context.Context, vtc *planetscalev2.VitessCell, mysqldImage string) (reconcile.Result, error) {
 	clusterName := vtc.Labels[planetscalev2.ClusterLabel]
 
 	key := client.ObjectKey{Namespace: vtc.Namespace, Name: vtgate.ServiceName(clusterName, vtc.Spec.Name)}
@@ -151,11 +151,11 @@ func (r *ReconcileVitessCell) reconcileVtgate(ctx context.Context, vtc *planetsc
 		Kind: &appsv1.Deployment{},
 
 		New: func(key client.ObjectKey) runtime.Object {
-			return vtgate.NewDeployment(key, spec)
+			return vtgate.NewDeployment(key, spec, mysqldImage)
 		},
 		UpdateInPlace: func(key client.ObjectKey, obj runtime.Object) {
 			newObj := obj.(*appsv1.Deployment)
-			vtgate.UpdateDeployment(newObj, spec)
+			vtgate.UpdateDeployment(newObj, spec, mysqldImage)
 		},
 		Status: func(key client.ObjectKey, obj runtime.Object) {
 			curObj := obj.(*appsv1.Deployment)
diff --git a/pkg/controller/vitesscell/vitesscell_controller.go b/pkg/controller/vitesscell/vitesscell_controller.go
index a0f62d75..025f7ef9 100644
--- a/pkg/controller/vitesscell/vitesscell_controller.go
+++ b/pkg/controller/vitesscell/vitesscell_controller.go
@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
+	apilabels "k8s.io/apimachinery/pkg/labels"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -195,8 +196,28 @@ func (r *ReconcileVitessCell) Reconcile(cctx context.Context, request reconcile.
 		resultBuilder.Error(err)
 	}
 
+	// List all VitessShard in the same cluster, we do this to determine what is the image used for the mysqld container
+	// Note that this is cheap because it comes from the local cache.
+	labels := map[string]string{
+		planetscalev2.ClusterLabel: vtc.Labels[planetscalev2.ClusterLabel],
+	}
+	opts := &client.ListOptions{
+		Namespace:     vtc.Namespace,
+		LabelSelector: apilabels.SelectorFromSet(labels),
+	}
+	vts := &planetscalev2.VitessShardList{}
+	if err := r.client.List(ctx, vts, opts); err != nil {
+		r.recorder.Eventf(vtc, corev1.EventTypeWarning, "ListFailed", "failed to list VitessShard objects: %v", err)
+		return resultBuilder.Error(err)
+	}
+
+	var mysqldImage string
+	if len(vts.Items) > 0 {
+		mysqldImage = vts.Items[0].Spec.Images.Mysqld.Image()
+	}
+
 	// Create/update vtgate deployments.
-	vtgateResult, err := r.reconcileVtgate(ctx, vtc)
+	vtgateResult, err := r.reconcileVtgate(ctx, vtc, mysqldImage)
 	resultBuilder.Merge(vtgateResult, err)
 
 	// Check which VitessKeyspaces are deployed to this cell.
diff --git a/pkg/controller/vitesscluster/reconcile_vtctld.go b/pkg/controller/vitesscluster/reconcile_vtctld.go
index 156a033f..3520544f 100644
--- a/pkg/controller/vitesscluster/reconcile_vtctld.go
+++ b/pkg/controller/vitesscluster/reconcile_vtctld.go
@@ -84,19 +84,19 @@ func (r *ReconcileVitessCluster) reconcileVtctld(ctx context.Context, vt *planet
 		Kind: &appsv1.Deployment{},
 
 		New: func(key client.ObjectKey) runtime.Object {
-			return vtctld.NewDeployment(key, specMap[key])
+			return vtctld.NewDeployment(key, specMap[key], vt.Spec.Images.Mysqld.Image())
 		},
 		UpdateInPlace: func(key client.ObjectKey, obj runtime.Object) {
 			newObj := obj.(*appsv1.Deployment)
 			if *vt.Spec.UpdateStrategy.Type == planetscalev2.ImmediateVitessClusterUpdateStrategyType {
-				vtctld.UpdateDeployment(newObj, specMap[key])
+				vtctld.UpdateDeployment(newObj, specMap[key], vt.Spec.Images.Mysqld.Image())
 				return
 			}
 			vtctld.UpdateDeploymentImmediate(newObj, specMap[key])
 		},
 		UpdateRollingInPlace: func(key client.ObjectKey, obj runtime.Object) {
 			newObj := obj.(*appsv1.Deployment)
-			vtctld.UpdateDeployment(newObj, specMap[key])
+			vtctld.UpdateDeployment(newObj, specMap[key], vt.Spec.Images.Mysqld.Image())
 		},
 		Status: func(key client.ObjectKey, obj runtime.Object) {
 			// This function will get called once for each Deployment.
diff --git a/pkg/controller/vitessshard/reconcile_backup_job.go b/pkg/controller/vitessshard/reconcile_backup_job.go
index a302fb04..ab341cba 100644
--- a/pkg/controller/vitessshard/reconcile_backup_job.go
+++ b/pkg/controller/vitessshard/reconcile_backup_job.go
@@ -141,7 +141,7 @@ func (r *ReconcileVitessShard) reconcileBackupJob(ctx context.Context, vts *plan
 		Kind: &corev1.Pod{},
 
 		New: func(key client.ObjectKey) runtime.Object {
-			return vttablet.NewBackupPod(key, specMap[key])
+			return vttablet.NewBackupPod(key, specMap[key], vts.Spec.Images.Mysqld.Image())
 		},
 		Status: func(key client.ObjectKey, obj runtime.Object) {
 			pod := obj.(*corev1.Pod)
diff --git a/pkg/controller/vitessshardreplication/reconcile_drain.go b/pkg/controller/vitessshardreplication/reconcile_drain.go
index a06ae282..3eca8fd4 100644
--- a/pkg/controller/vitessshardreplication/reconcile_drain.go
+++ b/pkg/controller/vitessshardreplication/reconcile_drain.go
@@ -19,13 +19,10 @@ package vitessshardreplication
 import (
 	"context"
 	"fmt"
-	"regexp"
-	"slices"
-	"strconv"
-	"strings"
 	"time"
 
 	"github.com/sirupsen/logrus"
+	"planetscale.dev/vitess-operator/pkg/operator/mysql"
 	"vitess.io/vitess/go/mysql/replication"
 	"vitess.io/vitess/go/vt/proto/tabletmanagerdata"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
@@ -508,7 +505,7 @@ func (r *ReconcileVitessShard) disableFastShutdown(
 			}
 		}
 
-		needsSafe, err := safeMysqldUpgrade(current, desiredImage)
+		needsSafe, err := mysql.DockerImageSafeUpgrade(current, desiredImage)
 		if err != nil {
 			return err
 		}
@@ -527,98 +524,6 @@ func (r *ReconcileVitessShard) disableFastShutdown(
 	return nil
 }
 
-var mysqlImageVersion = regexp.MustCompile(`^(\d+)\.(\d+)\.(\d+)`)
-
-func safeMysqldUpgrade(currentImage, desiredImage string) (bool, error) {
-	if currentImage == "" || desiredImage == "" {
-		// No action if we have unknown versions.
-		return false, nil
-	}
-
-	// Quick check so no regexp matching is needed for the most common
-	// case where nothing changes.
-	if desiredImage == currentImage {
-		return false, nil
-	}
-
-	currentParts := strings.SplitN(currentImage, ":", 2)
-	if len(currentParts) != 2 {
-		return false, nil
-	}
-
-	desiredParts := strings.SplitN(desiredImage, ":", 2)
-	if len(desiredParts) != 2 {
-		return false, nil
-	}
-
-	current := currentParts[1]
-	desired := desiredParts[1]
-
-	curStrParts := mysqlImageVersion.FindStringSubmatch(current)
-	if len(curStrParts) != 4 {
-		// Invalid version, assume that we need to do a safe upgrade.
-		return true, nil
-	}
-	dstStrParts := mysqlImageVersion.FindStringSubmatch(desired)
-	if len(dstStrParts) != 4 {
-		// Invalid version, assume that we need to do a safe upgrade.
-		return true, nil
-	}
-	if slices.Equal(curStrParts, dstStrParts) {
-		return false, nil
-	}
-	dstParts := make([]int, len(dstStrParts)-1)
-	curParts := make([]int, len(curStrParts)-1)
-	for i, part := range dstStrParts[1:] {
-		// We already matched with `\d_` so there's no
-		// way this can trigger an error.
-		dstParts[i], _ = strconv.Atoi(part)
-	}
-
-	for i, part := range curStrParts[1:] {
-		// We already matched with `\d_` so there's no
-		// way this can trigger an error.
-		curParts[i], _ = strconv.Atoi(part)
-	}
-
-	if dstParts[0] < curParts[0] {
-		return false, fmt.Errorf("cannot downgrade major version from %s to %s", current, desired)
-	}
-	if dstParts[0] == curParts[1] && dstParts[1] < curParts[1] {
-		return false, fmt.Errorf("cannot downgrade minor version from %s to %s", current, desired)
-	}
-
-	// Alright, here it gets more tricky. MySQL has had a complicated release history. For the 8.0 series,
-	// up to 8.0.34 at least (known at this point), it was not supported to downgrade patch releases
-	// as patch release could also include on-disk data format changes. This happened a number of times
-	// in practice as well, so this concern is real.
-	//
-	// MySQL though has announced a new release strategy, see:
-	// https://dev.mysql.com/blog-archive/introducing-mysql-innovation-and-long-term-support-lts-versions/
-	//
-	// With that release strategy, it will become possible that patch releases will be safe to downgrade
-	// as well and since the data format doesn't change on-disk anymore, it's also safe to upgrade with
-	// fast shutdown enabled.
-	// Specifically, it calls out that "MySQL 8.0.34+ will become bugfix only release (red)". This means
-	// that we use that version as a cut-off point here for when we need to disable fast shutdown or not.
-	if dstParts[0] == 8 && dstParts[1] == 0 && curParts[0] == 8 && curParts[1] == 0 {
-		// Our upgrade process stays within the 8.0.x version range.
-		if dstParts[2] >= 34 && curParts[2] >= 34 {
-			// No need for safe upgrade if both versions are 8.0.34 or higher.
-			return false, nil
-		}
-		// We can't downgrade within the 8.0.x series before 8.0.34.
-		if dstParts[2] < curParts[2] {
-			return false, fmt.Errorf("cannot downgrade patch version from %s to %s", current, desired)
-		}
-		// Always need safe upgrade if we change the patch release for 8.0.x before 8.0.34.
-		return dstParts[2] != curParts[2], nil
-	}
-
-	// For any major or minor version change we always need safe upgrade.
-	return dstParts[0] != curParts[0] || dstParts[1] != curParts[1], nil
-}
-
 type candidateInfo struct {
 	tablet   *topo.TabletInfo
 	position replication.Position
diff --git a/pkg/operator/environment/environment.go b/pkg/operator/environment/environment.go
index b712b8d0..6d48725d 100644
--- a/pkg/operator/environment/environment.go
+++ b/pkg/operator/environment/environment.go
@@ -32,7 +32,7 @@ import (
 
 var (
 	reconcileTimeout   time.Duration
-	mySQLServerVersion = "8.0.30-Vitess"
+	MySQLServerVersion = "8.0.30-Vitess"
 	// truncateUILen truncate queries in debug UIs to the given length. 0 means unlimited.
 	truncateUILen = 512
 	// truncateErrLen truncate queries in error logs to the given length. 0 means unlimited.
@@ -68,7 +68,7 @@ func ReconcileTimeout() time.Duration {
 // VtEnvironment gets the vitess environment to be used in the operator.
 func VtEnvironment() (*vtenv.Environment, error) {
 	return vtenv.New(vtenv.Options{
-		MySQLServerVersion: mySQLServerVersion,
+		MySQLServerVersion: MySQLServerVersion,
 		TruncateUILen:      truncateUILen,
 		TruncateErrLen:     truncateErrLen,
 	})
diff --git a/pkg/operator/mysql/mysql.go b/pkg/operator/mysql/mysql.go
new file mode 100644
index 00000000..16ffb3dd
--- /dev/null
+++ b/pkg/operator/mysql/mysql.go
@@ -0,0 +1,144 @@
+/*
+Copyright 2024 PlanetScale Inc.
+
+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 mysql
+
+import (
+	"fmt"
+	"regexp"
+	"slices"
+	"strconv"
+	"strings"
+
+	"planetscale.dev/vitess-operator/pkg/operator/environment"
+	"planetscale.dev/vitess-operator/pkg/operator/vitess"
+	"vitess.io/vitess/go/vt/sqlparser"
+)
+
+var imageVersionRegExp = regexp.MustCompile(`^(\d+)\.(\d+)\.(\d+)`)
+
+func UpdateMySQLServerVersion(flags vitess.Flags, mysqldImage string) {
+	value, err := dockerImageGetVersionToString(mysqldImage)
+	if err != nil {
+		return
+	}
+	flags["mysql_server_version"] = value
+	environment.MySQLServerVersion = value
+}
+
+func dockerImageGetVersionToString(currentVersionImage string) (string, error) {
+	currentVersionSlice := strings.SplitN(currentVersionImage, ":", 2)
+	if len(currentVersionSlice) != 2 {
+		return "", fmt.Errorf("could not parse the image name and image label, got: %s, but expected xx:xx", currentVersionImage)
+	}
+
+	label := currentVersionSlice[1]
+	_, err := sqlparser.ConvertMySQLVersionToCommentVersion(label)
+	if err != nil {
+		return "", err
+	}
+	return fmt.Sprintf("%s-Vitess", label), nil
+}
+
+func DockerImageSafeUpgrade(currentVersionImage, desiredVersionImage string) (bool, error) {
+	if currentVersionImage == "" || desiredVersionImage == "" {
+		// No action if we have unknown versions.
+		return false, nil
+	}
+
+	// Quick check so no regexp matching is needed for the most common
+	// case where nothing changes.
+	if desiredVersionImage == currentVersionImage {
+		return false, nil
+	}
+
+	currentParts := strings.SplitN(currentVersionImage, ":", 2)
+	if len(currentParts) != 2 {
+		return false, nil
+	}
+
+	desiredParts := strings.SplitN(desiredVersionImage, ":", 2)
+	if len(desiredParts) != 2 {
+		return false, nil
+	}
+
+	current := currentParts[1]
+	desired := desiredParts[1]
+
+	curStrParts := imageVersionRegExp.FindStringSubmatch(current)
+	if len(curStrParts) != 4 {
+		// Invalid version, assume that we need to do a safe upgrade.
+		return true, nil
+	}
+	dstStrParts := imageVersionRegExp.FindStringSubmatch(desired)
+	if len(dstStrParts) != 4 {
+		// Invalid version, assume that we need to do a safe upgrade.
+		return true, nil
+	}
+	if slices.Equal(curStrParts, dstStrParts) {
+		return false, nil
+	}
+	dstParts := make([]int, len(dstStrParts)-1)
+	curParts := make([]int, len(curStrParts)-1)
+	for i, part := range dstStrParts[1:] {
+		// We already matched with `\d_` so there's no
+		// way this can trigger an error.
+		dstParts[i], _ = strconv.Atoi(part)
+	}
+
+	for i, part := range curStrParts[1:] {
+		// We already matched with `\d_` so there's no
+		// way this can trigger an error.
+		curParts[i], _ = strconv.Atoi(part)
+	}
+
+	if dstParts[0] < curParts[0] {
+		return false, fmt.Errorf("cannot downgrade major version from %s to %s", current, desired)
+	}
+	if dstParts[0] == curParts[1] && dstParts[1] < curParts[1] {
+		return false, fmt.Errorf("cannot downgrade minor version from %s to %s", current, desired)
+	}
+
+	// Alright, here it gets more tricky. MySQL has had a complicated release history. For the 8.0 series,
+	// up to 8.0.34 at least (known at this point), it was not supported to downgrade patch releases
+	// as patch release could also include on-disk data format changes. This happened a number of times
+	// in practice as well, so this concern is real.
+	//
+	// MySQL though has announced a new release strategy, see:
+	// https://dev.mysql.com/blog-archive/introducing-mysql-innovation-and-long-term-support-lts-versions/
+	//
+	// With that release strategy, it will become possible that patch releases will be safe to downgrade
+	// as well and since the data format doesn't change on-disk anymore, it's also safe to upgrade with
+	// fast shutdown enabled.
+	// Specifically, it calls out that "MySQL 8.0.34+ will become bugfix only release (red)". This means
+	// that we use that version as a cut-off point here for when we need to disable fast shutdown or not.
+	if dstParts[0] == 8 && dstParts[1] == 0 && curParts[0] == 8 && curParts[1] == 0 {
+		// Our upgrade process stays within the 8.0.x version range.
+		if dstParts[2] >= 34 && curParts[2] >= 34 {
+			// No need for safe upgrade if both versions are 8.0.34 or higher.
+			return false, nil
+		}
+		// We can't downgrade within the 8.0.x series before 8.0.34.
+		if dstParts[2] < curParts[2] {
+			return false, fmt.Errorf("cannot downgrade patch version from %s to %s", current, desired)
+		}
+		// Always need safe upgrade if we change the patch release for 8.0.x before 8.0.34.
+		return dstParts[2] != curParts[2], nil
+	}
+
+	// For any major or minor version change we always need safe upgrade.
+	return dstParts[0] != curParts[0] || dstParts[1] != curParts[1], nil
+}
diff --git a/pkg/controller/vitessshardreplication/reconcile_drain_test.go b/pkg/operator/mysql/mysql_test.go
similarity index 97%
rename from pkg/controller/vitessshardreplication/reconcile_drain_test.go
rename to pkg/operator/mysql/mysql_test.go
index 734a4e89..c190dabf 100644
--- a/pkg/controller/vitessshardreplication/reconcile_drain_test.go
+++ b/pkg/operator/mysql/mysql_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package vitessshardreplication
+package mysql
 
 import (
 	"testing"
@@ -132,7 +132,7 @@ func TestSafeMysqldUpgrade(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			needsSafe, err := safeMysqldUpgrade(tt.current, tt.desired)
+			needsSafe, err := DockerImageSafeUpgrade(tt.current, tt.desired)
 			if tt.err != "" {
 				assert.EqualError(t, err, tt.err)
 			} else {
diff --git a/pkg/operator/vtctld/deployment.go b/pkg/operator/vtctld/deployment.go
index 391673e2..5901cf3e 100644
--- a/pkg/operator/vtctld/deployment.go
+++ b/pkg/operator/vtctld/deployment.go
@@ -24,6 +24,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/utils/pointer"
+	"planetscale.dev/vitess-operator/pkg/operator/mysql"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	planetscalev2 "planetscale.dev/vitess-operator/pkg/apis/planetscale/v2"
@@ -74,7 +75,7 @@ type Spec struct {
 }
 
 // NewDeployment creates a new Deployment object for vtctld.
-func NewDeployment(key client.ObjectKey, spec *Spec) *appsv1.Deployment {
+func NewDeployment(key client.ObjectKey, spec *Spec, mysqldImage string) *appsv1.Deployment {
 	// Fill in the immutable parts.
 	obj := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
@@ -88,7 +89,7 @@ func NewDeployment(key client.ObjectKey, spec *Spec) *appsv1.Deployment {
 		},
 	}
 	// Set everything else.
-	UpdateDeployment(obj, spec)
+	UpdateDeployment(obj, spec, mysqldImage)
 	return obj
 }
 
@@ -104,7 +105,7 @@ func UpdateDeploymentImmediate(obj *appsv1.Deployment, spec *Spec) {
 
 // UpdateDeployment updates the mutable parts of the vtctld Deployment
 // that should be changed as part of a gradual, rolling update.
-func UpdateDeployment(obj *appsv1.Deployment, spec *Spec) {
+func UpdateDeployment(obj *appsv1.Deployment, spec *Spec, mysqldImage string) {
 	UpdateDeploymentImmediate(obj, spec)
 
 	// Reset Pod template labels so we remove old ones.
@@ -131,6 +132,7 @@ func UpdateDeployment(obj *appsv1.Deployment, spec *Spec) {
 		key = strings.TrimLeft(key, "-")
 		flags[key] = value
 	}
+	mysql.UpdateMySQLServerVersion(flags, mysqldImage)
 
 	// Set only the Pod template fields we care about.
 	// Use functions from the `operator/update` package for lists
diff --git a/pkg/operator/vtgate/deployment.go b/pkg/operator/vtgate/deployment.go
index 3e34f831..a2882881 100644
--- a/pkg/operator/vtgate/deployment.go
+++ b/pkg/operator/vtgate/deployment.go
@@ -24,6 +24,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/utils/pointer"
+	"planetscale.dev/vitess-operator/pkg/operator/mysql"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	planetscalev2 "planetscale.dev/vitess-operator/pkg/apis/planetscale/v2"
@@ -86,7 +87,7 @@ type Spec struct {
 }
 
 // NewDeployment creates a new Deployment object for vtgate.
-func NewDeployment(key client.ObjectKey, spec *Spec) *appsv1.Deployment {
+func NewDeployment(key client.ObjectKey, spec *Spec, mysqldImage string) *appsv1.Deployment {
 	// Fill in the immutable parts.
 	obj := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
@@ -100,12 +101,12 @@ func NewDeployment(key client.ObjectKey, spec *Spec) *appsv1.Deployment {
 		},
 	}
 	// Set everything else.
-	UpdateDeployment(obj, spec)
+	UpdateDeployment(obj, spec, mysqldImage)
 	return obj
 }
 
 // UpdateDeployment updates the mutable parts of the vtgate Deployment.
-func UpdateDeployment(obj *appsv1.Deployment, spec *Spec) {
+func UpdateDeployment(obj *appsv1.Deployment, spec *Spec, mysqldImage string) {
 	// Set labels on the Deployment object.
 	update.Labels(&obj.Labels, spec.Labels)
 
@@ -223,6 +224,7 @@ func UpdateDeployment(obj *appsv1.Deployment, spec *Spec) {
 	flags := spec.baseFlags()
 
 	// Update the Pod template, container, and flags for various optional things.
+	mysql.UpdateMySQLServerVersion(flags, mysqldImage)
 	updateAuth(spec, flags, vtgateContainer, &obj.Spec.Template.Spec)
 	updateTransport(spec, flags, vtgateContainer, &obj.Spec.Template.Spec)
 	update.Volumes(&obj.Spec.Template.Spec.Volumes, spec.ExtraVolumes)
diff --git a/pkg/operator/vttablet/pod.go b/pkg/operator/vttablet/pod.go
index 392ea282..9dcc25f1 100644
--- a/pkg/operator/vttablet/pod.go
+++ b/pkg/operator/vttablet/pod.go
@@ -25,6 +25,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/utils/pointer"
+	"planetscale.dev/vitess-operator/pkg/operator/mysql"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	"vitess.io/vitess/go/vt/topo/topoproto"
@@ -86,6 +87,7 @@ func UpdatePod(obj *corev1.Pod, spec *Spec) {
 		key = strings.TrimLeft(key, "-")
 		vttabletAllFlags[key] = value
 	}
+	mysql.UpdateMySQLServerVersion(vttabletAllFlags, spec.Images.Mysqld.Image())
 
 	// Compute all operator-generated env vars first.
 	env := tabletEnvVars.Get(spec)
@@ -165,13 +167,16 @@ func UpdatePod(obj *corev1.Pod, spec *Spec) {
 	var mysqldContainer *corev1.Container
 	var mysqldExporterContainer *corev1.Container
 
+	mysqlctldAllFlags := mysqlctldFlags.Get(spec)
+	mysql.UpdateMySQLServerVersion(mysqlctldAllFlags, spec.Images.Mysqld.Image())
+
 	if spec.Mysqld != nil {
 		mysqldContainer = &corev1.Container{
 			Name:            MysqldContainerName,
 			Image:           spec.Images.Mysqld.Image(),
 			ImagePullPolicy: spec.ImagePullPolicies.Mysqld,
 			Command:         []string{mysqldCommand},
-			Args:            mysqlctldFlags.Get(spec).FormatArgs(),
+			Args:            mysqlctldAllFlags.FormatArgs(),
 			Ports: []corev1.ContainerPort{
 				{
 					Name:          planetscalev2.DefaultMysqlPortName,
diff --git a/pkg/operator/vttablet/vtbackup_pod.go b/pkg/operator/vttablet/vtbackup_pod.go
index 8dcd80d6..5823cdb1 100644
--- a/pkg/operator/vttablet/vtbackup_pod.go
+++ b/pkg/operator/vttablet/vtbackup_pod.go
@@ -23,6 +23,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
+	"planetscale.dev/vitess-operator/pkg/operator/mysql"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	planetscalev2 "planetscale.dev/vitess-operator/pkg/apis/planetscale/v2"
@@ -92,7 +93,7 @@ func InitialBackupPodName(clusterName, keyspaceName string, keyRange planetscale
 
 // NewBackupPod creates a new vtbackup Pod, which is like a special kind of
 // minimal tablet used to run backups as a batch process.
-func NewBackupPod(key client.ObjectKey, backupSpec *BackupSpec) *corev1.Pod {
+func NewBackupPod(key client.ObjectKey, backupSpec *BackupSpec, mysqldImage string) *corev1.Pod {
 	tabletSpec := backupSpec.TabletSpec
 
 	// Include vttablet env vars, since we run some vttablet code like backups.
@@ -132,6 +133,8 @@ func NewBackupPod(key client.ObjectKey, backupSpec *BackupSpec) *corev1.Pod {
 	// Make a copy of Resources since it contains pointers.
 	update.ResourceRequirements(&containerResources, &tabletSpec.Mysqld.Resources)
 
+	vtbackupAllFlags := vtbackupFlags.Get(backupSpec)
+	mysql.UpdateMySQLServerVersion(vtbackupAllFlags, mysqldImage)
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace:   key.Namespace,
@@ -171,7 +174,7 @@ func NewBackupPod(key client.ObjectKey, backupSpec *BackupSpec) *corev1.Pod {
 					Image:           tabletSpec.Images.Mysqld.Image(),
 					ImagePullPolicy: tabletSpec.ImagePullPolicies.Mysqld,
 					Command:         []string{vtbackupCommand},
-					Args:            vtbackupFlags.Get(backupSpec).FormatArgs(),
+					Args:            vtbackupAllFlags.FormatArgs(),
 					Resources:       containerResources,
 					SecurityContext: securityContext,
 					Env:             env,