diff --git a/.github/workflows/chart-release-dispatcher.yaml b/.github/workflows/chart-release-dispatcher.yaml
index 10fc6cbdb0..ecc4348768 100644
--- a/.github/workflows/chart-release-dispatcher.yaml
+++ b/.github/workflows/chart-release-dispatcher.yaml
@@ -29,7 +29,7 @@ jobs:
       # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it
       - uses: actions/checkout@v3
         with:
-          token: ${{ secrets.my_pat }}
+          token: ${{ secrets.ORG_PAT_GITHUB }}
           ref: ${{ steps.extract_branch.outputs.branch }}
           fetch-depth: 0
       
@@ -50,10 +50,10 @@ jobs:
       - name: Get PR url and PR User
         id: get_pr_url_user
         run: |
-          head_sha=$(curl -s -H "Authorization: Bearer ${{ secrets.my_pat }}" -H "Accept: application/vnd.github.v3+json" "https://api.github.com/repos/${{ github.repository }}/actions/runs/${{ github.event.workflow_run.id }}/jobs" | jq -r '.jobs[0].head_sha')
+          head_sha=$(curl -s -H "Authorization: Bearer ${{ secrets.ORG_PAT_GITHUB }}" -H "Accept: application/vnd.github.v3+json" "https://api.github.com/repos/${{ github.repository }}/actions/runs/${{ github.event.workflow_run.id }}/jobs" | jq -r '.jobs[0].head_sha')
           echo "Head SHA: $head_sha"
-          pr_url=$(curl -s -H "Authorization: Bearer ${{ secrets.my_pat }}" -H "Accept: application/vnd.github.v3+json" "https://api.github.com/search/issues?q=sha:$head_sha+type:pr" | jq -r '.items[0].html_url')
-          pr_user=$(curl -s -H "Authorization: Bearer ${{ secrets.my_pat }}" -H "Accept: application/vnd.github.v3+json" "https://api.github.com/search/issues?q=sha:$head_sha+type:pr" | jq -r '.items[0].user.login')
+          pr_url=$(curl -s -H "Authorization: Bearer ${{ secrets.ORG_PAT_GITHUB }}" -H "Accept: application/vnd.github.v3+json" "https://api.github.com/search/issues?q=sha:$head_sha+type:pr" | jq -r '.items[0].html_url')
+          pr_user=$(curl -s -H "Authorization: Bearer ${{ secrets.ORG_PAT_GITHUB }}" -H "Accept: application/vnd.github.v3+json" "https://api.github.com/search/issues?q=sha:$head_sha+type:pr" | jq -r '.items[0].user.login')
           echo "pr_url=$pr_url" >> $GITHUB_OUTPUT
           echo "pr_user=$pr_user" >> $GITHUB_OUTPUT
 
@@ -65,7 +65,7 @@ jobs:
       - name: Repository Dispatch
         uses: peter-evans/repository-dispatch@v2
         with:
-          token: ${{ secrets.my_pat }}
+          token: ${{ secrets.ORG_PAT_GITHUB }}
           repository: ${{ matrix.repo }}
           event-type: dispatch_chart_release_workflow
           client-payload: |-
diff --git a/.github/workflows/github-actions-pr-jira.yaml b/.github/workflows/github-actions-pr-jira.yaml
new file mode 100644
index 0000000000..76cd01ab38
--- /dev/null
+++ b/.github/workflows/github-actions-pr-jira.yaml
@@ -0,0 +1,14 @@
+name: GitHub-Jira Link Action
+run-name: ${{ github.actor }} is ensuring Jira ID is present in PR title
+on:
+  pull_request:
+    types: [opened, edited, synchronize, reopened]
+    branches: [main, staging, master, beta, develop, prod, development]
+
+jobs:
+  Enforce-GitHub-Jira-Link-Action:
+    runs-on: ubuntu-latest
+    if: ${{ !contains(fromJson('["main", "staging", "master", "beta", "develop", "prod", "development"]'), github.event.pull_request.head.ref) }}
+    steps:
+      - name: Enforce Pull Request Title includes Jira Issue Key
+        uses: ryanvade/enforce-pr-title-style-action@v2.1.1
\ No newline at end of file
diff --git a/.github/workflows/main-ecr.yml b/.github/workflows/main-ecr.yml
index acb8883f8d..2a64a38c27 100644
--- a/.github/workflows/main-ecr.yml
+++ b/.github/workflows/main-ecr.yml
@@ -196,4 +196,4 @@ jobs:
           ${{ steps.login-ecr.outputs.registry }}/atlanhq/${{ github.event.repository.name }}:${{ steps.get_branch.outputs.branch }}-${{ steps.semver_tag.outputs.new_tag }}
         build-args: |
           ACCESS_TOKEN_USR=$GITHUB_ACTOR
-          ACCESS_TOKEN_PWD=${{ secrets.my_pat }}  
\ No newline at end of file
+          ACCESS_TOKEN_PWD=${{ secrets.ORG_PAT_GITHUB }}  
\ No newline at end of file
diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml
index 0977cb36a2..f8a09b5589 100644
--- a/.github/workflows/maven.yml
+++ b/.github/workflows/maven.yml
@@ -58,7 +58,7 @@ jobs:
             [{
                 "id": "github",
                 "username": "atlan-ci",
-                "password": "${{ secrets.my_pat }}"
+                "password": "${{ secrets.ORG_PAT_GITHUB }}"
             }]
 
       - name: Build with Maven
@@ -77,7 +77,7 @@ jobs:
         shell: bash
 
       - name: Get version tag
-        run: echo "##[set-output name=version;]$(echo `git ls-remote https://${{ secrets.my_pat }}@github.com/atlanhq/${REPOSITORY_NAME}.git ${{ steps.get_branch.outputs.branch }} | awk '{ print $1}' | cut -c1-7`)abcd"
+        run: echo "##[set-output name=version;]$(echo `git ls-remote https://${{ secrets.ORG_PAT_GITHUB }}@github.com/atlanhq/${REPOSITORY_NAME}.git ${{ steps.get_branch.outputs.branch }} | awk '{ print $1}' | cut -c1-7`)abcd"
         id: get_version
 
       - name: Set up Buildx 
