ATLAS-2807: Re-evaluate classification propagation during entity delete

common/src/main/java/org/apache/atlas/repository/Constants.java

@@ -134,6 +134,7 @@ public final class Constants {
     public static final String TEMP_STRUCT_NAME_PREFIX       = "__tempQueryResultStruct";
 
     public static final String CLASSIFICATION_ENTITY_GUID                     = INTERNAL_PROPERTY_KEY_PREFIX + "entityGuid";
+    public static final String CLASSIFICATION_ENTITY_STATUS                   = INTERNAL_PROPERTY_KEY_PREFIX + "entityStatus";
     public static final String CLASSIFICATION_VALIDITY_PERIODS_KEY            = INTERNAL_PROPERTY_KEY_PREFIX + "validityPeriods";
     public static final String CLASSIFICATION_VERTEX_PROPAGATE_KEY            = INTERNAL_PROPERTY_KEY_PREFIX + "propagate";
     public static final String CLASSIFICATION_VERTEX_NAME_KEY                 = TYPE_NAME_PROPERTY_KEY;

intg/src/main/java/org/apache/atlas/AtlasErrorCode.java

@@ -128,7 +128,7 @@ public enum AtlasErrorCode {
     CLASSIFICATION_NOT_ASSOCIATED_WITH_ENTITY(400, "ATLAS-400-00-06D", "Classification {0} is not associated with entity"),
     UNKNOWN_GLOSSARY_TERM(400, "ATLAS-400-00-06E", "{0}: Unknown/invalid glossary term"),
     INVALID_CLASSIFICATION_PARAMS(400, "ATLAS-400-00-06F", "Invalid classification parameters passed for {0} operation for entity: {1}"),
-    PROPAGATED_CLASSIFICATION_NOT_ASSOCIATED_WITH_ENTITY(400, "ATLAS-400-00-070", "Propagated classification {0} is not associated with entity"),
+    PROPAGATED_CLASSIFICATION_NOT_ASSOCIATED_WITH_ENTITY(400, "ATLAS-400-00-070", "Propagated classification {0} is not associated with entity {2}, it is associated with entity {1}"),
     INVALID_BLOCKED_PROPAGATED_CLASSIFICATION(400, "ATLAS-400-00-071", "Invalid propagated classification: {0} with entityGuid: {1} added to blocked propagated classifications."),
     MISSING_MANDATORY_ANCHOR(400, "ATLAS-400-00-072", "Mandatory anchor attribute is missing"),
     MISSING_MANDATORY_QUALIFIED_NAME(400, "ATLAS-400-00-073", "Mandatory qualifiedName attribute is missing"),
@@ -152,6 +152,7 @@ public enum AtlasErrorCode {
     INVALID_TIMEBOUNDRY_START_TIME(400, "ATLAS-400-00-87B", "Invalid startTime {0}"),
     INVALID_TIMEBOUNDRY_END_TIME(400, "ATLAS-400-00-87C", "Invalid endTime {0}"),
     INVALID_TIMEBOUNDRY_DATERANGE(400, "ATLAS-400-00-87D", "Invalid dateRange: startTime {0} must be before endTime {1}"),
+    PROPAGATED_CLASSIFICATION_REMOVAL_NOT_SUPPORTED(400, "ATLAS-400-00-87E", "Removal of classification {0}, which is propagated from entity {1}, is not supported"),
 
     UNAUTHORIZED_ACCESS(403, "ATLAS-403-00-001", "{0} is not authorized to perform {1}"),
 
@@ -174,6 +175,7 @@ public enum AtlasErrorCode {
     INVALID_ENTITY_GUID_FOR_CLASSIFICATION_UPDATE(404, "ATLAS-404-00-010", "Updating entityGuid of classification is not allowed."),
     INSTANCE_GUID_NOT_DATASET(404, "ATLAS-404-00-011", "Given instance guid {0} is not a dataset"),
     INSTANCE_GUID_DELETED(404, "ATLAS-404-00-012", "Given instance guid {0} has been deleted"),
+    NO_PROPAGATED_CLASSIFICATIONS_FOUND_FOR_ENTITY(404, "ATLAS-404-00-013", "No propagated classifications associated with entity: {0}"),
 
     // All data conflict errors go here
     TYPE_ALREADY_EXISTS(409, "ATLAS-409-00-001", "Given type {0} already exists"),

intg/src/main/java/org/apache/atlas/model/instance/AtlasClassification.java

@@ -37,6 +37,7 @@ import javax.xml.bind.annotation.XmlSeeAlso;
 import org.apache.atlas.model.PList;
 import org.apache.atlas.model.SearchFilter.SortType;
 import org.apache.atlas.model.TimeBoundary;
+import org.apache.atlas.model.instance.AtlasEntity.Status;
 
 import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.NONE;
 import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.PUBLIC_ONLY;
@@ -54,6 +55,7 @@ public class AtlasClassification extends AtlasStruct implements Serializable {
     private static final long serialVersionUID = 1L;
 
     private String             entityGuid      = null;
+    private Status             entityStatus    = Status.ACTIVE;
     private Boolean            propagate       = null;
     private List<TimeBoundary> validityPeriods = null;
 
@@ -82,6 +84,7 @@ public class AtlasClassification extends AtlasStruct implements Serializable {
             setTypeName(other.getTypeName());
             setAttributes(other.getAttributes());
             setEntityGuid(other.getEntityGuid());
+            setEntityStatus(other.getEntityStatus());
             setPropagate(other.isPropagate());
             setValidityPeriods(other.getValidityPeriods());
         }
@@ -111,6 +114,14 @@ public class AtlasClassification extends AtlasStruct implements Serializable {
         this.validityPeriods = validityPeriods;
     }
 
+    public Status getEntityStatus() {
+        return entityStatus;
+    }
+
+    public void setEntityStatus(Status entityStatus) {
+        this.entityStatus = entityStatus;
+    }
+
     @JsonIgnore
     public void addValityPeriod(TimeBoundary validityPeriod) {
         List<TimeBoundary> vpList = this.validityPeriods;
@@ -132,12 +143,13 @@ public class AtlasClassification extends AtlasStruct implements Serializable {
         AtlasClassification that = (AtlasClassification) o;
         return Objects.equals(propagate, that.propagate) &&
                Objects.equals(entityGuid, that.entityGuid) &&
+               entityStatus == that.entityStatus &&
                Objects.equals(validityPeriods, that.validityPeriods);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(super.hashCode(), entityGuid, propagate);
+        return Objects.hash(super.hashCode(), entityGuid, entityStatus, propagate);
     }
 
     @Override
@@ -145,6 +157,7 @@ public class AtlasClassification extends AtlasStruct implements Serializable {
         final StringBuilder sb = new StringBuilder("AtlasClassification{");
         super.toString(sb);
         sb.append("entityGuid='").append(entityGuid).append('\'');
+        sb.append(", entityStatus=").append(entityStatus);
         sb.append(", propagate=").append(propagate);
         sb.append(", validityPeriods=").append(validityPeriods);
         sb.append('}');

repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java

@@ -486,6 +486,24 @@ public final class GraphHelper {
         return ret;
     }
 
+    public static List<AtlasEdge> getIncomingClassificationEdges(AtlasVertex classificationVertex) {
+        List<AtlasEdge> ret                = new ArrayList<>();
+        String          classificationName = getTypeName(classificationVertex);
+        Iterable        edges              = classificationVertex.query().direction(AtlasEdgeDirection.IN).label(CLASSIFICATION_LABEL)
+                                                                 .has(CLASSIFICATION_EDGE_NAME_PROPERTY_KEY, classificationName).edges();
+        if (edges != null) {
+            Iterator<AtlasEdge> iterator = edges.iterator();
+
+            while (iterator.hasNext()) {
+                AtlasEdge edge = iterator.next();
+
+                ret.add(edge);
+            }
+        }
+
+        return ret;
+    }
+
     public static List<AtlasVertex> getAllPropagatedEntityVertices(AtlasVertex classificationVertex) {
         List<AtlasVertex> ret = new ArrayList<>();
 
@@ -980,6 +998,10 @@ public final class GraphHelper {
         return getTraitNames(entityVertex, false);
     }
 
+    public static List<String> getPropagatedTraitNames(AtlasVertex entityVertex) {
+        return getTraitNames(entityVertex, true);
+    }
+
     public static List<String> getAllTraitNames(AtlasVertex entityVertex) {
         return getTraitNames(entityVertex, null);
     }
@@ -1198,6 +1220,21 @@ public final class GraphHelper {
         return ret;
     }
 
+    public static boolean isClassificationEdge(AtlasEdge edge) {
+        boolean ret = false;
+
+        if (edge != null) {
+            String  edgeLabel    = edge.getLabel();
+            Boolean isPropagated = edge.getProperty(Constants.CLASSIFICATION_EDGE_IS_PROPAGATED_PROPERTY_KEY, Boolean.class);
+
+            if (edgeLabel != null && isPropagated != null) {
+                ret = edgeLabel.equals(CLASSIFICATION_LABEL) && !isPropagated;
+            }
+        }
+
+        return ret;
+    }
+
     public static List<String> getBlockedClassificationIds(AtlasEdge edge) {
         List<String> ret = null;
 
@@ -1216,6 +1253,12 @@ public final class GraphHelper {
         return (propagateTags == null) ? null : PropagateTags.valueOf(propagateTags);
     }
 
+    public static Status getClassificationEntityStatus(AtlasElement element) {
+        String status = element.getProperty(Constants.CLASSIFICATION_ENTITY_STATUS, String.class);
+
+        return (status == null) ? null : Status.valueOf(status);
+    }
+
     //Added conditions in fetching system attributes to handle test failures in GremlinTest where these properties are not set
     public static String getCreatedByAsString(AtlasElement element){
         return element.getProperty(Constants.CREATED_BY_KEY, String.class);

repository/src/main/java/org/apache/atlas/repository/store/graph/AtlasEntityStore.java

@@ -197,9 +197,11 @@ public interface AtlasEntityStore {
     void addClassification(List<String> guids, AtlasClassification classification) throws AtlasBaseException;
 
     /**
-     * Delete classification(s)
+     * Delete classification
      */
-    void deleteClassifications(String guid, List<String> classificationNames) throws AtlasBaseException;
+    void deleteClassification(String guid, String classificationName) throws AtlasBaseException;
+
+    void deleteClassification(String guid, String classificationName, String associatedEntityGuid) throws AtlasBaseException;
 
     List<AtlasClassification> getClassifications(String guid) throws AtlasBaseException;
 

repository/src/main/java/org/apache/atlas/repository/store/graph/v1/DeleteHandlerV1.java

@@ -59,6 +59,7 @@ import static org.apache.atlas.model.instance.AtlasEntity.Status.ACTIVE;
 import static org.apache.atlas.model.instance.AtlasEntity.Status.DELETED;
 import static org.apache.atlas.model.typedef.AtlasRelationshipDef.PropagateTags.ONE_TO_TWO;
 import static org.apache.atlas.repository.Constants.CLASSIFICATION_EDGE_NAME_PROPERTY_KEY;
+import static org.apache.atlas.repository.Constants.CLASSIFICATION_ENTITY_STATUS;
 import static org.apache.atlas.repository.Constants.CLASSIFICATION_LABEL;
 import static org.apache.atlas.repository.Constants.PROPAGATED_TRAIT_NAMES_PROPERTY_KEY;
 import static org.apache.atlas.repository.Constants.RELATIONSHIP_GUID_PROPERTY_KEY;
@@ -277,7 +278,8 @@ public abstract class DeleteHandlerV1 {
             if (LOG.isDebugEnabled()) {
                 LOG.debug("Processing delete for typeCategory={}, isOwned={}", typeCategory, isOwned);
             }
-            //If the vertex is of type struct/trait, delete the edge and then the reference vertex as the vertex is not shared by any other entities.
+            //If the vertex is of type struct delete the edge and then the reference vertex as the vertex is not shared by any other entities.
+            //If the vertex is of type classification, delete the edge and then the reference vertex only if the vertex is not shared by any other propagated entities.
             //If the vertex is of type class, and its composite attribute, this reference vertex' lifecycle is controlled
             //through this delete, hence delete the edge and the reference vertex.
             AtlasVertex vertexForDelete = edge.getInVertex();
@@ -576,6 +578,37 @@ public abstract class DeleteHandlerV1 {
         }
     }
 
+    public void deletePropagatedClassification(AtlasVertex entityVertex, String classificationName, String associatedEntityGuid) throws AtlasBaseException {
+        AtlasEdge propagatedEdge = getPropagatedClassificationEdge(entityVertex, classificationName, associatedEntityGuid);
+
+        if (propagatedEdge == null) {
+            throw new AtlasBaseException(AtlasErrorCode.PROPAGATED_CLASSIFICATION_NOT_ASSOCIATED_WITH_ENTITY, classificationName, associatedEntityGuid, getGuid(entityVertex));
+        }
+
+        AtlasVertex classificationVertex = propagatedEdge.getInVertex();
+
+        // do not remove propagated classification with ACTIVE associated entity
+        if (getClassificationEntityStatus(classificationVertex) == ACTIVE) {
+            throw new AtlasBaseException(AtlasErrorCode.PROPAGATED_CLASSIFICATION_REMOVAL_NOT_SUPPORTED, classificationName, associatedEntityGuid);
+        }
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Removing propagated classification: [{} - associatedEntityGuid: {}] from: [{}][{}] with edge label: [{}]",
+                       classificationName, associatedEntityGuid, getTypeName(entityVertex), getGuid(entityVertex), CLASSIFICATION_LABEL);
+        }
+
+        AtlasClassification classification = entityRetriever.toAtlasClassification(classificationVertex);
+
+        // delete classification edge
+        deletePropagatedEdge(propagatedEdge);
+
+        // delete classification vertex
+        deleteClassificationVertex(classificationVertex, true);
+
+        // record remove propagation details to send notifications at the end
+        RequestContext.get().recordRemovedPropagation(getGuid(entityVertex), classification);
+    }
+
     public void deletePropagatedEdge(AtlasEdge edge) throws AtlasBaseException {
         String      classificationName = AtlasGraphUtilsV2.getProperty(edge, CLASSIFICATION_EDGE_NAME_PROPERTY_KEY, String.class);
         AtlasVertex entityVertex       = edge.getOutVertex();
@@ -616,6 +649,12 @@ public abstract class DeleteHandlerV1 {
             }
         }
 
+        if (isClassificationEdge(edge)) {
+            AtlasVertex classificationVertex = edge.getInVertex();
+
+            AtlasGraphUtilsV2.setProperty(classificationVertex, CLASSIFICATION_ENTITY_STATUS, DELETED.name());
+        }
+
         deleteEdge(edge, force);
     }
 
@@ -872,12 +911,17 @@ public abstract class DeleteHandlerV1 {
         _deleteVertex(instanceVertex, force);
     }
 
-    protected void deleteClassificationVertex(AtlasVertex classificationVertex, boolean force) {
+    public void deleteClassificationVertex(AtlasVertex classificationVertex, boolean force) {
         if (LOG.isDebugEnabled()) {
             LOG.debug("Deleting classification vertex", string(classificationVertex));
         }
 
-        _deleteVertex(classificationVertex, force);
+        List<AtlasEdge> incomingClassificationEdges = getIncomingClassificationEdges(classificationVertex);
+
+        // delete classification vertex only if it has no more entity references (direct or propagated)
+        if (CollectionUtils.isEmpty(incomingClassificationEdges)) {
+            _deleteVertex(classificationVertex, force);
+        }
     }
 
     private boolean isInternalType(final AtlasVertex instanceVertex) {
@@ -905,13 +949,9 @@ public abstract class DeleteHandlerV1 {
      * @throws AtlasException
      */
     private void deleteAllClassifications(AtlasVertex instanceVertex) throws AtlasBaseException {
-        List<AtlasEdge> classificationEdges = getClassificationEdges(instanceVertex);
+        List<AtlasEdge> classificationEdges = getAllClassificationEdges(instanceVertex);
 
         for (AtlasEdge edge : classificationEdges) {
-            AtlasVertex classificationVertex = edge.getInVertex();
-
-            removeTagPropagation(classificationVertex);
-
             deleteEdgeReference(edge, CLASSIFICATION, false, false, instanceVertex);
         }
     }

repository/src/main/java/org/apache/atlas/repository/store/graph/v1/HardDeleteHandlerV1.java

@@ -52,9 +52,6 @@ public class HardDeleteHandlerV1 extends DeleteHandlerV1 {
             LOG.debug("==> HardDeleteHandlerV1.deleteEdge({}, {})", GraphHelper.string(edge), force);
         }
 
-        // re-evaluate tag propagation
-        removeTagPropagation(edge);
-
         graphHelper.removeEdge(edge);
     }
 }

repository/src/main/java/org/apache/atlas/repository/store/graph/v1/SoftDeleteHandlerV1.java

@@ -69,9 +69,6 @@ public class SoftDeleteHandlerV1 extends DeleteHandlerV1 {
             LOG.debug("==> SoftDeleteHandlerV1.deleteEdge({}, {})",GraphHelper.string(edge), force);
         }
 
-        // re-evaluate tag propagation
-        removeTagPropagation(edge);
-
         if (force) {
             graphHelper.removeEdge(edge);
         } else {

repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java

@@ -539,27 +539,36 @@ public class AtlasEntityStoreV2 implements AtlasEntityStore {
 
     @Override
     @GraphTransaction
-    public void deleteClassifications(final String guid, final List<String> classificationNames) throws AtlasBaseException {
+    public void deleteClassification(final String guid, final String classificationName) throws AtlasBaseException {
+        deleteClassification(guid, classificationName, null);
+    }
+
+    @Override
+    @GraphTransaction
+    public void deleteClassification(final String guid, final String classificationName, final String associatedEntityGuid) throws AtlasBaseException {
         if (StringUtils.isEmpty(guid)) {
             throw new AtlasBaseException(AtlasErrorCode.INVALID_PARAMETERS, "Guid(s) not specified");
         }
-        if (CollectionUtils.isEmpty(classificationNames)) {
-            throw new AtlasBaseException(AtlasErrorCode.INVALID_PARAMETERS, "classifications(s) not specified");
+        if (StringUtils.isEmpty(classificationName)) {
+            throw new AtlasBaseException(AtlasErrorCode.INVALID_PARAMETERS, "classifications not specified");
         }
 
         AtlasEntityHeader entityHeader = entityRetriever.toAtlasEntityHeaderWithClassifications(guid);
 
-        for (String classification : classificationNames) {
-            AtlasAuthorizationUtils.verifyAccess(new AtlasEntityAccessRequest(typeRegistry, AtlasPrivilege.ENTITY_REMOVE_CLASSIFICATION, entityHeader, new AtlasClassification(classification)), "remove classification: guid=", guid, ", classification=", classification);
+        // verify authorization only for removal of directly associated classification and not propagated one.
+        if (StringUtils.isEmpty(associatedEntityGuid) || guid.equals(associatedEntityGuid)) {
+            AtlasAuthorizationUtils.verifyAccess(new AtlasEntityAccessRequest(typeRegistry, AtlasPrivilege.ENTITY_REMOVE_CLASSIFICATION,
+                                                 entityHeader, new AtlasClassification(classificationName)),
+                                                 "remove classification: guid=", guid, ", classification=", classificationName);
         }
 
         if (LOG.isDebugEnabled()) {
-            LOG.debug("Deleting classifications={} from entity={}", classificationNames, guid);
+            LOG.debug("Deleting classification={} from entity={}", classificationName, guid);
         }
 
         GraphTransactionInterceptor.lockObjectAndReleasePostCommit(guid);
 
-        entityGraphMapper.deleteClassifications(guid, classificationNames);
+        entityGraphMapper.deleteClassification(guid, classificationName, associatedEntityGuid);
     }
 
 

repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java

@@ -67,6 +67,7 @@ import java.util.*;
 import java.util.stream.Collectors;
 
 import static org.apache.atlas.model.TypeCategory.CLASSIFICATION;
+import static org.apache.atlas.model.instance.AtlasEntity.Status.ACTIVE;
 import static org.apache.atlas.model.instance.AtlasEntity.Status.DELETED;
 import static org.apache.atlas.model.instance.AtlasRelatedObjectId.KEY_RELATIONSHIP_ATTRIBUTES;
 import static org.apache.atlas.model.instance.EntityMutations.EntityOperation.CREATE;
@@ -83,6 +84,7 @@ import static org.apache.atlas.repository.graph.GraphHelper.getCollectionElement
 import static org.apache.atlas.repository.graph.GraphHelper.getClassificationEdge;
 import static org.apache.atlas.repository.graph.GraphHelper.getClassificationVertex;
 import static org.apache.atlas.repository.graph.GraphHelper.getMapElementsProperty;
+import static org.apache.atlas.repository.graph.GraphHelper.getPropagatedTraitNames;
 import static org.apache.atlas.repository.graph.GraphHelper.getStatus;
 import static org.apache.atlas.repository.graph.GraphHelper.getTraitLabel;
 import static org.apache.atlas.repository.graph.GraphHelper.getTraitNames;
@@ -271,6 +273,7 @@ public class EntityGraphMapper {
 
         AtlasGraphUtilsV2.addProperty(ret, Constants.SUPER_TYPES_PROPERTY_KEY, classificationType.getAllSuperTypes());
         AtlasGraphUtilsV2.setProperty(ret, Constants.CLASSIFICATION_ENTITY_GUID, classification.getEntityGuid());
+        AtlasGraphUtilsV2.setProperty(ret, Constants.CLASSIFICATION_ENTITY_STATUS, classification.getEntityStatus().name());
 
         return ret;
     }
@@ -1312,6 +1315,11 @@ public class EntityGraphMapper {
                     classification.setEntityGuid(guid);
                 }
 
+                // set associated entity status to classification
+                if (classification.getEntityStatus() == null) {
+                    classification.setEntityStatus(ACTIVE);
+                }
+
                 // ignore propagated classifications
 
                 if (LOG.isDebugEnabled()) {
@@ -1390,8 +1398,30 @@ public class EntityGraphMapper {
         }
     }
 
-    public void deleteClassifications(String entityGuid, List<String> classificationNames) throws AtlasBaseException {
-        if (CollectionUtils.isEmpty(classificationNames)) {
+    public void deleteClassification(String entityGuid, String classificationName, String associatedEntityGuid) throws AtlasBaseException {
+        if (StringUtils.isEmpty(associatedEntityGuid) || associatedEntityGuid.equals(entityGuid)) {
+            deleteClassification(entityGuid, classificationName);
+        } else {
+            deletePropagatedClassification(entityGuid, classificationName, associatedEntityGuid);
+        }
+    }
+
+    private void deletePropagatedClassification(String entityGuid, String classificationName, String associatedEntityGuid) throws AtlasBaseException {
+        if (StringUtils.isEmpty(classificationName)) {
+            throw new AtlasBaseException(AtlasErrorCode.INVALID_CLASSIFICATION_PARAMS, "delete", entityGuid);
+        }
+
+        AtlasVertex entityVertex = AtlasGraphUtilsV2.findByGuid(entityGuid);
+
+        if (entityVertex == null) {
+            throw new AtlasBaseException(AtlasErrorCode.INSTANCE_GUID_NOT_FOUND, entityGuid);
+        }
+
+        deleteHandler.deletePropagatedClassification(entityVertex, classificationName, associatedEntityGuid);
+    }
+
+    public void deleteClassification(String entityGuid, String classificationName) throws AtlasBaseException {
+        if (StringUtils.isEmpty(classificationName)) {
             throw new AtlasBaseException(AtlasErrorCode.INVALID_CLASSIFICATION_PARAMS, "delete", entityGuid);
         }
 
@@ -1407,57 +1437,55 @@ public class EntityGraphMapper {
             throw new AtlasBaseException(AtlasErrorCode.NO_CLASSIFICATIONS_FOUND_FOR_ENTITY, entityGuid);
         }
 
-        validateClassificationExists(traitNames, classificationNames);
+        validateClassificationExists(traitNames, classificationName);
 
         Map<AtlasVertex, List<AtlasClassification>> removedClassifications = new HashMap<>();
 
-        for (String classificationName : classificationNames) {
-            AtlasVertex         classificationVertex = getClassificationVertex(entityVertex, classificationName);
-            AtlasClassification classification       = entityRetriever.toAtlasClassification(classificationVertex);
+        AtlasVertex         classificationVertex = getClassificationVertex(entityVertex, classificationName);
+        AtlasClassification classification       = entityRetriever.toAtlasClassification(classificationVertex);
 
-            // remove classification from propagated entities if propagation is turned on
-            if (isPropagationEnabled(classificationVertex)) {
-                List<AtlasVertex> propagatedEntityVertices = deleteHandler.removeTagPropagation(classificationVertex);
+        // remove classification from propagated entities if propagation is turned on
+        if (isPropagationEnabled(classificationVertex)) {
+            List<AtlasVertex> propagatedEntityVertices = deleteHandler.removeTagPropagation(classificationVertex);
 
-                // add propagated entities and deleted classification details to removeClassifications map
-                if (CollectionUtils.isNotEmpty(propagatedEntityVertices)) {
-                    for (AtlasVertex propagatedEntityVertex : propagatedEntityVertices) {
-                        List<AtlasClassification> classifications = removedClassifications.get(propagatedEntityVertex);
+            // add propagated entities and deleted classification details to removeClassifications map
+            if (CollectionUtils.isNotEmpty(propagatedEntityVertices)) {
+                for (AtlasVertex propagatedEntityVertex : propagatedEntityVertices) {
+                    List<AtlasClassification> classifications = removedClassifications.get(propagatedEntityVertex);
 
-                        if (classifications == null) {
-                            classifications = new ArrayList<>();
+                    if (classifications == null) {
+                        classifications = new ArrayList<>();
 
-                            removedClassifications.put(propagatedEntityVertex, classifications);
-                        }
-
-                        classifications.add(classification);
+                        removedClassifications.put(propagatedEntityVertex, classifications);
                     }
+
+                    classifications.add(classification);
                 }
             }
+        }
 
-            // add associated entity and deleted classification details to removeClassifications map
-            List<AtlasClassification> classifications = removedClassifications.get(entityVertex);
+        // add associated entity and deleted classification details to removeClassifications map
+        List<AtlasClassification> classifications = removedClassifications.get(entityVertex);
 
-            if (classifications == null) {
-                classifications = new ArrayList<>();
+        if (classifications == null) {
+            classifications = new ArrayList<>();
 
-                removedClassifications.put(entityVertex, classifications);
-            }
+            removedClassifications.put(entityVertex, classifications);
+        }
 
-            classifications.add(classification);
+        classifications.add(classification);
 
-            // remove classifications from associated entity
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Removing classification: [{}] from: [{}][{}] with edge label: [{}]", classificationName,
-                           getTypeName(entityVertex), entityGuid, CLASSIFICATION_LABEL);
-            }
+        // remove classifications from associated entity
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Removing classification: [{}] from: [{}][{}] with edge label: [{}]", classificationName,
+                    getTypeName(entityVertex), entityGuid, CLASSIFICATION_LABEL);
+        }
 
-            AtlasEdge edge = getClassificationEdge(entityVertex, classificationVertex);
+        AtlasEdge edge = getClassificationEdge(entityVertex, classificationVertex);
 
-            deleteHandler.deleteEdgeReference(edge, CLASSIFICATION, false, true, entityVertex);
+        deleteHandler.deleteEdgeReference(edge, CLASSIFICATION, false, true, entityVertex);
 
-            traitNames.remove(classificationName);
-        }
+        traitNames.remove(classificationName);
 
         updateTraitNamesProperty(entityVertex, traitNames);
 
@@ -1670,7 +1698,9 @@ public class EntityGraphMapper {
         List<String> traitNames = getTraitNames(instanceVertex);
 
         if (CollectionUtils.isNotEmpty(traitNames)) {
-            deleteClassifications(guid, traitNames);
+            for (String traitName : traitNames) {
+                deleteClassification(guid, traitName);
+            }
         }
     }
 
@@ -1693,6 +1723,12 @@ public class EntityGraphMapper {
         }
     }
 
+    private void validateClassificationExists(List<String> existingClassifications, String suppliedClassificationName) throws AtlasBaseException {
+        if (!existingClassifications.contains(suppliedClassificationName)) {
+            throw new AtlasBaseException(AtlasErrorCode.CLASSIFICATION_NOT_ASSOCIATED_WITH_ENTITY, suppliedClassificationName);
+        }
+    }
+
     private AtlasEdge getOrCreateRelationship(AtlasVertex end1Vertex, AtlasVertex end2Vertex, String relationshipName,
                                               Map<String, Object> relationshipAttributes) throws AtlasBaseException {
         return relationshipStore.getOrCreate(end1Vertex, end2Vertex, new AtlasRelationship(relationshipName, relationshipAttributes));

repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphRetriever.java

@@ -100,6 +100,7 @@ import static org.apache.atlas.repository.graph.GraphHelper.getAllClassification
 import static org.apache.atlas.repository.graph.GraphHelper.getAllTraitNames;
 import static org.apache.atlas.repository.graph.GraphHelper.getBlockedClassificationIds;
 import static org.apache.atlas.repository.graph.GraphHelper.getArrayElementsProperty;
+import static org.apache.atlas.repository.graph.GraphHelper.getClassificationEntityStatus;
 import static org.apache.atlas.repository.graph.GraphHelper.getClassificationVertices;
 import static org.apache.atlas.repository.graph.GraphHelper.getGuid;
 import static org.apache.atlas.repository.graph.GraphHelper.getIncomingEdgesByLabel;
@@ -276,6 +277,7 @@ public final class EntityGraphRetriever {
         AtlasClassification ret = new AtlasClassification(getTypeName(classificationVertex));
 
         ret.setEntityGuid(AtlasGraphUtilsV2.getProperty(classificationVertex, CLASSIFICATION_ENTITY_GUID, String.class));
+        ret.setEntityStatus(getClassificationEntityStatus(classificationVertex));
         ret.setPropagate(isPropagationEnabled(classificationVertex));
 
         String strValidityPeriods = AtlasGraphUtilsV2.getProperty(classificationVertex, CLASSIFICATION_VALIDITY_PERIODS_KEY, String.class);

repository/src/test/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2Test.java

@@ -273,7 +273,7 @@ public class AtlasEntityStoreV2Test extends AtlasEntityTestBase {
         AtlasEntityHeader tableDefinition1 = response.getFirstUpdatedEntityByTypeName(TABLE_TYPE);
         AtlasEntity updatedTableDef1 = getEntityFromStore(tableDefinition1);
         validateEntity(entitiesInfo, updatedTableDef1);
-                
+
         Assert.assertTrue(partsMap.get("part0").equals(((Map<String, AtlasStruct>) updatedTableDef1.getAttribute("partitionsMap")).get("part0")));
 
         //update map - add a map key
@@ -905,7 +905,9 @@ public class AtlasEntityStoreV2Test extends AtlasEntityTestBase {
         assertTrue(classificationSet.contains(TAG_NAME));
         assertTrue(classificationSet.contains(TAG_NAME_2));
 
-        entityStore.deleteClassifications(dbEntityGuid, actualClassifications);
+        for (String actualClassificationName : actualClassifications) {
+            entityStore.deleteClassification(dbEntityGuid, actualClassificationName);
+        }
     }
 
     @Test(dependsOnMethods = "testCreate")
@@ -936,7 +938,7 @@ public class AtlasEntityStoreV2Test extends AtlasEntityTestBase {
         assertTrue(actualDBAssociatedEntityGuid.contains(dbEntityGuid));
         assertTrue(actualTblAssociatedEntityGuid.contains(tblEntityGuid));
 
-        entityStore.deleteClassifications(dbEntityGuid, Collections.singletonList(TAG_NAME));
-        entityStore.deleteClassifications(tblEntityGuid, Collections.singletonList(TAG_NAME));
+        entityStore.deleteClassification(dbEntityGuid, TAG_NAME);
+        entityStore.deleteClassification(tblEntityGuid, TAG_NAME);
     }
 }
\ No newline at end of file

repository/src/test/java/org/apache/atlas/repository/tagpropagation/ClassificationPropagationTest.java

@@ -17,6 +17,7 @@
  */
 package org.apache.atlas.repository.tagpropagation;
 
+import com.vividsolutions.jts.util.Assert;
 import org.apache.atlas.RequestContext;
 import org.apache.atlas.TestModules;
 import org.apache.atlas.discovery.AtlasLineageService;
@@ -25,6 +26,7 @@ import org.apache.atlas.model.instance.AtlasClassification;
 import org.apache.atlas.model.instance.AtlasEntity;
 import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
 import org.apache.atlas.model.instance.AtlasRelationship;
+import org.apache.atlas.model.instance.EntityMutationResponse;
 import org.apache.atlas.model.lineage.AtlasLineageInfo;
 import org.apache.atlas.model.lineage.AtlasLineageInfo.LineageRelation;
 import org.apache.atlas.model.typedef.AtlasClassificationDef;
@@ -57,6 +59,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
+import static org.apache.atlas.AtlasErrorCode.PROPAGATED_CLASSIFICATION_REMOVAL_NOT_SUPPORTED;
 import static org.apache.atlas.graph.GraphSandboxUtil.useLocalSolr;
 import static org.apache.atlas.model.lineage.AtlasLineageInfo.LineageDirection;
 import static org.apache.atlas.model.typedef.AtlasRelationshipDef.PropagateTags.BOTH;
@@ -79,6 +82,10 @@ public class ClassificationPropagationTest {
     public static final String EMPLOYEES1_PROCESS      = "EMPLOYEES1_PROCESS";
     public static final String EMPLOYEES2_PROCESS      = "EMPLOYEES2_PROCESS";
     public static final String EMPLOYEES_UNION_PROCESS = "EMPLOYEES_UNION_PROCESS";
+    public static final String IMPORT_FILE             = "tag-propagation-data-1.zip";
+    public static final String EMPLOYEES_TABLE         = "EMPLOYEES_TABLE";
+    public static final String US_EMPLOYEES_TABLE      = "US_EMPLOYEES2_TABLE";
+    public static final String EMPLOYEES_PROCESS       = "EMPLOYEES_PROCESS";
 
     @Inject
     private AtlasTypeDefStore typeDefStore;
@@ -121,13 +128,21 @@ public class ClassificationPropagationTest {
     /** This test uses the lineage graph:
      *
 
-                         [Process1] ----> [Employees1]
-                        /                              \
-                       /                                \
-       [hdfs_employees]                                  [Process3] ----> [ EmployeesUnion ]
-                       \                                /
-                        \                             /
-                         [Process2] ----> [Employees2]
+     Lineage - 1
+     -----------
+     [Process1] ----> [Employees1]
+     /                              \
+     /                                \
+     [hdfs_employees]                                  [Process3] ----> [ EmployeesUnion ]
+     \                                /
+     \                             /
+     [Process2] ----> [Employees2]
+
+
+     Lineage - 2
+     -----------
+
+     [Employees] ----> [Process] ----> [ US_Employees ]
 
      */
 
@@ -443,6 +458,31 @@ public class ClassificationPropagationTest {
         assertTrue(blockedClassifications.isEmpty());
     }
 
+    @Test(dependsOnMethods = {"removeBlockedPropagatedClassifications"})
+    public void addClassification_DeleteCase() throws AtlasBaseException {
+        AtlasEntity         employees = getEntity(EMPLOYEES_TABLE);
+        AtlasClassification tag1      = new AtlasClassification("tag1");
+
+        tag1.setEntityGuid(employees.getGuid());
+        tag1.setPropagate(true);
+
+        addClassification(employees, tag1);
+
+        List<String> propagatedEntities = Arrays.asList(EMPLOYEES_PROCESS, US_EMPLOYEES_TABLE);
+
+        assertClassificationExistInEntities(propagatedEntities, tag1);
+
+        AtlasEntity employees_process = getEntity(EMPLOYEES_PROCESS);
+        AtlasEntity us_employees      = getEntity(US_EMPLOYEES_TABLE);
+
+        deletePropagatedClassificationExpectFail(employees_process, tag1);
+        deletePropagatedClassificationExpectFail(us_employees, tag1);
+
+        deleteEntity(EMPLOYEES_PROCESS);
+        assertClassificationExistInEntity(EMPLOYEES_PROCESS, tag1);
+        assertClassificationExistInEntity(US_EMPLOYEES_TABLE, tag1);
+    }
+
     private void assertClassificationExistInList(Set<AtlasClassification> classifications, AtlasClassification classification) {
         String  classificationName  = classification.getTypeName();
         String  entityGuid          = classification.getEntityGuid();
@@ -521,7 +561,7 @@ public class ClassificationPropagationTest {
 
             loadSampleClassificationDefs();
 
-            runImportWithNoParameters(importService, getZipSource("tag-propagation-data.zip"));
+            runImportWithNoParameters(importService, getZipSource(IMPORT_FILE));
 
             initializeEntitiesMap();
         } catch (AtlasBaseException | IOException e) {
@@ -560,6 +600,10 @@ public class ClassificationPropagationTest {
         entitiesMap.put(EMPLOYEES2_PROCESS, "c0201260-dbeb-45f4-930d-5129eab31dc9");
         entitiesMap.put(EMPLOYEES_UNION_PROCESS, "470a2d1e-b1fd-47de-8f2d-8dfd0a0275a7");
 
+        entitiesMap.put(EMPLOYEES_TABLE, "b4edad46-d00f-4e94-be39-8d2619d17e6c");
+        entitiesMap.put(US_EMPLOYEES_TABLE, "44acef8e-fefe-491c-87d9-e2ea6a9ad3b0");
+        entitiesMap.put(EMPLOYEES_PROCESS, "a1c9a281-d30b-419c-8199-7434b245d7fe");
+
         lineageInfo = lineageService.getAtlasLineageInfo(entitiesMap.get(HDFS_PATH_EMPLOYEES), LineageDirection.BOTH, 3);
     }
 
@@ -570,6 +614,13 @@ public class ClassificationPropagationTest {
         return entityWithExtInfo.getEntity();
     }
 
+    private boolean deleteEntity(String entityName) throws AtlasBaseException {
+        String                 entityGuid = entitiesMap.get(entityName);
+        EntityMutationResponse response   = entityStore.deleteById(entityGuid);
+
+        return CollectionUtils.isNotEmpty(response.getDeletedEntities());
+    }
+
     private AtlasClassification getClassification(AtlasEntity hdfs_employees, AtlasClassification tag2) throws AtlasBaseException {
         return entityStore.getClassification(hdfs_employees.getGuid(), tag2.getTypeName());
     }
@@ -595,7 +646,26 @@ public class ClassificationPropagationTest {
     }
 
     private void deleteClassifications(AtlasEntity entity, List<String> classificationNames) throws AtlasBaseException {
-        entityStore.deleteClassifications(entity.getGuid(), classificationNames);
+        for (String classificationName : classificationNames) {
+            entityStore.deleteClassification(entity.getGuid(), classificationName);
+        }
+    }
+
+    private void deletePropagatedClassification(AtlasEntity entity, AtlasClassification classification) throws AtlasBaseException {
+        deletePropagatedClassification(entity, classification.getTypeName(), classification.getEntityGuid());
+    }
+
+    private void deletePropagatedClassification(AtlasEntity entity, String classificationName, String associatedEntityGuid) throws AtlasBaseException {
+        entityStore.deleteClassification(entity.getGuid(), classificationName, associatedEntityGuid);
+    }
+
+    private void deletePropagatedClassificationExpectFail(AtlasEntity entity, AtlasClassification classification) {
+        try {
+            deletePropagatedClassification(entity, classification);
+            fail();
+        } catch (AtlasBaseException ex) {
+            Assert.equals(ex.getAtlasErrorCode(), PROPAGATED_CLASSIFICATION_REMOVAL_NOT_SUPPORTED);
+        }
     }
 
     private AtlasRelationship getRelationship(String fromEntityName, String toEntityName) throws AtlasBaseException {

webapp/src/main/java/org/apache/atlas/web/resources/EntityResource.java

@@ -1083,7 +1083,7 @@ public class EntityResource {
                 perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "EntityResource.deleteTrait(" + guid + ", " + traitName + ")");
             }
 
-            entitiesStore.deleteClassifications(guid, new ArrayList<String>() {{ add(traitName); }});
+            entitiesStore.deleteClassification(guid, traitName);
 
             Map<String, Object> response = new HashMap<>();
             response.put(AtlasClient.REQUEST_ID, Servlets.getRequestId());

webapp/src/main/java/org/apache/atlas/web/rest/EntityREST.java

@@ -542,7 +542,7 @@ public class EntityREST {
                 throw new AtlasBaseException(AtlasErrorCode.INSTANCE_BY_UNIQUE_ATTRIBUTE_NOT_FOUND, typeName, attributes.toString());
             }
 
-            entitiesStore.deleteClassifications(guid, Collections.singletonList(classificationName));
+            entitiesStore.deleteClassification(guid, classificationName);
         } finally {
             AtlasPerfTracer.log(perf);
         }
@@ -557,15 +557,17 @@ public class EntityREST {
     @Path("/guid/{guid}/classification/{classificationName}")
     @Produces(Servlets.JSON_MEDIA_TYPE)
     public void deleteClassification(@PathParam("guid") String guid,
-                                     @PathParam("classificationName") final String classificationName) throws AtlasBaseException {
+                                     @PathParam("classificationName") final String classificationName,
+                                     @QueryParam("associatedEntityGuid") final String associatedEntityGuid) throws AtlasBaseException {
         Servlets.validateQueryParamLength("guid", guid);
         Servlets.validateQueryParamLength("classificationName", classificationName);
+        Servlets.validateQueryParamLength("associatedEntityGuid", associatedEntityGuid);
 
         AtlasPerfTracer perf = null;
 
         try {
             if (AtlasPerfTracer.isPerfTraceEnabled(PERF_LOG)) {
-                perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "EntityREST.deleteClassification(" + guid + "," + classificationName + ")");
+                perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "EntityREST.deleteClassification(" + guid + "," + classificationName + "," + associatedEntityGuid + ")");
             }
 
             if (StringUtils.isEmpty(guid)) {
@@ -574,7 +576,7 @@ public class EntityREST {
 
             ensureClassificationType(classificationName);
 
-            entitiesStore.deleteClassifications(guid, new ArrayList<String>() {{ add(classificationName);}} );
+            entitiesStore.deleteClassification(guid, classificationName, associatedEntityGuid);
         } finally {
             AtlasPerfTracer.log(perf);
         }

webapp/src/test/java/org/apache/atlas/web/adapters/TestEntityREST.java

@@ -20,6 +20,7 @@ package org.apache.atlas.web.adapters;
 import org.apache.atlas.RequestContext;
 import org.apache.atlas.TestModules;
 import org.apache.atlas.TestUtilsV2;
+import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.model.instance.AtlasClassification;
 import org.apache.atlas.model.instance.AtlasClassification.AtlasClassifications;
 import org.apache.atlas.model.instance.AtlasEntity;
@@ -177,7 +178,7 @@ public class TestEntityREST {
         Assert.assertNotNull(updatedClassification);
         Assert.assertEquals(updatedClassification.getAttribute("tag"), testClassification.getAttribute("tag"));
 
-        entityREST.deleteClassification(dbEntity.getGuid(), TestUtilsV2.PHI);
+        deleteClassification(dbEntity.getGuid(), TestUtilsV2.PHI);
     }
 
     @Test(dependsOnMethods = "testAddAndGetClassification")
@@ -193,7 +194,7 @@ public class TestEntityREST {
     @Test(dependsOnMethods = "testGetEntityWithAssociations")
     public void  testDeleteClassification() throws Exception {
 
-        entityREST.deleteClassification(dbEntity.getGuid(), TestUtilsV2.CLASSIFICATION);
+        deleteClassification(dbEntity.getGuid(), TestUtilsV2.CLASSIFICATION);
         final AtlasClassification.AtlasClassifications retrievedClassifications = entityREST.getClassifications(dbEntity.getGuid());
 
         Assert.assertNotNull(retrievedClassifications);
@@ -367,4 +368,8 @@ public class TestEntityREST {
             put(name, new String[] { value });
         }};
     }
+
+    private void deleteClassification(String guid, String classificationName) throws AtlasBaseException {
+        entityREST.deleteClassification(guid, classificationName, null);
+    }
 }