diff --git a/apis/core/v1alpha1/annotations.go b/apis/core/v1alpha1/annotations.go
index c2e8ea8..bf50ffd 100644
--- a/apis/core/v1alpha1/annotations.go
+++ b/apis/core/v1alpha1/annotations.go
@@ -35,6 +35,15 @@ const (
 	// TODO(jaypipes): Link to documentation on cross-account resource
 	// management
 	AnnotationOwnerAccountID = AnnotationPrefix + "owner-account-id"
+	// AnnotationTeamID is an annotation whose value is the identifier
+	// for the AWS team ID to manage the resources.  If this annotation
+	// is set on a CR, the Kubernetes user is indicating that the ACK service
+	// controller should create/patch/delete the resource in the specified AWS
+	// role for this team ID. In order for this cross-account resource management
+	// to succeed, the AWS IAM Role that the ACK service controller runs as needs
+	// to have the ability to call the AWS STS::AssumeRole API call and assume an
+	// IAM Role in the target AWS Account.
+	AnnotationTeamID = AnnotationPrefix + "team-id"
 	// AnnotationRegion is an annotation whose value is the identifier for the
 	// the AWS region in which the resources should be created. If this annotation
 	// is set on a CR metadata, that means the user is indicating to the ACK service
diff --git a/apis/core/v1alpha1/common.go b/apis/core/v1alpha1/common.go
index cbefbb7..b4519e4 100644
--- a/apis/core/v1alpha1/common.go
+++ b/apis/core/v1alpha1/common.go
@@ -19,6 +19,9 @@ type AWSRegion string
 // AWSAccountID represents an AWS account identifier
 type AWSAccountID string
 
+// TeamID represents a team ID identifier.
+type TeamID string
+
 // AWSResourceName represents an AWS Resource Name (ARN)
 type AWSResourceName string
 
diff --git a/pkg/featuregate/features.go b/pkg/featuregate/features.go
index 56275af..4e1bc5a 100644
--- a/pkg/featuregate/features.go
+++ b/pkg/featuregate/features.go
@@ -16,11 +16,16 @@
 // optionally overridden.
 package featuregate
 
+const (
+	// CARMv2 is the name of the CARMv2 feature.
+	CARMv2 = "CARMv2"
+)
+
 // defaultACKFeatureGates is a map of feature names to Feature structs
 // representing the default feature gates for ACK controllers.
 var defaultACKFeatureGates = FeatureGates{
 	// Set feature gates here
-	// "feature1": {Stage: Alpha, Enabled: false},
+	CARMv2: {Stage: Alpha, Enabled: false},
 }
 
 // FeatureStage represents the development stage of a feature.
diff --git a/pkg/runtime/adoption_reconciler.go b/pkg/runtime/adoption_reconciler.go
index 4f0f1cb..4c5f6fd 100644
--- a/pkg/runtime/adoption_reconciler.go
+++ b/pkg/runtime/adoption_reconciler.go
@@ -17,6 +17,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/aws/aws-sdk-go/aws/arn"
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
@@ -32,6 +33,7 @@ import (
 	ackv1alpha1 "github.com/aws-controllers-k8s/runtime/apis/core/v1alpha1"
 	ackcfg "github.com/aws-controllers-k8s/runtime/pkg/config"
 	ackerr "github.com/aws-controllers-k8s/runtime/pkg/errors"
+	"github.com/aws-controllers-k8s/runtime/pkg/featuregate"
 	ackmetrics "github.com/aws-controllers-k8s/runtime/pkg/metrics"
 	"github.com/aws-controllers-k8s/runtime/pkg/requeue"
 	ackrtcache "github.com/aws-controllers-k8s/runtime/pkg/runtime/cache"
@@ -115,20 +117,44 @@ func (r *adoptionReconciler) reconcile(ctx context.Context, req ctrlrt.Request)
 	// If the ConfigMap is not created, or not populated with an
 	// accountID to roleARN mapping, we need to properly requeue with a
 	// helpful message to the user.
-	var roleARN ackv1alpha1.AWSResourceName
 	acctID, needCARMLookup := r.getOwnerAccountID(res)
-	if needCARMLookup {
-		// This means that the user is specifying a namespace that is
-		// annotated with an owner account ID. We need to retrieve the
-		// roleARN from the ConfigMap and properly requeue if the roleARN
-		// is not available.
+
+	var roleARN ackv1alpha1.AWSResourceName
+	if r.cfg.FeatureGates.IsEnabled(featuregate.CARMv2) {
+		teamID := r.getTeamID(res)
+		if teamID != "" {
+			// The user is specifying a namespace that is annotated with a team ID.
+			// Requeue if the corresponding roleARN is not available in the CARMv2 configmap.
+			// Additionally, set the account ID to the role's account ID.
+			roleARN, err = r.getRoleARNv2(string(teamID))
+			if err != nil {
+				ackrtlog.InfoAdoptedResource(r.log, res, fmt.Sprintf("Unable to start adoption reconcilliation %s: %v", acctID, err))
+				return requeue.NeededAfter(err, roleARNNotAvailableRequeueDelay)
+			}
+			parsedARN, err := arn.Parse(string(roleARN))
+			if err != nil {
+				return fmt.Errorf("parsing role ARN %q from namespace annotation: %v", roleARN, err)
+			}
+			acctID = ackv1alpha1.AWSAccountID(parsedARN.AccountID)
+		} else if needCARMLookup {
+			// The user is specifying a namespace that is annotated with an owner account ID.
+			// Requeue if the corresponding roleARN is not available in the CARMv2 configmap.
+			roleARN, err = r.getRoleARNv2(string(acctID))
+			if err != nil {
+				ackrtlog.InfoAdoptedResource(r.log, res, fmt.Sprintf("Unable to start adoption reconcilliation %s: %v", acctID, err))
+				return requeue.NeededAfter(err, roleARNNotAvailableRequeueDelay)
+			}
+		}
+	} else if needCARMLookup {
+		// The user is specifying a namespace that is annotated with an owner account ID.
+		// Requeue if the corresponding roleARN is not available in the Accounts (CARMv1) configmap.
 		roleARN, err = r.getRoleARN(acctID)
 		if err != nil {
 			ackrtlog.InfoAdoptedResource(r.log, res, fmt.Sprintf("Unable to start adoption reconcilliation %s: %v", acctID, err))
-			// r.getRoleARN errors are not terminal, we should requeue.
 			return requeue.NeededAfter(err, roleARNNotAvailableRequeueDelay)
 		}
 	}
+
 	region := r.getRegion(res)
 	targetDescriptor := rmf.ResourceDescriptor()
 	endpointURL := r.getEndpointURL(res)
@@ -460,6 +486,19 @@ func (r *adoptionReconciler) getOwnerAccountID(
 	return ackv1alpha1.AWSAccountID(r.cfg.AccountID), false
 }
 
+// getTeamID returns the team ID that owns the supplied resource.
+func (r *adoptionReconciler) getTeamID(
+	res *ackv1alpha1.AdoptedResource,
+) ackv1alpha1.TeamID {
+	// look for team id in the namespace annotations
+	namespace := res.GetNamespace()
+	teamID, ok := r.cache.Namespaces.GetTeamID(namespace)
+	if ok {
+		return ackv1alpha1.TeamID(teamID)
+	}
+	return ""
+}
+
 // getEndpointURL returns the AWS account that owns the supplied resource.
 // We look for the namespace associated endpoint url, if that is set we use it.
 // Otherwise if none of these annotations are set we use the endpoint url specified
@@ -478,14 +517,28 @@ func (r *adoptionReconciler) getEndpointURL(
 	return r.cfg.EndpointURL
 }
 
-// getRoleARN return the Role ARN that should be assumed in order to manage
-// the resources.
-func (r *adoptionReconciler) getRoleARN(
-	acctID ackv1alpha1.AWSAccountID,
-) (ackv1alpha1.AWSResourceName, error) {
-	roleARN, err := r.cache.Accounts.GetAccountRoleARN(string(acctID))
+// getRoleARNv2 returns the Role ARN that should be assumed for the given account/team ID,
+// from the CARMv2 configmap, in order to manage the resources.
+func (r *adoptionReconciler) getRoleARNv2(id string) (ackv1alpha1.AWSResourceName, error) {
+	// use service level roleARN if present
+	serviceID := r.sc.GetMetadata().ServiceAlias + "." + id
+	if roleARN, err := r.cache.CARMMaps.GetValue(serviceID); err == nil {
+		return ackv1alpha1.AWSResourceName(roleARN), nil
+	}
+	// otherwise use account/team level roleARN
+	roleARN, err := r.cache.CARMMaps.GetValue(id)
+	if err != nil {
+		return "", fmt.Errorf("retrieving role ARN for account/team ID %q from %q configmap: %v", id, ackrtcache.ACKCARMMapV2, err)
+	}
+	return ackv1alpha1.AWSResourceName(roleARN), nil
+}
+
+// getRoleARN returns the Role ARN that should be assumed for the given account ID,
+// from the CARMv1 configmap, in order to manage the resources.
+func (r *adoptionReconciler) getRoleARN(acctID ackv1alpha1.AWSAccountID) (ackv1alpha1.AWSResourceName, error) {
+	roleARN, err := r.cache.Accounts.GetValue(string(acctID))
 	if err != nil {
-		return "", fmt.Errorf("unable to retrieve role ARN for account %s: %v", acctID, err)
+		return "", fmt.Errorf("retrieving role ARN for account ID %q from %q configMap: %v", acctID, ackrtcache.ACKRoleAccountMap, err)
 	}
 	return ackv1alpha1.AWSResourceName(roleARN), nil
 }
diff --git a/pkg/runtime/cache/account.go b/pkg/runtime/cache/account.go
index 56c95a0..72559fd 100644
--- a/pkg/runtime/cache/account.go
+++ b/pkg/runtime/cache/account.go
@@ -28,49 +28,54 @@ var (
 	// ErrCARMConfigMapNotFound is an error that is returned when the CARM
 	// configmap is not found.
 	ErrCARMConfigMapNotFound = errors.New("CARM configmap not found")
-	// ErrAccountIDNotFound is an error that is returned when the account ID
+	// ErrKeyNotFound is an error that is returned when the account ID
 	// is not found in the CARM configmap.
-	ErrAccountIDNotFound = errors.New("account ID not found in CARM configmap")
-	// ErrEmptyRoleARN is an error that is returned when the role ARN is empty
+	ErrKeyNotFound = errors.New("key not found in CARM configmap")
+	// ErrEmptyValue is an error that is returned when the role ARN is empty
 	// in the CARM configmap.
-	ErrEmptyRoleARN = errors.New("role ARN is empty in CARM configmap")
+	ErrEmptyValue = errors.New("role value is empty in CARM configmap")
 )
 
 const (
 	// ACKRoleAccountMap is the name of the configmap map object storing
 	// all the AWS Account IDs associated with their AWS Role ARNs.
 	ACKRoleAccountMap = "ack-role-account-map"
+
+	// ACKCARMMapV2 is the name of the v2 CARM map.
+	// It stores the mapping for:
+	// - Account ID to the AWS role ARNs.
+	ACKCARMMapV2 = "ack-carm-map"
 )
 
-// AccountCache is responsible for caching the CARM configmap
+// CARMMap is responsible for caching the CARM configmap
 // data. It is listening to all the events related to the CARM map and
 // make the changes accordingly.
-type AccountCache struct {
+type CARMMap struct {
 	sync.RWMutex
 	log              logr.Logger
-	roleARNs         map[string]string
+	data             map[string]string
 	configMapCreated bool
 	hasSynced        func() bool
 }
 
-// NewAccountCache instanciate a new AccountCache.
-func NewAccountCache(log logr.Logger) *AccountCache {
-	return &AccountCache{
-		log:              log.WithName("cache.account"),
-		roleARNs:         make(map[string]string),
+// NewCARMMapCache instanciate a new CARMMap.
+func NewCARMMapCache(log logr.Logger) *CARMMap {
+	return &CARMMap{
+		log:              log.WithName("cache.carm"),
+		data:             make(map[string]string),
 		configMapCreated: false,
 	}
 }
 
-// resourceMatchACKRoleAccountConfigMap verifies if a resource is
+// resourceMatchCARMConfigMap verifies if a resource is
 // the CARM configmap. It verifies the name, namespace and object type.
-func resourceMatchACKRoleAccountsConfigMap(raw interface{}) bool {
+func resourceMatchCARMConfigMap(raw interface{}, name string) bool {
 	object, ok := raw.(*corev1.ConfigMap)
-	return ok && object.ObjectMeta.Name == ACKRoleAccountMap
+	return ok && object.ObjectMeta.Name == name
 }
 
 // Run instantiate a new SharedInformer for ConfigMaps and runs it to begin processing items.
-func (c *AccountCache) Run(clientSet kubernetes.Interface, stopCh <-chan struct{}) {
+func (c *CARMMap) Run(name string, clientSet kubernetes.Interface, stopCh <-chan struct{}) {
 	c.log.V(1).Info("Starting shared informer for accounts cache", "targetConfigMap", ACKRoleAccountMap)
 	informer := informersv1.NewConfigMapInformer(
 		clientSet,
@@ -80,33 +85,33 @@ func (c *AccountCache) Run(clientSet kubernetes.Interface, stopCh <-chan struct{
 	)
 	informer.AddEventHandler(k8scache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
-			if resourceMatchACKRoleAccountsConfigMap(obj) {
+			if resourceMatchCARMConfigMap(obj, name) {
 				cm := obj.(*corev1.ConfigMap)
 				object := cm.DeepCopy()
 				// To avoid multiple mutex locks, we are updating the cache
 				// and the configmap existence flag in the same function.
 				configMapCreated := true
-				c.updateAccountRoleData(configMapCreated, object.Data)
+				c.updateData(configMapCreated, object.Data)
 				c.log.V(1).Info("created account config map", "name", cm.ObjectMeta.Name)
 			}
 		},
 		UpdateFunc: func(orig, desired interface{}) {
-			if resourceMatchACKRoleAccountsConfigMap(desired) {
+			if resourceMatchCARMConfigMap(desired, name) {
 				cm := desired.(*corev1.ConfigMap)
 				object := cm.DeepCopy()
 				//TODO(a-hilaly): compare data checksum before updating the cache
-				c.updateAccountRoleData(true, object.Data)
+				c.updateData(true, object.Data)
 				c.log.V(1).Info("updated account config map", "name", cm.ObjectMeta.Name)
 			}
 		},
 		DeleteFunc: func(obj interface{}) {
-			if resourceMatchACKRoleAccountsConfigMap(obj) {
+			if resourceMatchCARMConfigMap(obj, name) {
 				cm := obj.(*corev1.ConfigMap)
 				newMap := make(map[string]string)
 				// To avoid multiple mutex locks, we are updating the cache
 				// and the configmap existence flag in the same function.
 				configMapCreated := false
-				c.updateAccountRoleData(configMapCreated, newMap)
+				c.updateData(configMapCreated, newMap)
 				c.log.V(1).Info("deleted account config map", "name", cm.ObjectMeta.Name)
 			}
 		},
@@ -115,33 +120,33 @@ func (c *AccountCache) Run(clientSet kubernetes.Interface, stopCh <-chan struct{
 	c.hasSynced = informer.HasSynced
 }
 
-// GetAccountRoleARN queries the AWS accountID associated Role ARN
+// GetValue queries the value
 // from the cached CARM configmap. It will return an error if the
-// configmap is not found, the accountID is not found or the role ARN
+// configmap is not found, the key is not found or the value
 // is empty.
 //
 // This function is thread safe.
-func (c *AccountCache) GetAccountRoleARN(accountID string) (string, error) {
+func (c *CARMMap) GetValue(key string) (string, error) {
 	c.RLock()
 	defer c.RUnlock()
 
 	if !c.configMapCreated {
 		return "", ErrCARMConfigMapNotFound
 	}
-	roleARN, ok := c.roleARNs[accountID]
+	roleARN, ok := c.data[key]
 	if !ok {
-		return "", ErrAccountIDNotFound
+		return "", ErrKeyNotFound
 	}
 	if roleARN == "" {
-		return "", ErrEmptyRoleARN
+		return "", ErrEmptyValue
 	}
 	return roleARN, nil
 }
 
-// updateAccountRoleData updates the CARM map. This function is thread safe.
-func (c *AccountCache) updateAccountRoleData(exist bool, data map[string]string) {
+// updateData updates the CARM map. This function is thread safe.
+func (c *CARMMap) updateData(exist bool, data map[string]string) {
 	c.Lock()
 	defer c.Unlock()
-	c.roleARNs = data
+	c.data = data
 	c.configMapCreated = exist
 }
diff --git a/pkg/runtime/cache/account_test.go b/pkg/runtime/cache/account_test.go
index 9ae24cb..e0bdd01 100644
--- a/pkg/runtime/cache/account_test.go
+++ b/pkg/runtime/cache/account_test.go
@@ -64,13 +64,13 @@ func TestAccountCache(t *testing.T) {
 	fakeLogger := ctrlrtzap.New(ctrlrtzap.UseFlagOptions(&zapOptions))
 
 	// initlizing account cache
-	accountCache := ackrtcache.NewAccountCache(fakeLogger)
+	accountCache := ackrtcache.NewCARMMapCache(fakeLogger)
 	stopCh := make(chan struct{})
-	accountCache.Run(k8sClient, stopCh)
+	accountCache.Run(ackrtcache.ACKRoleAccountMap, k8sClient, stopCh)
 
 	// Before creating the configmap, the accountCache should error for any
 	// GetAccountRoleARN call.
-	_, err := accountCache.GetAccountRoleARN(testAccount1)
+	_, err := accountCache.GetValue(testAccount1)
 	require.NotNil(t, err)
 	require.Equal(t, err, ackrtcache.ErrCARMConfigMapNotFound)
 
@@ -90,12 +90,12 @@ func TestAccountCache(t *testing.T) {
 	time.Sleep(time.Second)
 
 	// Test with non existing account
-	_, err = accountCache.GetAccountRoleARN("random-account-not-exist")
+	_, err = accountCache.GetValue("random-account-not-exist")
 	require.NotNil(t, err)
 	require.Equal(t, err, ackrtcache.ErrCARMConfigMapNotFound)
 
 	// Test with existing account
-	_, err = accountCache.GetAccountRoleARN(testAccount1)
+	_, err = accountCache.GetValue(testAccount1)
 	require.NotNil(t, err)
 	require.Equal(t, err, ackrtcache.ErrCARMConfigMapNotFound)
 
@@ -114,17 +114,17 @@ func TestAccountCache(t *testing.T) {
 	time.Sleep(time.Second)
 
 	// Test with non existing account
-	_, err = accountCache.GetAccountRoleARN("random-account-not-exist")
+	_, err = accountCache.GetValue("random-account-not-exist")
 	require.NotNil(t, err)
-	require.Equal(t, err, ackrtcache.ErrAccountIDNotFound)
+	require.Equal(t, err, ackrtcache.ErrKeyNotFound)
 
 	// Test with existing account - but role ARN is empty
-	_, err = accountCache.GetAccountRoleARN(testAccount3)
+	_, err = accountCache.GetValue(testAccount3)
 	require.NotNil(t, err)
-	require.Equal(t, err, ackrtcache.ErrEmptyRoleARN)
+	require.Equal(t, err, ackrtcache.ErrEmptyValue)
 
 	// Test with existing account
-	roleARN, err := accountCache.GetAccountRoleARN(testAccount1)
+	roleARN, err := accountCache.GetValue(testAccount1)
 	require.Nil(t, err)
 	require.Equal(t, roleARN, testAccountARN1)
 
@@ -144,21 +144,21 @@ func TestAccountCache(t *testing.T) {
 	time.Sleep(time.Second)
 
 	// Test with non existing account
-	_, err = accountCache.GetAccountRoleARN("random-account-not-exist")
+	_, err = accountCache.GetValue("random-account-not-exist")
 	require.NotNil(t, err)
-	require.Equal(t, err, ackrtcache.ErrAccountIDNotFound)
+	require.Equal(t, err, ackrtcache.ErrKeyNotFound)
 
 	// Test that account was removed
-	_, err = accountCache.GetAccountRoleARN(testAccount3)
+	_, err = accountCache.GetValue(testAccount3)
 	require.NotNil(t, err)
-	require.Equal(t, err, ackrtcache.ErrAccountIDNotFound)
+	require.Equal(t, err, ackrtcache.ErrKeyNotFound)
 
 	// Test with existing account
-	roleARN, err = accountCache.GetAccountRoleARN(testAccount1)
+	roleARN, err = accountCache.GetValue(testAccount1)
 	require.Nil(t, err)
 	require.Equal(t, roleARN, testAccountARN1)
 
-	roleARN, err = accountCache.GetAccountRoleARN(testAccount2)
+	roleARN, err = accountCache.GetValue(testAccount2)
 	require.Nil(t, err)
 	require.Equal(t, roleARN, testAccountARN2)
 
@@ -172,15 +172,15 @@ func TestAccountCache(t *testing.T) {
 	time.Sleep(time.Second)
 
 	// Test that accounts ware removed
-	_, err = accountCache.GetAccountRoleARN(testAccount1)
+	_, err = accountCache.GetValue(testAccount1)
 	require.NotNil(t, err)
 	require.Equal(t, err, ackrtcache.ErrCARMConfigMapNotFound)
 
-	_, err = accountCache.GetAccountRoleARN(testAccount2)
+	_, err = accountCache.GetValue(testAccount2)
 	require.NotNil(t, err)
 	require.Equal(t, err, ackrtcache.ErrCARMConfigMapNotFound)
 
-	_, err = accountCache.GetAccountRoleARN(testAccount3)
+	_, err = accountCache.GetValue(testAccount3)
 	require.NotNil(t, err)
 	require.Equal(t, err, ackrtcache.ErrCARMConfigMapNotFound)
 }
diff --git a/pkg/runtime/cache/cache.go b/pkg/runtime/cache/cache.go
index 06275cf..7dd464e 100644
--- a/pkg/runtime/cache/cache.go
+++ b/pkg/runtime/cache/cache.go
@@ -75,7 +75,10 @@ type Caches struct {
 	stopCh chan struct{}
 
 	// Accounts cache
-	Accounts *AccountCache
+	Accounts *CARMMap
+
+	// CARMMaps v2 cache
+	CARMMaps *CARMMap
 
 	// Namespaces cache
 	Namespaces *NamespaceCache
@@ -84,7 +87,8 @@ type Caches struct {
 // New instantiate a new Caches object.
 func New(log logr.Logger, config Config) Caches {
 	return Caches{
-		Accounts:   NewAccountCache(log),
+		Accounts:   NewCARMMapCache(log),
+		CARMMaps:   NewCARMMapCache(log),
 		Namespaces: NewNamespaceCache(log, config.WatchScope, config.Ignored),
 	}
 }
@@ -93,7 +97,10 @@ func New(log logr.Logger, config Config) Caches {
 func (c Caches) Run(clientSet kubernetes.Interface) {
 	stopCh := make(chan struct{})
 	if c.Accounts != nil {
-		c.Accounts.Run(clientSet, stopCh)
+		c.Accounts.Run(ACKRoleAccountMap, clientSet, stopCh)
+	}
+	if c.CARMMaps != nil {
+		c.CARMMaps.Run(ACKCARMMapV2, clientSet, stopCh)
 	}
 	if c.Namespaces != nil {
 		c.Namespaces.Run(clientSet, stopCh)
diff --git a/pkg/runtime/cache/namespace.go b/pkg/runtime/cache/namespace.go
index ebe04ae..68bbb65 100644
--- a/pkg/runtime/cache/namespace.go
+++ b/pkg/runtime/cache/namespace.go
@@ -32,6 +32,8 @@ type namespaceInfo struct {
 	defaultRegion string
 	// services.k8s.aws/owner-account-id Annotation
 	ownerAccountID string
+	// services.k8s.aws/team-id Annotation
+	teamID string
 	// services.k8s.aws/endpoint-url Annotation
 	endpointURL string
 	// {service}.services.k8s.aws/deletion-policy Annotations (keyed by service)
@@ -54,6 +56,14 @@ func (n *namespaceInfo) getOwnerAccountID() string {
 	return n.ownerAccountID
 }
 
+// getTeamID returns the namespace team-id
+func (n *namespaceInfo) getTeamID() string {
+	if n == nil {
+		return ""
+	}
+	return n.teamID
+}
+
 // getEndpointURL returns the namespace Endpoint URL
 func (n *namespaceInfo) getEndpointURL() string {
 	if n == nil {
@@ -186,6 +196,16 @@ func (c *NamespaceCache) GetOwnerAccountID(namespace string) (string, bool) {
 	return "", false
 }
 
+// GetTeamID returns the team-id if it exists
+func (c *NamespaceCache) GetTeamID(namespace string) (string, bool) {
+	info, ok := c.getNamespaceInfo(namespace)
+	if ok {
+		a := info.getTeamID()
+		return a, a != ""
+	}
+	return "", false
+}
+
 // GetEndpointURL returns the endpoint URL if it exists
 func (c *NamespaceCache) GetEndpointURL(namespace string) (string, bool) {
 	info, ok := c.getNamespaceInfo(namespace)
@@ -229,6 +249,10 @@ func (c *NamespaceCache) setNamespaceInfoFromK8sObject(ns *corev1.Namespace) {
 	if ok {
 		nsInfo.ownerAccountID = OwnerAccountID
 	}
+	TeamID, ok := nsa[ackv1alpha1.AnnotationTeamID]
+	if ok {
+		nsInfo.teamID = TeamID
+	}
 	EndpointURL, ok := nsa[ackv1alpha1.AnnotationEndpointURL]
 	if ok {
 		nsInfo.endpointURL = EndpointURL
diff --git a/pkg/runtime/cache/namespace_test.go b/pkg/runtime/cache/namespace_test.go
index 86fb240..121878d 100644
--- a/pkg/runtime/cache/namespace_test.go
+++ b/pkg/runtime/cache/namespace_test.go
@@ -131,6 +131,101 @@ func TestNamespaceCache(t *testing.T) {
 	require.False(t, ok)
 }
 
+func TestNamespaceCacheWithRoleARN(t *testing.T) {
+	// create a fake k8s client and fake watcher
+	k8sClient := k8sfake.NewSimpleClientset()
+	watcher := watch.NewFake()
+	k8sClient.PrependWatchReactor("production", k8stesting.DefaultWatchReactor(watcher, nil))
+
+	// New logger writing to specific buffer
+	zapOptions := ctrlrtzap.Options{
+		Development: true,
+		Level:       zapcore.InfoLevel,
+	}
+	fakeLogger := ctrlrtzap.New(ctrlrtzap.UseFlagOptions(&zapOptions))
+
+	// initlizing account cache
+	namespaceCache := ackrtcache.NewNamespaceCache(fakeLogger, []string{}, []string{})
+	stopCh := make(chan struct{})
+
+	namespaceCache.Run(k8sClient, stopCh)
+
+	// Test create events
+	_, err := k8sClient.CoreV1().Namespaces().Create(
+		context.Background(),
+		&corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "production",
+				Annotations: map[string]string{
+					ackv1alpha1.AnnotationDefaultRegion: "us-west-2",
+					ackv1alpha1.AnnotationTeamID:        "team-a",
+					ackv1alpha1.AnnotationEndpointURL:   "https://amazon-service.region.amazonaws.com",
+				},
+			},
+		},
+		metav1.CreateOptions{},
+	)
+	require.Nil(t, err)
+
+	time.Sleep(time.Second)
+
+	defaultRegion, ok := namespaceCache.GetDefaultRegion("production")
+	require.True(t, ok)
+	require.Equal(t, "us-west-2", defaultRegion)
+
+	teamID, ok := namespaceCache.GetTeamID("production")
+	require.True(t, ok)
+	require.Equal(t, "team-a", teamID)
+
+	endpointURL, ok := namespaceCache.GetEndpointURL("production")
+	require.True(t, ok)
+	require.Equal(t, "https://amazon-service.region.amazonaws.com", endpointURL)
+
+	// Test update events
+	_, err = k8sClient.CoreV1().Namespaces().Update(
+		context.Background(),
+		&corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "production",
+				Annotations: map[string]string{
+					ackv1alpha1.AnnotationDefaultRegion: "us-est-1",
+					ackv1alpha1.AnnotationTeamID:        "team-b",
+					ackv1alpha1.AnnotationEndpointURL:   "https://amazon-other-service.region.amazonaws.com",
+				},
+			},
+		},
+		metav1.UpdateOptions{},
+	)
+	require.Nil(t, err)
+
+	time.Sleep(time.Second)
+
+	defaultRegion, ok = namespaceCache.GetDefaultRegion("production")
+	require.True(t, ok)
+	require.Equal(t, "us-est-1", defaultRegion)
+
+	teamID, ok = namespaceCache.GetTeamID("production")
+	require.True(t, ok)
+	require.Equal(t, "team-b", teamID)
+
+	endpointURL, ok = namespaceCache.GetEndpointURL("production")
+	require.True(t, ok)
+	require.Equal(t, "https://amazon-other-service.region.amazonaws.com", endpointURL)
+
+	// Test delete events
+	err = k8sClient.CoreV1().Namespaces().Delete(
+		context.Background(),
+		"production",
+		metav1.DeleteOptions{},
+	)
+	require.Nil(t, err)
+
+	time.Sleep(time.Second)
+
+	_, ok = namespaceCache.GetDefaultRegion(testNamespace1)
+	require.False(t, ok)
+}
+
 func TestScopedNamespaceCache(t *testing.T) {
 	defaultConfig := ackrtcache.Config{
 		WatchScope: []string{"watch-scope", "watch-scope-2"},
diff --git a/pkg/runtime/reconciler.go b/pkg/runtime/reconciler.go
index 4b6afaa..e76eaa8 100644
--- a/pkg/runtime/reconciler.go
+++ b/pkg/runtime/reconciler.go
@@ -20,6 +20,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws/arn"
 	backoff "github.com/cenkalti/backoff/v4"
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
@@ -38,6 +39,7 @@ import (
 	ackcondition "github.com/aws-controllers-k8s/runtime/pkg/condition"
 	ackcfg "github.com/aws-controllers-k8s/runtime/pkg/config"
 	ackerr "github.com/aws-controllers-k8s/runtime/pkg/errors"
+	"github.com/aws-controllers-k8s/runtime/pkg/featuregate"
 	ackmetrics "github.com/aws-controllers-k8s/runtime/pkg/metrics"
 	"github.com/aws-controllers-k8s/runtime/pkg/requeue"
 	ackrtcache "github.com/aws-controllers-k8s/runtime/pkg/runtime/cache"
@@ -232,24 +234,41 @@ func (r *resourceReconciler) Reconcile(ctx context.Context, req ctrlrt.Request)
 	// If the ConfigMap is not created, or not populated with an
 	// accountID to roleARN mapping, we need to properly requeue with a
 	// helpful message to the user.
-	var roleARN ackv1alpha1.AWSResourceName
 	acctID, needCARMLookup := r.getOwnerAccountID(desired)
-	if needCARMLookup {
-		// This means that the user is specifying a namespace that is
-		// annotated with an owner account ID. We need to retrieve the
-		// roleARN from the ConfigMap and properly requeue if the roleARN
-		// is not available.
+
+	var roleARN ackv1alpha1.AWSResourceName
+	if r.cfg.FeatureGates.IsEnabled(featuregate.CARMv2) {
+		teamID := r.getTeamID(desired)
+		if teamID != "" {
+			// The user is specifying a namespace that is annotated with a team ID.
+			// Requeue if the corresponding roleARN is not available in the CARMv2 configmap.
+			// Additionally, set the account ID to the role's account ID.
+			roleARN, err = r.getRoleARNv2(string(teamID))
+			if err != nil {
+				return r.handleCacheError(ctx, err, desired)
+			}
+			parsedARN, err := arn.Parse(string(roleARN))
+			if err != nil {
+				return ctrlrt.Result{}, fmt.Errorf("parsing role ARN %q from namespace annotation: %v", roleARN, err)
+			}
+			acctID = ackv1alpha1.AWSAccountID(parsedARN.AccountID)
+		} else if needCARMLookup {
+			// The user is specifying a namespace that is annotated with an owner account ID.
+			// Requeue if the corresponding roleARN is not available in the CARMv2 configmap.
+			roleARN, err = r.getRoleARNv2(string(acctID))
+			if err != nil {
+				return r.handleCacheError(ctx, err, desired)
+			}
+		}
+	} else if needCARMLookup {
+		// The user is specifying a namespace that is annotated with an owner account ID.
+		// Requeue if the corresponding roleARN is not available in the Accounts (CARMv1) configmap.
 		roleARN, err = r.getRoleARN(acctID)
 		if err != nil {
-			// TODO(a-hilaly): Refactor all the reconcile function to make it
-			// easier to understand and maintain.
-			reason := err.Error()
-			latest := desired.DeepCopy()
-			// set ResourceSynced condition to false with proper error message
-			condition.SetSynced(latest, corev1.ConditionFalse, &condition.UnavailableIAMRoleMessage, &reason)
-			return r.HandleReconcileError(ctx, desired, latest, requeue.NeededAfter(err, roleARNNotAvailableRequeueDelay))
+			return r.handleCacheError(ctx, err, desired)
 		}
 	}
+
 	region := r.getRegion(desired)
 	endpointURL := r.getEndpointURL(desired)
 	gvk := r.rd.GroupVersionKind()
@@ -275,6 +294,20 @@ func (r *resourceReconciler) Reconcile(ctx context.Context, req ctrlrt.Request)
 	return r.HandleReconcileError(ctx, desired, latest, err)
 }
 
+func (r *resourceReconciler) handleCacheError(
+	ctx context.Context,
+	err error,
+	desired acktypes.AWSResource,
+) (ctrlrt.Result, error) {
+	// TODO(a-hilaly): Refactor all the reconcile function to make it
+	// easier to understand and maintain.
+	reason := err.Error()
+	latest := desired.DeepCopy()
+	// set ResourceSynced condition to false with proper error message
+	condition.SetSynced(latest, corev1.ConditionFalse, &condition.UnavailableIAMRoleMessage, &reason)
+	return r.HandleReconcileError(ctx, desired, latest, requeue.NeededAfter(err, roleARNNotAvailableRequeueDelay))
+}
+
 // reconcile either cleans up a deleted resource or ensures that the supplied
 // AWSResource's backing API resource matches the supplied desired state.
 //
@@ -1088,14 +1121,41 @@ func (r *resourceReconciler) getOwnerAccountID(
 	return controllerAccountID, false
 }
 
-// getRoleARN return the Role ARN that should be assumed in order to manage
-// the resources.
-func (r *resourceReconciler) getRoleARN(
-	acctID ackv1alpha1.AWSAccountID,
-) (ackv1alpha1.AWSResourceName, error) {
-	roleARN, err := r.cache.Accounts.GetAccountRoleARN(string(acctID))
+// getTeamID gets the team-id from the namespace annotation.
+func (r *resourceReconciler) getTeamID(
+	res acktypes.AWSResource,
+) ackv1alpha1.TeamID {
+	// look for team ID in the namespace annotations
+	namespace := res.MetaObject().GetNamespace()
+	namespacedTeamID, ok := r.cache.Namespaces.GetTeamID(namespace)
+	if ok {
+		return ackv1alpha1.TeamID(namespacedTeamID)
+	}
+	return ackv1alpha1.TeamID("")
+}
+
+// getRoleARNv2 returns the Role ARN that should be assumed for the given account/team ID,
+// from the CARMv2 configmap, in order to manage the resources.
+func (r *resourceReconciler) getRoleARNv2(id string) (ackv1alpha1.AWSResourceName, error) {
+	// use service level roleARN if present
+	serviceID := r.sc.GetMetadata().ServiceAlias + "." + id
+	if roleARN, err := r.cache.CARMMaps.GetValue(serviceID); err == nil {
+		return ackv1alpha1.AWSResourceName(roleARN), nil
+	}
+	// otherwise use account/team level roleARN
+	roleARN, err := r.cache.CARMMaps.GetValue(id)
+	if err != nil {
+		return "", fmt.Errorf("retrieving role ARN for account/team ID %q from %q configmap: %v", id, ackrtcache.ACKCARMMapV2, err)
+	}
+	return ackv1alpha1.AWSResourceName(roleARN), nil
+}
+
+// getRoleARN returns the Role ARN that should be assumed for the given account ID,
+// from the CARMv1 configmap, in order to manage the resources.
+func (r *resourceReconciler) getRoleARN(acctID ackv1alpha1.AWSAccountID) (ackv1alpha1.AWSResourceName, error) {
+	roleARN, err := r.cache.Accounts.GetValue(string(acctID))
 	if err != nil {
-		return "", fmt.Errorf("unable to retrieve role ARN for account %s: %v", acctID, err)
+		return "", fmt.Errorf("retrieving role ARN for account ID %q from %q configMap: %v", acctID, ackrtcache.ACKRoleAccountMap, err)
 	}
 	return ackv1alpha1.AWSResourceName(roleARN), nil
 }