sheetalshah1007 commented on code in PR #729:
URL: https://github.com/apache/atlas/pull/729#discussion_r3850422998
##########
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:
Valid - Added a null guard when queuing the task: getPropagateTags(edge)
can return null on legacy edges without the property set; in that case the task
is queued without old propagation state and handled as a legacy task.
##########
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);
+ return;
+ }
+ }
+
+ // Step 6: Add/remove propagated tags on downstream entities. — inline
when tasks disabled, or in the background task.
+ List<AtlasVertex> affectedClassifications = new ArrayList<>();
+
+ if (propagationChanged) {
+ if (propagatableClassifications == null) {
+ propagatableClassifications =
getPropagatableClassifications(edge, oldTagPropagation);
+ }
+
+ affectedClassifications.addAll(propagatableClassifications);
+
affectedClassifications.addAll(getPropagatableClassifications(edge,
newTagPropagation));
+ } else if (blockedChanged) {
+ List<AtlasVertex> blockedDiff = (List<AtlasVertex>)
CollectionUtils.disjunction(currBlockedClassifications, classificationsToBlock);
+ affectedClassifications.addAll(blockedDiff);
+ }
+
+ Set<AtlasVertex> uniqueAffectedClassifications = new
HashSet<>(affectedClassifications);
+
+ for (AtlasVertex classificationVertex : uniqueAffectedClassifications)
{
+ List<AtlasVertex> propagationsToRemove = new ArrayList<>();
+ List<AtlasVertex> propagationsToAdd = new ArrayList<>();
+
+
entityRetriever.evaluateClassificationPropagation(classificationVertex,
propagationsToAdd, propagationsToRemove);
+
+ if (CollectionUtils.isNotEmpty(propagationsToAdd)) {
+ addTagPropagation(classificationVertex, propagationsToAdd);
+ }
+
+ if (CollectionUtils.isNotEmpty(propagationsToRemove)) {
+ removeTagPropagation(classificationVertex,
propagationsToRemove);
}
- } else {
Review Comment:
Restored in the Step 4 comment. Same behavior: we write propagateTags or the
blocked list, not both. If both change in one PUT, propagateTags wins.
--
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]
