Merge branch 'staging' into mlh-1240-improve-cm-refresh

sriram-atlan · web-flow · commit 96d902c8b3a5 · 2025-10-08T17:41:07.000+05:30
diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml
@@ -25,8 +25,6 @@ on:
       - beta
       - master
       - staging
-      - mlh-1226-denorm-attrs-push-kafka
-      - mlh-1240-improve-cm-refresh
 
 jobs:
   build:
diff --git a/auth-agents-common/src/main/java/org/apache/atlas/policytransformer/CachePolicyTransformerImpl.java b/auth-agents-common/src/main/java/org/apache/atlas/policytransformer/CachePolicyTransformerImpl.java
@@ -387,6 +387,9 @@ private List<RangerPolicyDelta> getRangerPolicyDelta(AtlasEntityHeader service,
         List<RangerPolicyDelta> deletedPolicyDeltas = new ArrayList<>();
         for (String policyGuid : policyGuids) {
             Integer deltaChangeType = auditEventToDeltaChangeType.get(policyChanges.get(policyGuid));
+            if (deltaChangeType == null) {
+                continue;
+            }
             if (deltaChangeType == RangerPolicyDelta.CHANGE_TYPE_POLICY_DELETE) {
                 RangerPolicy deletedPolicy = new RangerPolicy();
                 deletedPolicy.setGuid(policyGuid);
@@ -415,6 +418,10 @@ public Map<String, EntityAuditActionV2> createPolicyChangeMap(String serviceName
         Map<String, EntityAuditActionV2> policyChanges = new HashMap<>();
         for (EntityAuditEventV2 event : events) {
             if (POLICY_ENTITY_TYPE.equals(event.getTypeName()) && !policyChanges.containsKey(event.getEntityId())) {
+                if (auditEventToDeltaChangeType.get(event.getAction()) == null) {
+                    LOG.warn("PolicyDelta: {}: No delta type found for audit_event={} guid={}", serviceName, event.getAction(), event.getEntityId());
+                    continue;
+                }
                 policyChanges.put(event.getEntityId(), event.getAction());
             }
         }
diff --git a/auth-audits/src/main/java/org/apache/atlas/audit/destination/Log4JAuditDestination.java b/auth-audits/src/main/java/org/apache/atlas/audit/destination/Log4JAuditDestination.java
@@ -46,6 +46,7 @@ public class Log4JAuditDestination extends AuditDestination {
 	private static final String AUTH_AUDIT_RESOURCE= "resource";
 	private static final String AUTH_AUDIT_CLIENT_IP = "cliIP";
 	private static final String AUTH_AUDIT_AGENT = "agent";
+	private static final String AUTH_AUDIT_ENFORCER = "enforcer";
 
 
 	public Log4JAuditDestination() {
@@ -105,6 +106,7 @@ private void recordLogAttributes(AuditEventBase eventBase) {
 			MDC.put(AUTH_AUDIT_RESULT, String.valueOf(event.getAccessResult()));
 			MDC.put(AUTH_AUDIT_CLIENT_IP, event.getClientIP());
 			MDC.put(AUTH_AUDIT_AGENT, event.getAgentId());
+			MDC.put(AUTH_AUDIT_ENFORCER, event.getAclEnforcer());
 		}
 	}
 
@@ -118,6 +120,7 @@ private void clearLogAttributes(AuditEventBase event) {
 			MDC.remove(AUTH_AUDIT_RESULT);
 			MDC.remove(AUTH_AUDIT_CLIENT_IP);
 			MDC.remove(AUTH_AUDIT_AGENT);
+			MDC.remove(AUTH_AUDIT_ENFORCER);
 		}
 	}
 
diff --git a/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusElement.java b/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusElement.java
@@ -352,7 +352,7 @@ protected void recordInternalAttribute(String propertyName, Object finalValue) {
         }
     }
 
-    protected void recordInternalAttributeIncrementalAdd(String propertyName, Object value, Class cardinality) {
+    protected void recordInternalAttributeIncrementalAdd(String propertyName, Class cardinality) {
         if (propertyName.startsWith(INTERNAL_PROPERTY_KEY_PREFIX)) {
             String entityGuid = this.getProperty(GUID_PROPERTY_KEY, String.class);
 
diff --git a/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusVertex.java b/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusVertex.java
@@ -56,7 +56,7 @@ public AtlasJanusVertex(AtlasJanusGraph graph, Vertex source) {
     public <T> void addProperty(String propertyName, T value) {
         try {
             getWrappedElement().property(VertexProperty.Cardinality.set, propertyName, value);
-            recordInternalAttributeIncrementalAdd(propertyName, value, Set.class);
+            recordInternalAttributeIncrementalAdd(propertyName, Set.class);
         } catch(SchemaViolationException e) {
             throw new AtlasSchemaViolationException(e);
         }
@@ -66,7 +66,7 @@ public <T> void addProperty(String propertyName, T value) {
     public <T> void addListProperty(String propertyName, T value) {
         try {
             getWrappedElement().property(VertexProperty.Cardinality.list, propertyName, value);
-            recordInternalAttributeIncrementalAdd(propertyName, value, List.class);
+            recordInternalAttributeIncrementalAdd(propertyName, List.class);
         } catch(SchemaViolationException e) {
             throw new AtlasSchemaViolationException(e);
         }
diff --git a/intg/src/main/java/org/apache/atlas/model/typedef/AtlasStructDef.java b/intg/src/main/java/org/apache/atlas/model/typedef/AtlasStructDef.java
@@ -557,7 +557,7 @@ public void setOptions(Map<String, String> options) {
         public boolean isSoftReferenced() {
             return this.options != null &&
                     getOptions().containsKey(AtlasAttributeDef.ATTRDEF_OPTION_SOFT_REFERENCE) &&
-                    getOptions().get(AtlasAttributeDef.ATTRDEF_OPTION_SOFT_REFERENCE).equals(STRING_TRUE);
+                    STRING_TRUE.equalsIgnoreCase(getOptions().get(AtlasAttributeDef.ATTRDEF_OPTION_SOFT_REFERENCE));
         }
 
         @JsonIgnore
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v1/DeleteHandlerV1.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v1/DeleteHandlerV1.java
@@ -1841,13 +1841,11 @@ private void updateAssetHasLineageStatus(AtlasVertex assetVertex, AtlasEdge curr
         }
 
     }
-    private void updateAssetHasLineageStatusV1(AtlasVertex assetVertex, AtlasEdge currentEdge, Collection<AtlasEdge> removedEdges) {
-        AtlasPerfMetrics.MetricRecorder metricRecorder = RequestContext.get().startMetricRecord("updateAssetHasLineageStatusV1");
 
-        removedEdges.forEach(edge -> RequestContext.get().addToDeletedEdgesIdsForResetHasLineage(edge.getIdForDisplay()));
+    private boolean updateAssetHasLineageStatusWithDirection(AtlasVertex assetVertex, AtlasEdge currentEdge, AtlasEdgeDirection direction, Set<String> exclusionList) {
 
         Iterator<AtlasEdge> edgeIterator = assetVertex.query()
-                .direction(AtlasEdgeDirection.BOTH)
+                .direction(direction)
                 .label(PROCESS_EDGE_LABELS)
                 .has(STATE_PROPERTY_KEY, ACTIVE.name())
                 .edges()
@@ -1857,7 +1855,7 @@ private void updateAssetHasLineageStatusV1(AtlasVertex assetVertex, AtlasEdge cu
 
         while (edgeIterator.hasNext()) {
             AtlasEdge edge = edgeIterator.next();
-            if (!RequestContext.get().getDeletedEdgesIdsForResetHasLineage().contains(edge.getIdForDisplay()) && !currentEdge.equals(edge)) {
+            if (!exclusionList.contains(edge.getIdForDisplay()) && !currentEdge.equals(edge)) {
                 AtlasVertex relatedProcessVertex = edge.getOutVertex();
                 boolean processHasLineage = getEntityHasLineage(relatedProcessVertex);
                 if (processHasLineage) {
@@ -1866,8 +1864,41 @@ private void updateAssetHasLineageStatusV1(AtlasVertex assetVertex, AtlasEdge cu
                 }
             }
         }
+        return processHasLineageCount > 0;
+    }
 
-        if (processHasLineageCount == 0) {
+    private boolean updateAssetHasLineageStatusWithOUTDirection(AtlasVertex assetVertex, AtlasEdge currentEdge, Set<String> exclusionList) {
+        AtlasPerfMetrics.MetricRecorder metricRecorder = RequestContext.get().startMetricRecord("updateAssetHasLineageStatusWithOUTDirection");
+        boolean hasLineage = updateAssetHasLineageStatusWithDirection(assetVertex, currentEdge, AtlasEdgeDirection.OUT, exclusionList);
+        RequestContext.get().endMetricRecord(metricRecorder);
+        return hasLineage;
+    }
+
+    private boolean updateAssetHasLineageStatusWithINDirection(AtlasVertex assetVertex, AtlasEdge currentEdge, Set<String> exclusionList) {
+        AtlasPerfMetrics.MetricRecorder metricRecorder = RequestContext.get().startMetricRecord("updateAssetHasLineageStatusWithINDirection");
+        boolean hasLineage = updateAssetHasLineageStatusWithDirection(assetVertex, currentEdge, AtlasEdgeDirection.IN, exclusionList);
+        RequestContext.get().endMetricRecord(metricRecorder);
+        return hasLineage;
+    }
+
+    private void updateAssetHasLineageStatusV1(AtlasVertex assetVertex, AtlasEdge currentEdge, Collection<AtlasEdge> removedEdges) {
+        AtlasPerfMetrics.MetricRecorder metricRecorder = RequestContext.get().startMetricRecord("updateAssetHasLineageStatusV1");
+
+        // Add removed edges to the context
+        removedEdges.forEach(edge -> RequestContext.get().addToDeletedEdgesIdsForResetHasLineage(edge.getIdForDisplay()));
+        Set<String> exclusionList = RequestContext.get().getDeletedEdgesIdsForResetHasLineage();
+
+        // First check in OUT direction
+        boolean hasActiveLineage = updateAssetHasLineageStatusWithOUTDirection(assetVertex, currentEdge, exclusionList);
+
+        // If no active lineage found in OUT direction, check IN direction
+        if (!hasActiveLineage) {
+            hasActiveLineage = updateAssetHasLineageStatusWithINDirection(assetVertex, currentEdge, exclusionList);
+        }
+
+        if (hasActiveLineage) {
+            AtlasGraphUtilsV2.setEncodedProperty(assetVertex, HAS_LINEAGE, true);
+        } else {
             AtlasGraphUtilsV2.setEncodedProperty(assetVertex, HAS_LINEAGE, false);
         }
 
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java
@@ -377,7 +377,7 @@ public EntityMutationResponse mapAttributesAndClassifications(EntityMutationCont
 
                     setCustomAttributes(vertex, createdEntity);
                     setSystemAttributesToEntity(vertex, createdEntity);
-                    resp.addEntity(CREATE, constructHeader(createdEntity, vertex, entityType.getAllAttributes()));
+                    resp.addEntity(CREATE, constructHeader(createdEntity, vertex, entityType));
 
                     if (bulkRequestContext.isAppendTags()) {
                         if (CollectionUtils.isNotEmpty(createdEntity.getAddOrUpdateClassifications())) {
@@ -489,7 +489,7 @@ public EntityMutationResponse mapAttributesAndClassifications(EntityMutationCont
                     }
 
                     setSystemAttributesToEntity(vertex, updatedEntity);
-                    resp.addEntity(updateType, constructHeader(updatedEntity, vertex, entityType.getAllAttributes()));
+                    resp.addEntity(updateType, constructHeader(updatedEntity, vertex, entityType));
 
                     // Add hasLineage for newly created edges
                     Set<AtlasEdge> newlyCreatedEdges = getNewCreatedInputOutputEdges(guid);
@@ -3621,7 +3621,8 @@ private Set<AtlasEdge> getRemovedInputOutputEdges(String guid) {
     }
 
 
-    private AtlasEntityHeader constructHeader(AtlasEntity entity, AtlasVertex vertex, Map<String, AtlasAttribute> attributeMap ) throws AtlasBaseException {
+    private AtlasEntityHeader constructHeader(AtlasEntity entity, AtlasVertex vertex, AtlasEntityType entityType) throws AtlasBaseException {
+        Map<String, AtlasAttribute> attributeMap = entityType.getAllAttributes();
         AtlasEntityHeader header = entityRetriever.toAtlasEntityHeaderWithClassifications(vertex, attributeMap.keySet());
         if (entity.getClassifications() == null) {
             entity.setClassifications(header.getClassifications());
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/repair/AtlasRepairAttributeService.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/repair/AtlasRepairAttributeService.java
@@ -0,0 +1,70 @@
+package org.apache.atlas.repository.store.graph.v2.repair;
+
+import org.apache.atlas.exception.AtlasBaseException;
+import org.apache.atlas.AtlasErrorCode;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import javax.inject.Inject;
+import java.util.Set;
+
+@Component
+public class AtlasRepairAttributeService {
+
+    private static final Logger LOG = LoggerFactory.getLogger(AtlasRepairAttributeService.class);
+
+    private final RepairAttributeFactory repairAttributeFactory;
+
+    @Inject
+    public AtlasRepairAttributeService(RepairAttributeFactory repairAttributeFactory) {
+        this.repairAttributeFactory = repairAttributeFactory;
+    }
+
+    public void repairAttributes(String attributeName, String repairType, Set<String> entityGuids)
+            throws AtlasBaseException {
+
+        validateRequest(attributeName, repairType, entityGuids);
+
+        LOG.info("Starting attribute repair - Type: {}, Attribute: {}, Entities: {}",
+                repairType, attributeName, entityGuids.size());
+
+        try {
+            AtlasRepairAttributeStrategy strategy = repairAttributeFactory.getStrategy(repairType, entityGuids);
+
+            strategy.validate(entityGuids, attributeName);
+            strategy.repair(entityGuids, attributeName);
+
+            LOG.info("Successfully completed attribute repair - Type: {}, Attribute: {}, Entities: {}",
+                    repairType, attributeName, entityGuids.size());
+
+        } catch (Exception e) {
+            LOG.error("Error during attribute repair - Type: {}, Attribute: {}, Entities: {}",
+                    repairType, attributeName, entityGuids.size(), e);
+            throw e;
+        }
+    }
+
+    private void validateRequest(String attributeName, String repairType, Set<String> entityGuids)
+            throws AtlasBaseException {
+
+        if (StringUtils.isEmpty(attributeName)) {
+            throw new AtlasBaseException(AtlasErrorCode.BAD_REQUEST, "Attribute name cannot be empty");
+        }
+
+        if (StringUtils.isEmpty(repairType)) {
+            throw new AtlasBaseException(AtlasErrorCode.BAD_REQUEST, "Repair type cannot be empty");
+        }
+
+        if (CollectionUtils.isEmpty(entityGuids)) {
+            throw new AtlasBaseException(AtlasErrorCode.BAD_REQUEST, "Entity GUIDs cannot be empty");
+        }
+
+        if (entityGuids.size() > 1000) {
+            throw new AtlasBaseException(AtlasErrorCode.BAD_REQUEST,
+                    "Too many entities. Maximum allowed: 1000, provided: " + entityGuids.size());
+        }
+    }
+}
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/repair/AtlasRepairAttributeStrategy.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/repair/AtlasRepairAttributeStrategy.java
@@ -0,0 +1,16 @@
+package org.apache.atlas.repository.store.graph.v2.repair;
+
+
+import org.apache.atlas.exception.AtlasBaseException;
+
+import java.util.Set;
+
+public interface AtlasRepairAttributeStrategy {
+
+    String getRepairType();
+    void repair(Set<String> entityGuids, String attributeName) throws AtlasBaseException;
+    void validate (Set<String> entityGuids, String attributeName) throws AtlasBaseException;
+
+}
+
+
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/repair/RemoveInvalidGuidsRepairStrategy.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/repair/RemoveInvalidGuidsRepairStrategy.java
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/repair/RepairAttributeFactory.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/repair/RepairAttributeFactory.java
diff --git a/repository/src/main/java/org/apache/atlas/services/TagsV2AutoEnabler.java b/repository/src/main/java/org/apache/atlas/services/TagsV2AutoEnabler.java
diff --git a/webapp/src/main/java/org/apache/atlas/web/rest/EntityREST.java b/webapp/src/main/java/org/apache/atlas/web/rest/EntityREST.java

Original file line number	Diff line number	Diff line change
`@@ -46,6 +46,7 @@ public class Log4JAuditDestination extends AuditDestination {`
`46`	`46`	`private static final String AUTH_AUDIT_RESOURCE= "resource";`
`47`	`47`	`private static final String AUTH_AUDIT_CLIENT_IP = "cliIP";`
`48`	`48`	`private static final String AUTH_AUDIT_AGENT = "agent";`
	`49`	`+ private static final String AUTH_AUDIT_ENFORCER = "enforcer";`
`49`	`50`
`50`	`51`
`51`	`52`	`public Log4JAuditDestination() {`
`@@ -105,6 +106,7 @@ private void recordLogAttributes(AuditEventBase eventBase) {`
`105`	`106`	`MDC.put(AUTH_AUDIT_RESULT, String.valueOf(event.getAccessResult()));`
`106`	`107`	`MDC.put(AUTH_AUDIT_CLIENT_IP, event.getClientIP());`
`107`	`108`	`MDC.put(AUTH_AUDIT_AGENT, event.getAgentId());`
	`109`	`+ MDC.put(AUTH_AUDIT_ENFORCER, event.getAclEnforcer());`
`108`	`110`	`}`
`109`	`111`	`}`
`110`	`112`
`@@ -118,6 +120,7 @@ private void clearLogAttributes(AuditEventBase event) {`
`118`	`120`	`MDC.remove(AUTH_AUDIT_RESULT);`
`119`	`121`	`MDC.remove(AUTH_AUDIT_CLIENT_IP);`
`120`	`122`	`MDC.remove(AUTH_AUDIT_AGENT);`
	`123`	`+ MDC.remove(AUTH_AUDIT_ENFORCER);`
`121`	`124`	`}`
`122`	`125`	`}`
`123`	`126`
Original file line number	Diff line number	Diff line change
`@@ -352,7 +352,7 @@ protected void recordInternalAttribute(String propertyName, Object finalValue) {`
`352`	`352`	`}`
`353`	`353`	`}`
`354`	`354`
`355`		`- protected void recordInternalAttributeIncrementalAdd(String propertyName, Object value, Class cardinality) {`
	`355`	`+ protected void recordInternalAttributeIncrementalAdd(String propertyName, Class cardinality) {`
`356`	`356`	`if (propertyName.startsWith(INTERNAL_PROPERTY_KEY_PREFIX)) {`
`357`	`357`	`String entityGuid = this.getProperty(GUID_PROPERTY_KEY, String.class);`
`358`	`358`
Original file line number	Diff line number	Diff line change
`@@ -557,7 +557,7 @@ public void setOptions(Map<String, String> options) {`
`557`	`557`	`public boolean isSoftReferenced() {`
`558`	`558`	`return this.options != null &&`
`559`	`559`	`getOptions().containsKey(AtlasAttributeDef.ATTRDEF_OPTION_SOFT_REFERENCE) &&`
`560`		`- getOptions().get(AtlasAttributeDef.ATTRDEF_OPTION_SOFT_REFERENCE).equals(STRING_TRUE);`
	`560`	`+ STRING_TRUE.equalsIgnoreCase(getOptions().get(AtlasAttributeDef.ATTRDEF_OPTION_SOFT_REFERENCE));`
`561`	`561`	`}`
`562`	`562`
`563`	`563`	`@JsonIgnore`