Skip to content

ATLAS-5375: Fix stale PUT response for /api/atlas/v2/relationship block/unblock propagation #729

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chaitalicod merged 2 commits into from
Aug 26, 2026
Merged

ATLAS-5375: Fix stale PUT response for /api/atlas/v2/relationship block/unblock propagation #729

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5d3f181
ATLAS-5375: Fix stale PUT response for /api/atlas/v2/relationship blo…
sheetalshah1007 Aug 14, 2026
52bb813
ATLAS-5375: Review fixes — Step 4 comment and null guard for task queue
sheetalshah1007 Aug 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions intg/src/main/java/org/apache/atlas/model/instance/AtlasRelationship.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ public class AtlasRelationship extends AtlasStruct implements Serializable {
private Long version;
private Set<AtlasClassification> propagatedClassifications;
private Set<AtlasClassification> blockedPropagatedClassifications;
private Set<String> pendingTasks; // read-only field i.e. value provided is ignored during relationship create/update

public AtlasRelationship() {
super();
Expand Down Expand Up @@ -239,6 +240,7 @@ public AtlasRelationship(AtlasRelationship other) {
if (other != null) {
init(other.guid, other.homeId, other.provenanceType, other.end1, other.end2, other.label, other.propagateTags, other.status, other.createdBy, other.updatedBy,
other.createTime, other.updateTime, other.version, other.propagatedClassifications, other.blockedPropagatedClassifications);
setPendingTasks(other.getPendingTasks());
}
}

Expand Down Expand Up @@ -362,6 +364,14 @@ public void setBlockedPropagatedClassifications(Set<AtlasClassification> blocked
this.blockedPropagatedClassifications = blockedPropagatedClassifications;
}

public Set<String> getPendingTasks() {
return pendingTasks;
}

public void setPendingTasks(Set<String> pendingTasks) {
this.pendingTasks = pendingTasks;
}

@Override
public StringBuilder toString(StringBuilder sb) {
if (sb == null) {
Expand Down Expand Up @@ -389,6 +399,9 @@ public StringBuilder toString(StringBuilder sb) {
sb.append(", blockedPropagatedClassifications=[");
dumpObjects(blockedPropagatedClassifications, sb);
sb.append("]");
sb.append(", pendingTasks=[");
dumpObjects(pendingTasks, sb);
sb.append("]");
sb.append('}');

return sb;
Expand Down Expand Up @@ -460,6 +473,7 @@ private void init(String guid, String homeId, Integer provenanceType, AtlasObjec
setVersion(version);
setPropagatedClassifications(propagatedClassifications);
setBlockedPropagatedClassifications(blockedPropagatedClassifications);
setPendingTasks(null);
}

public enum Status { ACTIVE, DELETED }
Expand Down
10 changes: 8 additions & 2 deletions repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -428,10 +428,16 @@ public static List<String> getTraitNames(AtlasVertex entityVertex, Boolean propa
}

public static List<AtlasVertex> getPropagatableClassifications(AtlasEdge edge) {
if (edge != null && getStatus(edge) != DELETED) {
return getPropagatableClassifications(edge, getPropagateTags(edge));
}
return new ArrayList<>();
}

public static List<AtlasVertex> getPropagatableClassifications(AtlasEdge edge, PropagateTags propagateTags) {
List<AtlasVertex> ret = new ArrayList<>();

if (edge != null && getStatus(edge) != DELETED) {
PropagateTags propagateTags = getPropagateTags(edge);
if (edge != null && getStatus(edge) != DELETED && propagateTags != null) {
AtlasVertex outVertex = edge.getOutVertex();
AtlasVertex inVertex = edge.getInVertex();

Expand Down
142 changes: 98 additions & 44 deletions repository/src/main/java/org/apache/atlas/repository/store/graph/v1/DeleteHandlerV1.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,7 @@
import static org.apache.atlas.repository.store.graph.v2.AtlasGraphUtilsV2.isReference;
import static org.apache.atlas.repository.store.graph.v2.tasks.ClassificationPropagateTaskFactory.CLASSIFICATION_PROPAGATION_ADD;
import static org.apache.atlas.repository.store.graph.v2.tasks.ClassificationPropagateTaskFactory.CLASSIFICATION_PROPAGATION_DELETE;
import static org.apache.atlas.repository.store.graph.v2.tasks.ClassificationPropagateTaskFactory.CLASSIFICATION_PROPAGATION_RELATIONSHIP_UPDATE;
import static org.apache.atlas.type.AtlasStructType.AtlasAttribute.AtlasRelationshipEdgeDirection.OUT;
import static org.apache.atlas.type.Constants.PENDING_TASKS_PROPERTY_KEY;

Expand Down Expand Up @@ -923,59 +924,107 @@ 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 (not both; propagateTags wins if both changed). 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 != null ? oldTagPropagation.name() : null, 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 {
Comment thread
sheetalshah1007 marked this conversation as resolved.
// update blocked propagated classifications only if there is no change is tag propagation (don't update both)
handleBlockedClassifications(edge, relationship.getBlockedPropagatedClassifications());
}
}

Expand Down Expand Up @@ -1029,9 +1078,13 @@ public void createAndQueueTask(String taskType, AtlasVertex entityVertex, String
}

public void createAndQueueTask(String taskType, AtlasEdge relationshipEdge, AtlasRelationship relationship) {
createAndQueueTask(taskType, relationshipEdge, relationship, null, null);
}

public void createAndQueueTask(String taskType, AtlasEdge relationshipEdge, AtlasRelationship relationship, String oldTagPropagation, java.util.List<String> oldBlockedClassifications) {
String currentUser = RequestContext.getCurrentUser();
String relationshipEdgeId = relationshipEdge.getIdForDisplay();
Map<String, Object> taskParams = ClassificationTask.toParameters(relationshipEdgeId, relationship);
Map<String, Object> taskParams = ClassificationTask.toParameters(relationshipEdgeId, relationship, oldTagPropagation, oldBlockedClassifications);
AtlasTask task = taskManagement.createTask(taskType, currentUser, taskParams);

AtlasGraphUtilsV2.addItemToListProperty(relationshipEdge, EDGE_PENDING_TASKS_PROPERTY_KEY, task.getGuid());
Expand Down Expand Up @@ -1507,19 +1560,20 @@ private List<AtlasVertex> getVerticesForIds(List<AtlasVertex> vertices, List<Str

// propagated classifications should contain blocked propagated classification
private AtlasVertex validateBlockedPropagatedClassification(List<AtlasVertex> classificationVertices, AtlasClassification classification) {
AtlasVertex ret = null;
if (classification == null || CollectionUtils.isEmpty(classificationVertices)) {
return null;
}

for (AtlasVertex vertex : classificationVertices) {
String classificationName = getClassificationName(vertex);
String entityGuid = getClassificationEntityGuid(vertex);

if (classificationName.equals(classification.getTypeName()) && entityGuid.equals(classification.getEntityGuid())) {
ret = vertex;
break;
return vertex;
}
}

return ret;
return null;
}

private void setBlockedClassificationIds(AtlasEdge edge, List<String> classificationIds) {
Expand Down
20 changes: 13 additions & 7 deletions repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasGraphUtilsV2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -911,20 +911,26 @@ public static List<String> assignEnumValues(String bmAttributeValues, AtlasEnumT
public static void addItemToListProperty(AtlasEdge edge, String property, String value) {
List<String> list = (List<String>) getListFromProperty(edge, property);

list.add(value);
if (!list.contains(value)) {
List<String> newList = new ArrayList<>(list);
newList.add(value);

edge.setListProperty(property, list);
edge.setListProperty(property, newList);
}
}

public static void removeItemFromListProperty(AtlasEdge edge, String property, String value) {
List<String> list = (List<String>) getListFromProperty(edge, property);

list.remove(value);
if (list.contains(value)) {
List<String> newList = new ArrayList<>(list);
newList.remove(value);

if (CollectionUtils.isEmpty(list)) {
edge.removeProperty(property);
} else {
edge.setListProperty(property, list);
if (CollectionUtils.isEmpty(newList)) {
edge.removeProperty(property);
} else {
edge.setListProperty(property, newList);
}
}
}

Expand Down
13 changes: 1 addition & 12 deletions ...ry/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasRelationshipStoreV2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@
*/
package org.apache.atlas.repository.store.graph.v2;

import org.apache.atlas.AtlasConfiguration;
import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.RequestContext;
import org.apache.atlas.annotation.GraphTransaction;
Expand Down Expand Up @@ -83,15 +82,13 @@
import static org.apache.atlas.repository.Constants.VERSION_PROPERTY_KEY;
import static org.apache.atlas.repository.store.graph.v2.AtlasGraphUtilsV2.getState;
import static org.apache.atlas.repository.store.graph.v2.AtlasGraphUtilsV2.getTypeName;
import static org.apache.atlas.repository.store.graph.v2.tasks.ClassificationPropagateTaskFactory.CLASSIFICATION_PROPAGATION_RELATIONSHIP_UPDATE;

@Component
public class AtlasRelationshipStoreV2 implements AtlasRelationshipStore {
private static final Logger LOG = LoggerFactory.getLogger(AtlasRelationshipStoreV2.class);

private static final Long DEFAULT_RELATIONSHIP_VERSION = 0L;
private static final boolean NOTIFICATIONS_ENABLED = NOTIFICATION_RELATIONSHIPS_ENABLED.getBoolean();
private static final boolean DEFERRED_ACTION_ENABLED = AtlasConfiguration.TASKS_USE_ENABLED.getBoolean();

private final AtlasGraph graph;
private final AtlasTypeRegistry typeRegistry;
Expand Down Expand Up @@ -449,11 +446,7 @@ private AtlasRelationship updateRelationship(AtlasEdge relationshipEdge, AtlasRe
}

private void updateTagPropagations(AtlasEdge relationshipEdge, AtlasRelationship relationship) throws AtlasBaseException {
if (DEFERRED_ACTION_ENABLED) {
createAndQueueTask(CLASSIFICATION_PROPAGATION_RELATIONSHIP_UPDATE, relationshipEdge, relationship);
} else {
deleteDelegate.getHandler().updateTagPropagations(relationshipEdge, relationship);
}
deleteDelegate.getHandler().updateTagPropagations(relationshipEdge, relationship);
}

private void validateRelationship(AtlasRelationship relationship) throws AtlasBaseException {
Expand Down Expand Up @@ -771,10 +764,6 @@ private void sendNotifications(AtlasRelationship ret, OperationType relationship
}
}

private void createAndQueueTask(String taskType, AtlasEdge relationshipEdge, AtlasRelationship relationship) {
deleteDelegate.getHandler().createAndQueueTask(taskType, relationshipEdge, relationship);
}

private void verifyRelationshipReadAccess(AtlasEdge edge) throws AtlasBaseException {
AtlasEntityHeader end1Entity = entityRetriever.toAtlasEntityHeaderWithClassifications(edge.getOutVertex());
AtlasEntityHeader end2Entity = entityRetriever.toAtlasEntityHeaderWithClassifications(edge.getInVertex());
Expand Down
Loading
Loading