diff --git a/cmd/xtrabackup/run-backup/main.go b/cmd/xtrabackup/run-backup/main.go
index d18f4945d5..312d7b1205 100644
--- a/cmd/xtrabackup/run-backup/main.go
+++ b/cmd/xtrabackup/run-backup/main.go
@@ -8,19 +8,20 @@ import (
 	"log"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
 
-	"github.com/percona/percona-xtradb-cluster-operator/pkg/xtrabackup/api"
-	xbscapi "github.com/percona/percona-xtradb-cluster-operator/pkg/xtrabackup/api"
-	xbscserver "github.com/percona/percona-xtradb-cluster-operator/pkg/xtrabackup/server"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/status"
+
+	"github.com/percona/percona-xtradb-cluster-operator/pkg/xtrabackup/api"
+	xbscapi "github.com/percona/percona-xtradb-cluster-operator/pkg/xtrabackup/api"
+	xbscserver "github.com/percona/percona-xtradb-cluster-operator/pkg/xtrabackup/server"
 )
 
 func main() {
-
 	req := getRequestObject()
 	if req == nil {
 		log.Fatal("Failed to get request object")
@@ -91,18 +92,16 @@ func getRequestObject() *xbscapi.CreateBackupRequest {
 		BackupConfig: &api.BackupConfig{},
 	}
 
-	containerOptions := &xbscapi.ContainerOptions{}
-	if opts := os.Getenv("CONTAINER_OPTIONS"); opts != "" {
-		err := json.Unmarshal([]byte(opts), containerOptions)
-		if err != nil {
-			log.Fatalf("Failed to unmarshal container options: %v", err)
-		}
-	}
-
 	req.BackupName = os.Getenv("BACKUP_NAME")
 	req.BackupConfig.Destination = os.Getenv("BACKUP_DEST")
 	req.BackupConfig.VerifyTls = os.Getenv("VERIFY_TLS") == "true"
-	req.BackupConfig.ContainerOptions = containerOptions
+	req.BackupConfig.ContainerOptions = &xbscapi.ContainerOptions{
+		Args: &xbscapi.BackupContainerArgs{
+			Xtrabackup: strings.Fields(os.Getenv("XB_EXTRA_ARGS")),
+			Xbcloud:    strings.Fields(os.Getenv("XBCLOUD_EXTRA_ARGS")),
+			Xbstream:   strings.Fields(os.Getenv("XBSTREAM_EXTRA_ARGS")),
+		},
+	}
 
 	storageType := os.Getenv("STORAGE_TYPE")
 	switch storageType {
diff --git a/e2e-tests/demand-backup-cloud-pxb/run b/e2e-tests/demand-backup-cloud-pxb/run
index 31af3a2f2c..a0adc3e41b 100755
--- a/e2e-tests/demand-backup-cloud-pxb/run
+++ b/e2e-tests/demand-backup-cloud-pxb/run
@@ -27,6 +27,22 @@ get_container_options() {
 	fi
 }
 
+check_backup_container_options() {
+	local backup_name=$1
+	local cluster_name=$2
+
+	local pxc_pod="${cluster_name}-pxc-0"
+	if ! kubectl_bin logs "${pxc_pod}" -c xtrabackup | grep -E 'xbcloud put .*--parallel=2'; then
+		echo "xbcloud container options were not applied for backup ${backup_name}"
+		exit 1
+	fi
+
+	if ! kubectl_bin logs "${pxc_pod}" -c xtrabackup | grep -E 'xbstream .*--parallel=2'; then
+		echo "xbstream container options were not applied for backup ${backup_name}"
+		exit 1
+	fi
+}
+
 run_recovery_from_source() {
 	local storage_type=${1:-s3}
 	local backup_name=${2:-on-demand-backup-aws-s3}
@@ -91,19 +107,19 @@ wait_for_upload_to_start() {
 delete_backup_pod() {
 	local backup_name=$1
 	desc "Delete ${backup_name} pod during backup"
-	
-    local cluster_name=$(kubectl_bin get pxc-backup ${backup_name} -o jsonpath='{.spec.pxcCluster}')
-    if [[ -z $cluster_name ]]; then
-        echo "Cluster name is not set on backup ${backup_name}"
-        exit 1
-    fi
+
+	local cluster_name=$(kubectl_bin get pxc-backup ${backup_name} -o jsonpath='{.spec.pxcCluster}')
+	if [[ -z $cluster_name ]]; then
+		echo "Cluster name is not set on backup ${backup_name}"
+		exit 1
+	fi
 
 	sleep 2
 	backup_pod=$(kubectl_bin get pods --selector=percona.com/backup-job-name=xb-${backup_name} -o jsonpath='{.items[].metadata.name}')
 	echo "Waiting for pod/${backup_pod} to become Running"
 	kubectl_bin wait --for=jsonpath='{.status.phase}'=Running pod/${backup_pod} --timeout=120s
 
-    local pxc_pod="${cluster_name}-pxc-0"
+	local pxc_pod="${cluster_name}-pxc-0"
 
 	echo "Waiting for upload to start"
 	wait_for_upload_to_start ${pxc_pod}
@@ -119,20 +135,20 @@ check_cloud_storage_cleanup() {
 		exit 1
 	fi
 
-    local cluster_name=$(kubectl_bin get pxc-backup ${backup_name} -o jsonpath='{.spec.pxcCluster}')
-    if [[ -z $cluster_name ]]; then
-        echo "Cluster name is not set on backup ${backup_name}"
-        exit 1
-    fi
-
-    local pxc_pod="${cluster_name}-pxc-0"
-    if kubectl_bin logs ${pxc_pod} -c xtrabackup | grep 'Deleting Backup'; then
-        echo "Cleanup was performed."
-    else
-        echo "Something went wrong. Delete was not performed."
-        kubectl_bin logs ${pxc_pod} -c xtrabackup
-        exit 1
-    fi
+	local cluster_name=$(kubectl_bin get pxc-backup ${backup_name} -o jsonpath='{.spec.pxcCluster}')
+	if [[ -z $cluster_name ]]; then
+		echo "Cluster name is not set on backup ${backup_name}"
+		exit 1
+	fi
+
+	local pxc_pod="${cluster_name}-pxc-0"
+	if kubectl_bin logs ${pxc_pod} -c xtrabackup | grep 'Deleting Backup'; then
+		echo "Cleanup was performed."
+	else
+		echo "Something went wrong. Delete was not performed."
+		kubectl_bin logs ${pxc_pod} -c xtrabackup
+		exit 1
+	fi
 }
 
 run_backup_with_delete() {
@@ -181,6 +197,7 @@ main() {
 
 		desc "Run backup ${backup_name_aws} for $cluster cluster"
 		run_backup_with_delete "${backup_name_aws}"
+		check_backup_container_options "${backup_name_aws}" "$cluster"
 
 		desc "Run recovery from s3 for $cluster cluster"
 		run_recovery_check "$cluster" "${backup_name_aws}"
diff --git a/pkg/xtrabackup/job.go b/pkg/xtrabackup/job.go
index 0e16ff1855..7887b06ee0 100644
--- a/pkg/xtrabackup/job.go
+++ b/pkg/xtrabackup/job.go
@@ -3,14 +3,16 @@ package xtrabackup
 import (
 	"fmt"
 
-	pxcv1 "github.com/percona/percona-xtradb-cluster-operator/pkg/apis/pxc/v1"
-	"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc"
-	"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc/app"
-	"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc/app/statefulset"
+	"github.com/pkg/errors"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
+
+	pxcv1 "github.com/percona/percona-xtradb-cluster-operator/pkg/apis/pxc/v1"
+	"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc"
+	"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc/app"
+	"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc/app/statefulset"
 )
 
 func JobSpec(
@@ -42,9 +44,9 @@ func JobSpec(
 	var initContainers []corev1.Container
 	initContainers = append(initContainers, statefulset.BackupInitContainer(cluster, initImage, storage.ContainerSecurityContext))
 
-	envs, err := xtrabackupJobEnvVars(backup, storage, primaryPodHost)
+	envs, err := xtrabackupJobEnvVars(cluster, backup, storage, primaryPodHost)
 	if err != nil {
-		return batchv1.JobSpec{}, fmt.Errorf("failed to get xtrabackup job env vars: %w", err)
+		return batchv1.JobSpec{}, errors.Wrap(err, "failed to get xtrabackup job env vars")
 	}
 
 	container := corev1.Container{
@@ -94,6 +96,7 @@ func JobSpec(
 }
 
 func xtrabackupJobEnvVars(
+	cluster *pxcv1.PerconaXtraDBCluster,
 	backup *pxcv1.PerconaXtraDBClusterBackup,
 	storage *pxcv1.BackupStorageSpec,
 	primaryPodHost string,
@@ -116,5 +119,10 @@ func xtrabackupJobEnvVars(
 			Value: backup.Name,
 		},
 	}
+
+	if cluster.CompareVersionWith("1.20.0") >= 0 {
+		envs = append(envs, backup.Spec.ContainerOptions.GetEnvVar(cluster, backup.Spec.StorageName)...)
+	}
+
 	return envs, nil
 }
diff --git a/pkg/xtrabackup/job_test.go b/pkg/xtrabackup/job_test.go
index f50a1b3f66..f5e511200b 100644
--- a/pkg/xtrabackup/job_test.go
+++ b/pkg/xtrabackup/job_test.go
@@ -3,14 +3,16 @@ package xtrabackup
 import (
 	"testing"
 
-	pxcv1 "github.com/percona/percona-xtradb-cluster-operator/pkg/apis/pxc/v1"
-	"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc/app"
 	"github.com/stretchr/testify/assert"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
+
+	pxcv1 "github.com/percona/percona-xtradb-cluster-operator/pkg/apis/pxc/v1"
+	"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc/app"
+	"github.com/percona/percona-xtradb-cluster-operator/pkg/version"
 )
 
 func TestJobSpec(t *testing.T) {
@@ -28,10 +30,24 @@ func TestJobSpec(t *testing.T) {
 	spec := &pxcv1.PXCBackupSpec{
 		StorageName:           storageName,
 		ActiveDeadlineSeconds: &activeDeadlineSeconds,
+		ContainerOptions: &pxcv1.BackupContainerOptions{
+			Env: []corev1.EnvVar{
+				{
+					Name:  "CUSTOM_ENV",
+					Value: "custom-value",
+				},
+			},
+			Args: pxcv1.BackupContainerArgs{
+				Xtrabackup: []string{"--parallel=10"},
+				Xbcloud:    []string{"--parallel=10"},
+				Xbstream:   []string{"--parallel=10"},
+			},
+		},
 	}
 
 	cluster := &pxcv1.PerconaXtraDBCluster{
 		Spec: pxcv1.PerconaXtraDBClusterSpec{
+			CRVersion: version.Version(),
 			Backup: &pxcv1.BackupSpec{
 				Image:           backupImage,
 				ImagePullPolicy: corev1.PullIfNotPresent,
@@ -208,7 +224,7 @@ func TestJobSpec(t *testing.T) {
 	assert.Equal(t, app.BinVolumeMountPath, container.VolumeMounts[0].MountPath)
 
 	// Assert Environment Variables
-	assert.Len(t, container.Env, 4)
+	assert.Len(t, container.Env, 8)
 	envMap := make(map[string]string)
 	for _, env := range container.Env {
 		envMap[env.Name] = env.Value
@@ -216,4 +232,8 @@ func TestJobSpec(t *testing.T) {
 	assert.Equal(t, primaryPodHost, envMap["HOST"])
 	assert.Equal(t, string(pxcv1.BackupStorageS3), envMap["STORAGE_TYPE"])
 	assert.Equal(t, "true", envMap["VERIFY_TLS"])
+	assert.Equal(t, "--parallel=10", envMap["XB_EXTRA_ARGS"])
+	assert.Equal(t, "--parallel=10", envMap["XBCLOUD_EXTRA_ARGS"])
+	assert.Equal(t, "--parallel=10", envMap["XBSTREAM_EXTRA_ARGS"])
+	assert.Equal(t, "custom-value", envMap["CUSTOM_ENV"])
 }