diff --git a/.circleci/config.yml b/.circleci/config.yml
index 2bf464fc1a..7aba89af53 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -70,7 +70,7 @@ commands:
         type: string
       consul-k8s-image:
         type: string
-        default: "docker.mirror.hashicorp.services/hashicorpdev/consul-k8s-control-plane:latest"
+        default: "ishustava/consul-k8s-dev:02-08-2022-d21554d8"
       go-path:
         type: string
         default: "/home/circleci/.go_workspace"
diff --git a/.github/workflows/golangci-lint-cli.yml b/.github/workflows/golangci-lint-cli.yml
index f4c395b23c..50860bbabb 100644
--- a/.github/workflows/golangci-lint-cli.yml
+++ b/.github/workflows/golangci-lint-cli.yml
@@ -1,4 +1,4 @@
-name: golangci-lint-acceptance
+name: golangci-lint-cli
 on:
   push:
     tags:
diff --git a/acceptance/framework/consul/cli_cluster.go b/acceptance/framework/consul/cli_cluster.go
index 5493533909..71662b88d8 100644
--- a/acceptance/framework/consul/cli_cluster.go
+++ b/acceptance/framework/consul/cli_cluster.go
@@ -51,7 +51,7 @@ func NewCLICluster(
 	ctx environment.TestContext,
 	cfg *config.TestConfig,
 	releaseName string,
-) Cluster {
+) *CLICluster {
 
 	// Create the namespace so the PSPs, SCCs, and enterprise secret can be created in the right namespace.
 	createOrUpdateNamespace(t, ctx.KubernetesClient(t), consulNS)
diff --git a/acceptance/framework/consul/consul_cluster.go b/acceptance/framework/consul/consul_cluster.go
index cda22d9322..f2586b5c5b 100644
--- a/acceptance/framework/consul/consul_cluster.go
+++ b/acceptance/framework/consul/consul_cluster.go
@@ -40,6 +40,11 @@ type Cluster interface {
 // HelmCluster implements Cluster and uses Helm
 // to create, destroy, and upgrade consul
 type HelmCluster struct {
+	// ACLToken is an optional ACL token that will be used to create
+	// a Consul API client. If not provided, we will attempt to read
+	// a bootstrap token from a Kubernetes secret stored in the cluster.
+	ACLToken string
+
 	ctx                environment.TestContext
 	helmOptions        *helm.Options
 	releaseName        string
@@ -55,7 +60,7 @@ func NewHelmCluster(
 	ctx environment.TestContext,
 	cfg *config.TestConfig,
 	releaseName string,
-) Cluster {
+) *HelmCluster {
 
 	if cfg.EnablePodSecurityPolicies {
 		configurePodSecurityPolicies(t, ctx.KubernetesClient(t), cfg, ctx.KubectlOptions(t).Namespace)
@@ -234,21 +239,26 @@ func (h *HelmCluster) SetupConsulClient(t *testing.T, secure bool) *api.Client {
 		config.TLSConfig.InsecureSkipVerify = true
 		config.Scheme = "https"
 
-		// Get the ACL token. First, attempt to read it from the bootstrap token (this will be true in primary Consul servers).
-		// If the bootstrap token doesn't exist, it means we are running against a secondary cluster
-		// and will try to read the replication token from the federation secret.
-		// In secondary servers, we don't create a bootstrap token since ACLs are only bootstrapped in the primary.
-		// Instead, we provide a replication token that serves the role of the bootstrap token.
-		aclSecret, err := h.kubernetesClient.CoreV1().Secrets(namespace).Get(context.Background(), h.releaseName+"-consul-bootstrap-acl-token", metav1.GetOptions{})
-		if err != nil && errors.IsNotFound(err) {
-			federationSecret := fmt.Sprintf("%s-consul-federation", h.releaseName)
-			aclSecret, err = h.kubernetesClient.CoreV1().Secrets(namespace).Get(context.Background(), federationSecret, metav1.GetOptions{})
-			require.NoError(t, err)
-			config.Token = string(aclSecret.Data["replicationToken"])
-		} else if err == nil {
-			config.Token = string(aclSecret.Data["token"])
+		// If an ACL token is provided, we'll use that instead of trying to find it.
+		if h.ACLToken != "" {
+			config.Token = h.ACLToken
 		} else {
-			require.NoError(t, err)
+			// Get the ACL token. First, attempt to read it from the bootstrap token (this will be true in primary Consul servers).
+			// If the bootstrap token doesn't exist, it means we are running against a secondary cluster
+			// and will try to read the replication token from the federation secret.
+			// In secondary servers, we don't create a bootstrap token since ACLs are only bootstrapped in the primary.
+			// Instead, we provide a replication token that serves the role of the bootstrap token.
+			aclSecret, err := h.kubernetesClient.CoreV1().Secrets(namespace).Get(context.Background(), h.releaseName+"-consul-bootstrap-acl-token", metav1.GetOptions{})
+			if err != nil && errors.IsNotFound(err) {
+				federationSecret := fmt.Sprintf("%s-consul-federation", h.releaseName)
+				aclSecret, err = h.kubernetesClient.CoreV1().Secrets(namespace).Get(context.Background(), federationSecret, metav1.GetOptions{})
+				require.NoError(t, err)
+				config.Token = string(aclSecret.Data["replicationToken"])
+			} else if err == nil {
+				config.Token = string(aclSecret.Data["token"])
+			} else {
+				require.NoError(t, err)
+			}
 		}
 	}
 
diff --git a/acceptance/framework/consul/consul_cluster_test.go b/acceptance/framework/consul/consul_cluster_test.go
index 1a9ac604db..3fb6710c8b 100644
--- a/acceptance/framework/consul/consul_cluster_test.go
+++ b/acceptance/framework/consul/consul_cluster_test.go
@@ -65,7 +65,7 @@ func TestNewHelmCluster(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			cluster := NewHelmCluster(t, tt.helmValues, &ctx{}, &config.TestConfig{ConsulImage: "test-config-image"}, "test")
-			require.Equal(t, cluster.(*HelmCluster).helmOptions.SetValues, tt.want)
+			require.Equal(t, cluster.helmOptions.SetValues, tt.want)
 		})
 	}
 }
diff --git a/acceptance/framework/helpers/helpers.go b/acceptance/framework/helpers/helpers.go
index 410cf38a6c..17dd805123 100644
--- a/acceptance/framework/helpers/helpers.go
+++ b/acceptance/framework/helpers/helpers.go
@@ -148,46 +148,3 @@ func MergeMaps(a, b map[string]string) {
 		a[k] = v
 	}
 }
-
-// VerifyFederation checks that the WAN federation between servers is successful
-// by first checking members are alive from the perspective of both servers.
-// If secure is true, it will also check that the ACL replication is running on the secondary server.
-func VerifyFederation(t *testing.T, primaryClient, secondaryClient *api.Client, releaseName string, secure bool) {
-	retrier := &retry.Timer{Timeout: 5 * time.Minute, Wait: 1 * time.Second}
-	start := time.Now()
-
-	// Check that server in dc1 is healthy from the perspective of the server in dc2, and vice versa.
-	// We're calling the Consul health API, as opposed to checking serf membership status,
-	// because we need to make sure that the federated servers can make API calls and forward requests
-	// from one server to another. From running tests in CI for a while and using serf membership status before,
-	// we've noticed that the status could be "alive" as soon as the server in the secondary cluster joins the primary
-	// and then switch to "failed". This would require us to check that the status is "alive" is showing consistently for
-	// some amount of time, which could be quite flakey. Calling the API in another datacenter allows us to check that
-	// each server can forward calls to another, which is what we need for connect.
-	retry.RunWith(retrier, t, func(r *retry.R) {
-		secondaryServerHealth, _, err := primaryClient.Health().Node(fmt.Sprintf("%s-consul-server-0", releaseName), &api.QueryOptions{Datacenter: "dc2"})
-		require.NoError(r, err)
-		require.Equal(r, secondaryServerHealth.AggregatedStatus(), api.HealthPassing)
-
-		primaryServerHealth, _, err := secondaryClient.Health().Node(fmt.Sprintf("%s-consul-server-0", releaseName), &api.QueryOptions{Datacenter: "dc1"})
-		require.NoError(r, err)
-		require.Equal(r, primaryServerHealth.AggregatedStatus(), api.HealthPassing)
-
-		if secure {
-			replicationStatus, _, err := secondaryClient.ACL().Replication(nil)
-			require.NoError(r, err)
-			require.True(r, replicationStatus.Enabled)
-			require.True(r, replicationStatus.Running)
-		}
-	})
-
-	logger.Logf(t, "Took %s to verify federation", time.Since(start))
-}
-
-// MergeMaps will merge the values in b with values in a and save in a.
-// If there are conflicts, the values in b will overwrite the values in a.
-func MergeMaps(a, b map[string]string) {
-	for k, v := range b {
-		a[k] = v
-	}
-}
diff --git a/acceptance/tests/vault/helpers.go b/acceptance/tests/vault/helpers.go
index d1439ee32e..7a690ec67a 100644
--- a/acceptance/tests/vault/helpers.go
+++ b/acceptance/tests/vault/helpers.go
@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	"github.com/hashicorp/consul-k8s/acceptance/framework/logger"
+	"github.com/hashicorp/go-uuid"
 	vapi "github.com/hashicorp/vault/api"
 	"github.com/stretchr/testify/require"
 )
@@ -16,6 +17,10 @@ const (
 path "consul/data/secret/gossip" {
   capabilities = ["read"]
 }`
+	replicationTokenPolicy = `
+path "consul/data/secret/replication" {
+  capabilities = ["read", "update"]
+}`
 
 	// connectCAPolicy allows Consul to bootstrap all certificates for the service mesh in Vault.
 	// Adapted from https://www.consul.io/docs/connect/ca/vault#consul-managed-pki-paths.
@@ -113,7 +118,7 @@ func configureKubernetesAuthRoles(t *testing.T, vaultClient *vapi.Client, consul
 	params = map[string]interface{}{
 		"bound_service_account_names":      consulServerServiceAccountName,
 		"bound_service_account_namespaces": ns,
-		"policies":                         fmt.Sprintf("consul-gossip,connect-ca,consul-server-%s", datacenter),
+		"policies":                         fmt.Sprintf("consul-gossip,connect-ca,consul-server-%s,consul-replication-token", datacenter),
 		"ttl":                              "24h",
 	}
 	_, err = vaultClient.Logical().Write(fmt.Sprintf("auth/%s/role/consul-server", authPath), params)
@@ -178,3 +183,42 @@ path %q {
 
 	return certificateIssuePath
 }
+
+// configureReplicationTokenVaultSecret generates a replication token secret ID,
+// stores it in vault as a secret and configures a policy to access it.
+func configureReplicationTokenVaultSecret(t *testing.T, vaultClient *vapi.Client, consulReleaseName, ns string, authMethodPaths ...string) string {
+	// Create the Vault Policy for the replication token.
+	logger.Log(t, "Creating replication token policy")
+	err := vaultClient.Sys().PutPolicy("consul-replication-token", replicationTokenPolicy)
+	require.NoError(t, err)
+
+	// Generate the token secret.
+	token, err := uuid.GenerateUUID()
+	require.NoError(t, err)
+
+	// Create the replication token secret.
+	logger.Log(t, "Creating the replication token secret")
+	params := map[string]interface{}{
+		"data": map[string]interface{}{
+			"replication": token,
+		},
+	}
+	_, err = vaultClient.Logical().Write("consul/data/secret/replication", params)
+	require.NoError(t, err)
+
+	logger.Log(t, "Creating kubernetes auth role for the server-acl-init job")
+	serverACLInitSAName := fmt.Sprintf("%s-consul-server-acl-init", consulReleaseName)
+	params = map[string]interface{}{
+		"bound_service_account_names":      serverACLInitSAName,
+		"bound_service_account_namespaces": ns,
+		"policies":                         "consul-replication-token",
+		"ttl":                              "24h",
+	}
+
+	for _, authMethodPath := range authMethodPaths {
+		_, err := vaultClient.Logical().Write(fmt.Sprintf("auth/%s/role/server-acl-init", authMethodPath), params)
+		require.NoError(t, err)
+	}
+
+	return token
+}
diff --git a/acceptance/tests/vault/vault_wan_fed_test.go b/acceptance/tests/vault/vault_wan_fed_test.go
index 77689ef40a..ed25f7bfea 100644
--- a/acceptance/tests/vault/vault_wan_fed_test.go
+++ b/acceptance/tests/vault/vault_wan_fed_test.go
@@ -12,6 +12,7 @@ import (
 	"github.com/hashicorp/consul-k8s/acceptance/framework/k8s"
 	"github.com/hashicorp/consul-k8s/acceptance/framework/logger"
 	"github.com/hashicorp/consul-k8s/acceptance/framework/vault"
+	"github.com/hashicorp/consul/api"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
@@ -115,6 +116,8 @@ func TestVault_WANFederationViaGateways(t *testing.T) {
 	primaryCertPath := configurePKICertificates(t, vaultClient, consulReleaseName, ns, "dc1")
 	secondaryCertPath := configurePKICertificates(t, vaultClient, consulReleaseName, ns, "dc2")
 
+	replicationToken := configureReplicationTokenVaultSecret(t, vaultClient, consulReleaseName, ns, "kubernetes", "kubernetes-dc2")
+
 	// Move Vault CA secret from primary to secondary so that we can mount it to pods in the
 	// secondary cluster.
 	vaultCASecretName := vault.CASecretName(vaultReleaseName)
@@ -135,7 +138,6 @@ func TestVault_WANFederationViaGateways(t *testing.T) {
 
 		// TLS config.
 		"global.tls.enabled":           "true",
-		"global.tls.httpsOnly":         "false",
 		"global.tls.enableAutoEncrypt": "true",
 		"global.tls.caCert.secretName": "pki/cert/ca",
 		"server.serverCert.secretName": primaryCertPath,
@@ -144,6 +146,12 @@ func TestVault_WANFederationViaGateways(t *testing.T) {
 		"global.gossipEncryption.secretName": "consul/data/secret/gossip",
 		"global.gossipEncryption.secretKey":  "gossip",
 
+		// ACL config.
+		"global.acls.manageSystemACLs":            "true",
+		"global.acls.createReplicationToken":      "true",
+		"global.acls.replicationToken.secretName": "consul/data/secret/replication",
+		"global.acls.replicationToken.secretKey":  "replication",
+
 		// Mesh config.
 		"connectInject.enabled": "true",
 		"controller.enabled":    "true",
@@ -156,12 +164,13 @@ func TestVault_WANFederationViaGateways(t *testing.T) {
 		"server.extraVolumes[0].load": "false",
 
 		// Vault config.
-		"global.secretsBackend.vault.enabled":          "true",
-		"global.secretsBackend.vault.consulServerRole": "consul-server",
-		"global.secretsBackend.vault.consulClientRole": "consul-client",
-		"global.secretsBackend.vault.consulCARole":     "consul-ca",
-		"global.secretsBackend.vault.ca.secretName":    vaultCASecretName,
-		"global.secretsBackend.vault.ca.secretKey":     "tls.crt",
+		"global.secretsBackend.vault.enabled":              "true",
+		"global.secretsBackend.vault.consulServerRole":     "consul-server",
+		"global.secretsBackend.vault.consulClientRole":     "consul-client",
+		"global.secretsBackend.vault.consulCARole":         "consul-ca",
+		"global.secretsBackend.vault.manageSystemACLsRole": "server-acl-init",
+		"global.secretsBackend.vault.ca.secretName":        vaultCASecretName,
+		"global.secretsBackend.vault.ca.secretKey":         "tls.crt",
 	}
 
 	if cfg.UseKind {
@@ -182,7 +191,6 @@ func TestVault_WANFederationViaGateways(t *testing.T) {
 
 		// TLS config.
 		"global.tls.enabled":           "true",
-		"global.tls.httpsOnly":         "false",
 		"global.tls.enableAutoEncrypt": "true",
 		"global.tls.caCert.secretName": "pki/cert/ca",
 		"server.serverCert.secretName": secondaryCertPath,
@@ -191,6 +199,11 @@ func TestVault_WANFederationViaGateways(t *testing.T) {
 		"global.gossipEncryption.secretName": "consul/data/secret/gossip",
 		"global.gossipEncryption.secretKey":  "gossip",
 
+		// ACL config.
+		"global.acls.manageSystemACLs":            "true",
+		"global.acls.replicationToken.secretName": "consul/data/secret/replication",
+		"global.acls.replicationToken.secretKey":  "replication",
+
 		// Mesh config.
 		"connectInject.enabled": "true",
 		"meshGateway.enabled":   "true",
@@ -203,13 +216,14 @@ func TestVault_WANFederationViaGateways(t *testing.T) {
 		"server.extraConfig":          serverExtraConfig,
 
 		// Vault config.
-		"global.secretsBackend.vault.enabled":          "true",
-		"global.secretsBackend.vault.consulServerRole": "consul-server",
-		"global.secretsBackend.vault.consulClientRole": "consul-client",
-		"global.secretsBackend.vault.consulCARole":     "consul-ca",
-		"global.secretsBackend.vault.ca.secretName":    vaultCASecretName,
-		"global.secretsBackend.vault.ca.secretKey":     "tls.crt",
-		"global.secretsBackend.vault.agentAnnotations": fmt.Sprintf("vault.hashicorp.com/tls-server-name: %s-vault", vaultReleaseName),
+		"global.secretsBackend.vault.enabled":              "true",
+		"global.secretsBackend.vault.consulServerRole":     "consul-server",
+		"global.secretsBackend.vault.consulClientRole":     "consul-client",
+		"global.secretsBackend.vault.consulCARole":         "consul-ca",
+		"global.secretsBackend.vault.manageSystemACLsRole": "server-acl-init",
+		"global.secretsBackend.vault.ca.secretName":        vaultCASecretName,
+		"global.secretsBackend.vault.ca.secretKey":         "tls.crt",
+		"global.secretsBackend.vault.agentAnnotations":     fmt.Sprintf("vault.hashicorp.com/tls-server-name: %s-vault", vaultReleaseName),
 	}
 
 	if cfg.UseKind {
@@ -223,9 +237,10 @@ func TestVault_WANFederationViaGateways(t *testing.T) {
 
 	// Verify federation between servers.
 	logger.Log(t, "verifying federation was successful")
-	primaryClient := primaryConsulCluster.SetupConsulClient(t, false)
-	secondaryClient := secondaryConsulCluster.SetupConsulClient(t, false)
-	helpers.VerifyFederation(t, primaryClient, secondaryClient, consulReleaseName, false)
+	primaryClient := primaryConsulCluster.SetupConsulClient(t, true)
+	secondaryConsulCluster.ACLToken = replicationToken
+	secondaryClient := secondaryConsulCluster.SetupConsulClient(t, true)
+	helpers.VerifyFederation(t, primaryClient, secondaryClient, consulReleaseName, true)
 
 	// Create a ProxyDefaults resource to configure services to use the mesh
 	// gateways.
@@ -243,6 +258,19 @@ func TestVault_WANFederationViaGateways(t *testing.T) {
 	logger.Log(t, "creating static-client in dc1")
 	k8s.DeployKustomize(t, primaryCtx.KubectlOptions(t), cfg.NoCleanupOnFailure, cfg.DebugDirectory, "../fixtures/cases/static-client-multi-dc")
 
+	logger.Log(t, "creating intention")
+	_, _, err = primaryClient.ConfigEntries().Set(&api.ServiceIntentionsConfigEntry{
+		Kind: api.ServiceIntentions,
+		Name: "static-server",
+		Sources: []*api.SourceIntention{
+			{
+				Name:   "static-client",
+				Action: api.IntentionActionAllow,
+			},
+		},
+	}, nil)
+	require.NoError(t, err)
+
 	logger.Log(t, "checking that connection is successful")
 	k8s.CheckStaticServerConnectionSuccessful(t, primaryCtx.KubectlOptions(t), "http://localhost:1234")
 }
diff --git a/charts/consul/templates/_helpers.tpl b/charts/consul/templates/_helpers.tpl
index 277f744d08..28a3fdf0ed 100644
--- a/charts/consul/templates/_helpers.tpl
+++ b/charts/consul/templates/_helpers.tpl
@@ -59,6 +59,20 @@ as well as the global.name setting.
 {{- if .Values.global.tls -}}{{- if .Values.global.tls.serverAdditionalIPSANs -}}{{- range $ipsan := .Values.global.tls.serverAdditionalIPSANs }},{{ $ipsan }} {{- end -}}{{- end -}}{{- end -}}
 {{- end -}}
 
+{{- define "consul.vaultReplicationTokenTemplate" -}}
+|
+          {{ "{{" }}- with secret "{{ .Values.global.acls.replicationToken.secretName }}" -{{ "}}" }}
+          {{ "{{" }}- {{ printf ".Data.data.%s" .Values.global.acls.replicationToken.secretKey }} -{{ "}}" }}
+          {{ "{{" }}- end -{{ "}}" }}
+{{- end -}}
+
+{{- define "consul.vaultReplicationTokenConfigTemplate" -}}
+|
+          {{ "{{" }}- with secret "{{ .Values.global.acls.replicationToken.secretName }}" -{{ "}}" }}
+          acl { tokens { agent = "{{ "{{" }}- {{ printf ".Data.data.%s" .Values.global.acls.replicationToken.secretKey }} -{{ "}}" }}", replication = "{{ "{{" }}- {{ printf ".Data.data.%s" .Values.global.acls.replicationToken.secretKey }} -{{ "}}" }}" }}
+          {{ "{{" }}- end -{{ "}}" }}
+{{- end -}}
+
 {{/*
 Sets up the extra-from-values config file passed to consul and then uses sed to do any necessary
 substitution for HOST_IP/POD_IP/HOSTNAME. Useful for dogstats telemetry. The output file
diff --git a/charts/consul/templates/server-acl-init-job.yaml b/charts/consul/templates/server-acl-init-job.yaml
index 5195e7975e..27bc8c1bf0 100644
--- a/charts/consul/templates/server-acl-init-job.yaml
+++ b/charts/consul/templates/server-acl-init-job.yaml
@@ -4,7 +4,9 @@
 {{- if and .Values.global.acls.createReplicationToken (not .Values.global.acls.manageSystemACLs) }}{{ fail "if global.acls.createReplicationToken is true, global.acls.manageSystemACLs must be true" }}{{ end -}}
 {{- if .Values.global.bootstrapACLs }}{{ fail "global.bootstrapACLs was removed, use global.acls.manageSystemACLs instead" }}{{ end -}}
 {{- if .Values.global.acls.manageSystemACLs }}
-{{- /* We don't render this job when server.updatePartition > 0 because that
+{{- if and .Values.global.secretsBackend.vault.enabled .Values.global.acls.replicationToken.secretName (not .Values.global.secretsBackend.vault.manageSystemACLsRole) }}{{ fail "global.secretsBackend.vault.manageSystemACLsRole must be set if global.secretsBackend.vault.enabled is true and global.acls.replicationToken is provided" }}{{ end -}}
+{{- if or (and .Values.global.acls.replicationToken.secretName (not .Values.global.acls.replicationToken.secretKey))  (and .Values.global.acls.replicationToken.secretKey (not .Values.global.acls.replicationToken.secretName))}}{{ fail "both global.acls.replicationToken.secretKey and global.acls.replicationToken.secretName must be set if one of them is provided" }}{{ end -}}
+  {{- /* We don't render this job when server.updatePartition > 0 because that
     means a server rollout is in progress and this job won't complete unless
     the rollout is finished (which won't happen until the partition is 0).
     If we ran it in this case, then the job would not complete which would cause
@@ -37,13 +39,21 @@ spec:
         {{- if (and .Values.global.secretsBackend.vault.enabled .Values.global.tls.enabled) }}
         "vault.hashicorp.com/agent-pre-populate-only": "true"
         "vault.hashicorp.com/agent-inject": "true"
-        "vault.hashicorp.com/role": {{ .Values.global.secretsBackend.vault.consulCARole }}
         "vault.hashicorp.com/agent-inject-secret-serverca.crt": {{ .Values.global.tls.caCert.secretName }}
         "vault.hashicorp.com/agent-inject-template-serverca.crt": {{ template "consul.serverTLSCATemplate" . }}
+        {{- if .Values.global.secretsBackend.vault.manageSystemACLsRole }}
+        "vault.hashicorp.com/role": {{ .Values.global.secretsBackend.vault.manageSystemACLsRole }}
+        {{- else if .Values.global.tls.enabled }}
+        "vault.hashicorp.com/role": {{ .Values.global.secretsBackend.vault.consulCARole }}
+        {{- end }}
         {{- if and .Values.global.secretsBackend.vault.ca.secretName .Values.global.secretsBackend.vault.ca.secretKey }}
         "vault.hashicorp.com/agent-extra-secret": "{{ .Values.global.secretsBackend.vault.ca.secretName }}"
         "vault.hashicorp.com/ca-cert": "/vault/custom/{{ .Values.global.secretsBackend.vault.ca.secretKey }}"
         {{- end }}
+        {{- if .Values.global.acls.replicationToken.secretName }}
+        "vault.hashicorp.com/agent-inject-secret-replication-token": "{{ .Values.global.acls.replicationToken.secretName }}"
+        "vault.hashicorp.com/agent-inject-template-replication-token":  {{ template "consul.vaultReplicationTokenTemplate" . }}
+        {{- end }}
         {{- if .Values.global.secretsBackend.vault.agentAnnotations }}
         {{ tpl .Values.global.secretsBackend.vault.agentAnnotations . | nindent 8 | trim }}
         {{- end }}
@@ -51,7 +61,7 @@ spec:
     spec:
       restartPolicy: Never
       serviceAccountName: {{ template "consul.fullname" . }}-server-acl-init
-      {{- if (or .Values.global.tls.enabled (and .Values.global.acls.replicationToken.secretName .Values.global.acls.replicationToken.secretKey) (and .Values.global.acls.bootstrapToken.secretName .Values.global.acls.bootstrapToken.secretKey)) }}
+      {{- if (or .Values.global.tls.enabled .Values.global.acls.replicationToken.secretName (and .Values.global.acls.bootstrapToken.secretName .Values.global.acls.bootstrapToken.secretKey)) }}
       volumes:
         {{- if and .Values.global.tls.enabled (not .Values.global.secretsBackend.vault.enabled) }}
         - name: consul-ca-cert
@@ -72,7 +82,7 @@ spec:
             items:
               - key: {{ .Values.global.acls.bootstrapToken.secretKey }}
                 path: bootstrap-token
-        {{- else if (and .Values.global.acls.replicationToken.secretName .Values.global.acls.replicationToken.secretKey) }}
+        {{- else if and .Values.global.acls.replicationToken.secretName (not .Values.global.secretsBackend.vault.enabled) }}
         - name: acl-replication-token
           secret:
             secretName: {{ .Values.global.acls.replicationToken.secretName }}
@@ -89,7 +99,7 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
-          {{- if (or .Values.global.tls.enabled (and .Values.global.acls.replicationToken.secretName .Values.global.acls.replicationToken.secretKey) (and .Values.global.acls.bootstrapToken.secretName .Values.global.acls.bootstrapToken.secretKey)) }}
+          {{- if (or .Values.global.tls.enabled .Values.global.acls.replicationToken.secretName (and .Values.global.acls.bootstrapToken.secretName .Values.global.acls.bootstrapToken.secretKey)) }}
           volumeMounts:
             {{- if and .Values.global.tls.enabled (not .Values.global.secretsBackend.vault.enabled) }}
             - name: consul-ca-cert
@@ -100,7 +110,7 @@ spec:
             - name: bootstrap-token
               mountPath: /consul/acl/tokens
               readOnly: true
-            {{- else if (and .Values.global.acls.replicationToken.secretName .Values.global.acls.replicationToken.secretKey) }}
+            {{- else if and .Values.global.acls.replicationToken.secretName (not .Values.global.secretsBackend.vault.enabled) }}
             - name: acl-replication-token
               mountPath: /consul/acl/tokens
               readOnly: true
@@ -232,9 +242,13 @@ spec:
 
                 {{- if (and .Values.global.acls.bootstrapToken.secretName .Values.global.acls.bootstrapToken.secretKey) }}
                 -bootstrap-token-file=/consul/acl/tokens/bootstrap-token \
-                {{- else if (and .Values.global.acls.replicationToken.secretName .Values.global.acls.replicationToken.secretKey) }}
+                {{- else if .Values.global.acls.replicationToken.secretName }}
+                {{- if .Values.global.secretsBackend.vault.enabled }}
+                -acl-replication-token-file=/vault/secrets/replication-token \
+                {{- else }}
                 -acl-replication-token-file=/consul/acl/tokens/acl-replication-token \
                 {{- end }}
+                {{- end }}
 
                 {{- if .Values.controller.enabled }}
                 -create-controller-token=true \
diff --git a/charts/consul/templates/server-statefulset.yaml b/charts/consul/templates/server-statefulset.yaml
index 0a6c16778b..74e1b4f63c 100644
--- a/charts/consul/templates/server-statefulset.yaml
+++ b/charts/consul/templates/server-statefulset.yaml
@@ -75,6 +75,10 @@ spec:
         "vault.hashicorp.com/agent-inject-secret-serverca.crt": {{ .Values.global.tls.caCert.secretName }}
         "vault.hashicorp.com/agent-inject-template-serverca.crt": {{ include "consul.serverTLSCATemplate" . }}
         {{- end }}
+        {{- if (and .Values.global.acls.replicationToken.secretName (not .Values.global.acls.createReplicationToken)) }}
+        "vault.hashicorp.com/agent-inject-secret-replication-token-config.hcl": "{{ .Values.global.acls.replicationToken.secretName }}"
+        "vault.hashicorp.com/agent-inject-template-replication-token-config.hcl":  {{ template "consul.vaultReplicationTokenConfigTemplate" . }}
+        {{- end }}
         {{- if .Values.global.secretsBackend.vault.agentAnnotations }}
         {{ tpl .Values.global.secretsBackend.vault.agentAnnotations . | nindent 8 | trim }}
         {{- end }}
@@ -217,7 +221,7 @@ spec:
             - name: CONSUL_LICENSE_PATH
               value: /consul/license/{{ .Values.global.enterpriseLicense.secretKey }}
             {{- end }}
-            {{- if (and .Values.global.acls.replicationToken.secretName .Values.global.acls.replicationToken.secretKey) }}
+            {{- if (and .Values.global.acls.replicationToken.secretName .Values.global.acls.replicationToken.secretKey (not .Values.global.secretsBackend.vault.enabled)) }}
             - name: ACL_REPLICATION_TOKEN
               valueFrom:
                 secretKeyRef:
@@ -294,8 +298,12 @@ spec:
                 -hcl="connect { enable_mesh_gateway_wan_federation = true }" \
                 {{- end }}
                 {{- if (and .Values.global.acls.replicationToken.secretName .Values.global.acls.replicationToken.secretKey) }}
+                {{- if (and .Values.global.secretsBackend.vault.enabled (not .Values.global.acls.createReplicationToken)) }}
+                -config-file=/vault/secrets/replication-token-config.hcl \
+                {{- else }}
                 -hcl="acl { tokens { agent = \"${ACL_REPLICATION_TOKEN}\", replication = \"${ACL_REPLICATION_TOKEN}\" } }" \
                 {{- end }}
+                {{- end }}
                 {{- if .Values.ui.enabled }}
                 -ui \
                 {{- if .Values.ui.dashboardURLTemplates.service }}
diff --git a/charts/consul/test/unit/server-acl-init-job.bats b/charts/consul/test/unit/server-acl-init-job.bats
index 1b84b5860b..a096a8aeb8 100644
--- a/charts/consul/test/unit/server-acl-init-job.bats
+++ b/charts/consul/test/unit/server-acl-init-job.bats
@@ -726,6 +726,72 @@ load _helpers
   [ "${actual}" = "/vault/custom/tls.crt" ]
 }
 
+#--------------------------------------------------------------------
+# Replication token in Vault
+
+@test "serverACLInit/Job: vault replication token can be provided" {
+  cd `chart_dir`
+  local object=$(helm template \
+    -s templates/server-acl-init-job.yaml  \
+    --set 'global.acls.manageSystemACLs=true' \
+    --set 'global.tls.enabled=true' \
+    --set 'global.tls.enableAutoEncrypt=true' \
+    --set 'global.tls.caCert.secretName=foo' \
+    --set 'global.secretsBackend.vault.enabled=true' \
+    --set 'global.secretsBackend.vault.consulClientRole=foo' \
+    --set 'global.secretsBackend.vault.consulServerRole=test' \
+    --set 'global.secretsBackend.vault.consulCARole=carole' \
+    --set 'global.secretsBackend.vault.manageSystemACLsRole=acl-role' \
+    --set 'global.acls.replicationToken.secretName=/vault/secret' \
+    --set 'global.acls.replicationToken.secretKey=token' \
+    . | tee /dev/stderr |
+      yq -r '.spec.template' | tee /dev/stderr)
+
+  # Check that the role is set.
+  local actual=$(echo $object | yq -r '.metadata.annotations."vault.hashicorp.com/role"')
+  [ "${actual}" = "acl-role" ]
+
+  # Check Vault secret annotations.
+  local actual=$(echo $object | yq -r '.metadata.annotations."vault.hashicorp.com/agent-inject-secret-replication-token"')
+  [ "${actual}" = "/vault/secret" ]
+
+  local actual=$(echo $object | yq -r '.metadata.annotations."vault.hashicorp.com/agent-inject-template-replication-token"')
+  local expected=$'{{- with secret \"/vault/secret\" -}}\n{{- .Data.data.token -}}\n{{- end -}}'
+  [ "${actual}" = "${expected}" ]
+
+  # Check that replication token Kubernetes secret volumes and volumeMounts are not attached.
+  local actual=$(echo $object | yq -r '.metadata.annotations."vault.hashicorp.com/agent-inject-secret-replication-token"')
+  [ "${actual}" = "/vault/secret" ]
+
+  local actual=$(echo $object | jq -r '.spec.volumes')
+  [ "${actual}" = "null" ]
+
+  local actual=$(echo $object | jq -r '.spec.containers[] | select(.name="post-install-job").volumeMounts')
+  [ "${actual}" = "null" ]
+
+  # Check that the replication token flag is set to the path of the Vault secret.
+  local actual=$(echo $object | jq -r '.spec.containers[] | select(.name="post-install-job").command | any(contains("-acl-replication-token-file=/vault/secrets/replication-token"))')
+  [ "${actual}" = "true" ]
+}
+
+@test "serverACLInit/Job: manageSystemACLsRole is required when Vault is enabled and replication token is set" {
+  cd `chart_dir`
+  run helm template \
+      -s templates/server-acl-init-job.yaml  \
+      --set 'global.acls.manageSystemACLs=true' \
+      --set 'global.acls.replicationToken.secretName=/vault/secret' \
+      --set 'global.acls.replicationToken.secretKey=foo' \
+      --set 'global.tls.enabled=true' \
+      --set 'global.tls.enableAutoEncrypt=true' \
+      --set 'global.tls.caCert.secretName=foo' \
+      --set 'global.secretsBackend.vault.enabled=true' \
+      --set 'global.secretsBackend.vault.consulClientRole=foo' \
+      --set 'global.secretsBackend.vault.consulServerRole=test' \
+      --set 'global.secretsBackend.vault.consulCARole=carole' .
+  [ "$status" -eq 1 ]
+  [[ "$output" =~ "global.secretsBackend.vault.manageSystemACLsRole must be set if global.secretsBackend.vault.enabled is true and global.acls.replicationToken is provided" ]]
+}
+
 #--------------------------------------------------------------------
 # Vault agent annotations
 
@@ -1230,59 +1296,31 @@ load _helpers
 #--------------------------------------------------------------------
 # global.acls.replicationToken
 
-@test "serverACLInit/Job: -acl-replication-token-file is not set by default" {
+@test "serverACLInit/Job: replicationToken.secretKey is required when replicationToken.secretName is set" {
   cd `chart_dir`
-  local object=$(helm template \
+  run helm template \
       -s templates/server-acl-init-job.yaml  \
       --set 'global.acls.manageSystemACLs=true' \
-      . | tee /dev/stderr)
-
-  # Test the flag is not set.
-  local actual=$(echo "$object" |
-    yq '.spec.template.spec.containers[0].command | any(contains("-acl-replication-token-file"))' | tee /dev/stderr)
-  [ "${actual}" = "false" ]
-
-  # Test the volume doesn't exist
-  local actual=$(echo "$object" |
-    yq '.spec.template.spec.volumes | length == 0' | tee /dev/stderr)
-  [ "${actual}" = "true" ]
-
-  # Test the volume mount doesn't exist
-  local actual=$(echo "$object" |
-    yq '.spec.template.spec.containers[0].volumeMounts | length == 0' | tee /dev/stderr)
-  [ "${actual}" = "true" ]
+      --set 'global.acls.replicationToken.secretName=name' \ .
+  [ "$status" -eq 1 ]
+  [[ "$output" =~ "both global.acls.replicationToken.secretKey and global.acls.replicationToken.secretName must be set if one of them is provided" ]]
 }
 
-@test "serverACLInit/Job: -acl-replication-token-file is not set when acls.replicationToken.secretName is set but secretKey is not" {
+@test "serverACLInit/Job: replicationToken.secretName is required when replicationToken.secretKey is set" {
   cd `chart_dir`
-  local object=$(helm template \
+  run helm template \
       -s templates/server-acl-init-job.yaml  \
       --set 'global.acls.manageSystemACLs=true' \
-      --set 'global.acls.replicationToken.secretName=name' \
-      . | tee /dev/stderr)
-
-  # Test the flag is not set.
-  local actual=$(echo "$object" |
-    yq '.spec.template.spec.containers[0].command | any(contains("-acl-replication-token-file"))' | tee /dev/stderr)
-  [ "${actual}" = "false" ]
-
-  # Test the volume doesn't exist
-  local actual=$(echo "$object" |
-    yq '.spec.template.spec.volumes | length == 0' | tee /dev/stderr)
-  [ "${actual}" = "true" ]
-
-  # Test the volume mount doesn't exist
-  local actual=$(echo "$object" |
-    yq '.spec.template.spec.containers[0].volumeMounts | length == 0' | tee /dev/stderr)
-  [ "${actual}" = "true" ]
+      --set 'global.acls.replicationToken.secretKey=key' \ .
+  [ "$status" -eq 1 ]
+  [[ "$output" =~ "both global.acls.replicationToken.secretKey and global.acls.replicationToken.secretName must be set if one of them is provided" ]]
 }
 
-@test "serverACLInit/Job: -acl-replication-token-file is not set when acls.replicationToken.secretKey is set but secretName is not" {
+@test "serverACLInit/Job: -acl-replication-token-file is not set by default" {
   cd `chart_dir`
   local object=$(helm template \
       -s templates/server-acl-init-job.yaml  \
       --set 'global.acls.manageSystemACLs=true' \
-      --set 'global.acls.replicationToken.secretKey=key' \
       . | tee /dev/stderr)
 
   # Test the flag is not set.
diff --git a/charts/consul/test/unit/server-config-configmap.bats b/charts/consul/test/unit/server-config-configmap.bats
index cea3ef7eff..ecb96599e8 100755
--- a/charts/consul/test/unit/server-config-configmap.bats
+++ b/charts/consul/test/unit/server-config-configmap.bats
@@ -168,28 +168,6 @@ load _helpers
   [ "${actual}" = "null" ]
 }
 
-@test "server/ConfigMap: enable_token_replication is not set when acls.replicationToken.secretName is set but secretKey is not" {
-  cd `chart_dir`
-  local actual=$(helm template \
-      -s templates/server-config-configmap.yaml  \
-      --set 'global.acls.manageSystemACLs=true' \
-      --set 'global.acls.replicationToken.secretName=name' \
-      . | tee /dev/stderr |
-      yq -r '.data["acl-config.json"]' | yq -r '.acl.enable_token_replication' | tee /dev/stderr)
-  [ "${actual}" = "null" ]
-}
-
-@test "server/ConfigMap: enable_token_replication is not set when acls.replicationToken.secretKey is set but secretName is not" {
-  cd `chart_dir`
-  local actual=$(helm template \
-      -s templates/server-config-configmap.yaml  \
-      --set 'global.acls.manageSystemACLs=true' \
-      --set 'global.acls.replicationToken.secretKey=key' \
-      . | tee /dev/stderr |
-      yq -r '.data["acl-config.json"]' | yq -r '.acl.enable_token_replication' | tee /dev/stderr)
-  [ "${actual}" = "null" ]
-}
-
 @test "server/ConfigMap: enable_token_replication is set when acls.replicationToken.secretKey and secretName are set" {
   cd `chart_dir`
   local actual=$(helm template \
diff --git a/charts/consul/test/unit/server-statefulset.bats b/charts/consul/test/unit/server-statefulset.bats
index 65e3abe511..7c49a7a001 100755
--- a/charts/consul/test/unit/server-statefulset.bats
+++ b/charts/consul/test/unit/server-statefulset.bats
@@ -1872,6 +1872,40 @@ load _helpers
   [ "${actual}" = "bar" ]
 }
 
+#--------------------------------------------------------------------
+# Vault replication token
+
+@test "server/StatefulSet: vault replication token is configured when secret provided and createReplicationToken is false" {
+  cd `chart_dir`
+  local object=$(helm template \
+    -s templates/server-statefulset.yaml  \
+    --set 'global.secretsBackend.vault.enabled=true' \
+    --set 'global.secretsBackend.vault.consulClientRole=test' \
+    --set 'global.secretsBackend.vault.consulServerRole=foo' \
+    --set 'global.acls.replicationToken.secretName=vault/replication-token' \
+    --set 'global.acls.replicationToken.secretKey=token' \
+    . | tee /dev/stderr |
+      yq -r '.spec.template' | tee /dev/stderr)
+
+  # Check that Vault annotations are set.
+  local actual="$(echo $object |
+      yq -r '.metadata.annotations["vault.hashicorp.com/agent-inject-secret-replication-token-config.hcl"]' | tee /dev/stderr)"
+  [ "${actual}" = "vault/replication-token" ]
+
+  local actual="$(echo $object |
+      yq -r '.metadata.annotations["vault.hashicorp.com/agent-inject-template-replication-token-config.hcl"]' | tee /dev/stderr)"
+  local expected=$'{{- with secret \"vault/replication-token\" -}}\nacl { tokens { agent = \"{{- .Data.data.token -}}\", replication = \"{{- .Data.data.token -}}\" }}\n{{- end -}}'
+  [ "${actual}" = "${expected}" ]
+
+  # Check that ACL_REPLICATION_TOKEN env var is not provided.
+  local actual="$(echo $object | yq -r '.spec.containers[] | select(.name=="consul").env[] | select(.name=="ACL_REPLICATION_TOKEN")' | tee /dev/stderr)"
+  [ "${actual}" = "" ]
+
+  # Check that path to Vault secret config is provided to the command.
+  local actual="$(echo $object | yq -r '.spec.containers[] | select(.name=="consul").command | any(contains("-config-file=/vault/secrets/replication-token-config.hcl"))' | tee /dev/stderr)"
+  [ "${actual}" = "true" ]
+}
+
 #--------------------------------------------------------------------
 # ui.dashboardURLTemplates.service
 
diff --git a/charts/consul/values.yaml b/charts/consul/values.yaml
index fac5b836be..3f2e36cc38 100644
--- a/charts/consul/values.yaml
+++ b/charts/consul/values.yaml
@@ -138,7 +138,10 @@ global:
       # The Vault role for the Consul server.
       # The role must be connected to the Consul server's service account and
       # have a policy with read capabilities for the following secrets:
-      # - gossip encryption key defined by `global.gossipEncryption.secretName`.
+      # - gossip encryption key defined by `global.gossipEncryption.secretName`
+      # - certificate issue path defined by `server.serverCert.secretName`
+      # - CA certificate defined by `global.tls.caCert.secretName`
+      # - replication token defined by `global.acls.replicationToken.secretName` if `global.federation.enabled` is `true`
       # To discover the service account name of the Consul server, run
       #     ```shell-session
       #     $ helm template --show-only templates/server-serviceaccount.yaml <release-name> hashicorp/consul
@@ -150,13 +153,23 @@ global:
       # The role must be connected to the Consul client's service account and
       # have a policy with read capabilities for the following secrets:
       # - gossip encryption key defined by `global.gossipEncryption.secretName`.
-      # To discover the service account name of the Consul server, run
+      # To discover the service account name of the Consul client, run
       #     ```shell-session
-      #     $ helm template --show-only templates/client-daemonset.yaml <release-name> charts/consul
+      #     $ helm template --show-only templates/client-serviceaccount.yaml <release-name> charts/consul
       #     ```
       # and check the name of `metadata.name`.
       consulClientRole: ""
 
+      # A Vault role to allow Kubernetes job that manages ACLs for this Helm chart (`server-acl-init`)
+      # to read and update Vault secrets for the Consul's bootstrap and replication tokens.
+      # This role must be bound the `server-acl-init`'s service account.
+      # To discover the service account name of the `server-acl-init` job, run
+      #     ```shell-session
+      #     $ helm template --show-only templates/server-acl-init-serviceaccount.yaml <release-name> charts/consul
+      #     ```
+      # and check the name of `metadata.name`.
+      manageSystemACLsRole: ""
+
       # This value defines additional annotations for
       # Vault agent on any pods where it'll be running.
       # This should be formatted as a multi-line string.
@@ -376,9 +389,9 @@ global:
     # and create ACL tokens and policies.
     # This value is ignored if `bootstrapToken` is also set.
     replicationToken:
-      # The name of the Kubernetes secret.
+      # The name of the Kubernetes secret or the path of the secret in Vault.
       secretName: null
-      # The key of the Kubernetes secret.
+      # The key of the Kubernetes or Vault secret.
       secretKey: null
 
   # [Enterprise Only] This value refers to a Kubernetes secret that you have created
diff --git a/control-plane/subcommand/server-acl-init/command.go b/control-plane/subcommand/server-acl-init/command.go
index e665d1e3f0..64906d44ad 100644
--- a/control-plane/subcommand/server-acl-init/command.go
+++ b/control-plane/subcommand/server-acl-init/command.go
@@ -325,7 +325,7 @@ func (c *Command) Run(args []string) int {
 		// the provided token to create policies and tokens for the rest of the components.
 		c.log.Info("Bootstrap token is provided so skipping Consul server ACL bootstrapping")
 		bootstrapToken = providedBootstrapToken
-	} else if c.flagACLReplicationTokenFile != "" {
+	} else if c.flagACLReplicationTokenFile != "" && !c.flagCreateACLReplicationToken {
 		// If ACL replication is enabled, we don't need to ACL bootstrap the servers
 		// since they will be performing replication.
 		// We can use the replication token as our bootstrap token because it
@@ -694,7 +694,11 @@ func (c *Command) Run(args []string) int {
 		}
 		// Policy must be global because it replicates from the primary DC
 		// and so the primary DC needs to be able to accept the token.
-		err = c.createGlobalACL(common.ACLReplicationTokenName, rules, consulDC, isPrimary, consulClient)
+		if aclReplicationToken != "" {
+			err = c.createGlobalACLWithSecretID(common.ACLReplicationTokenName, rules, consulDC, isPrimary, consulClient, aclReplicationToken)
+		} else {
+			err = c.createGlobalACL(common.ACLReplicationTokenName, rules, consulDC, isPrimary, consulClient)
+		}
 		if err != nil {
 			c.log.Error(err.Error())
 			return 1
diff --git a/control-plane/subcommand/server-acl-init/command_test.go b/control-plane/subcommand/server-acl-init/command_test.go
index da3cfc3de3..bbccfcc20d 100644
--- a/control-plane/subcommand/server-acl-init/command_test.go
+++ b/control-plane/subcommand/server-acl-init/command_test.go
@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"os"
 	"strconv"
 	"strings"
 	"testing"
@@ -19,6 +20,7 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	v1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/fake"
 
@@ -325,6 +327,70 @@ func TestRun_TokensPrimaryDC(t *testing.T) {
 	}
 }
 
+func TestRun_ReplicationTokenPrimaryDC_WithProvidedSecretID(t *testing.T) {
+	t.Parallel()
+
+	k8s, testSvr := completeSetup(t)
+	defer testSvr.Stop()
+	require := require.New(t)
+
+	replicationToken := "123e4567-e89b-12d3-a456-426614174000"
+	replicationTokenFile, err := ioutil.TempFile("", "replicationtoken")
+	require.NoError(err)
+	defer os.Remove(replicationTokenFile.Name())
+
+	replicationTokenFile.WriteString(replicationToken)
+	// Run the command.
+	ui := cli.NewMockUi()
+	cmd := Command{
+		UI:        ui,
+		clientset: k8s,
+	}
+	cmd.init()
+	cmdArgs := []string{
+		"-timeout=1m",
+		"-k8s-namespace=" + ns,
+		"-server-address", strings.Split(testSvr.HTTPAddr, ":")[0],
+		"-server-port", strings.Split(testSvr.HTTPAddr, ":")[1],
+		"-resource-prefix=" + resourcePrefix,
+		"-create-acl-replication-token",
+		"-acl-replication-token-file", replicationTokenFile.Name(),
+	}
+
+	responseCode := cmd.Run(cmdArgs)
+	require.Equal(0, responseCode, ui.ErrorWriter.String())
+
+	// Check that this token is created.
+	consul, err := api.NewClient(&api.Config{
+		Address: testSvr.HTTPAddr,
+		Token:   replicationToken,
+	})
+	require.NoError(err)
+	token, _, err := consul.ACL().TokenReadSelf(nil)
+	require.NoError(err)
+
+	for _, policyLink := range token.Policies {
+		policy := policyExists(t, policyLink.Name, consul)
+		require.Nil(policy.Datacenters)
+
+		// Test that the token was not created as a Kubernetes Secret.
+		_, err := k8s.CoreV1().Secrets(ns).Get(context.Background(), resourcePrefix+"-acl-replication-acl-token", metav1.GetOptions{})
+		require.True(k8serrors.IsNotFound(err))
+	}
+
+	// Test that if the same command is run again, it doesn't error.
+	t.Run(t.Name()+"-retried", func(t *testing.T) {
+		ui = cli.NewMockUi()
+		cmd = Command{
+			UI:        ui,
+			clientset: k8s,
+		}
+		cmd.init()
+		responseCode = cmd.Run(cmdArgs)
+		require.Equal(0, responseCode, ui.ErrorWriter.String())
+	})
+}
+
 // Test creating each token type when replication is enabled.
 func TestRun_TokensReplicatedDC(t *testing.T) {
 	t.Parallel()
diff --git a/control-plane/subcommand/server-acl-init/create_or_update.go b/control-plane/subcommand/server-acl-init/create_or_update.go
index a17d78c6e3..80dca054bf 100644
--- a/control-plane/subcommand/server-acl-init/create_or_update.go
+++ b/control-plane/subcommand/server-acl-init/create_or_update.go
@@ -13,26 +13,34 @@ import (
 // createLocalACL creates a policy and acl token for this dc (datacenter), i.e.
 // the policy is only valid for this datacenter and the token is a local token.
 func (c *Command) createLocalACL(name, rules, dc string, isPrimary bool, consulClient *api.Client) error {
-	return c.createACL(name, rules, true, dc, isPrimary, consulClient)
+	return c.createACL(name, rules, true, dc, isPrimary, consulClient, "")
 }
 
 // createGlobalACL creates a global policy and acl token. The policy is valid
 // for all datacenters and the token is global. dc must be passed because the
 // policy name may have the datacenter name appended.
 func (c *Command) createGlobalACL(name, rules, dc string, isPrimary bool, consulClient *api.Client) error {
-	return c.createACL(name, rules, false, dc, isPrimary, consulClient)
+	return c.createACL(name, rules, false, dc, isPrimary, consulClient, "")
+}
+
+// createGlobalACLWithSecretID creates a global policy and acl token with provided secret ID.
+func (c *Command) createGlobalACLWithSecretID(name, rules, dc string, isPrimary bool, consulClient *api.Client, secretID string) error {
+	return c.createACL(name, rules, false, dc, isPrimary, consulClient, secretID)
 }
 
 // createACL creates a policy with rules and name. If localToken is true then
 // the token will be a local token and the policy will be scoped to only dc.
 // If localToken is false, the policy will be global.
 // The token will be written to a Kubernetes secret.
-func (c *Command) createACL(name, rules string, localToken bool, dc string, isPrimary bool, consulClient *api.Client) error {
+// When secretID is provided, we will use that value for the created token and
+// will skip writing it to a Kubernetes secret (because in this case we assume that
+// this value already exists in some secrets storage).
+func (c *Command) createACL(name, rules string, localToken bool, dc string, isPrimary bool, consulClient *api.Client, secretID string) error {
 	// Create policy with the given rules.
 	policyName := fmt.Sprintf("%s-token", name)
 	if c.flagFederation && !isPrimary {
 		// If performing ACL replication, we must ensure policy names are
-		// globally unique so we append the datacenter name but only in secondary datacenters..
+		// globally unique so we append the datacenter name but only in secondary datacenters.
 		policyName += fmt.Sprintf("-%s", dc)
 	}
 	var datacenters []string
@@ -53,21 +61,37 @@ func (c *Command) createACL(name, rules string, localToken bool, dc string, isPr
 		return err
 	}
 
-	// Check if the secret already exists, if so, we assume the ACL has already been
-	// created and return.
-	secretName := c.withPrefix(name + "-acl-token")
-	_, err = c.clientset.CoreV1().Secrets(c.flagK8sNamespace).Get(c.ctx, secretName, metav1.GetOptions{})
-	if err == nil {
-		c.log.Info(fmt.Sprintf("Secret %q already exists", secretName))
-		return nil
-	}
-
 	// Create token for the policy if the secret did not exist previously.
 	tokenTmpl := api.ACLToken{
 		Description: fmt.Sprintf("%s Token", policyTmpl.Name),
 		Policies:    []*api.ACLTokenPolicyLink{{Name: policyTmpl.Name}},
 		Local:       localToken,
 	}
+
+	// Check if the replication token already exists in some form.
+	secretName := c.withPrefix(name + "-acl-token")
+	// When secretID is not provided, we assume that replication token should exist
+	// as a Kubernetes secret.
+	if secretID == "" {
+		// Check if the secret already exists, if so, we assume the ACL has already been
+		// created and return.
+		_, err = c.clientset.CoreV1().Secrets(c.flagK8sNamespace).Get(c.ctx, secretName, metav1.GetOptions{})
+		if err == nil {
+			c.log.Info(fmt.Sprintf("Secret %q already exists", secretName))
+			return nil
+		}
+	} else {
+		// If secretID is provided, we check if the token with secretID already exists in Consul
+		// and exit if it does. Otherwise, set the secretID to the provided value.
+		_, _, err = consulClient.ACL().TokenReadSelf(&api.QueryOptions{Token: secretID})
+		if err == nil {
+			c.log.Info("ACL replication token already exists; skipping creation")
+			return nil
+		} else {
+			tokenTmpl.SecretID = secretID
+		}
+	}
+
 	var token string
 	err = c.untilSucceeds(fmt.Sprintf("creating token for policy %s", policyTmpl.Name),
 		func() error {
@@ -81,25 +105,28 @@ func (c *Command) createACL(name, rules string, localToken bool, dc string, isPr
 		return err
 	}
 
-	// Write token to a Kubernetes secret.
-	return c.untilSucceeds(fmt.Sprintf("writing Secret for token %s", policyTmpl.Name),
-		func() error {
-			secret := &apiv1.Secret{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:   secretName,
-					Labels: map[string]string{common.CLILabelKey: common.CLILabelValue},
-				},
-				Data: map[string][]byte{
-					common.ACLTokenSecretKey: []byte(token),
-				},
-			}
-			_, err := c.clientset.CoreV1().Secrets(c.flagK8sNamespace).Create(c.ctx, secret, metav1.CreateOptions{})
-			return err
-		})
+	if secretID == "" {
+		// Write token to a Kubernetes secret.
+		return c.untilSucceeds(fmt.Sprintf("writing Secret for token %s", policyTmpl.Name),
+			func() error {
+				secret := &apiv1.Secret{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:   secretName,
+						Labels: map[string]string{common.CLILabelKey: common.CLILabelValue},
+					},
+					Data: map[string][]byte{
+						common.ACLTokenSecretKey: []byte(token),
+					},
+				}
+				_, err := c.clientset.CoreV1().Secrets(c.flagK8sNamespace).Create(c.ctx, secret, metav1.CreateOptions{})
+				return err
+			})
+	}
+	return nil
 }
 
 func (c *Command) createOrUpdateACLPolicy(policy api.ACLPolicy, consulClient *api.Client) error {
-	// Attempt to create the ACL policy
+	// Attempt to create the ACL policy.
 	_, _, err := consulClient.ACL().PolicyCreate(&policy, &api.WriteOptions{})
 
 	// With the introduction of Consul namespaces, if someone upgrades into a