diff --git a/charts/consul/templates/partition-init-job.yaml b/charts/consul/templates/partition-init-job.yaml
index b9d438c436..053478ed66 100644
--- a/charts/consul/templates/partition-init-job.yaml
+++ b/charts/consul/templates/partition-init-job.yaml
@@ -28,8 +28,9 @@ spec:
     spec:
       restartPolicy: Never
       serviceAccountName: {{ template "consul.fullname" . }}-partition-init
-      {{- if .Values.global.tls.enabled }}
+      {{- if (or .Values.global.tls.enabled (and .Values.global.acls.bootstrapToken.secretName .Values.global.acls.bootstrapToken.secretKey)) }}
       volumes:
+      {{- if .Values.global.tls.enabled }}
         - name: consul-ca-cert
           secret:
             {{- if .Values.global.tls.caCert.secretName }}
@@ -41,6 +42,15 @@ spec:
               - key: {{ default "tls.crt" .Values.global.tls.caCert.secretKey }}
                 path: tls.crt
       {{- end }}
+      {{- if (and .Values.global.acls.bootstrapToken.secretName .Values.global.acls.bootstrapToken.secretKey) }}
+        - name: partition-token
+          secret:
+            secretName: {{ .Values.global.acls.bootstrapToken.secretName }}
+            items:
+            - key: {{ .Values.global.acls.bootstrapToken.secretKey }}
+              path: partition-token
+      {{- end }}
+      {{- end }}
       containers:
         - name: post-install-job
           image: {{ .Values.global.imageK8S }}
@@ -49,11 +59,18 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
-          {{- if .Values.global.tls.enabled }}
+          {{- if (or .Values.global.tls.enabled (and .Values.global.acls.bootstrapToken.secretName .Values.global.acls.bootstrapToken.secretKey)) }}
           volumeMounts:
+            {{- if .Values.global.tls.enabled }}
             - name: consul-ca-cert
               mountPath: /consul/tls/ca
               readOnly: true
+            {{- end }}
+            {{- if (and .Values.global.acls.bootstrapToken.secretName .Values.global.acls.bootstrapToken.secretKey) }}
+            - name: partition-token
+              mountPath: /consul/acl/tokens
+              readOnly: true
+           {{- end }}
            {{- end }}
           command:
             - "/bin/sh"
@@ -81,6 +98,9 @@ spec:
                 -consul-tls-server-name={{ .Values.externalServers.tlsServerName }} \
                 {{- end }}
                 {{- end }}
+                {{- if (and .Values.global.acls.bootstrapToken.secretName .Values.global.acls.bootstrapToken.secretKey) }}
+                -token-file=/consul/acl/tokens/partition-token \
+                {{- end }}
                 -partition-name={{ .Values.global.adminPartitions.name }}
           resources:
             requests:
diff --git a/charts/consul/templates/server-acl-init-job.yaml b/charts/consul/templates/server-acl-init-job.yaml
index 752774064a..07b9ab8e6b 100644
--- a/charts/consul/templates/server-acl-init-job.yaml
+++ b/charts/consul/templates/server-acl-init-job.yaml
@@ -134,7 +134,10 @@ spec:
                 -sync-consul-node-name={{ .Values.syncCatalog.consulNodeName }} \
                 {{- end }}
                 {{- end }}
-
+                {{- if .Values.global.adminPartitions.enabled }}
+                -enable-partitions=true \
+                -partition={{ .Values.global.adminPartitions.name }} \
+                {{- end }}
                 {{- if (or (and (ne (.Values.dns.enabled | toString) "-") .Values.dns.enabled) (and (eq (.Values.dns.enabled | toString) "-") .Values.global.enabled)) }}
                 -allow-dns=true \
                 {{- end }}
diff --git a/charts/consul/test/unit/client-daemonset.bats b/charts/consul/test/unit/client-daemonset.bats
index cbbbc10ef9..d34cc46de8 100755
--- a/charts/consul/test/unit/client-daemonset.bats
+++ b/charts/consul/test/unit/client-daemonset.bats
@@ -1407,7 +1407,7 @@ rollingUpdate:
 #--------------------------------------------------------------------
 # partitions
 