@@ -89,7 +89,7 @@ jobs:
         with:
           registry: ghcr.io
           username: $GITHUB_ACTOR
-          password: ${{ secrets.my_pat }}
+          password: ${{ secrets.ORG_PAT_GITHUB }}
 
       - name: Build and push
         id: docker_build
diff --git a/.github/workflows/trivy-docker-scan.yml b/.github/workflows/trivy-docker-scan.yml
index 6be78e7552..f910348903 100644
--- a/.github/workflows/trivy-docker-scan.yml
+++ b/.github/workflows/trivy-docker-scan.yml
@@ -29,7 +29,7 @@ jobs:
           output: 'trivy-results-docker.sarif'
           exit-code: '1'
           #ignore-unfixed: true
-          severity: 'CRITICAL,HIGH,MEDIUM'
+          severity: 'CRITICAL,HIGH'
 
       - name: Upload Trivy Docker Scan Results To GitHub Security tab
         uses: github/codeql-action/upload-sarif@v2
diff --git a/common/src/main/java/org/apache/atlas/repository/Constants.java b/common/src/main/java/org/apache/atlas/repository/Constants.java
index accea8ec88..a5529cf397 100644
--- a/common/src/main/java/org/apache/atlas/repository/Constants.java
+++ b/common/src/main/java/org/apache/atlas/repository/Constants.java
@@ -366,6 +366,10 @@ public final class Constants {
     public static final String IMPALA_SOURCE      = "impala";
     public static final String STORM_SOURCE       = "storm";
     public static final String FILE_SPOOL_SOURCE  = "file_spool";
+    public static final String ASSET_POLICY_GUIDS  = "assetPolicyGUIDs";
+    public static final String ASSET_POLICIES_COUNT  = "assetPoliciesCount";
+
+
 
     /*
      * All supported file-format extensions for Bulk Imports through file upload
diff --git a/intg/src/main/java/org/apache/atlas/model/instance/LinkBusinessPolicyRequest.java b/intg/src/main/java/org/apache/atlas/model/instance/LinkBusinessPolicyRequest.java
new file mode 100644
index 0000000000..e42fd4ad55
--- /dev/null
+++ b/intg/src/main/java/org/apache/atlas/model/instance/LinkBusinessPolicyRequest.java
@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.atlas.model.instance;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+import java.io.Serializable;
+import java.util.List;
+import java.util.Set;
+
+import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.NONE;
+import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.PUBLIC_ONLY;
+
+/**
+ * Request to link/unlink policies from asset.
+ */
+@JsonAutoDetect(getterVisibility = PUBLIC_ONLY, setterVisibility = PUBLIC_ONLY, fieldVisibility = NONE)
+@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@XmlRootElement
+@XmlAccessorType(XmlAccessType.PROPERTY)
+public class LinkBusinessPolicyRequest implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    private Set<String> linkGuids;
+    private Set<String> unlinkGuids;
+
+    public Set<String> getLinkGuids() {
+        return linkGuids;
+    }
+
+    public void setLinkGuids(Set<String> linkGuids) {
+        this.linkGuids = linkGuids;
+    }
+
+    public Set<String> getUnlinkGuids() {
+        return unlinkGuids;
+    }
+
+    public void setUnlinkGuids(Set<String> unlinkGuids) {
+        this.unlinkGuids = unlinkGuids;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder("LinkBusinessPolicyRequest{");
+        sb.append("linkGuids=").append(linkGuids);
+        sb.append(", unlinkGuids=").append(unlinkGuids);
+        sb.append('}');
+        return sb.toString();
+    }
+}
diff --git a/repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java b/repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java
index 179d915df9..d5926e8a00 100755
--- a/repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java
+++ b/repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java
@@ -540,7 +540,8 @@ public static List<String> getPropagatedVerticesIds (AtlasVertex classificationV
     }
 
     public static boolean hasEntityReferences(AtlasVertex classificationVertex) {
-        return classificationVertex.hasEdges(AtlasEdgeDirection.IN, CLASSIFICATION_LABEL);
+        Iterator edgeIterator = classificationVertex.query().direction(AtlasEdgeDirection.IN).label(CLASSIFICATION_LABEL).edges(1).iterator();
+        return edgeIterator != null && edgeIterator.hasNext();
     }
 
     public static List<AtlasVertex> getAllPropagatedEntityVertices(AtlasVertex classificationVertex) {
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/aliasstore/ESAliasStore.java b/repository/src/main/java/org/apache/atlas/repository/store/aliasstore/ESAliasStore.java
index ddd57e2664..2d272cb8fc 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/aliasstore/ESAliasStore.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/aliasstore/ESAliasStore.java
@@ -65,6 +65,7 @@
 @Component
 public class ESAliasStore implements IndexAliasStore {
     private static final Logger LOG = LoggerFactory.getLogger(ESAliasStore.class);
+    public static final String NEW_WILDCARD_DOMAIN_SUPER = "default/domain/*/super";
 
     private final AtlasGraph graph;
     private final EntityGraphRetriever entityRetriever;
@@ -214,7 +215,11 @@ private void personaPolicyToESDslClauses(List<AtlasEntity> policies,
                 } else if (getPolicyActions(policy).contains(ACCESS_READ_PERSONA_DOMAIN)) {
 
                     for (String asset : assets) {
-                        terms.add(asset);
+                        if(!isAllDomain(asset)) {
+                            terms.add(asset);
+                        } else {
+                            asset = NEW_WILDCARD_DOMAIN_SUPER;
+                        }
                         allowClauseList.add(mapOf("wildcard", mapOf(QUALIFIED_NAME, asset + "*")));
                     }
 
@@ -246,6 +251,9 @@ private void personaPolicyToESDslClauses(List<AtlasEntity> policies,
         allowClauseList.add(mapOf("terms", mapOf(QUALIFIED_NAME, terms)));
     }
 
+    private boolean isAllDomain(String asset) {
+        return asset.equals("*/super") || asset.equals("*") || asset.equals(NEW_WILDCARD_DOMAIN_SUPER);
+    }
     private Map<String, Object> esClausesToFilter(List<Map<String, Object>> allowClauseList) {
         if (CollectionUtils.isNotEmpty(allowClauseList)) {
             return mapOf("bool", mapOf("should", allowClauseList));
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/AtlasEntityStore.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/AtlasEntityStore.java
index 912799cdd6..24ca97d731 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/AtlasEntityStore.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/AtlasEntityStore.java
@@ -365,4 +365,10 @@ EntityMutationResponse deleteByUniqueAttributes(List<AtlasObjectId> objectIds)
 
     void repairAccesscontrolAlias(String guid) throws AtlasBaseException;
 
+
+    void linkBusinessPolicy(String policyId, Set<String> linkGuids) throws AtlasBaseException;
+
+
+    void unlinkBusinessPolicy(String policyId, Set<String> unlinkGuids) throws AtlasBaseException;
+
 }
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
index 83df0fe2ef..a5b168f1e7 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
@@ -20,16 +20,12 @@
 
 
 import com.google.common.annotations.VisibleForTesting;
-import org.apache.atlas.AtlasErrorCode;
-import org.apache.atlas.DeleteType;
-import org.apache.atlas.GraphTransactionInterceptor;
-import org.apache.atlas.RequestContext;
-import org.apache.atlas.AtlasException;
-import org.apache.atlas.AtlasConfiguration;
+import org.apache.atlas.*;
 import org.apache.atlas.annotation.GraphTransaction;
 import org.apache.atlas.authorize.*;
 import org.apache.atlas.authorize.AtlasEntityAccessRequest.AtlasEntityAccessRequestBuilder;
-import org.apache.atlas.authorize.AtlasPrivilege;
+import org.apache.atlas.bulkimport.BulkImportResponse;
+import org.apache.atlas.bulkimport.BulkImportResponse.ImportInfo;
 import org.apache.atlas.discovery.EntityDiscoveryService;
 import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.featureflag.FeatureFlagStore;
@@ -55,40 +51,31 @@
 import org.apache.atlas.repository.store.graph.EntityGraphDiscovery;
 import org.apache.atlas.repository.store.graph.EntityGraphDiscoveryContext;
 import org.apache.atlas.repository.store.graph.v1.DeleteHandlerDelegate;
-import org.apache.atlas.repository.store.graph.v2.AtlasEntityComparator.AtlasEntityDiffResult;
 import org.apache.atlas.repository.store.graph.v1.RestoreHandlerV1;
+import org.apache.atlas.repository.store.graph.v2.AtlasEntityComparator.AtlasEntityDiffResult;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.AuthPolicyPreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.ConnectionPreProcessor;
-import org.apache.atlas.repository.store.graph.v2.preprocessor.accesscontrol.StakeholderPreProcessor;
-import org.apache.atlas.repository.store.graph.v2.preprocessor.contract.ContractPreProcessor;
-import org.apache.atlas.repository.store.graph.v2.preprocessor.datamesh.StakeholderTitlePreProcessor;
-import org.apache.atlas.repository.store.graph.v2.preprocessor.resource.LinkPreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.PreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.accesscontrol.PersonaPreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.accesscontrol.PurposePreProcessor;
-import org.apache.atlas.repository.store.graph.v2.preprocessor.datamesh.DataProductPreProcessor;
+import org.apache.atlas.repository.store.graph.v2.preprocessor.accesscontrol.StakeholderPreProcessor;
+import org.apache.atlas.repository.store.graph.v2.preprocessor.contract.ContractPreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.datamesh.DataDomainPreProcessor;
+import org.apache.atlas.repository.store.graph.v2.preprocessor.datamesh.DataProductPreProcessor;
+import org.apache.atlas.repository.store.graph.v2.preprocessor.datamesh.StakeholderTitlePreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.glossary.CategoryPreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.glossary.GlossaryPreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.glossary.TermPreProcessor;
+import org.apache.atlas.repository.store.graph.v2.preprocessor.resource.LinkPreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.resource.ReadmePreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.sql.QueryCollectionPreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.sql.QueryFolderPreProcessor;
 import org.apache.atlas.repository.store.graph.v2.preprocessor.sql.QueryPreProcessor;
 import org.apache.atlas.repository.store.graph.v2.tasks.MeaningsTask;
 import org.apache.atlas.tasks.TaskManagement;
-import org.apache.atlas.type.AtlasArrayType;
-import org.apache.atlas.type.AtlasBusinessMetadataType;
+import org.apache.atlas.type.*;
 import org.apache.atlas.type.AtlasBusinessMetadataType.AtlasBusinessAttribute;
-import org.apache.atlas.type.AtlasClassificationType;
-import org.apache.atlas.type.AtlasEntityType;
-import org.apache.atlas.type.AtlasEnumType;
 import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
-import org.apache.atlas.type.AtlasType;
-import org.apache.atlas.type.AtlasTypeRegistry;
-import org.apache.atlas.type.AtlasTypeUtil;
-import org.apache.atlas.bulkimport.BulkImportResponse;
-import org.apache.atlas.bulkimport.BulkImportResponse.ImportInfo;
 import org.apache.atlas.util.FileUtils;
 import org.apache.atlas.utils.AtlasEntityUtil;
 import org.apache.atlas.utils.AtlasPerfMetrics;
@@ -111,18 +98,15 @@
 import static org.apache.atlas.bulkimport.BulkImportResponse.ImportStatus.FAILED;
 import static org.apache.atlas.model.instance.AtlasEntity.Status.ACTIVE;
 import static org.apache.atlas.model.instance.EntityMutations.EntityOperation.*;
+import static org.apache.atlas.repository.Constants.IS_INCOMPLETE_PROPERTY_KEY;
+import static org.apache.atlas.repository.Constants.STATE_PROPERTY_KEY;
 import static org.apache.atlas.repository.Constants.*;
 import static org.apache.atlas.repository.graph.GraphHelper.*;
-import static org.apache.atlas.repository.graph.GraphHelper.getStatus;
 import static org.apache.atlas.repository.store.graph.v2.EntityGraphMapper.validateLabels;
-import static org.apache.atlas.repository.store.graph.v2.tasks.MeaningsTaskFactory.*;
+import static org.apache.atlas.repository.store.graph.v2.tasks.MeaningsTaskFactory.UPDATE_ENTITY_MEANINGS_ON_TERM_HARD_DELETE;
+import static org.apache.atlas.repository.store.graph.v2.tasks.MeaningsTaskFactory.UPDATE_ENTITY_MEANINGS_ON_TERM_SOFT_DELETE;
 import static org.apache.atlas.repository.util.AccessControlUtils.REL_ATTR_POLICIES;
-import static org.apache.atlas.type.Constants.HAS_LINEAGE;
-import static org.apache.atlas.type.Constants.HAS_LINEAGE_VALID;
-import static org.apache.atlas.type.Constants.MEANINGS_TEXT_PROPERTY_KEY;
-import static org.apache.atlas.type.Constants.MEANINGS_PROPERTY_KEY;
-import static org.apache.atlas.type.Constants.MEANING_NAMES_PROPERTY_KEY;
-import static org.apache.atlas.type.Constants.PENDING_TASKS_PROPERTY_KEY;
+import static org.apache.atlas.type.Constants.*;
 
 
 
@@ -151,10 +135,12 @@ public class AtlasEntityStoreV2 implements AtlasEntityStore {
 
     private final ESAliasStore esAliasStore;
 
+    private final IAtlasMinimalChangeNotifier atlasAlternateChangeNotifier;
     @Inject
     public AtlasEntityStoreV2(AtlasGraph graph, DeleteHandlerDelegate deleteDelegate, RestoreHandlerV1 restoreHandlerV1, AtlasTypeRegistry typeRegistry,
                               IAtlasEntityChangeNotifier entityChangeNotifier, EntityGraphMapper entityGraphMapper, TaskManagement taskManagement,
-                              AtlasRelationshipStore atlasRelationshipStore, FeatureFlagStore featureFlagStore) {
+                              AtlasRelationshipStore atlasRelationshipStore, FeatureFlagStore featureFlagStore,
+                              IAtlasMinimalChangeNotifier atlasAlternateChangeNotifier) {
         this.graph                = graph;
         this.deleteDelegate       = deleteDelegate;
         this.restoreHandlerV1     = restoreHandlerV1;
@@ -168,7 +154,7 @@ public AtlasEntityStoreV2(AtlasGraph graph, DeleteHandlerDelegate deleteDelegate
         this.atlasRelationshipStore = atlasRelationshipStore;
         this.featureFlagStore = featureFlagStore;
         this.esAliasStore = new ESAliasStore(graph, entityRetriever);
-
+        this.atlasAlternateChangeNotifier = atlasAlternateChangeNotifier;
         try {
             this.discovery = new EntityDiscoveryService(typeRegistry, graph, null, null, null, null);
         } catch (AtlasException e) {
@@ -2737,6 +2723,54 @@ public void repairAccesscontrolAlias(String guid) throws AtlasBaseException {
 
         RequestContext.get().endMetricRecord(metric);
     }
+
+    @Override
+    @GraphTransaction
+    public void linkBusinessPolicy(String policyGuid, Set<String> linkGuids) throws AtlasBaseException {
+        AtlasPerfMetrics.MetricRecorder metric = RequestContext.get().startMetricRecord("linkBusinessPolicy.GraphTransaction");
+
+        try {
+            List<AtlasVertex> vertices = this.entityGraphMapper.linkBusinessPolicy(policyGuid, linkGuids);
+            if (CollectionUtils.isEmpty(vertices)) {
+                return;
+            }
+
+            handleBusinessPolicyMutation(vertices);
+        } catch (Exception e) {
+            LOG.error("Error during linkBusinessPolicy for policyGuid: {}", policyGuid, e);
+            throw e;
+        } finally {
+            RequestContext.get().endMetricRecord(metric);
+        }
+    }
+
+    @Override
+    @GraphTransaction
+    public void unlinkBusinessPolicy(String policyGuid, Set<String> unlinkGuids) throws AtlasBaseException {
+        AtlasPerfMetrics.MetricRecorder metric = RequestContext.get().startMetricRecord("unlinkBusinessPolicy.GraphTransaction");
+        try {
+            List<AtlasVertex> vertices = this.entityGraphMapper.unlinkBusinessPolicy(policyGuid, unlinkGuids);
+            if (CollectionUtils.isEmpty(vertices)) {
+                return;
+            }
+
+            handleBusinessPolicyMutation(vertices);
+        } catch (Exception e) {
+            LOG.error("Error during unlinkBusinessPolicy for policyGuid: {}", policyGuid, e);
+            throw e;
+        } finally {
+            RequestContext.get().endMetricRecord(metric);
+        }
+    }
+
+    private void handleBusinessPolicyMutation(List<AtlasVertex> vertices) throws AtlasBaseException {
+        AtlasPerfMetrics.MetricRecorder metricRecorder = RequestContext.get().startMetricRecord("handleBusinessPolicyMutation");
+        this.atlasAlternateChangeNotifier.onEntitiesMutation(vertices);
+        RequestContext.get().endMetricRecord(metricRecorder);
+    }
+
+
 }
 
 
+
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/BusinessPolicyNotifierImpl.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/BusinessPolicyNotifierImpl.java
new file mode 100644
index 0000000000..fab059cfbe
--- /dev/null
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/BusinessPolicyNotifierImpl.java
@@ -0,0 +1,64 @@
+package org.apache.atlas.repository.store.graph.v2;
+
+import org.apache.atlas.RequestContext;
+import org.apache.atlas.exception.AtlasBaseException;
+import org.apache.atlas.listener.EntityChangeListenerV2;
+import org.apache.atlas.model.instance.AtlasEntity;
+import org.apache.atlas.repository.graphdb.AtlasVertex;
+import org.apache.atlas.utils.AtlasPerfMetrics;
+import org.springframework.stereotype.Component;
+
+import javax.inject.Inject;
+import java.util.*;
+
+import static org.apache.atlas.repository.Constants.*;
+import static org.apache.atlas.repository.graph.GraphHelper.*;
+
+
+@Component
+public class BusinessPolicyNotifierImpl implements IAtlasMinimalChangeNotifier {
+
+    private final Set<EntityChangeListenerV2> entityChangeListenersV2;
+
+    @Inject
+    public BusinessPolicyNotifierImpl(Set<EntityChangeListenerV2> entityChangeListenersV2) {
+        this.entityChangeListenersV2 = entityChangeListenersV2;
+
+    }
+
+    @Override
+    public void onEntitiesMutation(final List<AtlasVertex> vertices) throws AtlasBaseException {
+        AtlasPerfMetrics.MetricRecorder metricRecorder = RequestContext.get().startMetricRecord("onEntitiesMutation");
+        final List<AtlasEntity> entities = new ArrayList<>(0);
+        vertices.forEach(item -> entities.add(createAtlasEntity(item)));
+        for (EntityChangeListenerV2 listener : entityChangeListenersV2) {
+            listener.onEntitiesUpdated(entities, false);
+        }
+
+        RequestContext.get().endMetricRecord(metricRecorder);
+    }
+
+    private AtlasEntity createAtlasEntity(AtlasVertex vertex) {
+        AtlasEntity atlasEntity = new AtlasEntity();
+        atlasEntity.setAttribute(QUALIFIED_NAME, vertex.getProperty(QUALIFIED_NAME, String.class));
+        atlasEntity.setAttribute(NAME, vertex.getProperty(NAME, String.class));
+
+        atlasEntity.setGuid(vertex.getProperty(GUID_PROPERTY_KEY, String.class));
+        atlasEntity.setTypeName(vertex.getProperty(TYPE_NAME_PROPERTY_KEY, String.class));
+        atlasEntity.setCreatedBy(vertex.getProperty(CREATED_BY_KEY, String.class));
+        atlasEntity.setUpdatedBy(vertex.getProperty(MODIFIED_BY_KEY, String.class));
+        atlasEntity.setCreateTime(new Date(vertex.getProperty(TIMESTAMP_PROPERTY_KEY, Long.class)));
+        atlasEntity.setUpdateTime(new Date(vertex.getProperty(MODIFICATION_TIMESTAMP_PROPERTY_KEY, Long.class)));
+        atlasEntity.setIsProxy(vertex.getProperty(IS_PROXY_KEY, Boolean.class));
+        atlasEntity.setIsIncomplete(vertex.getProperty(IS_INCOMPLETE_PROPERTY_KEY, Boolean.class));
+        atlasEntity.setStatus(getStatus(vertex));
+        atlasEntity.setProvenanceType(getProvenanceType(vertex));
+        atlasEntity.setHomeId(getHomeId(vertex));
+        atlasEntity.setVersion(getVersion(vertex));
+
+
+        return atlasEntity;
+    }
+
+
+}
\ No newline at end of file
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java
index 596420696d..f52ef2f56d 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java
@@ -3590,6 +3590,7 @@ public void updateClassifications(EntityMutationContext context, String guid, Li
         if (CollectionUtils.isEmpty(classifications)) {
             throw new AtlasBaseException(AtlasErrorCode.INVALID_CLASSIFICATION_PARAMS, "update", guid);
         }
+        entityRetriever.verifyClassificationsPropagationMode(classifications);
 
         AtlasVertex entityVertex = AtlasGraphUtilsV2.findByGuid(this.graph, guid);
 
@@ -3711,7 +3712,21 @@ public void updateClassifications(EntityMutationContext context, String guid, Li
             Boolean updatedRestrictPropagationThroughLineage = classification.getRestrictPropagationThroughLineage();
             Boolean currentRestrictPropagationThroughHierarchy = currentClassification.getRestrictPropagationThroughHierarchy();
             Boolean updatedRestrictPropagationThroughHierarchy = classification.getRestrictPropagationThroughHierarchy();
-            String propagationMode = entityRetriever.determinePropagationMode(updatedRestrictPropagationThroughLineage, updatedRestrictPropagationThroughHierarchy);
+            if (updatedRestrictPropagationThroughLineage == null) {
+                updatedRestrictPropagationThroughLineage = currentRestrictPropagationThroughLineage;
+                classification.setRestrictPropagationThroughLineage(updatedRestrictPropagationThroughLineage);
+            }
+            if (updatedRestrictPropagationThroughHierarchy == null) {
+                updatedRestrictPropagationThroughHierarchy = currentRestrictPropagationThroughHierarchy;
+                classification.setRestrictPropagationThroughHierarchy(updatedRestrictPropagationThroughHierarchy);
+            }
+
+            String propagationMode = CLASSIFICATION_PROPAGATION_MODE_DEFAULT;
+            if (updatedTagPropagation) {
+                // determinePropagationMode also validates the propagation restriction option values
+                propagationMode = entityRetriever.determinePropagationMode(updatedRestrictPropagationThroughLineage, updatedRestrictPropagationThroughHierarchy);
+            }
+
             if ((!Objects.equals(updatedRemovePropagations, currentRemovePropagations) ||
                     !Objects.equals(currentTagPropagation, updatedTagPropagation) ||
                     !Objects.equals(currentRestrictPropagationThroughLineage, updatedRestrictPropagationThroughLineage)) &&
@@ -3733,7 +3748,6 @@ public void updateClassifications(EntityMutationContext context, String guid, Li
                 if (updatedTagPropagation) {
                     if (updatedRestrictPropagationThroughLineage != null && !currentRestrictPropagationThroughLineage && updatedRestrictPropagationThroughLineage) {
                         deleteDelegate.getHandler().removeTagPropagation(classificationVertex);
-
                     }
                     if (updatedRestrictPropagationThroughHierarchy != null && !currentRestrictPropagationThroughHierarchy && updatedRestrictPropagationThroughHierarchy) {
                         deleteDelegate.getHandler().removeTagPropagation(classificationVertex);
@@ -4567,4 +4581,51 @@ public void addHasLineage(Set<AtlasEdge> inputOutputEdges, boolean isRestoreEnti
         RequestContext.get().endMetricRecord(metricRecorder);
     }
 
-}
\ No newline at end of file
+
+    public List<AtlasVertex> linkBusinessPolicy(String policyId, Set<String> linkGuids) {
+        return linkGuids.stream().map(guid -> findByGuid(graph, guid)).filter(Objects::nonNull).filter(ev -> {
+            Set<String> existingValues = ev.getMultiValuedSetProperty(ASSET_POLICY_GUIDS, String.class);
+            return !existingValues.contains(policyId);
+        }).peek(ev -> {
+            Set<String> existingValues = ev.getMultiValuedSetProperty(ASSET_POLICY_GUIDS, String.class);
+            existingValues.add(policyId);
+            ev.setProperty(ASSET_POLICY_GUIDS, policyId);
+            ev.setProperty(ASSET_POLICIES_COUNT, existingValues.size());
+
+            updateModificationMetadata(ev);
+
+            cacheDifferentialEntity(ev, existingValues);
+        }).collect(Collectors.toList());
+    }
+
+
+    public List<AtlasVertex> unlinkBusinessPolicy(String policyId, Set<String> unlinkGuids) {
+        return unlinkGuids.stream().map(guid -> AtlasGraphUtilsV2.findByGuid(graph, guid)).filter(Objects::nonNull).filter(ev -> {
+            Set<String> existingValues = ev.getMultiValuedSetProperty(ASSET_POLICY_GUIDS, String.class);
+            return existingValues.contains(policyId);
+        }).peek(ev -> {
+            Set<String> existingValues = ev.getMultiValuedSetProperty(ASSET_POLICY_GUIDS, String.class);
+            existingValues.remove(policyId);
+            ev.removePropertyValue(ASSET_POLICY_GUIDS, policyId);
+            ev.setProperty(ASSET_POLICIES_COUNT, existingValues.size());
+
+            updateModificationMetadata(ev);
+
+            cacheDifferentialEntity(ev, existingValues);
+        }).collect(Collectors.toList());
+    }
+
+
+    private void cacheDifferentialEntity(AtlasVertex ev, Set<String> existingValues) {
+        AtlasEntity diffEntity = new AtlasEntity(ev.getProperty(TYPE_NAME_PROPERTY_KEY, String.class));
+        diffEntity.setGuid(ev.getProperty(GUID_PROPERTY_KEY, String.class));
+        diffEntity.setAttribute(ASSET_POLICY_GUIDS, existingValues);
+        diffEntity.setAttribute(ASSET_POLICIES_COUNT, existingValues.size());
+        diffEntity.setUpdatedBy(ev.getProperty(MODIFIED_BY_KEY, String.class));
+        diffEntity.setUpdateTime(new Date(RequestContext.get().getRequestTime()));
+
+        RequestContext requestContext = RequestContext.get();
+        requestContext.cacheDifferentialEntity(diffEntity);
+    }
+
+}
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/IAtlasMinimalChangeNotifier.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/IAtlasMinimalChangeNotifier.java
new file mode 100644
index 0000000000..35c2d9e757
--- /dev/null
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/IAtlasMinimalChangeNotifier.java
@@ -0,0 +1,27 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.atlas.repository.store.graph.v2;
+
+import org.apache.atlas.exception.AtlasBaseException;
+import org.apache.atlas.repository.graphdb.AtlasVertex;
+
+import java.util.List;
+
+public interface IAtlasMinimalChangeNotifier {
+    void onEntitiesMutation(final List<AtlasVertex> vertices) throws AtlasBaseException;
+}
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/bulkimport/MigrationImport.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/bulkimport/MigrationImport.java
index 9edbfc1cc3..84e7948790 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/bulkimport/MigrationImport.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/bulkimport/MigrationImport.java
@@ -129,7 +129,7 @@ private AtlasEntityStoreV2 createEntityStore(AtlasGraph graph, AtlasTypeRegistry
                 graph, relationshipStore, entityChangeNotifier, getInstanceConverter(graph), fullTextMapperV2, null, null);
         AtlasRelationshipStoreV2 atlasRelationshipStoreV2 = new AtlasRelationshipStoreV2(graph, typeRegistry, deleteDelegate, entityChangeNotifier);
 
-        return new AtlasEntityStoreV2(graph, deleteDelegate, restoreHandlerV1, typeRegistry, entityChangeNotifier, entityGraphMapper, null, atlasRelationshipStoreV2, null);
+        return new AtlasEntityStoreV2(graph, deleteDelegate, restoreHandlerV1, typeRegistry, entityChangeNotifier, entityGraphMapper, null, atlasRelationshipStoreV2, null, null);
     }
 
     private void shutdownEntityCreationManager(EntityCreationManager creationManager) {
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/AuthPolicyPreProcessor.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/AuthPolicyPreProcessor.java
index 62adf8119a..58fb516564 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/AuthPolicyPreProcessor.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/AuthPolicyPreProcessor.java
@@ -42,10 +42,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
 
 import static org.apache.atlas.AtlasErrorCode.BAD_REQUEST;
@@ -67,6 +64,7 @@
 
 public class AuthPolicyPreProcessor implements PreProcessor {
     private static final Logger LOG = LoggerFactory.getLogger(AuthPolicyPreProcessor.class);
+    public static final String ENTITY_DEFAULT_DOMAIN_SUPER = "entity:default/domain/*/super";
 
     private final AtlasGraph graph;
     private final AtlasTypeRegistry typeRegistry;
@@ -127,6 +125,8 @@ private void processCreatePolicy(AtlasStruct entity) throws AtlasBaseException {
             if (!POLICY_SUB_CATEGORY_DOMAIN.equals(policySubCategory)) {
                 validator.validate(policy, null, parentEntity, CREATE);
                 validateConnectionAdmin(policy);
+            } else {
+                validateAndReduce(policy);
             }
 
             policy.setAttribute(QUALIFIED_NAME, String.format("%s/%s", getEntityQualifiedName(parentEntity), getUUID()));
@@ -165,6 +165,21 @@ private void processCreatePolicy(AtlasStruct entity) throws AtlasBaseException {
         RequestContext.get().endMetricRecord(metricRecorder);
     }
 
+
+    private void validateAndReduce(AtlasEntity policy) {
+        List<String> resources = (List<String>) policy.getAttribute(ATTR_POLICY_RESOURCES);
+        boolean hasAllDomainPattern = resources.stream().anyMatch(resource ->
+                resource.equals("entity:*") ||
+                        resource.equals("entity:*/super") ||
+                        resource.equals(ENTITY_DEFAULT_DOMAIN_SUPER)
+        );
+
+        if (hasAllDomainPattern) {
+            policy.setAttribute(ATTR_POLICY_RESOURCES, Collections.singletonList(ENTITY_DEFAULT_DOMAIN_SUPER));
+        }
+    }
+
+
     private void processUpdatePolicy(AtlasStruct entity, AtlasVertex vertex) throws AtlasBaseException {
         AtlasPerfMetrics.MetricRecorder metricRecorder = RequestContext.get().startMetricRecord("processUpdatePolicy");
         AtlasEntity policy = (AtlasEntity) entity;
@@ -182,6 +197,8 @@ private void processUpdatePolicy(AtlasStruct entity, AtlasVertex vertex) throws
             if (!POLICY_SUB_CATEGORY_DOMAIN.equals(policySubCategory)) {
                 validator.validate(policy, existingPolicy, parentEntity, UPDATE);
                 validateConnectionAdmin(policy);
+            } else {
+                validateAndReduce(policy);
             }
 
             String qName = getEntityQualifiedName(existingPolicy);
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/accesscontrol/StakeholderPreProcessor.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/accesscontrol/StakeholderPreProcessor.java
index 1adbb8ec47..1cba29f935 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/accesscontrol/StakeholderPreProcessor.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/accesscontrol/StakeholderPreProcessor.java
@@ -58,8 +58,7 @@
 import static org.apache.atlas.repository.Constants.STAKEHOLDER_ENTITY_TYPE;
 import static org.apache.atlas.repository.Constants.STAKEHOLDER_TITLE_ENTITY_TYPE;
 import static org.apache.atlas.repository.store.graph.v2.preprocessor.PreProcessorUtils.indexSearchPaginated;
-import static org.apache.atlas.repository.store.graph.v2.preprocessor.datamesh.StakeholderTitlePreProcessor.ATTR_DOMAIN_QUALIFIED_NAMES;
-import static org.apache.atlas.repository.store.graph.v2.preprocessor.datamesh.StakeholderTitlePreProcessor.STAR;
+import static org.apache.atlas.repository.store.graph.v2.preprocessor.datamesh.StakeholderTitlePreProcessor.*;
 import static org.apache.atlas.repository.util.AccessControlUtils.ATTR_ACCESS_CONTROL_ENABLED;
 import static org.apache.atlas.repository.util.AccessControlUtils.ATTR_PERSONA_ROLE_ID;
 import static org.apache.atlas.repository.util.AccessControlUtils.REL_ATTR_POLICIES;
@@ -290,7 +289,7 @@ protected void ensureTitleAvailableForDomain(String domainQualifiedName, String
 
             List<String> domainQualifiedNames = (List<String>) stakeholderTitleHeader.getAttribute(ATTR_DOMAIN_QUALIFIED_NAMES);
 
-            if (!domainQualifiedNames.contains(STAR)) {
+            if (!domainQualifiedNames.contains(STAR) && !domainQualifiedNames.contains(NEW_STAR)) {
                 Optional parentDomain = domainQualifiedNames.stream().filter(x -> domainQualifiedName.startsWith(x)).findFirst();
 
                 if (!parentDomain.isPresent()) {
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/contract/DataContract.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/contract/DataContract.java
index 4ceea2853c..4dce70108a 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/contract/DataContract.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/contract/DataContract.java
@@ -37,10 +37,9 @@ public class DataContract {
 
     @Valid @NotNull
     public String                               kind;
-    public Status                               status;
-    @JsonProperty(value = "template_version", defaultValue = "0.0.1")
-    public String                               templateVersion;
-    @Valid @NotNull
+    public Status                               status = Status.DRAFT;
+    @JsonProperty(value = "template_version")
+    public String                               templateVersion = "0.0.1";
     public String                               data_source;
     @Valid @NotNull
     public String                               dataset;
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/datamesh/StakeholderTitlePreProcessor.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/datamesh/StakeholderTitlePreProcessor.java
index 97685d1ed6..1de5fe685e 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/datamesh/StakeholderTitlePreProcessor.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/preprocessor/datamesh/StakeholderTitlePreProcessor.java
@@ -48,6 +48,7 @@ public class StakeholderTitlePreProcessor implements PreProcessor {
 
 
     public static final String STAR = "*/super";
+    public static final String NEW_STAR = "default/domain/*/super";
     public static final String ATTR_DOMAIN_QUALIFIED_NAMES = "stakeholderTitleDomainQualifiedNames";
 
     public static final String REL_ATTR_STAKEHOLDERS = "stakeholders";
@@ -114,13 +115,13 @@ private void processCreateStakeholderTitle(AtlasEntity entity) throws AtlasBaseE
             if (CollectionUtils.isEmpty(domainQualifiedNames)) {
                 throw new AtlasBaseException(BAD_REQUEST, "Please provide attribute " + ATTR_DOMAIN_QUALIFIED_NAMES);
             }
-
-            if (domainQualifiedNames.contains(STAR)) {
+            if (domainQualifiedNames.contains(NEW_STAR) || domainQualifiedNames.contains(STAR)) {
                 if (domainQualifiedNames.size() > 1) {
-
                     domainQualifiedNames.clear();
-                    domainQualifiedNames.add(STAR);
+                    domainQualifiedNames.add(NEW_STAR);
                     entity.setAttribute(ATTR_DOMAIN_QUALIFIED_NAMES, domainQualifiedNames);
+                }else {
+                    domainQualifiedNames.replaceAll(s -> s.equals(STAR) ? NEW_STAR : s);
                 }
 
                 String qualifiedName = format(PATTERN_QUALIFIED_NAME_ALL_DOMAINS, getUUID());
@@ -211,8 +212,8 @@ private void authorizeDomainAccess(List<String> domainQualifiedNames) throws Atl
         for (String domainQualifiedName: domainQualifiedNames) {
             String domainQualifiedNameToAuth;
 
-            if (domainQualifiedNames.contains(STAR)) {
-                domainQualifiedNameToAuth = "*/super";
+            if (domainQualifiedNames.contains(STAR) || domainQualifiedNames.contains(NEW_STAR)) {
+                domainQualifiedNameToAuth = NEW_STAR;
             } else {
                 domainQualifiedNameToAuth = domainQualifiedName;
             }
diff --git a/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationSender.java b/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationSender.java
index c039cc837a..ece56f294e 100644
--- a/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationSender.java
+++ b/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationSender.java
@@ -18,6 +18,7 @@
 package org.apache.atlas.notification;
 
 import org.apache.atlas.GraphTransactionInterceptor;
+import org.apache.atlas.RequestContext;
 import org.apache.atlas.model.notification.EntityNotification;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.configuration.Configuration;
@@ -95,10 +96,8 @@ public void send(EntityNotification.EntityNotificationV2.OperationType operation
                 notificationHook = new PostCommitNotificationHook(operationType, notifications);
                 postCommitNotificationHooks.set(notificationHook);
             } else {
-                if (isRelationshipEvent(operationType))
-                    notificationHook.addRelationshipNotifications(notifications);
-                else
-                    notificationHook.addNotifications(notifications);
+                if (isRelationshipEvent(operationType)) notificationHook.addRelationshipNotifications(notifications);
+                else notificationHook.addNotifications(notifications);
             }
         }
 
diff --git a/webapp/src/main/java/org/apache/atlas/web/rest/BusinessPolicyREST.java b/webapp/src/main/java/org/apache/atlas/web/rest/BusinessPolicyREST.java
new file mode 100644
index 0000000000..ef252293a0
--- /dev/null
+++ b/webapp/src/main/java/org/apache/atlas/web/rest/BusinessPolicyREST.java
@@ -0,0 +1,113 @@
+package org.apache.atlas.web.rest;
+
+import org.apache.atlas.AtlasErrorCode;
+import org.apache.atlas.RequestContext;
+import org.apache.atlas.annotation.Timed;
+import org.apache.atlas.exception.AtlasBaseException;
+import org.apache.atlas.model.instance.LinkBusinessPolicyRequest;
+import org.apache.atlas.repository.store.graph.AtlasEntityStore;
+import org.apache.atlas.utils.AtlasPerfMetrics;
+import org.apache.atlas.utils.AtlasPerfTracer;
+import org.apache.atlas.web.util.Servlets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import javax.ws.rs.*;
+import javax.ws.rs.core.MediaType;
+
+import static org.apache.atlas.repository.util.AccessControlUtils.ARGO_SERVICE_USER_NAME;
+
+@Path("business-policy")
+@Singleton
+@Service
+@Consumes({Servlets.JSON_MEDIA_TYPE, MediaType.APPLICATION_JSON})
+@Produces({Servlets.JSON_MEDIA_TYPE, MediaType.APPLICATION_JSON})
+public class BusinessPolicyREST {
+
+    private static final Logger LOG = LoggerFactory.getLogger(BusinessPolicyREST.class);
+    private static final Logger PERF_LOG = AtlasPerfTracer.getPerfLogger("rest.BusinessPolicyREST");
+
+    private final AtlasEntityStore entitiesStore;
+
+    @Inject
+    public BusinessPolicyREST(AtlasEntityStore entitiesStore) {
+        this.entitiesStore = entitiesStore;
+    }
+
+    /**
+     * Links a business policy to entities.
+     *
+     * @param policyGuid the ID of the policy to be linked
+     * @param request    the request containing the GUIDs of the assets to link the policy to
+     * @throws AtlasBaseException if there is an error during the linking process
+     */
+    @POST
+    @Path("/{policyId}/link-business-policy")
+    @Timed
+    public void linkBusinessPolicy(@PathParam("policyId") final String policyGuid, final LinkBusinessPolicyRequest request) throws AtlasBaseException {
+        AtlasPerfMetrics.MetricRecorder metric = RequestContext.get().startMetricRecord("linkBusinessPolicy");
+        // Ensure the current user is authorized to link policies
+        if (!ARGO_SERVICE_USER_NAME.equals(RequestContext.getCurrentUser())) {
+            throw new AtlasBaseException(AtlasErrorCode.UNAUTHORIZED_ACCESS, RequestContext.getCurrentUser(), "Policy linking");
+        }
+
+        // Set request context parameters
+        RequestContext.get().setIncludeClassifications(false);
+        RequestContext.get().setIncludeMeanings(false);
+
+        AtlasPerfTracer perf = null;
+        try {
+            // Start performance tracing if enabled
+            if (AtlasPerfTracer.isPerfTraceEnabled(PERF_LOG)) {
+                perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "BusinessPolicyREST.linkBusinessPolicy(" + policyGuid + ")");
+            }
+
+            // Link the business policy to the specified entities
+            entitiesStore.linkBusinessPolicy(policyGuid, request.getLinkGuids());
+        } finally {
+            // Log performance metrics
+            AtlasPerfTracer.log(perf);
+            RequestContext.get().endMetricRecord(metric);
+        }
+    }
+
+    /**
+     * Unlinks a business policy from entities.
+     *
+     * @param policyGuid the ID of the policy to be unlinked
+     * @param request    the request containing the GUIDs of the assets to unlink the policy from
+     * @throws AtlasBaseException if there is an error during the unlinking process
+     */
+    @POST
+    @Path("/{policyId}/unlink-business-policy")
+    @Timed
+    public void unlinkBusinessPolicy(@PathParam("policyId") final String policyGuid, final LinkBusinessPolicyRequest request) throws AtlasBaseException {
+        AtlasPerfMetrics.MetricRecorder metric = RequestContext.get().startMetricRecord("unlinkBusinessPolicy");
+        // Ensure the current user is authorized to unlink policies
+        if (!ARGO_SERVICE_USER_NAME.equals(RequestContext.getCurrentUser())) {
+            throw new AtlasBaseException(AtlasErrorCode.UNAUTHORIZED_ACCESS, RequestContext.getCurrentUser(), "Policy unlinking");
+        }
+
+        // Set request context parameters
+        RequestContext.get().setIncludeClassifications(false);
+        RequestContext.get().setIncludeMeanings(false);
+
+        AtlasPerfTracer perf = null;
+        try {
+            // Start performance tracing if enabled
+            if (AtlasPerfTracer.isPerfTraceEnabled(PERF_LOG)) {
+                perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "BusinessPolicyREST.unlinkBusinessPolicy(" + policyGuid + ")");
+            }
+
+            // Unlink the business policy from the specified entities
+            entitiesStore.unlinkBusinessPolicy(policyGuid, request.getUnlinkGuids());
+        } finally {
+            // Log performance metrics
+            AtlasPerfTracer.log(perf);
+            RequestContext.get().endMetricRecord(metric);
+        }
+    }
+}