in repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java [1098:1294]
public void updateClassifications(EntityMutationContext context, String guid, List<AtlasClassification> classifications) throws AtlasBaseException {
if (CollectionUtils.isEmpty(classifications)) {
throw new AtlasBaseException(AtlasErrorCode.INVALID_CLASSIFICATION_PARAMS, "update", guid);
}
AtlasVertex entityVertex = AtlasGraphUtilsV2.findByGuid(this.graph, guid);
if (entityVertex == null) {
throw new AtlasBaseException(AtlasErrorCode.INSTANCE_GUID_NOT_FOUND, guid);
}
AtlasPerfTracer perf = null;
if (AtlasPerfTracer.isPerfTraceEnabled(PERF_LOG)) {
perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "EntityGraphMapper.updateClassifications");
}
String entityTypeName = AtlasGraphUtilsV2.getTypeName(entityVertex);
AtlasEntityType entityType = typeRegistry.getEntityTypeByName(entityTypeName);
List<AtlasClassification> updatedClassifications = new ArrayList<>();
List<AtlasVertex> entitiesToPropagateTo = new ArrayList<>();
Set<AtlasVertex> notificationVertices = new HashSet<>(Collections.singleton(entityVertex));
Map<AtlasVertex, List<AtlasClassification>> addedPropagations = null;
Map<AtlasClassification, List<AtlasVertex>> removedPropagations = new HashMap<>();
for (AtlasClassification classification : classifications) {
String classificationName = classification.getTypeName();
String classificationEntityGuid = classification.getEntityGuid();
if (StringUtils.isEmpty(classificationEntityGuid)) {
classification.setEntityGuid(guid);
}
if (StringUtils.isNotEmpty(classificationEntityGuid) && !StringUtils.equalsIgnoreCase(guid, classificationEntityGuid)) {
throw new AtlasBaseException(AtlasErrorCode.CLASSIFICATION_UPDATE_FROM_PROPAGATED_ENTITY, classificationName);
}
AtlasVertex classificationVertex = getClassificationVertex(entityVertex, classificationName);
if (classificationVertex == null) {
throw new AtlasBaseException(AtlasErrorCode.CLASSIFICATION_NOT_ASSOCIATED_WITH_ENTITY, classificationName);
}
LOG.debug("Updating classification {} for entity {}", classification, guid);
AtlasClassification currentClassification = entityRetriever.toAtlasClassification(classificationVertex);
if (currentClassification == null) {
continue;
}
validateAndNormalizeForUpdate(classification);
boolean isClassificationUpdated = false;
// check for attribute update
Map<String, Object> updatedAttributes = classification.getAttributes();
if (MapUtils.isNotEmpty(updatedAttributes)) {
for (String attributeName : updatedAttributes.keySet()) {
currentClassification.setAttribute(attributeName, updatedAttributes.get(attributeName));
}
isClassificationUpdated = true;
}
// check for validity period update
List<TimeBoundary> currentValidityPeriods = currentClassification.getValidityPeriods();
List<TimeBoundary> updatedValidityPeriods = classification.getValidityPeriods();
if (!Objects.equals(currentValidityPeriods, updatedValidityPeriods)) {
currentClassification.setValidityPeriods(updatedValidityPeriods);
isClassificationUpdated = true;
}
// check for removePropagationsOnEntityDelete update
Boolean currentRemovePropagations = currentClassification.getRemovePropagationsOnEntityDelete();
Boolean updatedRemovePropagations = classification.getRemovePropagationsOnEntityDelete();
if (updatedRemovePropagations != null && (updatedRemovePropagations != currentRemovePropagations)) {
AtlasGraphUtilsV2.setEncodedProperty(classificationVertex, CLASSIFICATION_VERTEX_REMOVE_PROPAGATIONS_KEY, updatedRemovePropagations);
isClassificationUpdated = true;
}
if (isClassificationUpdated) {
List<AtlasVertex> propagatedEntityVertices = graphHelper.getAllPropagatedEntityVertices(classificationVertex);
notificationVertices.addAll(propagatedEntityVertices);
}
if (LOG.isDebugEnabled()) {
LOG.debug("updating vertex {} for trait {}", string(classificationVertex), classificationName);
}
mapClassification(EntityOperation.UPDATE, context, classification, entityType, entityVertex, classificationVertex);
updateModificationMetadata(entityVertex);
// handle update of 'propagate' flag
Boolean currentTagPropagation = currentClassification.isPropagate();
Boolean updatedTagPropagation = classification.isPropagate();
/* -----------------------------
| Current Tag | Updated Tag |
| Propagation | Propagation |
|-------------|-------------|
| true | true | => no-op
|-------------|-------------|
| false | false | => no-op
|-------------|-------------|
| false | true | => Add Tag Propagation (send ADD classification notifications)
|-------------|-------------|
| true | false | => Remove Tag Propagation (send REMOVE classification notifications)
|-------------|-------------| */
if (updatedTagPropagation != null && taskManagement != null && deferredActionEnabled) {
String propagationType = updatedTagPropagation ? CLASSIFICATION_PROPAGATION_ADD : CLASSIFICATION_PROPAGATION_DELETE;
createAndQueueTask(propagationType, entityVertex, classificationVertex.getIdForDisplay(), classificationName);
updatedTagPropagation = null;
}
// compute propagatedEntityVertices once and use it for subsequent iterations and notifications
if (updatedTagPropagation != null && currentTagPropagation != updatedTagPropagation) {
if (updatedTagPropagation) {
if (CollectionUtils.isEmpty(entitiesToPropagateTo)) {
entitiesToPropagateTo = entityRetriever.getImpactedVerticesV2(entityVertex, null, classificationVertex.getIdForDisplay());
}
if (CollectionUtils.isNotEmpty(entitiesToPropagateTo)) {
if (addedPropagations == null) {
addedPropagations = new HashMap<>(entitiesToPropagateTo.size());
for (AtlasVertex entityToPropagateTo : entitiesToPropagateTo) {
addedPropagations.put(entityToPropagateTo, new ArrayList<>());
}
}
List<AtlasVertex> entitiesPropagatedTo = deleteDelegate.getHandler().addTagPropagation(classificationVertex, entitiesToPropagateTo);
if (entitiesPropagatedTo != null) {
for (AtlasVertex entityPropagatedTo : entitiesPropagatedTo) {
addedPropagations.get(entityPropagatedTo).add(classification);
}
}
}
} else {
List<AtlasVertex> impactedVertices = deleteDelegate.getHandler().removeTagPropagation(classificationVertex);
if (CollectionUtils.isNotEmpty(impactedVertices)) {
/*
removedPropagations is a HashMap of entity against list of classifications i.e. for each entity 1 entry in the map.
Maintaining classification wise entity list lets us send the audit request in bulk,
since 1 classification is applied to many entities (including the child entities).
Eg. If a classification is being propagated to 1000 entities, its edge count would be 2000, as per removedPropagations map
we would have 2000 entries and value would always be 1 classification wrapped in a list.
By this rearrangement we maintain an entity list against each classification, as of now its entry size would be 1 (as per request from UI)
instead of 2000. Moreover this allows us to send audit request classification wise instead of separate requests for each entities.
This reduces audit calls from 2000 to 1.
*/
removedPropagations.put(classification, impactedVertices);
}
}
}
updatedClassifications.add(currentClassification);
}
if (CollectionUtils.isNotEmpty(entitiesToPropagateTo)) {
notificationVertices.addAll(entitiesToPropagateTo);
}
for (AtlasVertex vertex : notificationVertices) {
String entityGuid = graphHelper.getGuid(vertex);
AtlasEntity entity = instanceConverter.getAndCacheEntity(entityGuid, ENTITY_CHANGE_NOTIFY_IGNORE_RELATIONSHIP_ATTRIBUTES);
if (isActive(entity)) {
vertex.setProperty(CLASSIFICATION_TEXT_KEY, fullTextMapperV2.getClassificationTextForEntity(entity));
entityChangeNotifier.onClassificationUpdatedToEntity(entity, updatedClassifications);
}
}
if (MapUtils.isNotEmpty(removedPropagations)) {
for (AtlasClassification classification : removedPropagations.keySet()) {
List<AtlasVertex> propagatedVertices = removedPropagations.get(classification);
List<AtlasEntity> propagatedEntities = updateClassificationText(classification, propagatedVertices);
//Sending audit request for all entities at once
entityChangeNotifier.onClassificationsDeletedFromEntities(propagatedEntities, Collections.singletonList(classification));
}
}
AtlasPerfTracer.log(perf);
}