diff --git a/.changelog/9351.txt b/.changelog/9351.txt
new file mode 100644
index 000000000..e676591d4
--- /dev/null
+++ b/.changelog/9351.txt
@@ -0,0 +1,3 @@
+```release-note:bug
+acl: global tokens created by auth methods now correctly replicate to secondary datacenters
+```
diff --git a/agent/consul/acl_endpoint_test.go b/agent/consul/acl_endpoint_test.go
index 56448443f..a265f79f4 100644
--- a/agent/consul/acl_endpoint_test.go
+++ b/agent/consul/acl_endpoint_test.go
@@ -4775,19 +4775,32 @@ func TestACLEndpoint_Login_with_MaxTokenTTL(t *testing.T) {
 func TestACLEndpoint_Login_with_TokenLocality(t *testing.T) {
 	go t.Parallel()
 
-	dir1, s1 := testServerWithConfig(t, func(c *Config) {
-		c.ACLDatacenter = "dc1"
-		c.ACLsEnabled = true
-		c.ACLMasterToken = "root"
-	})
-	defer os.RemoveAll(dir1)
-	defer s1.Shutdown()
-	codec := rpcClient(t, s1)
-	defer codec.Close()
+	_, s1, codec := testACLServerWithConfig(t, func(c *Config) {
+		c.ACLTokenMinExpirationTTL = 10 * time.Millisecond
+		c.ACLTokenMaxExpirationTTL = 5 * time.Second
+	}, false)
+
+	_, s2, _ := testACLServerWithConfig(t, func(c *Config) {
+		c.Datacenter = "dc2"
+		c.ACLTokenMinExpirationTTL = 10 * time.Millisecond
+		c.ACLTokenMaxExpirationTTL = 5 * time.Second
+		// token replication is required to test deleting non-local tokens in secondary dc
+		c.ACLTokenReplication = true
+	}, true)
 
 	waitForLeaderEstablishment(t, s1)
+	waitForLeaderEstablishment(t, s2)
+
+	joinWAN(t, s2, s1)
+
+	waitForNewACLs(t, s1)
+	waitForNewACLs(t, s2)
+
+	// Ensure s2 is authoritative.
+	waitForNewACLReplication(t, s2, structs.ACLReplicateTokens, 1, 1, 0)
 
 	acl := ACL{srv: s1}
+	acl2 := ACL{srv: s2}
 
 	testSessionID := testauth.StartSession()
 	defer testauth.ResetSession(testSessionID)
@@ -4799,18 +4812,20 @@ func TestACLEndpoint_Login_with_TokenLocality(t *testing.T) {
 	)
 
 	cases := map[string]struct {
-		tokenLocality string
-		expectLocal   bool
+		tokenLocality     string
+		expectLocal       bool
+		deleteFromReplica bool
 	}{
-		"empty":  {tokenLocality: "", expectLocal: true},
-		"local":  {tokenLocality: "local", expectLocal: true},
-		"global": {tokenLocality: "global", expectLocal: false},
+		"empty":             {tokenLocality: "", expectLocal: true},
+		"local":             {tokenLocality: "local", expectLocal: true},
+		"global dc1 delete": {tokenLocality: "global", expectLocal: false, deleteFromReplica: false},
+		"global dc2 delete": {tokenLocality: "global", expectLocal: false, deleteFromReplica: true},
 	}
 
 	for name, tc := range cases {
 		tc := tc
 		t.Run(name, func(t *testing.T) {
-			method, err := upsertTestCustomizedAuthMethod(codec, "root", "dc1", func(method *structs.ACLAuthMethod) {
+			method, err := upsertTestCustomizedAuthMethod(codec, TestDefaultMasterToken, "dc1", func(method *structs.ACLAuthMethod) {
 				method.TokenLocality = tc.tokenLocality
 				method.Config = map[string]interface{}{
 					"SessionID": testSessionID,
@@ -4819,7 +4834,7 @@ func TestACLEndpoint_Login_with_TokenLocality(t *testing.T) {
 			require.NoError(t, err)
 
 			_, err = upsertTestBindingRule(
-				codec, "root", "dc1", method.Name,
+				codec, TestDefaultMasterToken, "dc1", method.Name,
 				"",
 				structs.BindingRuleBindTypeService,
 				"web",
@@ -4841,7 +4856,7 @@ func TestACLEndpoint_Login_with_TokenLocality(t *testing.T) {
 
 			secretID := resp.SecretID
 
-			got := &resp
+			got := resp.Clone()
 			got.CreateIndex = 0
 			got.ModifyIndex = 0
 			got.AccessorID = ""
@@ -4863,13 +4878,38 @@ func TestACLEndpoint_Login_with_TokenLocality(t *testing.T) {
 			require.Equal(t, got, expect)
 
 			// Now turn around and nuke it.
+			var (
+				logoutACL ACL
+				logoutDC  string
+			)
+			if tc.deleteFromReplica {
+				// wait for it to show up
+				retry.Run(t, func(r *retry.R) {
+					var resp structs.ACLTokenResponse
+					require.NoError(r, acl2.TokenRead(&structs.ACLTokenGetRequest{
+						TokenID:        secretID,
+						TokenIDType:    structs.ACLTokenSecret,
+						Datacenter:     "dc2",
+						EnterpriseMeta: *defaultEntMeta,
+						QueryOptions:   structs.QueryOptions{Token: TestDefaultMasterToken},
+					}, &resp))
+					require.NotNil(r, resp.Token, "cannot lookup token with secretID %q", secretID)
+				})
+
+				logoutACL = acl2
+				logoutDC = "dc2"
+			} else {
+				logoutACL = acl
+				logoutDC = "dc1"
+			}
+
 			logoutReq := structs.ACLLogoutRequest{
-				Datacenter:   "dc1",
+				Datacenter:   logoutDC,
 				WriteRequest: structs.WriteRequest{Token: secretID},
 			}
 
 			var ignored bool
-			require.NoError(t, acl.Logout(&logoutReq, &ignored))
+			require.NoError(t, logoutACL.Logout(&logoutReq, &ignored))
 		})
 	}
 }
diff --git a/agent/consul/acl_replication_test.go b/agent/consul/acl_replication_test.go
index d1b39c26f..5629fabf6 100644
--- a/agent/consul/acl_replication_test.go
+++ b/agent/consul/acl_replication_test.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/agent/consul/authmethod/testauth"
 	"github.com/hashicorp/consul/agent/structs"
 	tokenStore "github.com/hashicorp/consul/agent/token"
 	"github.com/hashicorp/consul/sdk/testutil/retry"
@@ -349,6 +350,32 @@ func TestACLReplication_Tokens(t *testing.T) {
 		tokens = append(tokens, &token)
 	}
 
+	// Create an auth method in the primary that can create global tokens
+	// so that we ensure that these replicate correctly.
+	testSessionID := testauth.StartSession()
+	defer testauth.ResetSession(testSessionID)
+	testauth.InstallSessionToken(testSessionID, "fake-token", "default", "demo", "abc123")
+	method1, err := upsertTestCustomizedAuthMethod(client, "root", "dc1", func(method *structs.ACLAuthMethod) {
+		method.TokenLocality = "global"
+		method.Config = map[string]interface{}{
+			"SessionID": testSessionID,
+		}
+	})
+	require.NoError(t, err)
+	_, err = upsertTestBindingRule(client, "root", "dc1", method1.Name, "", structs.BindingRuleBindTypeService, "demo")
+	require.NoError(t, err)
+
+	// Create one token via this process.
+	methodToken := structs.ACLToken{}
+	require.NoError(t, s1.RPC("ACL.Login", &structs.ACLLoginRequest{
+		Auth: &structs.ACLLoginParams{
+			AuthMethod:  method1.Name,
+			BearerToken: "fake-token",
+		},
+		Datacenter: "dc1",
+	}, &methodToken))
+	tokens = append(tokens, &methodToken)
+
 	checkSame := func(t *retry.R) {
 		// only account for global tokens - local tokens shouldn't be replicated
 		index, remote, err := s1.fsm.State().ACLTokenList(nil, false, true, "", "", "", nil, nil)
@@ -359,6 +386,11 @@ func TestACLReplication_Tokens(t *testing.T) {
 		require.Len(t, local, len(remote))
 		for i, token := range remote {
 			require.Equal(t, token.Hash, local[i].Hash)
+
+			if token.AccessorID == methodToken.AccessorID {
+				require.Equal(t, method1.Name, token.AuthMethod)
+				require.Equal(t, method1.Name, local[i].AuthMethod)
+			}
 		}
 
 		s2.aclReplicationStatusLock.RLock()
diff --git a/agent/consul/acl_replication_types.go b/agent/consul/acl_replication_types.go
index c1e29d818..06bfe4e50 100644
--- a/agent/consul/acl_replication_types.go
+++ b/agent/consul/acl_replication_types.go
@@ -113,6 +113,7 @@ func (r *aclTokenReplicator) UpdateLocalBatch(ctx context.Context, srv *Server,
 		Tokens:            r.updated[start:end],
 		CAS:               false,
 		AllowMissingLinks: true,
+		FromReplication:   true,
 	}
 
 	resp, err := srv.raftApply(structs.ACLTokenSetRequestType, &req)
diff --git a/agent/consul/fsm/commands_oss.go b/agent/consul/fsm/commands_oss.go
index d637a366d..4ea12e7a6 100644
--- a/agent/consul/fsm/commands_oss.go
+++ b/agent/consul/fsm/commands_oss.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/armon/go-metrics"
 	"github.com/armon/go-metrics/prometheus"
+	"github.com/hashicorp/consul/agent/consul/state"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/api"
 )
@@ -509,7 +510,14 @@ func (c *FSM) applyACLTokenSetOperation(buf []byte, index uint64) interface{} {
 	defer metrics.MeasureSinceWithLabels([]string{"fsm", "acl", "token"}, time.Now(),
 		[]metrics.Label{{Name: "op", Value: "upsert"}})
 
-	return c.state.ACLTokenBatchSet(index, req.Tokens, req.CAS, req.AllowMissingLinks, req.ProhibitUnprivileged)
+	opts := state.ACLTokenSetOptions{
+		CAS:                          req.CAS,
+		AllowMissingPolicyAndRoleIDs: req.AllowMissingLinks,
+		ProhibitUnprivileged:         req.ProhibitUnprivileged,
+		Legacy:                       false,
+		FromReplication:              req.FromReplication,
+	}
+	return c.state.ACLTokenBatchSet(index, req.Tokens, opts)
 }
 
 func (c *FSM) applyACLTokenDeleteOperation(buf []byte, index uint64) interface{} {
diff --git a/agent/consul/state/acl.go b/agent/consul/state/acl.go
index 61f0b54ba..bff97c718 100644
--- a/agent/consul/state/acl.go
+++ b/agent/consul/state/acl.go
@@ -305,7 +305,7 @@ func (s *Store) ACLBootstrap(idx, resetIndex uint64, token *structs.ACLToken, le
 		}
 	}
 
-	if err := aclTokenSetTxn(tx, idx, token, false, false, false, legacy); err != nil {
+	if err := aclTokenSetTxn(tx, idx, token, ACLTokenSetOptions{Legacy: legacy}); err != nil {
 		return fmt.Errorf("failed inserting bootstrap token: %v", err)
 	}
 	if err := tx.Insert("index", &IndexEntry{"acl-token-bootstrap", idx}); err != nil {
@@ -632,19 +632,31 @@ func (s *Store) ACLTokenSet(idx uint64, token *structs.ACLToken, legacy bool) er
 	defer tx.Abort()
 
 	// Call set on the ACL
-	if err := aclTokenSetTxn(tx, idx, token, false, false, false, legacy); err != nil {
+	if err := aclTokenSetTxn(tx, idx, token, ACLTokenSetOptions{Legacy: legacy}); err != nil {
 		return err
 	}
 
 	return tx.Commit()
 }
 
-func (s *Store) ACLTokenBatchSet(idx uint64, tokens structs.ACLTokens, cas, allowMissingPolicyAndRoleIDs, prohibitUnprivileged bool) error {
+type ACLTokenSetOptions struct {
+	CAS                          bool
+	AllowMissingPolicyAndRoleIDs bool
+	ProhibitUnprivileged         bool
+	Legacy                       bool
+	FromReplication              bool
+}
+
+func (s *Store) ACLTokenBatchSet(idx uint64, tokens structs.ACLTokens, opts ACLTokenSetOptions) error {
+	if opts.Legacy {
+		return fmt.Errorf("failed inserting acl token: cannot use this endpoint to persist legacy tokens")
+	}
+
 	tx := s.db.WriteTxn(idx)
 	defer tx.Abort()
 
 	for _, token := range tokens {
-		if err := aclTokenSetTxn(tx, idx, token, cas, allowMissingPolicyAndRoleIDs, prohibitUnprivileged, false); err != nil {
+		if err := aclTokenSetTxn(tx, idx, token, opts); err != nil {
 			return err
 		}
 	}
@@ -654,16 +666,20 @@ func (s *Store) ACLTokenBatchSet(idx uint64, tokens structs.ACLTokens, cas, allo
 
 // aclTokenSetTxn is the inner method used to insert an ACL token with the
 // proper indexes into the state store.
-func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMissingPolicyAndRoleIDs, prohibitUnprivileged, legacy bool) error {
+func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, opts ACLTokenSetOptions) error {
 	// Check that the ID is set
 	if token.SecretID == "" {
 		return ErrMissingACLTokenSecret
 	}
 
-	if !legacy && token.AccessorID == "" {
+	if !opts.Legacy && token.AccessorID == "" {
 		return ErrMissingACLTokenAccessor
 	}
 
+	if opts.FromReplication && token.Local {
+		return fmt.Errorf("Cannot replicate local tokens")
+	}
+
 	// DEPRECATED (ACL-Legacy-Compat)
 	if token.Rules != "" {
 		// When we update a legacy acl token we may have to correct old HCL to
@@ -688,7 +704,7 @@ func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMiss
 		original = existing.(*structs.ACLToken)
 	}
 
-	if cas {
+	if opts.CAS {
 		// set-if-unset case
 		if token.ModifyIndex == 0 && original != nil {
 			return nil
@@ -703,7 +719,7 @@ func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMiss
 		}
 	}
 
-	if legacy && original != nil {
+	if opts.Legacy && original != nil {
 		if original.UsesNonLegacyFields() {
 			return fmt.Errorf("failed inserting acl token: cannot use legacy endpoint to modify a non-legacy token")
 		}
@@ -716,16 +732,16 @@ func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMiss
 	}
 
 	var numValidPolicies int
-	if numValidPolicies, err = resolveTokenPolicyLinks(tx, token, allowMissingPolicyAndRoleIDs); err != nil {
+	if numValidPolicies, err = resolveTokenPolicyLinks(tx, token, opts.AllowMissingPolicyAndRoleIDs); err != nil {
 		return err
 	}
 
 	var numValidRoles int
-	if numValidRoles, err = resolveTokenRoleLinks(tx, token, allowMissingPolicyAndRoleIDs); err != nil {
+	if numValidRoles, err = resolveTokenRoleLinks(tx, token, opts.AllowMissingPolicyAndRoleIDs); err != nil {
 		return err
 	}
 
-	if token.AuthMethod != "" {
+	if token.AuthMethod != "" && !opts.FromReplication {
 		method, err := getAuthMethodWithTxn(tx, nil, token.AuthMethod, token.ACLAuthMethodEnterpriseMeta.ToEnterpriseMeta())
 		if err != nil {
 			return err
@@ -749,7 +765,7 @@ func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMiss
 		}
 	}
 
-	if prohibitUnprivileged {
+	if opts.ProhibitUnprivileged {
 		if numValidRoles == 0 && numValidPolicies == 0 && len(token.ServiceIdentities) == 0 && len(token.NodeIdentities) == 0 {
 			return ErrTokenHasNoPrivileges
 		}
@@ -1059,7 +1075,7 @@ func aclTokenDeleteTxn(tx *txn, idx uint64, value, index string, entMeta *struct
 	return aclTokenDeleteWithToken(tx, token.(*structs.ACLToken), idx)
 }
 
-func aclTokenDeleteAllForAuthMethodTxn(tx *txn, idx uint64, methodName string, methodMeta *structs.EnterpriseMeta) error {
+func aclTokenDeleteAllForAuthMethodTxn(tx *txn, idx uint64, methodName string, methodGlobalLocality bool, methodMeta *structs.EnterpriseMeta) error {
 	// collect all the tokens linked with the given auth method.
 	iter, err := aclTokenListByAuthMethod(tx, methodName, methodMeta, structs.WildcardEnterpriseMeta())
 	if err != nil {
@@ -1069,7 +1085,15 @@ func aclTokenDeleteAllForAuthMethodTxn(tx *txn, idx uint64, methodName string, m
 	var tokens structs.ACLTokens
 	for raw := iter.Next(); raw != nil; raw = iter.Next() {
 		token := raw.(*structs.ACLToken)
-		tokens = append(tokens, token)
+		tokenIsGlobal := !token.Local
+
+		// Need to ensure that if we have an auth method named "blah" in the
+		// primary and secondary datacenters, and the primary instance has
+		// TokenLocality==global that when we delete the secondary instance we
+		// don't also blow away replicated tokens from the primary.
+		if methodGlobalLocality == tokenIsGlobal {
+			tokens = append(tokens, token)
+		}
 	}
 
 	if len(tokens) > 0 {
@@ -1877,7 +1901,7 @@ func aclAuthMethodDeleteTxn(tx *txn, idx uint64, name string, entMeta *structs.E
 		return err
 	}
 
-	if err := aclTokenDeleteAllForAuthMethodTxn(tx, idx, method.Name, entMeta); err != nil {
+	if err := aclTokenDeleteAllForAuthMethodTxn(tx, idx, method.Name, method.TokenLocality == "global", entMeta); err != nil {
 		return err
 	}
 
diff --git a/agent/consul/state/acl_events_test.go b/agent/consul/state/acl_events_test.go
index 8f10514d2..495602996 100644
--- a/agent/consul/state/acl_events_test.go
+++ b/agent/consul/state/acl_events_test.go
@@ -20,28 +20,28 @@ func TestACLChangeUnsubscribeEvent(t *testing.T) {
 		{
 			Name: "token create",
 			Mutate: func(tx *txn) error {
-				return aclTokenSetTxn(tx, tx.Index, newACLToken(1), false, false, false, false)
+				return aclTokenSetTxn(tx, tx.Index, newACLToken(1), ACLTokenSetOptions{})
 			},
 			expected: stream.NewCloseSubscriptionEvent(newSecretIDs(1)),
 		},
 		{
 			Name: "token update",
 			Setup: func(tx *txn) error {
-				return aclTokenSetTxn(tx, tx.Index, newACLToken(1), false, false, false, false)
+				return aclTokenSetTxn(tx, tx.Index, newACLToken(1), ACLTokenSetOptions{})
 			},
 			Mutate: func(tx *txn) error {
 				// Add a policy to the token (never mind it doesn't exist for now) we
 				// allow it in the set command below.
 				token := newACLToken(1)
 				token.Policies = []structs.ACLTokenPolicyLink{{ID: "33333333-1111-1111-1111-111111111111"}}
-				return aclTokenSetTxn(tx, tx.Index, token, false, true, false, false)
+				return aclTokenSetTxn(tx, tx.Index, token, ACLTokenSetOptions{AllowMissingPolicyAndRoleIDs: true})
 			},
 			expected: stream.NewCloseSubscriptionEvent(newSecretIDs(1)),
 		},
 		{
 			Name: "token delete",
 			Setup: func(tx *txn) error {
-				return aclTokenSetTxn(tx, tx.Index, newACLToken(1), false, false, false, false)
+				return aclTokenSetTxn(tx, tx.Index, newACLToken(1), ACLTokenSetOptions{})
 			},
 			Mutate: func(tx *txn) error {
 				token := newACLToken(1)
@@ -144,7 +144,7 @@ func newACLRoleWithSingleToken(tx *txn) error {
 	}
 	token := newACLToken(1)
 	token.Roles = append(token.Roles, structs.ACLTokenRoleLink{ID: role.ID})
-	return aclTokenSetTxn(tx, tx.Index, token, false, false, false, false)
+	return aclTokenSetTxn(tx, tx.Index, token, ACLTokenSetOptions{})
 }
 
 func newACLPolicyWithSingleToken(tx *txn) error {
@@ -154,7 +154,7 @@ func newACLPolicyWithSingleToken(tx *txn) error {
 	}
 	token := newACLToken(1)
 	token.Policies = append(token.Policies, structs.ACLTokenPolicyLink{ID: policy.ID})
-	return aclTokenSetTxn(tx, tx.Index, token, false, false, false, false)
+	return aclTokenSetTxn(tx, tx.Index, token, ACLTokenSetOptions{})
 }
 
 func newSecretIDs(ids ...int) []string {
diff --git a/agent/consul/state/acl_test.go b/agent/consul/state/acl_test.go
index 5aed5287f..441ddfbbb 100644
--- a/agent/consul/state/acl_test.go
+++ b/agent/consul/state/acl_test.go
@@ -621,7 +621,7 @@ func TestStateStore_ACLTokens_UpsertBatchRead(t *testing.T) {
 			},
 		}
 
-		require.NoError(t, s.ACLTokenBatchSet(2, tokens, true, false, false))
+		require.NoError(t, s.ACLTokenBatchSet(2, tokens, ACLTokenSetOptions{CAS: true}))
 
 		_, token, err := s.ACLTokenGetByAccessor(nil, tokens[0].AccessorID, nil)
 		require.NoError(t, err)
@@ -642,7 +642,7 @@ func TestStateStore_ACLTokens_UpsertBatchRead(t *testing.T) {
 			},
 		}
 
-		require.NoError(t, s.ACLTokenBatchSet(5, tokens, true, false, false))
+		require.NoError(t, s.ACLTokenBatchSet(5, tokens, ACLTokenSetOptions{CAS: true}))
 
 		updated := structs.ACLTokens{
 			&structs.ACLToken{
@@ -653,7 +653,7 @@ func TestStateStore_ACLTokens_UpsertBatchRead(t *testing.T) {
 			},
 		}
 
-		require.NoError(t, s.ACLTokenBatchSet(6, updated, true, false, false))
+		require.NoError(t, s.ACLTokenBatchSet(6, updated, ACLTokenSetOptions{CAS: true}))
 
 		_, token, err := s.ACLTokenGetByAccessor(nil, tokens[0].AccessorID, nil)
 		require.NoError(t, err)
@@ -672,7 +672,7 @@ func TestStateStore_ACLTokens_UpsertBatchRead(t *testing.T) {
 			},
 		}
 
-		require.NoError(t, s.ACLTokenBatchSet(5, tokens, true, false, false))
+		require.NoError(t, s.ACLTokenBatchSet(5, tokens, ACLTokenSetOptions{CAS: true}))
 
 		updated := structs.ACLTokens{
 			&structs.ACLToken{
@@ -682,7 +682,7 @@ func TestStateStore_ACLTokens_UpsertBatchRead(t *testing.T) {
 			},
 		}
 
-		require.NoError(t, s.ACLTokenBatchSet(6, updated, true, false, false))
+		require.NoError(t, s.ACLTokenBatchSet(6, updated, ACLTokenSetOptions{CAS: true}))
 
 		_, token, err := s.ACLTokenGetByAccessor(nil, tokens[0].AccessorID, nil)
 		require.NoError(t, err)
@@ -705,7 +705,7 @@ func TestStateStore_ACLTokens_UpsertBatchRead(t *testing.T) {
 			},
 		}
 
-		require.NoError(t, s.ACLTokenBatchSet(2, tokens, false, false, false))
+		require.NoError(t, s.ACLTokenBatchSet(2, tokens, ACLTokenSetOptions{}))
 
 		idx, rtokens, err := s.ACLTokenBatchGet(nil, []string{
 			"a4f68bd6-3af5-4f56-b764-3c6f20247879",
@@ -736,7 +736,7 @@ func TestStateStore_ACLTokens_UpsertBatchRead(t *testing.T) {
 			},
 		}
 
-		require.NoError(t, s.ACLTokenBatchSet(2, tokens, false, false, false))
+		require.NoError(t, s.ACLTokenBatchSet(2, tokens, ACLTokenSetOptions{}))
 
 		updates := structs.ACLTokens{
 			&structs.ACLToken{
@@ -761,7 +761,7 @@ func TestStateStore_ACLTokens_UpsertBatchRead(t *testing.T) {
 			},
 		}
 
-		require.NoError(t, s.ACLTokenBatchSet(3, updates, false, false, false))
+		require.NoError(t, s.ACLTokenBatchSet(3, updates, ACLTokenSetOptions{}))
 
 		idx, rtokens, err := s.ACLTokenBatchGet(nil, []string{
 			"a4f68bd6-3af5-4f56-b764-3c6f20247879",
@@ -808,9 +808,9 @@ func TestStateStore_ACLTokens_UpsertBatchRead(t *testing.T) {
 			},
 		}
 
-		require.Error(t, s.ACLTokenBatchSet(2, tokens, false, false, false))
+		require.Error(t, s.ACLTokenBatchSet(2, tokens, ACLTokenSetOptions{}))
 
-		require.NoError(t, s.ACLTokenBatchSet(2, tokens, false, true, false))
+		require.NoError(t, s.ACLTokenBatchSet(2, tokens, ACLTokenSetOptions{AllowMissingPolicyAndRoleIDs: true}))
 
 		idx, rtokens, err := s.ACLTokenBatchGet(nil, []string{
 			"a4f68bd6-3af5-4f56-b764-3c6f20247879",
@@ -847,9 +847,9 @@ func TestStateStore_ACLTokens_UpsertBatchRead(t *testing.T) {
 			},
 		}
 
-		require.Error(t, s.ACLTokenBatchSet(2, tokens, false, false, false))
+		require.Error(t, s.ACLTokenBatchSet(2, tokens, ACLTokenSetOptions{}))
 
-		require.NoError(t, s.ACLTokenBatchSet(2, tokens, false, true, false))
+		require.NoError(t, s.ACLTokenBatchSet(2, tokens, ACLTokenSetOptions{AllowMissingPolicyAndRoleIDs: true}))
 
 		idx, rtokens, err := s.ACLTokenBatchGet(nil, []string{
 			"a4f68bd6-3af5-4f56-b764-3c6f20247879",
@@ -953,7 +953,7 @@ func TestStateStore_ACLTokens_ListUpgradeable(t *testing.T) {
 		},
 	}
 
-	require.NoError(t, s.ACLTokenBatchSet(7, updates, false, false, false))
+	require.NoError(t, s.ACLTokenBatchSet(7, updates, ACLTokenSetOptions{}))
 
 	tokens, _, err = s.ACLTokenListUpgradeable(10)
 	require.NoError(t, err)
@@ -1044,7 +1044,7 @@ func TestStateStore_ACLToken_List(t *testing.T) {
 		},
 	}
 
-	require.NoError(t, s.ACLTokenBatchSet(2, tokens, false, false, false))
+	require.NoError(t, s.ACLTokenBatchSet(2, tokens, ACLTokenSetOptions{}))
 
 	type testCase struct {
 		name       string
@@ -1565,7 +1565,7 @@ func TestStateStore_ACLToken_Delete(t *testing.T) {
 			},
 		}
 
-		require.NoError(t, s.ACLTokenBatchSet(2, tokens, false, false, false))
+		require.NoError(t, s.ACLTokenBatchSet(2, tokens, ACLTokenSetOptions{}))
 
 		_, rtoken, err := s.ACLTokenGetByAccessor(nil, "f1093997-b6c7-496d-bfb8-6b1b1895641b", nil)
 		require.NoError(t, err)
@@ -2829,6 +2829,123 @@ func TestStateStore_ACLAuthMethod_SetGet(t *testing.T) {
 	})
 }
 
+func TestStateStore_ACLAuthMethod_GlobalNameShadowing_TokenTest(t *testing.T) {
+	t.Parallel()
+
+	// This ensures that when a primary DC and secondary DC create identically
+	// named auth methods, and the primary instance has a tokenLocality==global
+	// that operations in the secondary correctly can target one or the other.
+
+	s := testACLStateStore(t)
+	lastIndex := uint64(1)
+
+	// For this test our state machine will simulate the SECONDARY(DC2), so
+	// we'll create our auth method here that shadows the global-token-minting
+	// one in the primary.
+
+	defaultEntMeta := structs.DefaultEnterpriseMeta()
+
+	lastIndex++
+	require.NoError(t, s.ACLAuthMethodSet(lastIndex, &structs.ACLAuthMethod{
+		Name:           "test",
+		Type:           "testing",
+		Description:    "test",
+		EnterpriseMeta: *defaultEntMeta,
+	}))
+
+	const ( // accessors
+		methodDC1_tok1 = "6d020c5d-c4fd-4348-ba79-beac37ed0b9c"
+		methodDC1_tok2 = "169160dc-34ab-45c6-aba7-ff65e9ace9cb"
+		methodDC2_tok1 = "8e14628e-7dde-4573-aca1-6386c0f2095d"
+		methodDC2_tok2 = "291e5af9-c68e-4dd3-8824-b2bdfdcc89e6"
+	)
+
+	lastIndex++
+	require.NoError(t, s.ACLTokenBatchSet(lastIndex, structs.ACLTokens{
+		&structs.ACLToken{
+			AccessorID:     methodDC2_tok1,
+			SecretID:       "d9399b7d-6c34-46bd-a675-c1352fadb6fd",
+			Description:    "test-dc2-t1",
+			AuthMethod:     "test",
+			Local:          true,
+			EnterpriseMeta: *defaultEntMeta,
+		},
+		&structs.ACLToken{
+			AccessorID:     methodDC2_tok2,
+			SecretID:       "3b72fc27-9230-42ab-a1e8-02cb489ab177",
+			Description:    "test-dc2-t2",
+			AuthMethod:     "test",
+			Local:          true,
+			EnterpriseMeta: *defaultEntMeta,
+		},
+	}, ACLTokenSetOptions{}))
+
+	lastIndex++
+	require.NoError(t, s.ACLTokenBatchSet(lastIndex, structs.ACLTokens{
+		&structs.ACLToken{
+			AccessorID:     methodDC1_tok1,
+			SecretID:       "7a1950c6-79dc-441c-acd2-e22cd3db0240",
+			Description:    "test-dc1-t1",
+			AuthMethod:     "test",
+			Local:          false,
+			EnterpriseMeta: *defaultEntMeta,
+		},
+		&structs.ACLToken{
+			AccessorID:     methodDC1_tok2,
+			SecretID:       "442cee4c-353f-4957-adbb-33db2f9e267f",
+			Description:    "test-dc1-t2",
+			AuthMethod:     "test",
+			Local:          false,
+			EnterpriseMeta: *defaultEntMeta,
+		},
+	}, ACLTokenSetOptions{FromReplication: true}))
+
+	toList := func(tokens structs.ACLTokens) []string {
+		var ret []string
+		for _, tok := range tokens {
+			ret = append(ret, tok.AccessorID)
+		}
+		return ret
+	}
+
+	require.True(t, t.Run("list local only", func(t *testing.T) {
+		_, got, err := s.ACLTokenList(nil, true, false, "", "", "test", defaultEntMeta, defaultEntMeta)
+		require.NoError(t, err)
+		require.ElementsMatch(t, []string{methodDC2_tok1, methodDC2_tok2}, toList(got))
+	}))
+	require.True(t, t.Run("list global only", func(t *testing.T) {
+		_, got, err := s.ACLTokenList(nil, false, true, "", "", "test", defaultEntMeta, defaultEntMeta)
+		require.NoError(t, err)
+		require.ElementsMatch(t, []string{methodDC1_tok1, methodDC1_tok2}, toList(got))
+	}))
+	require.True(t, t.Run("list both", func(t *testing.T) {
+		_, got, err := s.ACLTokenList(nil, true, true, "", "", "test", defaultEntMeta, defaultEntMeta)
+		require.NoError(t, err)
+		require.ElementsMatch(t, []string{methodDC1_tok1, methodDC1_tok2, methodDC2_tok1, methodDC2_tok2}, toList(got))
+	}))
+
+	lastIndex++
+	require.True(t, t.Run("delete dc2 auth method", func(t *testing.T) {
+		require.NoError(t, s.ACLAuthMethodDeleteByName(lastIndex, "test", nil))
+	}))
+
+	require.True(t, t.Run("list local only (after dc2 delete)", func(t *testing.T) {
+		_, got, err := s.ACLTokenList(nil, true, false, "", "", "test", defaultEntMeta, defaultEntMeta)
+		require.NoError(t, err)
+		require.Empty(t, got)
+	}))
+	require.True(t, t.Run("list global only (after dc2 delete)", func(t *testing.T) {
+		_, got, err := s.ACLTokenList(nil, false, true, "", "", "test", defaultEntMeta, defaultEntMeta)
+		require.NoError(t, err)
+		require.ElementsMatch(t, []string{methodDC1_tok1, methodDC1_tok2}, toList(got))
+	}))
+	require.True(t, t.Run("list both (after dc2 delete)", func(t *testing.T) {
+		_, got, err := s.ACLTokenList(nil, true, true, "", "", "test", defaultEntMeta, defaultEntMeta)
+		require.NoError(t, err)
+		require.ElementsMatch(t, []string{methodDC1_tok1, methodDC1_tok2}, toList(got))
+	}))
+}
+
 func TestStateStore_ACLAuthMethods_UpsertBatchRead(t *testing.T) {
 	t.Parallel()
 
@@ -3092,27 +3209,31 @@ func TestStateStore_ACLAuthMethod_Delete_RuleAndTokenCascade(t *testing.T) {
 			SecretID:    "7a1950c6-79dc-441c-acd2-e22cd3db0240",
 			Description: "test-m1-t1",
 			AuthMethod:  "test-1",
+			Local:       true,
 		},
 		&structs.ACLToken{
 			AccessorID:  method1_tok2,
 			SecretID:    "442cee4c-353f-4957-adbb-33db2f9e267f",
 			Description: "test-m1-t2",
 			AuthMethod:  "test-1",
+			Local:       true,
 		},
 		&structs.ACLToken{
 			AccessorID:  method2_tok1,
 			SecretID:    "d9399b7d-6c34-46bd-a675-c1352fadb6fd",
 			Description: "test-m2-t1",
 			AuthMethod:  "test-2",
+			Local:       true,
 		},
 		&structs.ACLToken{
 			AccessorID:  method2_tok2,
 			SecretID:    "3b72fc27-9230-42ab-a1e8-02cb489ab177",
 			Description: "test-m2-t2",
 			AuthMethod:  "test-2",
+			Local:       true,
 		},
 	}
-	require.NoError(t, s.ACLTokenBatchSet(4, tokens, false, false, false))
+	require.NoError(t, s.ACLTokenBatchSet(4, tokens, ACLTokenSetOptions{}))
 
 	// Delete one method.
 	require.NoError(t, s.ACLAuthMethodDeleteByName(4, "test-1", nil))
@@ -3570,7 +3691,7 @@ func TestStateStore_ACLTokens_Snapshot_Restore(t *testing.T) {
 		},
 	}
 
-	require.NoError(t, s.ACLTokenBatchSet(4, tokens, false, false, false))
+	require.NoError(t, s.ACLTokenBatchSet(4, tokens, ACLTokenSetOptions{}))
 
 	// Snapshot the ACLs.
 	snap := s.Snapshot()
diff --git a/agent/structs/acl.go b/agent/structs/acl.go
index fef8d0875..d509abee6 100644
--- a/agent/structs/acl.go
+++ b/agent/structs/acl.go
@@ -1331,6 +1331,7 @@ type ACLTokenBatchSetRequest struct {
 	CAS                  bool
 	AllowMissingLinks    bool
 	ProhibitUnprivileged bool
+	FromReplication      bool
 }
 
 // ACLTokenBatchDeleteRequest is used only at the Raft layer