-@test "client/DaemonSet: -partitions can be set by global.adminPartition.enabled" {
+@test "client/DaemonSet: -partitions can be set by global.adminPartitions.enabled" {
   cd `chart_dir`
   local actual=$(helm template \
       -s templates/client-daemonset.yaml  \
@@ -1417,7 +1417,7 @@ rollingUpdate:
   [ "${actual}" = "true" ]
 }
 
-@test "client/DaemonSet: -partitions can be overridden by global.adminPartition.name" {
+@test "client/DaemonSet: -partitions can be overridden by global.adminPartitions.name" {
   cd `chart_dir`
   local actual=$(helm template \
       -s templates/client-daemonset.yaml  \
diff --git a/charts/consul/test/unit/partition-init-job.bats b/charts/consul/test/unit/partition-init-job.bats
index ac6d5ccd27..c49fd0b744 100644
--- a/charts/consul/test/unit/partition-init-job.bats
+++ b/charts/consul/test/unit/partition-init-job.bats
@@ -93,4 +93,53 @@ load _helpers
   # check that the volume uses the provided secret key
   actual=$(echo $ca_cert_volume | jq -r '.secret.items[0].key' | tee /dev/stderr)
   [ "${actual}" = "key" ]
-}
\ No newline at end of file
+}
+
+#--------------------------------------------------------------------
+# global.acls.bootstrapToken
+
+@test "partitionInit/Job: partition-token volume is created when global.acls.bootstrapToken is provided" {
+  cd `chart_dir`
+  local actual=$(helm template \
+      -s templates/partition-init-job.yaml  \
+      --set 'global.enabled=false' \
+      --set 'global.adminPartitions.enabled=true' \
+      --set 'global.acls.bootstrapToken.secretName=partition-token' \
+      --set 'global.acls.bootstrapToken.secretKey=token' \
+      . | tee /dev/stderr |
+      yq '.spec.template.spec.volumes[0].name == "partition-token"' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+}
+
+@test "partitionInit/Job: partition-token volumeMount is created when global.acls.bootstrapToken is provided" {
+  cd `chart_dir`
+  local object=$(helm template \
+      -s templates/partition-init-job.yaml  \
+      --set 'global.enabled=false' \
+      --set 'global.adminPartitions.enabled=true' \
+      --set 'global.acls.bootstrapToken.secretName=partition-token' \
+      --set 'global.acls.bootstrapToken.secretKey=token' \
+      . | tee /dev/stderr |
+      yq '.spec.template.spec.containers[0].volumeMounts[0]' | tee /dev/stderr)
+
+  local actual=$(echo $object |
+      yq -r '.name' | tee /dev/stderr)
+  [ "${actual}" = "partition-token" ]
+
+  local actual=$(echo $object |
+      yq -r '.mountPath' | tee /dev/stderr)
+  [ "${actual}" = "/consul/acl/tokens" ]
+}
+
+@test "partitionInit/Job: command includes partition-token dir when global.acls.bootstrapToken is provided" {
+  cd `chart_dir`
+  local actual=$(helm template \
+      -s templates/partition-init-job.yaml  \
+      --set 'global.enabled=false' \
+      --set 'global.adminPartitions.enabled=true' \
+      --set 'global.acls.bootstrapToken.secretName=partition-token' \
+      --set 'global.acls.bootstrapToken.secretKey=token' \
+      . | tee /dev/stderr |
+      yq '.spec.template.spec.containers[0].command | any(contains("/consul/acl/tokens/partition-token"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+}
diff --git a/charts/consul/test/unit/server-acl-init-job.bats b/charts/consul/test/unit/server-acl-init-job.bats
index 4fd6867d17..a0ae9a34c5 100644
--- a/charts/consul/test/unit/server-acl-init-job.bats
+++ b/charts/consul/test/unit/server-acl-init-job.bats
@@ -995,6 +995,45 @@ load _helpers
   [ "${actual}" = "true" ]
 }
 
+#--------------------------------------------------------------------
+# admin partitions
+
+@test "serverACLInit/Job: admin partitions disabled by default" {
+  cd `chart_dir`
+  local object=$(helm template \
+      -s templates/server-acl-init-job.yaml  \
+      --set 'global.acls.manageSystemACLs=true' \
+      . | tee /dev/stderr |
+      yq '.spec.template.spec.containers[0].command' | tee /dev/stderr)
+
+  local actual=$(echo $object |
+    yq 'any(contains("enable-partitions"))' | tee /dev/stderr)
+  [ "${actual}" = "false" ]
+
+  local actual=$(echo $object |
+    yq 'any(contains("partition"))' | tee /dev/stderr)
+  [ "${actual}" = "false" ]
+}
+
+@test "serverACLInit/Job: admin partitions enabled when admin partitions are enabled" {
+  cd `chart_dir`
+  local object=$(helm template \
+      -s templates/server-acl-init-job.yaml  \
+      --set 'global.acls.manageSystemACLs=true' \
+      --set 'global.adminPartitions.enabled=true' \
+      --set 'global.enableConsulNamespaces=true' \
+      . | tee /dev/stderr |
+      yq '.spec.template.spec.containers[0].command' | tee /dev/stderr)
+
+  local actual=$(echo $object |
+    yq 'any(contains("enable-partitions"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+
+  local actual=$(echo $object |
+    yq 'any(contains("partition"))' | tee /dev/stderr)
+  [ "${actual}" = "true" ]
+}
+
 #--------------------------------------------------------------------
 # global.acls.createReplicationToken
 
diff --git a/control-plane/subcommand/partition-init/command.go b/control-plane/subcommand/partition-init/command.go
index bdbb4df21d..492d4fd666 100644
--- a/control-plane/subcommand/partition-init/command.go
+++ b/control-plane/subcommand/partition-init/command.go
@@ -5,6 +5,8 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"io/ioutil"
+	"strings"
 	"sync"
 	"time"
 
@@ -32,6 +34,7 @@ type Command struct {
 	flagConsulCACert        string
 	flagConsulTLSServerName string
 	flagUseHTTPS            bool
+	flagTokenFile           string
 
 	flagLogLevel string
 	flagLogJSON  bool
@@ -65,7 +68,8 @@ func (c *Command) init() {
 		"The server name to set as the SNI header when sending HTTPS requests to Consul.")
 	c.flags.BoolVar(&c.flagUseHTTPS, "use-https", false,
 		"Toggle for using HTTPS for all API calls to Consul.")
-
+	c.flags.StringVar(&c.flagTokenFile, "token-file", "",
+		"Path to file containing ACL token for creating partitions. This token must have 'acl:write' and 'operator:write' permissions.")
 	c.flags.DurationVar(&c.flagTimeout, "timeout", 10*time.Minute,
 		"How long we'll try to bootstrap Partitions for before timing out, e.g. 1ms, 2s, 3m")
 	c.flags.StringVar(&c.flagLogLevel, "log-level", "info",
@@ -120,6 +124,21 @@ func (c *Command) Run(args []string) int {
 		return 1
 	}
 
+	var providedToken string
+	if c.flagTokenFile != "" {
+		// Load the bootstrap token from file.
+		tokenBytes, err := ioutil.ReadFile(c.flagTokenFile)
+		if err != nil {
+			c.UI.Error(fmt.Sprintf("Unable to read bootstrap token from file %q: %s", c.flagTokenFile, err))
+			return 1
+		}
+		if len(tokenBytes) == 0 {
+			c.UI.Error(fmt.Sprintf("Bootstrap token file %q is empty", c.flagTokenFile))
+			return 1
+		}
+		providedToken = strings.TrimSpace(string(tokenBytes))
+	}
+
 	serverAddresses, err := common.GetResolvedServerAddresses(c.flagServerAddresses, c.providers, c.log)
 	if err != nil {
 		c.UI.Error(fmt.Sprintf("Unable to discover any Consul addresses from %q: %s", c.flagServerAddresses[0], err))
@@ -132,14 +151,18 @@ func (c *Command) Run(args []string) int {
 	}
 	// For all of the next operations we'll need a Consul client.
 	serverAddr := fmt.Sprintf("%s:%d", serverAddresses[0], c.flagServerPort)
-	consulClient, err := consul.NewClient(&api.Config{
+	config := &api.Config{
 		Address: serverAddr,
 		Scheme:  scheme,
 		TLSConfig: api.TLSConfig{
 			Address: c.flagConsulTLSServerName,
 			CAFile:  c.flagConsulCACert,
 		},
-	})
+	}
+	if providedToken != "" {
+		config.Token = providedToken
+	}
+	consulClient, err := consul.NewClient(config)
 	if err != nil {
 		c.UI.Error(fmt.Sprintf("Error creating Consul client for addr %q: %s", serverAddr, err))
 		return 1
diff --git a/control-plane/subcommand/partition-init/command_ent_test.go b/control-plane/subcommand/partition-init/command_ent_test.go
index 39e031a8b7..b2258e7b71 100644
--- a/control-plane/subcommand/partition-init/command_ent_test.go
+++ b/control-plane/subcommand/partition-init/command_ent_test.go
@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hashicorp/consul-k8s/control-plane/subcommand/common"
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/sdk/testutil"
 	"github.com/mitchellh/cli"
@@ -48,75 +49,108 @@ func TestRun_FlagValidation(t *testing.T) {
 
 func TestRun_PartitionCreate(t *testing.T) {
 	partitionName := "test-partition"
-
-	server, err := testutil.NewTestServerConfigT(t, nil)
-	require.NoError(t, err)
-	server.WaitForLeader(t)
-	defer server.Stop()
-
-	consul, err := api.NewClient(&api.Config{
-		Address: server.HTTPAddr,
-	})
-	require.NoError(t, err)
-
-	ui := cli.NewMockUi()
-	cmd := Command{
-		UI: ui,
-	}
-	cmd.init()
-	args := []string{
-		"-server-address=" + strings.Split(server.HTTPAddr, ":")[0],
-		"-server-port=" + strings.Split(server.HTTPAddr, ":")[1],
-		"-partition-name", partitionName,
+	acls := []bool{true, false}
+	var callback testutil.ServerConfigCallback
+	var tokenFile string
+	masterToken := "b78d37c7-0ca7-5f4d-99ee-6d9975ce4586"
+
+	for _, aclsEnabled := range acls {
+		if aclsEnabled {
+			callback = func(cfg *testutil.TestServerConfig) {
+				cfg.ACL.Enabled = true
+				cfg.ACL.Tokens.Master = masterToken
+			}
+			tokenFile = common.WriteTempFile(t, masterToken)
+		}
+		server, err := testutil.NewTestServerConfigT(t, callback)
+		require.NoError(t, err)
+		server.WaitForLeader(t)
+		defer server.Stop()
+
+		consul, err := api.NewClient(&api.Config{
+			Address: server.HTTPAddr,
+		})
+		require.NoError(t, err)
+
+		ui := cli.NewMockUi()
+		cmd := Command{
+			UI: ui,
+		}
+		cmd.init()
+		args := []string{
+			"-server-address=" + strings.Split(server.HTTPAddr, ":")[0],
+			"-server-port=" + strings.Split(server.HTTPAddr, ":")[1],
+			"-partition-name", partitionName,
+		}
+		if aclsEnabled {
+			args = append(args, "-token-file", tokenFile)
+		}
+
+		responseCode := cmd.Run(args)
+
+		require.Equal(t, 0, responseCode)
+
+		partition, _, err := consul.Partitions().Read(context.Background(), partitionName, nil)
+		require.NoError(t, err)
+		require.NotNil(t, partition)
+		require.Equal(t, partitionName, partition.Name)
 	}
-
-	responseCode := cmd.Run(args)
-
-	require.Equal(t, 0, responseCode)
-
-	partition, _, err := consul.Partitions().Read(context.Background(), partitionName, nil)
-	require.NoError(t, err)
-	require.NotNil(t, partition)
-	require.Equal(t, partitionName, partition.Name)
 }
 
 func TestRun_PartitionExists(t *testing.T) {
 	partitionName := "test-partition"
-
-	server, err := testutil.NewTestServerConfigT(t, nil)
-	require.NoError(t, err)
-	server.WaitForLeader(t)
-	defer server.Stop()
-
-	consul, err := api.NewClient(&api.Config{
-		Address: server.HTTPAddr,
-	})
-	require.NoError(t, err)
-
-	// Create the Admin Partition before the test runs.
-	_, _, err = consul.Partitions().Create(context.Background(), &api.AdminPartition{Name: partitionName, Description: "Created before test"}, nil)
-	require.NoError(t, err)
-
-	ui := cli.NewMockUi()
-	cmd := Command{
-		UI: ui,
-	}
-	cmd.init()
-	args := []string{
-		"-server-address=" + strings.Split(server.HTTPAddr, ":")[0],
-		"-server-port=" + strings.Split(server.HTTPAddr, ":")[1],
-		"-partition-name", partitionName,
+	acls := []bool{true, false}
+	var callback testutil.ServerConfigCallback
+	var tokenFile string
+	masterToken := "b78d37c7-0ca7-5f4d-99ee-6d9975ce4586"
+
+	for _, aclsEnabled := range acls {
+		if aclsEnabled {
+			callback = func(cfg *testutil.TestServerConfig) {
+				cfg.ACL.Enabled = true
+				cfg.ACL.Tokens.Master = masterToken
+			}
+			tokenFile = common.WriteTempFile(t, masterToken)
+		}
+
+		server, err := testutil.NewTestServerConfigT(t, callback)
+		require.NoError(t, err)
+		server.WaitForLeader(t)
+		defer server.Stop()
+
+		consul, err := api.NewClient(&api.Config{
+			Address: server.HTTPAddr,
+		})
+		require.NoError(t, err)
+
+		// Create the Admin Partition before the test runs.
+		_, _, err = consul.Partitions().Create(context.Background(), &api.AdminPartition{Name: partitionName, Description: "Created before test"}, nil)
+		require.NoError(t, err)
+
+		ui := cli.NewMockUi()
+		cmd := Command{
+			UI: ui,
+		}
+		cmd.init()
+		args := []string{
+			"-server-address=" + strings.Split(server.HTTPAddr, ":")[0],
+			"-server-port=" + strings.Split(server.HTTPAddr, ":")[1],
+			"-partition-name", partitionName,
+		}
+		if aclsEnabled {
+			args = append(args, "-token-file", tokenFile)
+		}
+
+		responseCode := cmd.Run(args)
+
+		require.Equal(t, 0, responseCode)
+
+		partition, _, err := consul.Partitions().Read(context.Background(), partitionName, nil)
+		require.NoError(t, err)
+		require.NotNil(t, partition)
+		require.Equal(t, partitionName, partition.Name)
+		require.Equal(t, "Created before test", partition.Description)
 	}
-
-	responseCode := cmd.Run(args)
-
-	require.Equal(t, 0, responseCode)
-
-	partition, _, err := consul.Partitions().Read(context.Background(), partitionName, nil)
-	require.NoError(t, err)
-	require.NotNil(t, partition)
-	require.Equal(t, partitionName, partition.Name)
-	require.Equal(t, "Created before test", partition.Description)
 }
 
 func TestRun_ExitsAfterTimeout(t *testing.T) {
diff --git a/control-plane/subcommand/server-acl-init/command.go b/control-plane/subcommand/server-acl-init/command.go
index b69b66edbb..5a372d5e29 100644
--- a/control-plane/subcommand/server-acl-init/command.go
+++ b/control-plane/subcommand/server-acl-init/command.go
@@ -67,6 +67,10 @@ type Command struct {
 	flagCreateACLReplicationToken bool
 	flagACLReplicationTokenFile   string
 
+	// Flags to support partitions
+	flagEnablePartitions bool   // true if Admin Partitions are enabled
+	flagPartitionName    string // name of the Admin Partition
+
 	// Flags to support namespaces
 	flagEnableNamespaces                 bool   // Use namespacing on all components
 	flagConsulSyncDestinationNamespace   string // Consul namespace to register all catalog sync services into if not mirroring
@@ -169,6 +173,11 @@ func (c *Command) init() {
 	c.flags.BoolVar(&c.flagUseHTTPS, "use-https", false,
 		"Toggle for using HTTPS for all API calls to Consul.")
 
+	c.flags.BoolVar(&c.flagEnablePartitions, "enable-partitions", false,
+		"[Enterprise Only] Enables Admin Partitions [Enterprise only feature]")
+	c.flags.StringVar(&c.flagPartitionName, "partition", "",
+		"[Enterprise Only] Name of the Admin Partition")
+
 	c.flags.BoolVar(&c.flagEnableNamespaces, "enable-namespaces", false,
 		"[Enterprise Only] Enables namespaces, in either a single Consul namespace or mirrored [Enterprise only feature]")
 	c.flags.StringVar(&c.flagConsulSyncDestinationNamespace, "consul-sync-destination-namespace", consulDefaultNamespace,
@@ -382,6 +391,15 @@ func (c *Command) Run(args []string) int {
 		}
 	}
 
+	if c.flagEnablePartitions && c.flagPartitionName == "default" && isPrimary {
+		// Partition token must be local because only the Primary datacenter can have Admin Partitions.
+		err := c.createLocalACL("partitions", partitionRules, consulDC, isPrimary, consulClient)
+		if err != nil {
+			c.log.Error(err.Error())
+			return 1
+		}
+	}
+
 	// If namespaces are enabled, to allow cross-Consul-namespace permissions
 	// for services from k8s, the Consul `default` namespace needs a policy
 	// allowing service discovery in all namespaces. Each namespace that is
@@ -389,12 +407,17 @@ func (c *Command) Run(args []string) int {
 	// connect inject) needs to reference this policy on namespace creation
 	// to finish the cross namespace permission setup.
 	if c.flagEnableNamespaces {
+		crossNamespaceRule, err := c.crossNamespaceRule()
+		if err != nil {
+			c.log.Error("Error templating cross namespace rules", "err", err)
+			return 1
+		}
 		policyTmpl := api.ACLPolicy{
 			Name:        "cross-namespace-policy",
 			Description: "Policy to allow permissions to cross Consul namespaces for k8s services",
-			Rules:       crossNamespaceRules,
+			Rules:       crossNamespaceRule,
 		}
-		err := c.untilSucceeds(fmt.Sprintf("creating %s policy", policyTmpl.Name),
+		err = c.untilSucceeds(fmt.Sprintf("creating %s policy", policyTmpl.Name),
 			func() error {
 				return c.createOrUpdateACLPolicy(policyTmpl, consulClient)
 			})
@@ -827,6 +850,12 @@ func (c *Command) validateFlags() error {
 		)
 	}
 
+	if c.flagEnablePartitions && c.flagPartitionName == "" {
+		return errors.New("-partition must be set if -enable-partitions is true")
+	}
+	if !c.flagEnablePartitions && c.flagPartitionName != "" {
+		return fmt.Errorf("-enable-partitions must be 'true' if setting -partition to %s", c.flagPartitionName)
+	}
 	return nil
 }
 
diff --git a/control-plane/subcommand/server-acl-init/command_ent_test.go b/control-plane/subcommand/server-acl-init/command_ent_test.go
index 57270644b6..96659caab8 100644
--- a/control-plane/subcommand/server-acl-init/command_ent_test.go
+++ b/control-plane/subcommand/server-acl-init/command_ent_test.go
@@ -19,7 +19,6 @@ import (
 // and there's a single consul destination namespace.
 func TestRun_ConnectInject_SingleDestinationNamespace(t *testing.T) {
 	t.Parallel()
-
 	consulDestNamespaces := []string{"default", "destination"}
 	for _, consulDestNamespace := range consulDestNamespaces {
 		t.Run(consulDestNamespace, func(tt *testing.T) {
@@ -40,6 +39,8 @@ func TestRun_ConnectInject_SingleDestinationNamespace(t *testing.T) {
 				"-resource-prefix=" + resourcePrefix,
 				"-k8s-namespace=" + ns,
 				"-create-inject-token",
+				"-enable-partitions",
+				"-partition=default",
 				"-enable-namespaces",
 				"-consul-inject-destination-namespace", consulDestNamespace,
 				"-acl-binding-rule-selector=serviceaccount.name!=default",
@@ -160,6 +161,8 @@ func TestRun_ConnectInject_NamespaceMirroring(t *testing.T) {
 				"-resource-prefix=" + resourcePrefix,
 				"-k8s-namespace=" + ns,
 				"-create-inject-token",
+				"-enable-partitions",
+				"-partition=default",
 				"-enable-namespaces",
 				"-enable-inject-k8s-namespace-mirroring",
 				"-inject-k8s-namespace-mirroring-prefix", c.MirroringPrefix,
@@ -203,7 +206,7 @@ func TestRun_ConnectInject_NamespaceMirroring(t *testing.T) {
 	}
 }
 
-// Test that ACL policies get updated if namespaces config changes.
+// Test that ACL policies get updated if namespaces/partition config changes.
 func TestRun_ACLPolicyUpdates(t *testing.T) {
 	t.Parallel()
 
@@ -234,9 +237,11 @@ func TestRun_ACLPolicyUpdates(t *testing.T) {
 				"-terminating-gateway-name=anothergw",
 				"-create-controller-token",
 			}
-			// Our second run, we're going to update from namespaces disabled to
-			// namespaces enabled with a single destination ns.
+			// Our second run, we're going to update from partitions and namespaces disabled to
+			// namespaces enabled with a single destination ns and partitions enabled.
 			secondRunArgs := append(firstRunArgs,
+				"-enable-partitions",
+				"-partition=default",
 				"-enable-namespaces",
 				"-consul-sync-destination-namespace=sync",
 				"-consul-inject-destination-namespace=dest")
@@ -322,6 +327,7 @@ func TestRun_ACLPolicyUpdates(t *testing.T) {
 				"gw-terminating-gateway-token",
 				"anothergw-terminating-gateway-token",
 				"controller-token",
+				"partitions-token",
 			}
 			policies, _, err = consul.ACL().PolicyList(nil)
 			require.NoError(err)
@@ -348,10 +354,12 @@ func TestRun_ACLPolicyUpdates(t *testing.T) {
 				case "connect-inject-token":
 					// The connect inject token doesn't have namespace config,
 					// but does change to operator:write from an empty string.
-					require.Contains(actRules, "operator = \"write\"")
+					require.Contains(actRules, "acl = \"write\"")
 				case "client-snapshot-agent-token", "enterprise-license-token":
 					// The snapshot agent and enterprise license tokens shouldn't change.
 					require.NotContains(actRules, "namespace")
+				case "partitions-token":
+					require.Contains(actRules, "acl = \"write\"\noperator = \"write\"")
 				default:
 					// Assert that the policies have the word namespace in them. This
 					// tests that they were updated. The actual contents are tested
@@ -528,6 +536,8 @@ func TestRun_ConnectInject_Updates(t *testing.T) {
 				"-server-port=" + strings.Split(testAgent.HTTPAddr, ":")[1],
 				"-resource-prefix=" + resourcePrefix,
 				"-k8s-namespace=" + ns,
+				"-enable-partitions",
+				"-partition=default",
 				"-create-inject-token",
 			}
 
@@ -693,6 +703,13 @@ func TestRun_TokensWithNamespacesEnabled(t *testing.T) {
 			SecretNames: []string{resourcePrefix + "-controller-acl-token"},
 			LocalToken:  false,
 		},
+		"partitions token": {
+			TokenFlags:  []string{"-enable-partitions", "-partition=default"},
+			PolicyNames: []string{"partitions-token"},
+			PolicyDCs:   []string{"dc1"},
+			SecretNames: []string{resourcePrefix + "-partitions-acl-token"},
+			LocalToken:  true,
+		},
 	}
 	for testName, c := range cases {
 		t.Run(testName, func(t *testing.T) {
@@ -713,6 +730,8 @@ func TestRun_TokensWithNamespacesEnabled(t *testing.T) {
 				"-server-port", strings.Split(testSvr.HTTPAddr, ":")[1],
 				"-resource-prefix=" + resourcePrefix,
 				"-k8s-namespace=" + ns,
+				"-enable-partitions",
+				"-partition=default",
 				"-enable-namespaces",
 			}, c.TokenFlags...)
 
@@ -779,37 +798,43 @@ func TestRun_GatewayNamespaceParsing(t *testing.T) {
 				"gateway-ingress-gateway-token",
 				"another-gateway-ingress-gateway-token"},
 			ExpectedPolicies: []string{`
-namespace "default" {
-  service "ingress" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
-  }
-  service_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "default" {
+    service "ingress" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }
   }
 }`, `
-namespace "default" {
-  service "gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
-  }
-  service_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "default" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }
   }
 }`, `
-namespace "default" {
-  service "another-gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
-  }
-  service_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "default" {
+    service "another-gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }
   }
 }`},
 		},
@@ -822,37 +847,43 @@ namespace "default" {
 				"gateway-ingress-gateway-token",
 				"another-gateway-ingress-gateway-token"},
 			ExpectedPolicies: []string{`
-namespace "default" {
-  service "ingress" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
-  }
-  service_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "default" {
+    service "ingress" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }
   }
 }`, `
-namespace "namespace1" {
-  service "gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
-  }
-  service_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "namespace1" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }
   }
 }`, `
-namespace "namespace2" {
-  service "another-gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
-  }
-  service_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "namespace2" {
+    service "another-gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }
   }
 }`},
 		},
@@ -865,28 +896,34 @@ namespace "namespace2" {
 				"gateway-terminating-gateway-token",
 				"another-gateway-terminating-gateway-token"},
 			ExpectedPolicies: []string{`
-namespace "default" {
-  service "terminating" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "default" {
+    service "terminating" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
   }
 }`, `
-namespace "default" {
-  service "gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "default" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
   }
 }`, `
-namespace "default" {
-  service "another-gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "default" {
+    service "another-gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
   }
 }`},
 		},
@@ -899,28 +936,34 @@ namespace "default" {
 				"gateway-terminating-gateway-token",
 				"another-gateway-terminating-gateway-token"},
 			ExpectedPolicies: []string{`
-namespace "default" {
-  service "terminating" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "default" {
+    service "terminating" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
   }
 }`, `
-namespace "namespace1" {
-  service "gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "namespace1" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
   }
 }`, `
-namespace "namespace2" {
-  service "another-gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
+partition "default" {
+  namespace "namespace2" {
+    service "another-gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
   }
 }`},
 		},
@@ -944,6 +987,8 @@ namespace "namespace2" {
 				"-server-port", strings.Split(testSvr.HTTPAddr, ":")[1],
 				"-resource-prefix=" + resourcePrefix,
 				"-enable-namespaces=true",
+				"-enable-partitions",
+				"-partition=default",
 			}, c.TokenFlags...)
 
 			responseCode := cmd.Run(cmdArgs)
diff --git a/control-plane/subcommand/server-acl-init/rules.go b/control-plane/subcommand/server-acl-init/rules.go
index 1b8a64b704..b729621873 100644
--- a/control-plane/subcommand/server-acl-init/rules.go
+++ b/control-plane/subcommand/server-acl-init/rules.go
@@ -7,6 +7,8 @@ import (
 )
 
 type rulesData struct {
+	EnablePartitions        bool
+	PartitionName           string
 	EnableNamespaces        bool
 	SyncConsulDestNS        string
 	SyncEnableNSMirroring   bool
@@ -35,28 +37,49 @@ service "consul-snapshot" {
 }`
 
 const entLicenseRules = `operator = "write"`
+const partitionRules = `acl = "write"
+operator = "write"
+agent_prefix "" {
+  policy = "read"
+}`
 
-const crossNamespaceRules = `namespace_prefix "" {
-  service_prefix "" {
-    policy = "read"
-  }
-  node_prefix "" {
-    policy = "read"
+func (c *Command) crossNamespaceRule() (string, error) {
+	crossNamespaceRulesTpl := `{{- if .EnablePartitions }}
+partition_prefix "" {
+{{- end }}
+  namespace_prefix "" {
+    service_prefix "" {
+      policy = "read"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
   }
-} `
+{{- if .EnablePartitions }}
+}
+{{- end }}`
+
+	return c.renderRules(crossNamespaceRulesTpl)
+}
 
 func (c *Command) agentRules() (string, error) {
 	agentRulesTpl := `
+{{- if .EnablePartitions }}
+partition "{{ .PartitionName }}" {
+{{- end }}
   node_prefix "" {
     policy = "write"
   }
 {{- if .EnableNamespaces }}
-namespace_prefix "" {
+  namespace_prefix "" {
 {{- end }}
-  service_prefix "" {
-    policy = "read"
-  }
+    service_prefix "" {
+      policy = "read"
+    }
 {{- if .EnableNamespaces }}
+  }
+{{- end }}
+{{- if .EnablePartitions }}
 }
 {{- end }}
 `
@@ -80,16 +103,22 @@ func (c *Command) anonymousTokenRules() (string, error) {
 	// ACL token. Thus the anonymous policy must
 	// allow reading all services.
 	anonTokenRulesTpl := `
+{{- if .EnablePartitions }}
+partition_prefix "" {
+{{- end }}
 {{- if .EnableNamespaces }}
-namespace_prefix "" {
+  namespace_prefix "" {
 {{- end }}
-  node_prefix "" {
-     policy = "read"
-  }
-  service_prefix "" {
-     policy = "read"
-  }
+    node_prefix "" {
+       policy = "read"
+    }
+    service_prefix "" {
+       policy = "read"
+    }
 {{- if .EnableNamespaces }}
+  }
+{{- end }}
+{{- if .EnablePartitions }}
 }
 {{- end }}
 `
@@ -133,19 +162,25 @@ namespace_prefix "" {
 
 func (c *Command) ingressGatewayRules(name, namespace string) (string, error) {
 	ingressGatewayRulesTpl := `
+{{- if .EnablePartitions }}
+partition "{{ .PartitionName }}" {
+{{- end }}
 {{- if .EnableNamespaces }}
-namespace "{{ .GatewayNamespace }}" {
+  namespace "{{ .GatewayNamespace }}" {
 {{- end }}
-  service "{{ .GatewayName }}" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
-  }
-  service_prefix "" {
-    policy = "read"
-  }
+    service "{{ .GatewayName }}" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }
 {{- if .EnableNamespaces }}
+  }
+{{- end }}
+{{- if .EnablePartitions }}
 }
 {{- end }}
 `
@@ -159,16 +194,22 @@ namespace "{{ .GatewayNamespace }}" {
 // of the initial implementation
 func (c *Command) terminatingGatewayRules(name, namespace string) (string, error) {
 	terminatingGatewayRulesTpl := `
+{{- if .EnablePartitions }}
+partition "{{ .PartitionName }}" {
+{{- end }}
 {{- if .EnableNamespaces }}
-namespace "{{ .GatewayNamespace }}" {
+  namespace "{{ .GatewayNamespace }}" {
 {{- end }}
-  service "{{ .GatewayName }}" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
-  }
+    service "{{ .GatewayName }}" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
 {{- if .EnableNamespaces }}
+  }
+{{- end }}
+{{- if .EnablePartitions }}
 }
 {{- end }}
 `
@@ -209,20 +250,28 @@ func (c *Command) injectRules() (string, error) {
 	// When ACLs are enabled, the endpoints controller needs "acl:write" permissions
 	// to delete ACL tokens created via "consul login".
 	injectRulesTpl := `
+{{- if .EnablePartitions }}
+partition "{{ .PartitionName }}" {
+  acl = "write"
+{{- else }}
 {{- if .EnableNamespaces }}
-operator = "write"
+  operator = "write"
 {{- end }}
-node_prefix "" {
-  policy = "write"
-}
-{{- if .EnableNamespaces }}
-namespace_prefix "" {
 {{- end }}
-  acl = "write"
-  service_prefix "" {
+  node_prefix "" {
     policy = "write"
   }
 {{- if .EnableNamespaces }}
+  namespace_prefix "" {
+{{- end }}
+    acl = "write"
+    service_prefix "" {
+      policy = "write"
+    }
+{{- if .EnableNamespaces }}
+  }
+{{- end }}
+{{- if .EnablePartitions }}
 }
 {{- end }}`
 	return c.renderRules(injectRulesTpl)
@@ -237,22 +286,28 @@ func (c *Command) aclReplicationRules() (string, error) {
 	// datacenters during federation since in order to start ACL replication,
 	// we need a token with both replication and agent permissions.
 	aclReplicationRulesTpl := `
-operator = "write"
-agent_prefix "" {
-  policy = "read"
-}
-node_prefix "" {
-  policy = "write"
-}
-{{- if .EnableNamespaces }}
-namespace_prefix "" {
+{{- if .EnablePartitions }}
+partition "default" {
 {{- end }}
-  acl = "write"
-  service_prefix "" {
+  operator = "write"
+  agent_prefix "" {
     policy = "read"
-    intentions = "read"
   }
+  node_prefix "" {
+    policy = "write"
+  }
+{{- if .EnableNamespaces }}
+  namespace_prefix "" {
+{{- end }}
+    acl = "write"
+    service_prefix "" {
+      policy = "read"
+      intentions = "read"
+    }
 {{- if .EnableNamespaces }}
+  }
+{{- end }}
+{{- if .EnablePartitions }}
 }
 {{- end }}
 `
@@ -261,19 +316,28 @@ namespace_prefix "" {
 
 func (c *Command) controllerRules() (string, error) {
 	controllerRules := `
-operator = "write"
+{{- if .EnablePartitions }}
+partition "{{ .PartitionName }}" {
+  mesh = "write"
+  acl = "write"
+{{- else }}
+  operator = "write"
+{{- end }}
 {{- if .EnableNamespaces }}
 {{- if .InjectEnableNSMirroring }}
-namespace_prefix "{{ .InjectNSMirroringPrefix }}" {
+  namespace_prefix "{{ .InjectNSMirroringPrefix }}" {
 {{- else }}
-namespace "{{ .InjectConsulDestNS }}" {
+  namespace "{{ .InjectConsulDestNS }}" {
 {{- end }}
 {{- end }}
-  service_prefix "" {
-    policy = "write"
-    intentions = "write"
-  }
+    service_prefix "" {
+      policy = "write"
+      intentions = "write"
+    }
 {{- if .EnableNamespaces }}
+  }
+{{- end }}
+{{- if .EnablePartitions }}
 }
 {{- end }}
 `
@@ -282,6 +346,8 @@ namespace "{{ .InjectConsulDestNS }}" {
 
 func (c *Command) rulesData() rulesData {
 	return rulesData{
+		EnablePartitions:        c.flagEnablePartitions,
+		PartitionName:           c.flagPartitionName,
 		EnableNamespaces:        c.flagEnableNamespaces,
 		SyncConsulDestNS:        c.flagConsulSyncDestinationNamespace,
 		SyncEnableNSMirroring:   c.flagEnableSyncK8SNSMirroring,
diff --git a/control-plane/subcommand/server-acl-init/rules_test.go b/control-plane/subcommand/server-acl-init/rules_test.go
index 1160236eef..54a7d51837 100644
--- a/control-plane/subcommand/server-acl-init/rules_test.go
+++ b/control-plane/subcommand/server-acl-init/rules_test.go
@@ -10,28 +10,48 @@ import (
 func TestAgentRules(t *testing.T) {
 	cases := []struct {
 		Name             string
+		EnablePartitions bool
+		PartitionName    string
 		EnableNamespaces bool
 		Expected         string
 	}{
 		{
-			"Namespaces are disabled",
-			false,
-			`node_prefix "" {
+			Name: "Namespaces and Partitions are disabled",
+			Expected: `
+  node_prefix "" {
     policy = "write"
   }
-  service_prefix "" {
-    policy = "read"
+    service_prefix "" {
+      policy = "read"
+    }`,
+		},
+		{
+			Name:             "Namespaces are enabled, Partitions are disabled",
+			EnableNamespaces: true,
+			Expected: `
+  node_prefix "" {
+    policy = "write"
+  }
+  namespace_prefix "" {
+    service_prefix "" {
+      policy = "read"
+    }
   }`,
 		},
 		{
-			"Namespaces are enabled",
-			true,
-			`node_prefix "" {
+			Name:             "Namespaces and Partitions are enabled",
+			EnablePartitions: true,
+			PartitionName:    "part-1",
+			EnableNamespaces: true,
+			Expected: `
+partition "part-1" {
+  node_prefix "" {
     policy = "write"
   }
-namespace_prefix "" {
-  service_prefix "" {
-    policy = "read"
+  namespace_prefix "" {
+    service_prefix "" {
+      policy = "read"
+    }
   }
 }`,
 		},
@@ -39,16 +59,16 @@ namespace_prefix "" {
 
 	for _, tt := range cases {
 		t.Run(tt.Name, func(t *testing.T) {
-			require := require.New(t)
-
 			cmd := Command{
+				flagEnablePartitions: tt.EnablePartitions,
+				flagPartitionName:    tt.PartitionName,
 				flagEnableNamespaces: tt.EnableNamespaces,
 			}
 
 			agentRules, err := cmd.agentRules()
 
-			require.NoError(err)
-			require.Equal(tt.Expected, agentRules)
+			require.NoError(t, err)
+			require.Equal(t, tt.Expected, agentRules)
 		})
 	}
 }
@@ -56,30 +76,48 @@ namespace_prefix "" {
 func TestAnonymousTokenRules(t *testing.T) {
 	cases := []struct {
 		Name             string
+		EnablePartitions bool
+		PartitionName    string
 		EnableNamespaces bool
 		Expected         string
 	}{
 		{
-			"Namespaces are disabled",
-			false,
-			`
-  node_prefix "" {
-     policy = "read"
-  }
-  service_prefix "" {
-     policy = "read"
+			Name: "Namespaces and Partitions are disabled",
+			Expected: `
+    node_prefix "" {
+       policy = "read"
+    }
+    service_prefix "" {
+       policy = "read"
+    }`,
+		},
+		{
+			Name:             "Namespaces are enabled, Partitions are disabled",
+			EnableNamespaces: true,
+			Expected: `
+  namespace_prefix "" {
+    node_prefix "" {
+       policy = "read"
+    }
+    service_prefix "" {
+       policy = "read"
+    }
   }`,
 		},
 		{
-			"Namespaces are enabled",
-			true,
-			`
-namespace_prefix "" {
-  node_prefix "" {
-     policy = "read"
-  }
-  service_prefix "" {
-     policy = "read"
+			Name:             "Namespaces and Partitions are enabled",
+			EnablePartitions: true,
+			PartitionName:    "part-2",
+			EnableNamespaces: true,
+			Expected: `
+partition_prefix "" {
+  namespace_prefix "" {
+    node_prefix "" {
+       policy = "read"
+    }
+    service_prefix "" {
+       policy = "read"
+    }
   }
 }`,
 		},
@@ -87,16 +125,16 @@ namespace_prefix "" {
 
 	for _, tt := range cases {
 		t.Run(tt.Name, func(t *testing.T) {
-			require := require.New(t)
-
 			cmd := Command{
+				flagEnablePartitions: tt.EnablePartitions,
+				flagPartitionName:    tt.PartitionName,
 				flagEnableNamespaces: tt.EnableNamespaces,
 			}
 
 			rules, err := cmd.anonymousTokenRules()
 
-			require.NoError(err)
-			require.Equal(tt.Expected, rules)
+			require.NoError(t, err)
+			require.Equal(t, tt.Expected, rules)
 		})
 	}
 }
@@ -108,9 +146,8 @@ func TestMeshGatewayRules(t *testing.T) {
 		Expected         string
 	}{
 		{
-			"Namespaces are disabled",
-			false,
-			`agent_prefix "" {
+			Name: "Namespaces are disabled",
+			Expected: `agent_prefix "" {
   	policy = "read"
   }
   service "mesh-gateway" {
@@ -124,9 +161,9 @@ func TestMeshGatewayRules(t *testing.T) {
   }`,
 		},
 		{
-			"Namespaces are enabled",
-			true,
-			`agent_prefix "" {
+			Name:             "Namespaces are enabled",
+			EnableNamespaces: true,
+			Expected: `agent_prefix "" {
   	policy = "read"
   }
 namespace "default" {
@@ -147,16 +184,14 @@ namespace_prefix "" {
 
 	for _, tt := range cases {
 		t.Run(tt.Name, func(t *testing.T) {
-			require := require.New(t)
-
 			cmd := Command{
 				flagEnableNamespaces: tt.EnableNamespaces,
 			}
 
 			meshGatewayRules, err := cmd.meshGatewayRules()
 
-			require.NoError(err)
-			require.Equal(tt.Expected, meshGatewayRules)
+			require.NoError(t, err)
+			require.Equal(t, tt.Expected, meshGatewayRules)
 		})
 	}
 }
@@ -166,58 +201,102 @@ func TestIngressGatewayRules(t *testing.T) {
 		Name             string
 		GatewayName      string
 		GatewayNamespace string
+		EnablePartitions bool
+		PartitionName    string
 		EnableNamespaces bool
 		Expected         string
 	}{
 		{
-			"Namespaces are disabled",
-			"ingress-gateway",
-			"",
-			false,
-			`
-  service "ingress-gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
-  }
-  service_prefix "" {
-    policy = "read"
+			Name:        "Namespaces and Partitions are disabled",
+			GatewayName: "ingress-gateway",
+			Expected: `
+    service "ingress-gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }`,
+		},
+		{
+			Name:             "Namespaces are enabled, Partitions are disabled",
+			GatewayName:      "gateway",
+			GatewayNamespace: "default",
+			EnableNamespaces: true,
+			Expected: `
+  namespace "default" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }
   }`,
 		},
 		{
-			"Namespaces are enabled",
-			"gateway",
-			"default",
-			true,
-			`
-namespace "default" {
-  service "gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
-  }
-  service_prefix "" {
-    policy = "read"
+			Name:             "Namespaces are enabled, non-default namespace, Partitions are disabled",
+			GatewayName:      "gateway",
+			GatewayNamespace: "non-default",
+			EnableNamespaces: true,
+			Expected: `
+  namespace "non-default" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }
+  }`,
+		},
+		{
+			Name:             "Namespaces and Partitions are enabled",
+			GatewayName:      "gateway",
+			GatewayNamespace: "default",
+			EnableNamespaces: true,
+			EnablePartitions: true,
+			PartitionName:    "part-1",
+			Expected: `
+partition "part-1" {
+  namespace "default" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }
   }
 }`,
 		},
 		{
-			"Namespaces are enabled, non-default namespace",
-			"gateway",
-			"non-default",
-			true,
-			`
-namespace "non-default" {
-  service "gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
-  }
-  service_prefix "" {
-    policy = "read"
+			Name:             "Namespaces and Partitions are enabled, non-default namespace",
+			GatewayName:      "gateway",
+			GatewayNamespace: "non-default",
+			EnableNamespaces: true,
+			EnablePartitions: true,
+			PartitionName:    "default",
+			Expected: `
+partition "default" {
+  namespace "non-default" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+    service_prefix "" {
+      policy = "read"
+    }
   }
 }`,
 		},
@@ -225,16 +304,16 @@ namespace "non-default" {
 
 	for _, tt := range cases {
 		t.Run(tt.Name, func(t *testing.T) {
-			require := require.New(t)
-
 			cmd := Command{
+				flagEnablePartitions: tt.EnablePartitions,
+				flagPartitionName:    tt.PartitionName,
 				flagEnableNamespaces: tt.EnableNamespaces,
 			}
 
 			ingressGatewayRules, err := cmd.ingressGatewayRules(tt.GatewayName, tt.GatewayNamespace)
 
-			require.NoError(err)
-			require.Equal(tt.Expected, ingressGatewayRules)
+			require.NoError(t, err)
+			require.Equal(t, tt.Expected, ingressGatewayRules)
 		})
 	}
 }
@@ -245,48 +324,86 @@ func TestTerminatingGatewayRules(t *testing.T) {
 		GatewayName      string
 		GatewayNamespace string
 		EnableNamespaces bool
+		EnablePartitions bool
+		PartitionName    string
 		Expected         string
 	}{
 		{
-			"Namespaces are disabled",
-			"terminating-gateway",
-			"",
-			false,
-			`
-  service "terminating-gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
+			Name:        "Namespaces and Partitions are disabled",
+			GatewayName: "terminating-gateway",
+			Expected: `
+    service "terminating-gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }`,
+		},
+		{
+			Name:             "Namespaces are enabled, Partitions are disabled",
+			GatewayName:      "gateway",
+			GatewayNamespace: "default",
+			EnableNamespaces: true,
+			Expected: `
+  namespace "default" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
   }`,
 		},
 		{
-			"Namespaces are enabled",
-			"gateway",
-			"default",
-			true,
-			`
-namespace "default" {
-  service "gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
+			Name:             "Namespaces are enabled, non-default namespace, Partitions are disabled",
+			GatewayName:      "gateway",
+			GatewayNamespace: "non-default",
+			EnableNamespaces: true,
+			Expected: `
+  namespace "non-default" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
+  }`,
+		},
+		{
+			Name:             "Namespaces and Partitions are enabled",
+			GatewayName:      "gateway",
+			GatewayNamespace: "default",
+			EnableNamespaces: true,
+			EnablePartitions: true,
+			PartitionName:    "part-1",
+			Expected: `
+partition "part-1" {
+  namespace "default" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
   }
 }`,
 		},
 		{
-			"Namespaces are enabled, non-default namespace",
-			"gateway",
-			"non-default",
-			true,
-			`
-namespace "non-default" {
-  service "gateway" {
-     policy = "write"
-  }
-  node_prefix "" {
-    policy = "read"
+			Name:             "Namespaces and Partitions are enabled, non-default namespace",
+			GatewayName:      "gateway",
+			GatewayNamespace: "non-default",
+			EnableNamespaces: true,
+			EnablePartitions: true,
+			PartitionName:    "default",
+			Expected: `
+partition "default" {
+  namespace "non-default" {
+    service "gateway" {
+       policy = "write"
+    }
+    node_prefix "" {
+      policy = "read"
+    }
   }
 }`,
 		},
@@ -294,16 +411,16 @@ namespace "non-default" {
 
 	for _, tt := range cases {
 		t.Run(tt.Name, func(t *testing.T) {
-			require := require.New(t)
-
 			cmd := Command{
+				flagEnablePartitions: tt.EnablePartitions,
+				flagPartitionName:    tt.PartitionName,
 				flagEnableNamespaces: tt.EnableNamespaces,
 			}
 
 			terminatingGatewayRules, err := cmd.terminatingGatewayRules(tt.GatewayName, tt.GatewayNamespace)
 
-			require.NoError(err)
-			require.Equal(tt.Expected, terminatingGatewayRules)
+			require.NoError(t, err)
+			require.Equal(t, tt.Expected, terminatingGatewayRules)
 		})
 	}
 }
@@ -319,13 +436,12 @@ func TestSyncRules(t *testing.T) {
 		Expected                       string
 	}{
 		{
-			"Namespaces are disabled",
-			false,
-			"sync-namespace",
-			true,
-			"prefix-",
-			"k8s-sync",
-			`node "k8s-sync" {
+			Name:                           "Namespaces are disabled",
+			ConsulSyncDestinationNamespace: "sync-namespace",
+			EnableSyncK8SNSMirroring:       true,
+			SyncK8SNSMirroringPrefix:       "prefix-",
+			SyncConsulNodeName:             "k8s-sync",
+			Expected: `node "k8s-sync" {
     policy = "write"
   }
   node_prefix "" {
@@ -336,13 +452,12 @@ func TestSyncRules(t *testing.T) {
   }`,
 		},
 		{
-			"Namespaces are disabled, non-default node name",
-			false,
-			"sync-namespace",
-			true,
-			"prefix-",
-			"new-node-name",
-			`node "new-node-name" {
+			Name:                           "Namespaces are disabled, non-default node name",
+			ConsulSyncDestinationNamespace: "sync-namespace",
+			EnableSyncK8SNSMirroring:       true,
+			SyncK8SNSMirroringPrefix:       "prefix-",
+			SyncConsulNodeName:             "new-node-name",
+			Expected: `node "new-node-name" {
     policy = "write"
   }
   node_prefix "" {
@@ -353,13 +468,12 @@ func TestSyncRules(t *testing.T) {
   }`,
 		},
 		{
-			"Namespaces are enabled, mirroring disabled",
-			true,
-			"sync-namespace",
-			false,
-			"prefix-",
-			"k8s-sync",
-			`node "k8s-sync" {
+			Name:                           "Namespaces are enabled, mirroring disabled",
+			EnableNamespaces:               true,
+			ConsulSyncDestinationNamespace: "sync-namespace",
+			SyncK8SNSMirroringPrefix:       "prefix-",
+			SyncConsulNodeName:             "k8s-sync",
+			Expected: `node "k8s-sync" {
     policy = "write"
   }
 operator = "write"
@@ -373,13 +487,12 @@ namespace "sync-namespace" {
 }`,
 		},
 		{
-			"Namespaces are enabled, mirroring disabled, non-default node name",
-			true,
-			"sync-namespace",
-			false,
-			"prefix-",
-			"new-node-name",
-			`node "new-node-name" {
+			Name:                           "Namespaces are enabled, mirroring disabled, non-default node name",
+			EnableNamespaces:               true,
+			ConsulSyncDestinationNamespace: "sync-namespace",
+			SyncK8SNSMirroringPrefix:       "prefix-",
+			SyncConsulNodeName:             "new-node-name",
+			Expected: `node "new-node-name" {
     policy = "write"
   }
 operator = "write"
@@ -393,13 +506,12 @@ namespace "sync-namespace" {
 }`,
 		},
 		{
-			"Namespaces are enabled, mirroring enabled, prefix empty",
-			true,
-			"sync-namespace",
-			true,
-			"",
-			"k8s-sync",
-			`node "k8s-sync" {
+			Name:                           "Namespaces are enabled, mirroring enabled, prefix empty",
+			EnableNamespaces:               true,
+			ConsulSyncDestinationNamespace: "sync-namespace",
+			EnableSyncK8SNSMirroring:       true,
+			SyncConsulNodeName:             "k8s-sync",
+			Expected: `node "k8s-sync" {
     policy = "write"
   }
 operator = "write"
@@ -413,13 +525,12 @@ namespace_prefix "" {
 }`,
 		},
 		{
-			"Namespaces are enabled, mirroring enabled, prefix empty, non-default node name",
-			true,
-			"sync-namespace",
-			true,
-			"",
-			"new-node-name",
-			`node "new-node-name" {
+			Name:                           "Namespaces are enabled, mirroring enabled, prefix empty, non-default node name",
+			EnableNamespaces:               true,
+			ConsulSyncDestinationNamespace: "sync-namespace",
+			EnableSyncK8SNSMirroring:       true,
+			SyncConsulNodeName:             "new-node-name",
+			Expected: `node "new-node-name" {
     policy = "write"
   }
 operator = "write"
@@ -433,13 +544,13 @@ namespace_prefix "" {
 }`,
 		},
 		{
-			"Namespaces are enabled, mirroring enabled, prefix defined",
-			true,
-			"sync-namespace",
-			true,
-			"prefix-",
-			"k8s-sync",
-			`node "k8s-sync" {
+			Name:                           "Namespaces are enabled, mirroring enabled, prefix defined",
+			EnableNamespaces:               true,
+			ConsulSyncDestinationNamespace: "sync-namespace",
+			EnableSyncK8SNSMirroring:       true,
+			SyncK8SNSMirroringPrefix:       "prefix-",
+			SyncConsulNodeName:             "k8s-sync",
+			Expected: `node "k8s-sync" {
     policy = "write"
   }
 operator = "write"
@@ -453,13 +564,13 @@ namespace_prefix "prefix-" {
 }`,
 		},
 		{
-			"Namespaces are enabled, mirroring enabled, prefix defined, non-default node name",
-			true,
-			"sync-namespace",
-			true,
-			"prefix-",
-			"new-node-name",
-			`node "new-node-name" {
+			Name:                           "Namespaces are enabled, mirroring enabled, prefix defined, non-default node name",
+			EnableNamespaces:               true,
+			ConsulSyncDestinationNamespace: "sync-namespace",
+			EnableSyncK8SNSMirroring:       true,
+			SyncK8SNSMirroringPrefix:       "prefix-",
+			SyncConsulNodeName:             "new-node-name",
+			Expected: `node "new-node-name" {
     policy = "write"
   }
 operator = "write"
@@ -476,8 +587,6 @@ namespace_prefix "prefix-" {
 
 	for _, tt := range cases {
 		t.Run(tt.Name, func(t *testing.T) {
-			require := require.New(t)
-
 			cmd := Command{
 				flagEnableNamespaces:               tt.EnableNamespaces,
 				flagConsulSyncDestinationNamespace: tt.ConsulSyncDestinationNamespace,
@@ -488,8 +597,8 @@ namespace_prefix "prefix-" {
 
 			syncRules, err := cmd.syncRules()
 
-			require.NoError(err)
-			require.Equal(tt.Expected, syncRules)
+			require.NoError(t, err)
+			require.Equal(t, tt.Expected, syncRules)
 		})
 	}
 }
@@ -498,48 +607,71 @@ namespace_prefix "prefix-" {
 func TestInjectRules(t *testing.T) {
 	cases := []struct {
 		EnableNamespaces bool
+		EnablePartitions bool
+		PartitionName    string
 		Expected         string
 	}{
 		{
 			EnableNamespaces: false,
+			EnablePartitions: false,
 			Expected: `
-node_prefix "" {
-  policy = "write"
-}
-  acl = "write"
-  service_prefix "" {
+  node_prefix "" {
+    policy = "write"
+  }
+    acl = "write"
+    service_prefix "" {
+      policy = "write"
+    }`,
+		},
+		{
+			EnableNamespaces: true,
+			EnablePartitions: false,
+			Expected: `
+  operator = "write"
+  node_prefix "" {
     policy = "write"
+  }
+  namespace_prefix "" {
+    acl = "write"
+    service_prefix "" {
+      policy = "write"
+    }
   }`,
 		},
 		{
 			EnableNamespaces: true,
+			EnablePartitions: true,
+			PartitionName:    "part-1",
 			Expected: `
-operator = "write"
-node_prefix "" {
-  policy = "write"
-}
-namespace_prefix "" {
+partition "part-1" {
   acl = "write"
-  service_prefix "" {
+  node_prefix "" {
     policy = "write"
   }
+  namespace_prefix "" {
+    acl = "write"
+    service_prefix "" {
+      policy = "write"
+    }
+  }
 }`,
 		},
 	}
 
 	for _, tt := range cases {
-		caseName := fmt.Sprintf("ns=%t", tt.EnableNamespaces)
+		caseName := fmt.Sprintf("ns=%t, partition=%t", tt.EnableNamespaces, tt.EnablePartitions)
 		t.Run(caseName, func(t *testing.T) {
-			require := require.New(t)
 
 			cmd := Command{
+				flagEnablePartitions: tt.EnablePartitions,
+				flagPartitionName:    tt.PartitionName,
 				flagEnableNamespaces: tt.EnableNamespaces,
 			}
 
 			injectorRules, err := cmd.injectRules()
 
-			require.NoError(err)
-			require.Equal(tt.Expected, injectorRules)
+			require.NoError(t, err)
+			require.Equal(t, tt.Expected, injectorRules)
 		})
 	}
 }
@@ -548,39 +680,65 @@ func TestReplicationTokenRules(t *testing.T) {
 	cases := []struct {
 		Name             string
 		EnableNamespaces bool
+		EnablePartitions bool
+		PartitionName    string
 		Expected         string
 	}{
 		{
-			"Namespaces are disabled",
-			false,
-			`operator = "write"
-agent_prefix "" {
-  policy = "read"
-}
-node_prefix "" {
-  policy = "write"
-}
-  acl = "write"
-  service_prefix "" {
+			Name: "Namespaces and Partitions are disabled",
+			Expected: `
+  operator = "write"
+  agent_prefix "" {
     policy = "read"
-    intentions = "read"
+  }
+  node_prefix "" {
+    policy = "write"
+  }
+    acl = "write"
+    service_prefix "" {
+      policy = "read"
+      intentions = "read"
+    }`,
+		},
+		{
+			Name:             "Namespaces are enabled, Partitions are disabled",
+			EnableNamespaces: true,
+			Expected: `
+  operator = "write"
+  agent_prefix "" {
+    policy = "read"
+  }
+  node_prefix "" {
+    policy = "write"
+  }
+  namespace_prefix "" {
+    acl = "write"
+    service_prefix "" {
+      policy = "read"
+      intentions = "read"
+    }
   }`,
 		},
 		{
-			"Namespaces are enabled",
-			true,
-			`operator = "write"
-agent_prefix "" {
-  policy = "read"
-}
-node_prefix "" {
-  policy = "write"
-}
-namespace_prefix "" {
-  acl = "write"
-  service_prefix "" {
+			Name:             "Namespaces and Partitions are enabled, default partition",
+			EnableNamespaces: true,
+			EnablePartitions: true,
+			PartitionName:    "default",
+			Expected: `
+partition "default" {
+  operator = "write"
+  agent_prefix "" {
     policy = "read"
-    intentions = "read"
+  }
+  node_prefix "" {
+    policy = "write"
+  }
+  namespace_prefix "" {
+    acl = "write"
+    service_prefix "" {
+      policy = "read"
+      intentions = "read"
+    }
   }
 }`,
 		},
@@ -588,13 +746,14 @@ namespace_prefix "" {
 
 	for _, tt := range cases {
 		t.Run(tt.Name, func(t *testing.T) {
-			require := require.New(t)
 			cmd := Command{
+				flagEnablePartitions: tt.EnablePartitions,
+				flagPartitionName:    tt.PartitionName,
 				flagEnableNamespaces: tt.EnableNamespaces,
 			}
 			replicationTokenRules, err := cmd.aclReplicationRules()
-			require.NoError(err)
-			require.Equal(tt.Expected, replicationTokenRules)
+			require.NoError(t, err)
+			require.Equal(t, tt.Expected, replicationTokenRules)
 		})
 	}
 }
@@ -602,6 +761,8 @@ namespace_prefix "" {
 func TestControllerRules(t *testing.T) {
 	cases := []struct {
 		Name             string
+		EnablePartitions bool
+		PartitionName    string
 		EnableNamespaces bool
 		DestConsulNS     string
 		Mirroring        bool
@@ -609,48 +770,106 @@ func TestControllerRules(t *testing.T) {
 		Expected         string
 	}{
 		{
-			Name:             "namespaces=disabled",
-			EnableNamespaces: false,
-			Expected: `operator = "write"
-  service_prefix "" {
-    policy = "write"
-    intentions = "write"
+			Name: "namespaces=disabled, partitions=disabled",
+			Expected: `
+  operator = "write"
+    service_prefix "" {
+      policy = "write"
+      intentions = "write"
+    }`,
+		},
+		{
+			Name:             "namespaces=enabled, consulDestNS=consul, partitions=disabled",
+			EnableNamespaces: true,
+			DestConsulNS:     "consul",
+			Expected: `
+  operator = "write"
+  namespace "consul" {
+    service_prefix "" {
+      policy = "write"
+      intentions = "write"
+    }
   }`,
 		},
 		{
-			Name:             "namespaces=enabled, consulDestNS=consul",
+			Name:             "namespaces=enabled, mirroring=true, partitions=disabled",
+			EnableNamespaces: true,
+			Mirroring:        true,
+			Expected: `
+  operator = "write"
+  namespace_prefix "" {
+    service_prefix "" {
+      policy = "write"
+      intentions = "write"
+    }
+  }`,
+		},
+		{
+			Name:             "namespaces=enabled, mirroring=true, mirroringPrefix=prefix-, partitions=disabled",
+			EnableNamespaces: true,
+			Mirroring:        true,
+			MirroringPrefix:  "prefix-",
+			Expected: `
+  operator = "write"
+  namespace_prefix "prefix-" {
+    service_prefix "" {
+      policy = "write"
+      intentions = "write"
+    }
+  }`,
+		},
+		{
+			Name:             "namespaces=enabled, consulDestNS=consul, partitions=enabled",
+			EnablePartitions: true,
+			PartitionName:    "part-1",
 			EnableNamespaces: true,
 			DestConsulNS:     "consul",
-			Expected: `operator = "write"
-namespace "consul" {
-  service_prefix "" {
-    policy = "write"
-    intentions = "write"
+			Expected: `
+partition "part-1" {
+  mesh = "write"
+  acl = "write"
+  namespace "consul" {
+    service_prefix "" {
+      policy = "write"
+      intentions = "write"
+    }
   }
 }`,
 		},
 		{
-			Name:             "namespaces=enabled, mirroring=true",
+			Name:             "namespaces=enabled, mirroring=true, partitions=enabled",
+			EnablePartitions: true,
+			PartitionName:    "part-1",
 			EnableNamespaces: true,
 			Mirroring:        true,
-			Expected: `operator = "write"
-namespace_prefix "" {
-  service_prefix "" {
-    policy = "write"
-    intentions = "write"
+			Expected: `
+partition "part-1" {
+  mesh = "write"
+  acl = "write"
+  namespace_prefix "" {
+    service_prefix "" {
+      policy = "write"
+      intentions = "write"
+    }
   }
 }`,
 		},
 		{
-			Name:             "namespaces=enabled, mirroring=true, mirroringPrefix=prefix-",
+			Name:             "namespaces=enabled, mirroring=true, mirroringPrefix=prefix-, partitions=enabled",
+			EnablePartitions: true,
+			PartitionName:    "part-1",
 			EnableNamespaces: true,
 			Mirroring:        true,
 			MirroringPrefix:  "prefix-",
-			Expected: `operator = "write"
-namespace_prefix "prefix-" {
-  service_prefix "" {
-    policy = "write"
-    intentions = "write"
+			Expected: `
+partition "part-1" {
+  mesh = "write"
+  acl = "write"
+  namespace_prefix "prefix-" {
+    service_prefix "" {
+      policy = "write"
+      intentions = "write"
+    }
   }
 }`,
 		},
@@ -658,19 +877,19 @@ namespace_prefix "prefix-" {
 
 	for _, tt := range cases {
 		t.Run(tt.Name, func(t *testing.T) {
-			require := require.New(t)
-
 			cmd := Command{
 				flagEnableNamespaces:                 tt.EnableNamespaces,
 				flagConsulInjectDestinationNamespace: tt.DestConsulNS,
 				flagEnableInjectK8SNSMirroring:       tt.Mirroring,
 				flagInjectK8SNSMirroringPrefix:       tt.MirroringPrefix,
+				flagEnablePartitions:                 tt.EnablePartitions,
+				flagPartitionName:                    tt.PartitionName,
 			}
 
 			rules, err := cmd.controllerRules()
 
-			require.NoError(err)
-			require.Equal(tt.Expected, rules)
+			require.NoError(t, err)
+			require.Equal(t, tt.Expected, rules)
 		})
 	}
 }