MLH-1226 track denorm attributes and add to kafka

.github/workflows/maven.yml

@@ -25,6 +25,7 @@ on:
       - beta
       - master
       - staging
+      - mlh-1226-denorm-attrs-push-kafka
 
 jobs:
   build:

graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusElement.java

@@ -19,12 +19,14 @@
 
 import java.util.*;
 
+import org.apache.atlas.RequestContext;
 import org.apache.atlas.repository.graphdb.AtlasEdge;
 import org.apache.atlas.repository.graphdb.AtlasElement;
 import org.apache.atlas.repository.graphdb.AtlasSchemaViolationException;
 import org.apache.atlas.repository.graphdb.AtlasVertex;
 import org.apache.atlas.repository.graphdb.janus.graphson.AtlasGraphSONMode;
 import org.apache.atlas.repository.graphdb.janus.graphson.AtlasGraphSONUtility;
+import org.apache.commons.lang.StringUtils;
 import org.apache.tinkerpop.gremlin.structure.Element;
 import org.apache.tinkerpop.gremlin.structure.Property;
 import org.codehaus.jettison.json.JSONException;
@@ -35,6 +37,9 @@
 import org.janusgraph.core.SchemaViolationException;
 import org.janusgraph.core.JanusGraphElement;
 
+import static org.apache.atlas.type.Constants.GUID_PROPERTY_KEY;
+import static org.apache.atlas.type.Constants.INTERNAL_PROPERTY_KEY_PREFIX;
+
 /**
  * Janus implementation of AtlasElement.
  *
@@ -109,36 +114,49 @@ public void removeProperty(String propertyName) {
         while(it.hasNext()) {
             Property<String> property = it.next();
             property.remove();
+            recordInternalAttribute(propertyName, null);
         }
     }
 
     @Override
     public void removePropertyValue(String propertyName, Object propertyValue) {
         Iterator<? extends Property<Object>> it = getWrappedElement().properties(propertyName);
+        List<Object> finalValues = new ArrayList<>();
+        boolean removedFirst = false;
 
         while (it.hasNext()) {
             Property currentProperty      = it.next();
             Object   currentPropertyValue = currentProperty.value();
 
-            if (Objects.equals(currentPropertyValue, propertyValue)) {
+            if (!removedFirst && Objects.equals(currentPropertyValue, propertyValue)) {
                 currentProperty.remove();
-                break;
+                removedFirst = true;
+            } else {
+                finalValues.add(currentPropertyValue);
             }
         }
+
+        recordInternalAttribute(propertyName, finalValues);
     }
 
     @Override
     public void removeAllPropertyValue(String propertyName, Object propertyValue) {
         Iterator<? extends Property<Object>> it = getWrappedElement().properties(propertyName);
+        List<Object> finalValues = new ArrayList<>();
+
 
         while (it.hasNext()) {
             Property currentProperty      = it.next();
             Object   currentPropertyValue = currentProperty.value();
 
             if (Objects.equals(currentPropertyValue, propertyValue)) {
                 currentProperty.remove();
+            } else {
+                finalValues.add(currentPropertyValue);
             }
         }
+
+        recordInternalAttribute(propertyName, finalValues);
     }
 
     @Override
@@ -151,6 +169,7 @@ public void setProperty(String propertyName, Object value) {
                 }
             } else {
                 getWrappedElement().property(propertyName, value);
+                recordInternalAttribute(propertyName, value);
             }
         } catch(SchemaViolationException e) {
             throw new AtlasSchemaViolationException(e);
@@ -308,7 +327,6 @@ public void setPropertyFromElementsIds(String propertyName, List<AtlasElement> v
     @Override
     public void setPropertyFromElementId(String propertyName, AtlasElement value) {
         setProperty(propertyName, value.getId().toString());
-
     }
 
 
@@ -317,4 +335,20 @@ public boolean isIdAssigned() {
         return true;
     }
 
+    private void recordInternalAttribute(String propertyName, Object finalValue) {
+        if (propertyName.startsWith(INTERNAL_PROPERTY_KEY_PREFIX)) {
+            RequestContext context = RequestContext.get();
+            String entityGuid = this.getProperty(GUID_PROPERTY_KEY, String.class);
+
+            if (StringUtils.isNotEmpty(entityGuid)) {
+                if (context.getAllInternalAttributesMap().get(entityGuid) != null) {
+                    context.getAllInternalAttributesMap().get(entityGuid).put(propertyName, finalValue);
+                } else {
+                    Map<String, Object> map = new HashMap<>();
+                    map.put(propertyName, finalValue);
+                    context.getAllInternalAttributesMap().put(entityGuid, map);
+                }
+            }
+        }
+    }
 }

server-api/src/main/java/org/apache/atlas/RequestContext.java

@@ -50,6 +50,7 @@ public class RequestContext {
     private final Map<String, AtlasEntityHeader>         updatedEntities      = new HashMap<>();
     private final Map<String, AtlasEntityHeader>         deletedEntities      = new HashMap<>();
     private final Map<String, AtlasEntityHeader>         restoreEntities      = new HashMap<>();
+    private final Map<String, Map<String, Object>> allInternalAttributesMap     = new HashMap<>();
 
 
     private       Map<String, String>                    lexoRankCache        = null;
@@ -200,6 +201,7 @@ public void clearCache() {
         addedClassificationAndVertices.clear();
         esDeferredOperations.clear();
         this.cassandraTagOperations.clear();
+        this.allInternalAttributesMap.clear();
 
         if (metrics != null && !metrics.isEmpty()) {
             METRICS.debug(metrics.toString());
@@ -877,6 +879,10 @@ public boolean isInvokedByLineage() {
         return isInvokedByLineage;
     }
 
+    public Map<String, Map<String, Object>> getAllInternalAttributesMap() {
+        return allInternalAttributesMap;
+    }
+
     public class EntityGuidPair {
         private final Object entity;
         private final String guid;

webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java

@@ -307,6 +307,16 @@ private AtlasEntityHeaderWithRelations toNotificationHeader(AtlasEntity entity)
                 }
             }
 
+            // fill the internal properties like __glossary, __meanings etc. (ES Isolation)
+            RequestContext context = RequestContext.get();
+            Map<String, Object> allInternalAttributesMap = context.getAllInternalAttributesMap().get(entity.getGuid());
+            if (MapUtils.isNotEmpty(allInternalAttributesMap)) {
+                for (String key : allInternalAttributesMap.keySet()) {
+                    Object value = allInternalAttributesMap.get(key);
+                    ret.setAttribute(key, value);
+                }
+            }
+
             //Add relationship attributes which has isOptional as false
             Map<String, Object> rel = new HashMap<>();
             for (Map<String, AtlasAttribute> attrs : entityType.getRelationshipAttributes().values()) {