chaitalicod commented on code in PR #729:
URL: https://github.com/apache/atlas/pull/729#discussion_r3849819858
##########
repository/src/main/java/org/apache/atlas/repository/store/graph/v1/DeleteHandlerV1.java:
##########
@@ -923,59 +924,106 @@ public void
removeFromPropagatedClassificationNames(AtlasVertex entityVertex, St
}
public void updateTagPropagations(AtlasEdge edge, AtlasRelationship
relationship) throws AtlasBaseException {
- PropagateTags oldTagPropagation = getPropagateTags(edge);
- PropagateTags newTagPropagation = relationship.getPropagateTags();
+ updateTagPropagations(edge, relationship, false, null, null);
+ }
- if (newTagPropagation != oldTagPropagation) {
- List<AtlasVertex> currentClassificationVertices
= getPropagatableClassifications(edge);
- Map<AtlasVertex, List<AtlasVertex>> currentClassificationsMap
=
entityRetriever.getClassificationPropagatedEntitiesMapping(currentClassificationVertices);
+ public void updateTagPropagations(AtlasEdge edge, AtlasRelationship
relationship, boolean isAsyncExecution, String oldTagPropStr,
java.util.List<String> oldBlockedList) throws AtlasBaseException {
+ // isAsyncExecution=false: relationship update — write edge, queue
background task (or update entities inline if tasks disabled).
+ // isAsyncExecution=true: background task — edge already updated;
update entity tags using old state from task-params (oldTagPropStr /
oldBlockedList).
+ boolean isLegacyTask = isAsyncExecution && oldTagPropStr == null;
- // Update propagation edge
- AtlasGraphUtilsV2.setEncodedProperty(edge,
RELATIONSHIPTYPE_TAG_PROPAGATION_KEY, newTagPropagation.name());
+ // Step 1: Capture the edge's propagation settings before the update
(propagateTags and blocked IDs).
+ // Read from the graph on a relationship update, or from task params
on a background task.
+ PropagateTags oldTagPropagation = isAsyncExecution && !isLegacyTask ?
PropagateTags.valueOf(oldTagPropStr) : getPropagateTags(edge);
+ List<String> oldBlocked = isAsyncExecution && !isLegacyTask ?
oldBlockedList : getBlockedClassificationIds(edge);
- List<AtlasVertex> updatedClassificationVertices
= getPropagatableClassifications(edge);
- List<AtlasVertex> classificationVerticesUnion
= (List<AtlasVertex>) CollectionUtils.union(currentClassificationVertices,
updatedClassificationVertices);
- Map<AtlasVertex, List<AtlasVertex>> updatedClassificationsMap
=
entityRetriever.getClassificationPropagatedEntitiesMapping(classificationVerticesUnion);
+ if (oldBlocked == null) {
+ oldBlocked = Collections.emptyList();
+ }
- // compute add/remove propagations list
- Map<AtlasVertex, List<AtlasVertex>> addPropagationsMap = new
HashMap<>();
- Map<AtlasVertex, List<AtlasVertex>> removePropagationsMap = new
HashMap<>();
+ PropagateTags newTagPropagation = relationship.getPropagateTags();
+ boolean propagationChanged = oldTagPropagation !=
newTagPropagation;
- if (MapUtils.isEmpty(currentClassificationsMap) &&
MapUtils.isNotEmpty(updatedClassificationsMap)) {
- addPropagationsMap.putAll(updatedClassificationsMap);
- } else if (MapUtils.isNotEmpty(currentClassificationsMap) &&
MapUtils.isEmpty(updatedClassificationsMap)) {
- removePropagationsMap.putAll(currentClassificationsMap);
- } else {
- for (AtlasVertex classificationVertex :
updatedClassificationsMap.keySet()) {
- List<AtlasVertex> currentPropagatingEntities =
currentClassificationsMap.getOrDefault(classificationVertex,
Collections.emptyList());
- List<AtlasVertex> updatedPropagatingEntities =
updatedClassificationsMap.getOrDefault(classificationVertex,
Collections.emptyList());
- List<AtlasVertex> entitiesAdded =
(List<AtlasVertex>) CollectionUtils.subtract(updatedPropagatingEntities,
currentPropagatingEntities);
- List<AtlasVertex> entitiesRemoved =
(List<AtlasVertex>) CollectionUtils.subtract(currentPropagatingEntities,
updatedPropagatingEntities);
-
- if (CollectionUtils.isNotEmpty(entitiesAdded)) {
- addPropagationsMap.put(classificationVertex,
entitiesAdded);
- }
+ // Step 2: When the client sends blockedPropagatedClassifications;
validate those tags against
+ // propagatable classifications, compare to the pre-update blocked
list, and set blockedChanged plus the
+ // vertex lists needed for edge write (Step 4) and entity updates
(Step 6). Null field = omit, preserve existing blocks.
+ List<AtlasVertex> classificationsToBlock = new ArrayList<>();
+ List<String> classificationIdsToBlock = new ArrayList<>();
- if (CollectionUtils.isNotEmpty(entitiesRemoved)) {
- removePropagationsMap.put(classificationVertex,
entitiesRemoved);
- }
+ Set<AtlasClassification> newBlockedClassifications =
relationship.getBlockedPropagatedClassifications();
+ boolean blockedChanged = false;
+ List<AtlasVertex> propagatableClassifications = null;
+ List<AtlasVertex> currBlockedClassifications =
Collections.emptyList();
+
+ if (newBlockedClassifications != null) {
+ propagatableClassifications = getPropagatableClassifications(edge,
oldTagPropagation);
+
+ for (AtlasClassification blockedClassification :
newBlockedClassifications) {
+ AtlasVertex classificationVertex =
validateBlockedPropagatedClassification(propagatableClassifications,
blockedClassification);
+
+ if (classificationVertex != null) {
+ classificationsToBlock.add(classificationVertex);
+
classificationIdsToBlock.add(classificationVertex.getIdForDisplay());
}
}
- for (AtlasVertex classificationVertex :
addPropagationsMap.keySet()) {
- List<AtlasVertex> entitiesToAddPropagation =
addPropagationsMap.get(classificationVertex);
+ blockedChanged = !CollectionUtils.isEqualCollection(oldBlocked,
classificationIdsToBlock);
- addTagPropagation(classificationVertex,
entitiesToAddPropagation);
+ if (blockedChanged) {
+ currBlockedClassifications =
getVerticesForIds(propagatableClassifications, oldBlocked);
}
+ }
- for (AtlasVertex classificationVertex :
removePropagationsMap.keySet()) {
- List<AtlasVertex> entitiesToRemovePropagation =
removePropagationsMap.get(classificationVertex);
+ // Step 3: No-op — nothing changed; skip edge write, task queue, and
entity propagation.
+ if (!propagationChanged && !blockedChanged) {
+ return;
+ }
- removeTagPropagation(classificationVertex,
entitiesToRemovePropagation);
+ // Step 4: Write propagateTags or blocked list to the edge; skipped on
background task unless legacy.
+ if (!isAsyncExecution || isLegacyTask) {
+ if (propagationChanged) {
+ AtlasGraphUtilsV2.setEncodedProperty(edge,
RELATIONSHIPTYPE_TAG_PROPAGATION_KEY, newTagPropagation.name());
+ } else if (blockedChanged) {
+ setBlockedClassificationIds(edge, classificationIdsToBlock);
+ }
+
+ // Step 5: Tasks enabled — edge written above; defer entity tags
to background task with pre-change state.
+ if (!isAsyncExecution && DEFERRED_ACTION_ENABLED) {
+
createAndQueueTask(CLASSIFICATION_PROPAGATION_RELATIONSHIP_UPDATE, edge,
relationship, oldTagPropagation.name(), oldBlocked);
Review Comment:
NPE guard on oldTagPropagation.name())
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
