Repository: atlas
Updated Branches:
  refs/heads/branch-1.0 21b5ec402 -> d3feb0409


ATLAS-3002: added instrumentation to collect time taken for sub-tasks during 
entity create/update

(cherry picked from commit beb34506a15379af4a306902da049f37b445a2f5)


Project: http://git-wip-us.apache.org/repos/asf/atlas/repo
Commit: http://git-wip-us.apache.org/repos/asf/atlas/commit/d3feb040
Tree: http://git-wip-us.apache.org/repos/asf/atlas/tree/d3feb040
Diff: http://git-wip-us.apache.org/repos/asf/atlas/diff/d3feb040

Branch: refs/heads/branch-1.0
Commit: d3feb04090c5a3398693efb6b90dcf8cc369d0b6
Parents: 21b5ec4
Author: Madhan Neethiraj <mad...@apache.org>
Authored: Tue Dec 18 09:17:29 2018 -0800
Committer: Madhan Neethiraj <mad...@apache.org>
Committed: Tue Dec 18 15:06:44 2018 -0800

----------------------------------------------------------------------
 .../hive/hook/events/CreateHiveProcess.java     |   2 +-
 .../apache/atlas/utils/AtlasPerfMetrics.java    | 121 +++++++++++++++++++
 distro/src/conf/atlas-log4j.xml                 |  19 ++-
 .../apache/atlas/kafka/KafkaNotification.java   |   3 +
 .../atlas/GraphTransactionInterceptor.java      |   6 +
 .../repository/audit/EntityAuditListener.java   |  33 +++++
 .../repository/audit/EntityAuditListenerV2.java |  34 +++++-
 .../repository/graph/FullTextMapperV2.java      |  86 +++++++++----
 .../graph/v2/AtlasEntityChangeNotifier.java     |  10 +-
 .../graph/v2/AtlasEntityGraphDiscoveryV2.java   |  10 ++
 .../store/graph/v2/AtlasEntityStoreV2.java      |   9 ++
 .../store/graph/v2/AtlasGraphUtilsV2.java       |  14 +++
 .../store/graph/v2/EntityGraphMapper.java       |  13 ++
 .../java/org/apache/atlas/RequestContext.java   |  22 +++-
 .../EntityNotificationListenerV2.java           |   5 +
 .../NotificationEntityChangeListener.java       |   6 +
 .../notification/NotificationHookConsumer.java  |  12 ++
 17 files changed, 378 insertions(+), 27 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/events/CreateHiveProcess.java
----------------------------------------------------------------------
diff --git 
a/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/events/CreateHiveProcess.java
 
b/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/events/CreateHiveProcess.java
index 2ccfff4..d61f1d7 100644
--- 
a/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/events/CreateHiveProcess.java
+++ 
b/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/events/CreateHiveProcess.java
@@ -185,7 +185,7 @@ public class CreateHiveProcess extends BaseHiveEvent {
 
             AtlasEntity columnLineageProcess = new 
AtlasEntity(HIVE_TYPE_COLUMN_LINEAGE);
 
-            columnLineageProcess.setAttribute(ATTRIBUTE_NAME, 
hiveProcess.getAttribute(ATTRIBUTE_NAME) + ":" + 
outputColumn.getAttribute(ATTRIBUTE_NAME));
+            columnLineageProcess.setAttribute(ATTRIBUTE_NAME, 
hiveProcess.getAttribute(ATTRIBUTE_QUALIFIED_NAME) + ":" + 
outputColumn.getAttribute(ATTRIBUTE_NAME));
             columnLineageProcess.setAttribute(ATTRIBUTE_QUALIFIED_NAME, 
hiveProcess.getAttribute(ATTRIBUTE_QUALIFIED_NAME) + ":" + 
outputColumn.getAttribute(ATTRIBUTE_NAME));
             columnLineageProcess.setAttribute(ATTRIBUTE_INPUTS, 
getObjectIds(inputColumns));
             columnLineageProcess.setAttribute(ATTRIBUTE_OUTPUTS, 
Collections.singletonList(getObjectId(outputColumn)));

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/common/src/main/java/org/apache/atlas/utils/AtlasPerfMetrics.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/atlas/utils/AtlasPerfMetrics.java 
b/common/src/main/java/org/apache/atlas/utils/AtlasPerfMetrics.java
new file mode 100644
index 0000000..c72b2c3
--- /dev/null
+++ b/common/src/main/java/org/apache/atlas/utils/AtlasPerfMetrics.java
@@ -0,0 +1,121 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.atlas.utils;
+
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+
+public class AtlasPerfMetrics {
+    private final Map<String, Metric> metrics = new LinkedHashMap<>();
+
+
+    public MetricRecorder getMetricRecorder(String name) {
+        return new MetricRecorder(name);
+    }
+
+    public void recordMetric(MetricRecorder recorder) {
+        if (recorder != null) {
+            final String name      = recorder.name;
+            final long   timeTaken = recorder.getElapsedTime();
+
+            Metric metric = metrics.get(name);
+
+            if (metric == null) {
+                metric = new Metric(name);
+
+                metrics.put(name, metric);
+            }
+
+            metric.invocations++;
+            metric.totalTimeMSecs += timeTaken;
+        }
+    }
+
+    public void clear() {
+        metrics.clear();
+    }
+
+    public boolean isEmpty() {
+        return metrics.isEmpty();
+    }
+
+    public Set<String> getMetricsNames() {
+        return metrics.keySet();
+    }
+
+    public Metric getMetric(String name) {
+        return metrics.get(name);
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder();
+
+        sb.append("{");
+
+        if (!metrics.isEmpty()) {
+            for (Metric metric : metrics.values()) {
+                
sb.append("\"").append(metric.getName()).append("\":{\"count\":").append(metric.getInvocations()).append(",\"timeTaken\":").append(metric.getTotalTimeMSecs()).append("},");
+            }
+
+            sb.setLength(sb.length() - 1); // remove last ","
+        }
+
+        sb.append("}");
+
+        return sb.toString();
+    }
+
+    public class MetricRecorder {
+        private final String name;
+        private final long   startTimeMs = System.currentTimeMillis();
+
+        MetricRecorder(String name) {
+            this.name = name;
+        }
+
+        long getElapsedTime() {
+            return System.currentTimeMillis() - startTimeMs;
+        }
+    }
+
+    public static class Metric {
+        private final String name;
+        private       short  invocations    = 0;
+        private       long   totalTimeMSecs = 0;
+
+        public Metric(String name) {
+            this.name = name;
+        }
+
+        public String getName() {
+            return name;
+        }
+
+        public short getInvocations() {
+            return invocations;
+        }
+
+        public long getTotalTimeMSecs() {
+            return totalTimeMSecs;
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/distro/src/conf/atlas-log4j.xml
----------------------------------------------------------------------
diff --git a/distro/src/conf/atlas-log4j.xml b/distro/src/conf/atlas-log4j.xml
index c183871..9b4fa04 100755
--- a/distro/src/conf/atlas-log4j.xml
+++ b/distro/src/conf/atlas-log4j.xml
@@ -37,6 +37,16 @@
         </layout>
     </appender>
 
+    <appender name="LARGE_MESSAGES" 
class="org.apache.log4j.RollingFileAppender">
+        <param name="File" value="{{log_dir}}/large_messages.log"/>
+        <param name="Append" value="true"/>
+        <param name="MaxFileSize" value="100MB" />
+        <param name="MaxBackupIndex" value="20" />
+        <layout class="org.apache.log4j.PatternLayout">
+            <param name="ConversionPattern" value="%d %m%n"/>
+        </layout>
+    </appender>
+
     <appender name="AUDIT" class="org.apache.log4j.RollingFileAppender">
         <param name="File" value="${atlas.log.dir}/audit.log"/>
         <param name="Append" value="true"/>
@@ -60,7 +70,7 @@
         <param name="File" value="${atlas.log.dir}/failed.log"/>
         <param name="Append" value="true"/>
         <layout class="org.apache.log4j.PatternLayout">
-            <param name="ConversionPattern" value="%d %m"/>
+            <param name="ConversionPattern" value="%d %m%n"/>
             <param name="maxFileSize" value="100MB" />
             <param name="maxBackupIndex" value="20" />
         </layout>
@@ -119,6 +129,11 @@
         <appender-ref ref="AUDIT"/>
     </logger>
 
+    <logger name="LARGE_MESSAGES" additivity="false">
+        <level value="warn"/>
+        <appender-ref ref="LARGE_MESSAGES"/>
+    </logger>
+
     <logger name="METRICS" additivity="false">
         <level value="debug"/>
         <appender-ref ref="METRICS"/>
@@ -126,7 +141,7 @@
 
     <logger name="FAILED" additivity="false">
         <level value="info"/>
-        <appender-ref ref="AUDIT"/>
+        <appender-ref ref="FAILED"/>
     </logger>
 
     <root>

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
----------------------------------------------------------------------
diff --git 
a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java 
b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
index 4bec917..1d0a273 100644
--- a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
+++ b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
@@ -106,6 +106,9 @@ public class KafkaNotification extends AbstractNotification 
implements Service {
         properties.put("enable.auto.commit", 
kafkaConf.getBoolean("enable.auto.commit", oldApiCommitEnableFlag));
         properties.put("session.timeout.ms", 
kafkaConf.getString("session.timeout.ms", "30000"));
 
+        // if no value is specified for max.poll.records, set to 1
+        properties.put("max.poll.records", 
kafkaConf.getInt("max.poll.records", 1));
+
         LOG.info("<== KafkaNotification()");
     }
 

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/repository/src/main/java/org/apache/atlas/GraphTransactionInterceptor.java
----------------------------------------------------------------------
diff --git 
a/repository/src/main/java/org/apache/atlas/GraphTransactionInterceptor.java 
b/repository/src/main/java/org/apache/atlas/GraphTransactionInterceptor.java
index 4c43677..cbd2226 100644
--- a/repository/src/main/java/org/apache/atlas/GraphTransactionInterceptor.java
+++ b/repository/src/main/java/org/apache/atlas/GraphTransactionInterceptor.java
@@ -23,6 +23,7 @@ import org.aopalliance.intercept.MethodInvocation;
 import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.exception.NotFoundException;
 import org.apache.atlas.repository.graphdb.AtlasGraph;
+import org.apache.atlas.utils.AtlasPerfMetrics.MetricRecorder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
@@ -69,6 +70,7 @@ public class GraphTransactionInterceptor implements 
MethodInterceptor {
         }
 
         boolean isSuccess = false;
+        MetricRecorder metric = null;
 
         try {
             try {
@@ -79,6 +81,8 @@ public class GraphTransactionInterceptor implements 
MethodInterceptor {
                         LOG.debug("Ignoring commit for nested/inner 
transaction {}.{}", invokingClass, invokedMethodName);
                     }
                 } else {
+                    metric = 
RequestContext.get().startMetricRecord("graphCommit");
+
                     doCommitOrRollback(invokingClass, invokedMethodName);
                 }
 
@@ -97,6 +101,8 @@ public class GraphTransactionInterceptor implements 
MethodInterceptor {
                 throw t;
             }
         } finally {
+            RequestContext.get().endMetricRecord(metric);
+
             // Only outer txn can mark as closed
             if (!isInnerTxn) {
                 if (LOG.isDebugEnabled()) {

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java
----------------------------------------------------------------------
diff --git 
a/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java
 
b/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java
index dfacb38..69d373d 100644
--- 
a/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java
+++ 
b/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java
@@ -24,6 +24,7 @@ import org.apache.atlas.EntityAuditEvent.EntityAuditAction;
 import org.apache.atlas.RequestContext;
 import org.apache.atlas.listener.EntityChangeListener;
 import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
+import org.apache.atlas.utils.AtlasPerfMetrics.MetricRecorder;
 import org.apache.atlas.v1.model.instance.Referenceable;
 import org.apache.atlas.v1.model.instance.Struct;
 import org.apache.atlas.type.AtlasEntityType;
@@ -65,6 +66,8 @@ public class EntityAuditListener implements 
EntityChangeListener {
 
     @Override
     public void onEntitiesAdded(Collection<Referenceable> entities, boolean 
isImport) throws AtlasException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
         List<EntityAuditEvent> events = new ArrayList<>();
         for (Referenceable entity : entities) {
             EntityAuditEvent event = createEvent(entity, isImport ? 
EntityAuditAction.ENTITY_IMPORT_CREATE : EntityAuditAction.ENTITY_CREATE);
@@ -72,10 +75,14 @@ public class EntityAuditListener implements 
EntityChangeListener {
         }
 
         auditRepository.putEventsV1(events);
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     @Override
     public void onEntitiesUpdated(Collection<Referenceable> entities, boolean 
isImport) throws AtlasException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
         List<EntityAuditEvent> events = new ArrayList<>();
         for (Referenceable entity : entities) {
             EntityAuditEvent event = createEvent(entity, isImport ? 
EntityAuditAction.ENTITY_IMPORT_UPDATE : EntityAuditAction.ENTITY_UPDATE);
@@ -83,45 +90,61 @@ public class EntityAuditListener implements 
EntityChangeListener {
         }
 
         auditRepository.putEventsV1(events);
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     @Override
     public void onTraitsAdded(Referenceable entity, Collection<? extends 
Struct> traits) throws AtlasException {
         if (traits != null) {
+            MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
             for (Struct trait : traits) {
                 EntityAuditEvent event = createEvent(entity, 
EntityAuditAction.TAG_ADD,
                                                      "Added trait: " + 
AtlasType.toV1Json(trait));
 
                 auditRepository.putEventsV1(event);
             }
+
+            RequestContext.get().endMetricRecord(metric);
         }
     }
 
     @Override
     public void onTraitsDeleted(Referenceable entity, Collection<? extends 
Struct> traits) throws AtlasException {
         if (traits != null) {
+            MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
             for (Struct trait : traits) {
                 EntityAuditEvent event = createEvent(entity, 
EntityAuditAction.TAG_DELETE, "Deleted trait: " + trait.getTypeName());
 
                 auditRepository.putEventsV1(event);
             }
+
+            RequestContext.get().endMetricRecord(metric);
         }
     }
 
     @Override
     public void onTraitsUpdated(Referenceable entity, Collection<? extends 
Struct> traits) throws AtlasException {
         if (traits != null) {
+            MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
             for (Struct trait : traits) {
                 EntityAuditEvent event = createEvent(entity, 
EntityAuditAction.TAG_UPDATE,
                                                      "Updated trait: " + 
AtlasType.toV1Json(trait));
 
                 auditRepository.putEventsV1(event);
             }
+
+            RequestContext.get().endMetricRecord(metric);
         }
     }
 
     @Override
     public void onEntitiesDeleted(Collection<Referenceable> entities, boolean 
isImport) throws AtlasException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
         List<EntityAuditEvent> events = new ArrayList<>();
         for (Referenceable entity : entities) {
             EntityAuditEvent event = createEvent(entity, isImport ? 
EntityAuditAction.ENTITY_IMPORT_DELETE : EntityAuditAction.ENTITY_DELETE, 
"Deleted entity");
@@ -129,10 +152,14 @@ public class EntityAuditListener implements 
EntityChangeListener {
         }
 
         auditRepository.putEventsV1(events);
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     @Override
     public void onTermAdded(Collection<Referenceable> entities, 
AtlasGlossaryTerm term) throws AtlasException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
         List<EntityAuditEvent> events = new ArrayList<>();
 
         for (Referenceable entity : entities) {
@@ -140,10 +167,14 @@ public class EntityAuditListener implements 
EntityChangeListener {
         }
 
         auditRepository.putEventsV1(events);
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     @Override
     public void onTermDeleted(Collection<Referenceable> entities, 
AtlasGlossaryTerm term) throws AtlasException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
         List<EntityAuditEvent> events = new ArrayList<>();
 
         for (Referenceable entity : entities) {
@@ -151,6 +182,8 @@ public class EntityAuditListener implements 
EntityChangeListener {
         }
 
         auditRepository.putEventsV1(events);
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     public List<EntityAuditEvent> getAuditEvents(String guid) throws 
AtlasException{

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListenerV2.java
----------------------------------------------------------------------
diff --git 
a/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListenerV2.java
 
b/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListenerV2.java
index 8ca8c9a..16d8879 100644
--- 
a/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListenerV2.java
+++ 
b/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListenerV2.java
@@ -26,13 +26,13 @@ import org.apache.atlas.listener.EntityChangeListenerV2;
 import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
 import org.apache.atlas.model.instance.AtlasClassification;
 import org.apache.atlas.model.instance.AtlasEntity;
-import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
 import org.apache.atlas.model.instance.AtlasRelatedObjectId;
 import org.apache.atlas.repository.converters.AtlasInstanceConverter;
 import org.apache.atlas.type.AtlasEntityType;
 import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
 import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.atlas.utils.AtlasPerfMetrics.MetricRecorder;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -79,6 +79,8 @@ public class EntityAuditListenerV2 implements 
EntityChangeListenerV2 {
 
     @Override
     public void onEntitiesAdded(List<AtlasEntity> entities, boolean isImport) 
throws AtlasBaseException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
         List<EntityAuditEventV2> events = new ArrayList<>();
 
         for (AtlasEntity entity : entities) {
@@ -88,10 +90,14 @@ public class EntityAuditListenerV2 implements 
EntityChangeListenerV2 {
         }
 
         auditRepository.putEventsV2(events);
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     @Override
     public void onEntitiesUpdated(List<AtlasEntity> entities, boolean 
isImport) throws AtlasBaseException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
         List<EntityAuditEventV2> events = new ArrayList<>();
 
         for (AtlasEntity entity : entities) {
@@ -101,10 +107,14 @@ public class EntityAuditListenerV2 implements 
EntityChangeListenerV2 {
         }
 
         auditRepository.putEventsV2(events);
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     @Override
     public void onEntitiesDeleted(List<AtlasEntity> entities, boolean 
isImport) throws AtlasBaseException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
         List<EntityAuditEventV2> events = new ArrayList<>();
 
         for (AtlasEntity entity : entities) {
@@ -114,11 +124,15 @@ public class EntityAuditListenerV2 implements 
EntityChangeListenerV2 {
         }
 
         auditRepository.putEventsV2(events);
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     @Override
     public void onClassificationsAdded(AtlasEntity entity, 
List<AtlasClassification> classifications) throws AtlasBaseException {
         if (CollectionUtils.isNotEmpty(classifications)) {
+            MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
             List<EntityAuditEventV2> events = new ArrayList<>();
 
             for (AtlasClassification classification : classifications) {
@@ -130,12 +144,16 @@ public class EntityAuditListenerV2 implements 
EntityChangeListenerV2 {
             }
 
             auditRepository.putEventsV2(events);
+
+            RequestContext.get().endMetricRecord(metric);
         }
     }
 
     @Override
     public void onClassificationsUpdated(AtlasEntity entity, 
List<AtlasClassification> classifications) throws AtlasBaseException {
         if (CollectionUtils.isNotEmpty(classifications)) {
+            MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
             List<EntityAuditEventV2> events = new ArrayList<>();
             String                   guid   = entity.getGuid();
 
@@ -154,12 +172,16 @@ public class EntityAuditListenerV2 implements 
EntityChangeListenerV2 {
             }
 
             auditRepository.putEventsV2(events);
+
+            RequestContext.get().endMetricRecord(metric);
         }
     }
 
     @Override
     public void onClassificationsDeleted(AtlasEntity entity, 
List<AtlasClassification> classifications) throws AtlasBaseException {
         if (CollectionUtils.isNotEmpty(classifications)) {
+            MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
             List<EntityAuditEventV2> events = new ArrayList<>();
 
             for (AtlasClassification classification : classifications) {
@@ -171,12 +193,16 @@ public class EntityAuditListenerV2 implements 
EntityChangeListenerV2 {
             }
 
             auditRepository.putEventsV2(events);
+
+            RequestContext.get().endMetricRecord(metric);
         }
     }
 
     @Override
     public void onTermAdded(AtlasGlossaryTerm term, List<AtlasRelatedObjectId> 
entities) throws AtlasBaseException {
         if (term != null && CollectionUtils.isNotEmpty(entities)) {
+            MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
             List<EntityAuditEventV2> events = new ArrayList<>();
 
             for (AtlasRelatedObjectId relatedObjectId : entities) {
@@ -188,12 +214,16 @@ public class EntityAuditListenerV2 implements 
EntityChangeListenerV2 {
             }
 
             auditRepository.putEventsV2(events);
+
+            RequestContext.get().endMetricRecord(metric);
         }
     }
 
     @Override
     public void onTermDeleted(AtlasGlossaryTerm term, 
List<AtlasRelatedObjectId> entities) throws AtlasBaseException {
         if (term != null && CollectionUtils.isNotEmpty(entities)) {
+            MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityAudit");
+
             List<EntityAuditEventV2> events = new ArrayList<>();
 
             for (AtlasRelatedObjectId relatedObjectId : entities) {
@@ -205,6 +235,8 @@ public class EntityAuditListenerV2 implements 
EntityChangeListenerV2 {
             }
 
             auditRepository.putEventsV2(events);
+
+            RequestContext.get().endMetricRecord(metric);
         }
     }
 

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/repository/src/main/java/org/apache/atlas/repository/graph/FullTextMapperV2.java
----------------------------------------------------------------------
diff --git 
a/repository/src/main/java/org/apache/atlas/repository/graph/FullTextMapperV2.java
 
b/repository/src/main/java/org/apache/atlas/repository/graph/FullTextMapperV2.java
index 429eec5..90d39fa 100644
--- 
a/repository/src/main/java/org/apache/atlas/repository/graph/FullTextMapperV2.java
+++ 
b/repository/src/main/java/org/apache/atlas/repository/graph/FullTextMapperV2.java
@@ -26,6 +26,13 @@ import 
org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
 import org.apache.atlas.model.instance.AtlasObjectId;
 import org.apache.atlas.model.instance.AtlasStruct;
 import org.apache.atlas.repository.store.graph.v2.EntityGraphRetriever;
+import org.apache.atlas.type.AtlasArrayType;
+import org.apache.atlas.type.AtlasBuiltInTypes;
+import org.apache.atlas.type.AtlasClassificationType;
+import org.apache.atlas.type.AtlasEntityType;
+import org.apache.atlas.type.AtlasStructType;
+import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
+import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
@@ -37,6 +44,7 @@ import org.springframework.stereotype.Component;
 
 import javax.inject.Inject;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -52,15 +60,18 @@ public class FullTextMapperV2 {
     private static final String FULL_TEXT_FOLLOW_REFERENCES          = 
"atlas.search.fulltext.followReferences";
     private static final String FULL_TEXT_EXCLUDE_ATTRIBUTE_PROPERTY = 
"atlas.search.fulltext.type";
 
-    private final EntityGraphRetriever     entityGraphRetriever;
+    private final AtlasTypeRegistry        typeRegistry;
     private final Configuration            configuration;
+    private final EntityGraphRetriever     entityGraphRetriever;
     private final boolean                  followReferences;
     private final Map<String, Set<String>> excludeAttributesCache = new 
HashMap<>();
 
 
     @Inject
     public FullTextMapperV2(AtlasTypeRegistry typeRegistry, Configuration 
configuration) {
+        this.typeRegistry  = typeRegistry;
         this.configuration = configuration;
+
         followReferences = this.configuration != null && 
this.configuration.getBoolean(FULL_TEXT_FOLLOW_REFERENCES, false);
         // If followReferences = false then ignore relationship attr loading
         entityGraphRetriever = new EntityGraphRetriever(typeRegistry, 
!followReferences);
@@ -90,11 +101,12 @@ public class FullTextMapperV2 {
 
             if (CollectionUtils.isNotEmpty(classifications)) {
                 for (AtlasClassification classification : classifications) {
-                    
sb.append(classification.getTypeName()).append(FULL_TEXT_DELIMITER);
+                    final AtlasClassificationType classificationType = 
typeRegistry.getClassificationTypeByName(classification.getTypeName());
+                    final Set<String>             excludeAttributes  = 
getExcludeAttributesForIndexText(classification.getTypeName());
 
-                    Set<String> excludeAttributes = 
getExcludeAttributesForIndexText(classification.getTypeName());
+                    
sb.append(classification.getTypeName()).append(FULL_TEXT_DELIMITER);
 
-                    mapAttributes(classification.getAttributes(), 
entityWithExtInfo, sb, new HashSet<String>(), excludeAttributes);
+                    mapAttributes(classificationType, 
classification.getAttributes(), entityWithExtInfo, sb, new HashSet<String>(), 
excludeAttributes);
                 }
             }
 
@@ -109,13 +121,24 @@ public class FullTextMapperV2 {
     }
 
     public String getIndexTextForEntity(String guid) throws AtlasBaseException 
{
-        String      ret    = null;
-        AtlasEntity entity = getAndCacheEntity(guid);
+        String                   ret    = null;
+        final AtlasEntity        entity;
+        final AtlasEntityExtInfo entityExtInfo;
+
+        if (followReferences) {
+            AtlasEntityWithExtInfo entityWithExtInfo = 
getAndCacheEntityWithExtInfo(guid);
+
+            entity        = entityWithExtInfo != null ? 
entityWithExtInfo.getEntity() : null;
+            entityExtInfo = entityWithExtInfo;
+        } else {
+            entity        = getAndCacheEntity(guid);
+            entityExtInfo = null;
+        }
 
         if (entity != null) {
             StringBuilder sb = new StringBuilder();
 
-            map(entity, null, sb, new HashSet<String>());
+            map(entity, entityExtInfo, sb, new HashSet<String>());
 
             ret = sb.toString();
         }
@@ -132,27 +155,30 @@ public class FullTextMapperV2 {
             return;
         }
 
+        final AtlasEntityType entityType        = 
typeRegistry.getEntityTypeByName(entity.getTypeName());
+        final Set<String>     excludeAttributes = 
getExcludeAttributesForIndexText(entity.getTypeName());
+
         processedGuids.add(entity.getGuid());
 
         sb.append(entity.getTypeName()).append(FULL_TEXT_DELIMITER);
 
-        Set<String> excludeAttributes = 
getExcludeAttributesForIndexText(entity.getTypeName());
+        mapAttributes(entityType, entity.getAttributes(), entityExtInfo, sb, 
processedGuids, excludeAttributes);
 
-        mapAttributes(entity.getAttributes(), entityExtInfo, sb, 
processedGuids, excludeAttributes);
-
-        List<AtlasClassification> classifications = 
entity.getClassifications();
+        final List<AtlasClassification> classifications = 
entity.getClassifications();
         if (CollectionUtils.isNotEmpty(classifications)) {
             for (AtlasClassification classification : classifications) {
-                
sb.append(classification.getTypeName()).append(FULL_TEXT_DELIMITER);
+                final AtlasClassificationType classificationType              
= typeRegistry.getClassificationTypeByName(classification.getTypeName());
+                final Set<String>             excludeClassificationAttributes 
= getExcludeAttributesForIndexText(classification.getTypeName());
 
-                Set<String> excludeClassificationAttributes = 
getExcludeAttributesForIndexText(classification.getTypeName());
 
-                mapAttributes(classification.getAttributes(), entityExtInfo, 
sb, processedGuids, excludeClassificationAttributes);
+                
sb.append(classification.getTypeName()).append(FULL_TEXT_DELIMITER);
+
+                mapAttributes(classificationType, 
classification.getAttributes(), entityExtInfo, sb, processedGuids, 
excludeClassificationAttributes);
             }
         }
     }
 
-    private void mapAttributes(Map<String, Object> attributes, 
AtlasEntityExtInfo entityExtInfo, StringBuilder sb,
+    private void mapAttributes(AtlasStructType structType, Map<String, Object> 
attributes, AtlasEntityExtInfo entityExtInfo, StringBuilder sb,
                                Set<String> processedGuids, Set<String> 
excludeAttributes) throws AtlasBaseException {
         if (MapUtils.isEmpty(attributes)) {
             return;
@@ -162,10 +188,28 @@ public class FullTextMapperV2 {
             String attribKey = attributeEntry.getKey();
             Object attrValue = attributeEntry.getValue();
 
-            if (attrValue == null || isExcludedAttribute(excludeAttributes, 
attribKey)) {
+            if (attrValue == null || excludeAttributes.contains(attribKey)) {
                 continue;
             }
 
+            if (!followReferences) {
+                AtlasAttribute attribute     = structType != null ? 
structType.getAttribute(attribKey) : null;
+                AtlasType      attributeType = attribute != null ? 
attribute.getAttributeType() : null;
+
+                if (attributeType == null) {
+                    continue;
+                }
+
+                if (attributeType instanceof AtlasArrayType) {
+                    attributeType = ((AtlasArrayType) 
attributeType).getElementType();
+                }
+
+                if (attributeType instanceof AtlasEntityType || attributeType 
instanceof AtlasBuiltInTypes.AtlasObjectIdType) {
+                    continue;
+                }
+            }
+
+
             sb.append(attribKey).append(FULL_TEXT_DELIMITER);
 
             mapAttribute(attrValue, entityExtInfo, sb, processedGuids);
@@ -249,12 +293,8 @@ public class FullTextMapperV2 {
         return entityWithExtInfo;
     }
 
-    private boolean isExcludedAttribute(Set<String> excludeAttributes, String 
attributeName) {
-        return CollectionUtils.isNotEmpty(excludeAttributes) && 
excludeAttributes.contains(attributeName);
-    }
-
     private Set<String> getExcludeAttributesForIndexText(String typeName) {
-        Set<String> ret = null;
+        final Set<String> ret;
 
         if (excludeAttributesCache.containsKey(typeName)) {
             ret = excludeAttributesCache.get(typeName);
@@ -265,9 +305,13 @@ public class FullTextMapperV2 {
 
             if (ArrayUtils.isNotEmpty(excludeAttributes)) {
                 ret = new HashSet<>(Arrays.asList(excludeAttributes));
+            } else {
+                ret = Collections.emptySet();
             }
 
             excludeAttributesCache.put(typeName, ret);
+        } else {
+            ret = Collections.emptySet();
         }
 
         return ret;

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityChangeNotifier.java
----------------------------------------------------------------------
diff --git 
a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityChangeNotifier.java
 
b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityChangeNotifier.java
index a8c3363..ca3179a 100644
--- 
a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityChangeNotifier.java
+++ 
b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityChangeNotifier.java
@@ -29,13 +29,13 @@ import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
 import org.apache.atlas.model.instance.AtlasClassification;
 import org.apache.atlas.model.instance.AtlasEntity;
 
-import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
 import org.apache.atlas.model.instance.AtlasEntityHeader;
 import org.apache.atlas.model.instance.AtlasRelatedObjectId;
 import org.apache.atlas.model.instance.EntityMutationResponse;
 import org.apache.atlas.model.instance.EntityMutations.EntityOperation;
 import org.apache.atlas.type.AtlasEntityType;
 import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.atlas.utils.AtlasPerfMetrics.MetricRecorder;
 import org.apache.atlas.v1.model.instance.Referenceable;
 import org.apache.atlas.v1.model.instance.Struct;
 import org.apache.atlas.repository.converters.AtlasInstanceConverter;
@@ -437,6 +437,8 @@ public class AtlasEntityChangeNotifier {
             LOG.warn("Unable to determine if FullText is disabled. Proceeding 
with FullText mapping");
         }
 
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("fullTextMapping");
+
         for (AtlasEntityHeader entityHeader : entityHeaders) {
             if(GraphHelper.isInternalType(entityHeader.getTypeName())) {
                 continue;
@@ -457,6 +459,8 @@ public class AtlasEntityChangeNotifier {
                 LOG.error("FullText mapping failed for Vertex[ guid = {} ]", 
guid, e);
             }
         }
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     private void updateFullTextMapping(String entityId, 
List<AtlasClassification> classifications) {
@@ -477,6 +481,8 @@ public class AtlasEntityChangeNotifier {
             return;
         }
 
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("fullTextMapping");
+
         try {
             String classificationFullText = 
fullTextMapperV2.getIndexTextForClassifications(entityId, classifications);
             String existingFullText       = 
AtlasGraphUtilsV2.getEncodedProperty(atlasVertex, ENTITY_TEXT_PROPERTY_KEY, 
String.class);
@@ -486,6 +492,8 @@ public class AtlasEntityChangeNotifier {
         } catch (AtlasBaseException e) {
             LOG.error("FullText mapping failed for Vertex[ guid = {} ]", 
entityId, e);
         }
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     private void doFullTextMapping(String guid) {

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityGraphDiscoveryV2.java
----------------------------------------------------------------------
diff --git 
a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityGraphDiscoveryV2.java
 
b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityGraphDiscoveryV2.java
index 6580bee..ddab2bf 100644
--- 
a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityGraphDiscoveryV2.java
+++ 
b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityGraphDiscoveryV2.java
@@ -18,6 +18,7 @@
 package org.apache.atlas.repository.store.graph.v2;
 
 import org.apache.atlas.AtlasErrorCode;
+import org.apache.atlas.RequestContext;
 import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.model.TypeCategory;
 import org.apache.atlas.model.instance.AtlasEntity;
@@ -35,6 +36,7 @@ import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
 import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
 import org.apache.atlas.type.AtlasTypeUtil;
+import org.apache.atlas.utils.AtlasPerfMetrics.MetricRecorder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -127,6 +129,8 @@ public class AtlasEntityGraphDiscoveryV2 implements 
EntityGraphDiscovery {
 
 
     protected void discover() throws AtlasBaseException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("walkEntityGraph");
+
         EntityStream entityStream = discoveryContext.getEntityStream();
 
         Set<String> walkedEntities = new HashSet<>();
@@ -162,9 +166,13 @@ public class AtlasEntityGraphDiscoveryV2 implements 
EntityGraphDiscovery {
                 walkedEntities.add(entity.getGuid());
             }
         }
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     protected void resolveReferences() throws AtlasBaseException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("resolveReferences");
+
         EntityResolver[] entityResolvers = new EntityResolver[] { new 
IDBasedEntityResolver(typeRegistry),
                                                                   new 
UniqAttrBasedEntityResolver(typeRegistry)
                                                                 };
@@ -172,6 +180,8 @@ public class AtlasEntityGraphDiscoveryV2 implements 
EntityGraphDiscovery {
         for (EntityResolver resolver : entityResolvers) {
             resolver.resolveEntityReferences(discoveryContext);
         }
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     private void visitReference(AtlasObjectIdType type, Object val) throws 
AtlasBaseException {

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
----------------------------------------------------------------------
diff --git 
a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
 
b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
index d97b74d..35aa3af 100644
--- 
a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
+++ 
b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
@@ -41,6 +41,7 @@ import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
 import org.apache.atlas.type.AtlasTypeUtil;
 import org.apache.atlas.utils.AtlasEntityUtil;
+import org.apache.atlas.utils.AtlasPerfMetrics.MetricRecorder;
 import org.apache.atlas.utils.AtlasPerfTracer;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
@@ -671,6 +672,8 @@ public class AtlasEntityStoreV2 implements AtlasEntityStore 
{
             perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "createOrUpdate()");
         }
 
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("createOrUpdate");
+
         try {
             final EntityMutationContext context = 
preCreateOrUpdate(entityStream, entityGraphMapper, isPartialUpdate);
 
@@ -730,11 +733,15 @@ public class AtlasEntityStoreV2 implements 
AtlasEntityStore {
 
             return ret;
         } finally {
+            RequestContext.get().endMetricRecord(metric);
+
             AtlasPerfTracer.log(perf);
         }
     }
 
     private EntityMutationContext preCreateOrUpdate(EntityStream entityStream, 
EntityGraphMapper entityGraphMapper, boolean isPartialUpdate) throws 
AtlasBaseException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("preCreateOrUpdate");
+
         EntityGraphDiscovery        graphDiscoverer  = new 
AtlasEntityGraphDiscoveryV2(typeRegistry, entityStream);
         EntityGraphDiscoveryContext discoveryContext = 
graphDiscoverer.discoverEntities();
         EntityMutationContext       context          = new 
EntityMutationContext(discoveryContext);
@@ -797,6 +804,8 @@ public class AtlasEntityStoreV2 implements AtlasEntityStore 
{
             }
         }
 
+        RequestContext.get().endMetricRecord(metric);
+
         return context;
     }
 

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasGraphUtilsV2.java
----------------------------------------------------------------------
diff --git 
a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasGraphUtilsV2.java
 
b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasGraphUtilsV2.java
index 25770a3..69d3f63 100644
--- 
a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasGraphUtilsV2.java
+++ 
b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasGraphUtilsV2.java
@@ -20,6 +20,7 @@ package org.apache.atlas.repository.store.graph.v2;
 
 import org.apache.atlas.ApplicationProperties;
 import org.apache.atlas.AtlasErrorCode;
+import org.apache.atlas.RequestContext;
 import org.apache.atlas.SortOrder;
 import org.apache.atlas.discovery.SearchProcessor;
 import org.apache.atlas.exception.AtlasBaseException;
@@ -38,6 +39,7 @@ import org.apache.atlas.type.AtlasEntityType;
 import org.apache.atlas.type.AtlasStructType;
 import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
 import org.apache.atlas.type.AtlasType;
+import org.apache.atlas.utils.AtlasPerfMetrics.MetricRecorder;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.configuration.Configuration;
 import org.apache.commons.lang.StringUtils;
@@ -268,6 +270,8 @@ public class AtlasGraphUtilsV2 {
     }
 
     public static AtlasVertex findByUniqueAttributes(AtlasEntityType 
entityType, Map<String, Object> attrValues) {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("findByUniqueAttributes");
+
         AtlasVertex vertex = null;
 
         final Map<String, AtlasAttribute> uniqueAttributes = 
entityType.getUniqAttributes();
@@ -302,6 +306,8 @@ public class AtlasGraphUtilsV2 {
             }
         }
 
+        RequestContext.get().endMetricRecord(metric);
+
         return vertex;
     }
 
@@ -345,6 +351,8 @@ public class AtlasGraphUtilsV2 {
     }
 
     public static AtlasVertex findByTypeAndPropertyName(String typeName, 
String propertyName, Object attrVal) {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("findByTypeAndPropertyName");
+
         AtlasGraphQuery query = AtlasGraphProvider.getGraphInstance().query()
                                                     
.has(Constants.ENTITY_TYPE_PROPERTY_KEY, typeName)
                                                     .has(propertyName, attrVal)
@@ -354,10 +362,14 @@ public class AtlasGraphUtilsV2 {
 
         AtlasVertex vertex = results.hasNext() ? results.next() : null;
 
+        RequestContext.get().endMetricRecord(metric);
+
         return vertex;
     }
 
     public static AtlasVertex findBySuperTypeAndPropertyName(String typeName, 
String propertyName, Object attrVal) {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("findBySuperTypeAndPropertyName");
+
         AtlasGraphQuery query = AtlasGraphProvider.getGraphInstance().query()
                                                     
.has(Constants.SUPER_TYPES_PROPERTY_KEY, typeName)
                                                     .has(propertyName, attrVal)
@@ -367,6 +379,8 @@ public class AtlasGraphUtilsV2 {
 
         AtlasVertex vertex = results.hasNext() ? results.next() : null;
 
+        RequestContext.get().endMetricRecord(metric);
+
         return vertex;
     }
 

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java
----------------------------------------------------------------------
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
index 6c830ba..98734cc 100644
--- 
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
@@ -55,6 +55,7 @@ import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
 import org.apache.atlas.type.AtlasTypeUtil;
 import org.apache.atlas.utils.AtlasJson;
+import org.apache.atlas.utils.AtlasPerfMetrics.MetricRecorder;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
@@ -175,6 +176,8 @@ public class EntityGraphMapper {
     }
 
     public EntityMutationResponse 
mapAttributesAndClassifications(EntityMutationContext context, final boolean 
isPartialUpdate, final boolean replaceClassifications) throws 
AtlasBaseException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("mapAttributesAndClassifications");
+
         EntityMutationResponse resp = new EntityMutationResponse();
 
         Collection<AtlasEntity> createdEntities = context.getCreatedEntities();
@@ -237,6 +240,8 @@ public class EntityGraphMapper {
             }
         }
 
+        RequestContext.get().endMetricRecord(metric);
+
         return resp;
     }
 
@@ -284,6 +289,8 @@ public class EntityGraphMapper {
         }
 
         if (MapUtils.isNotEmpty(struct.getAttributes())) {
+            MetricRecorder metric = 
RequestContext.get().startMetricRecord("mapAttributes");
+
             AtlasStructType structType = getStructType(struct.getTypeName());
 
             if (op.equals(CREATE)) {
@@ -308,6 +315,8 @@ public class EntityGraphMapper {
             }
 
             updateModificationMetadata(vertex);
+
+            RequestContext.get().endMetricRecord(metric);
         }
 
         if (LOG.isDebugEnabled()) {
@@ -322,6 +331,8 @@ public class EntityGraphMapper {
         }
 
         if (MapUtils.isNotEmpty(entity.getRelationshipAttributes())) {
+            MetricRecorder metric = 
RequestContext.get().startMetricRecord("mapRelationshipAttributes");
+
             AtlasEntityType entityType = getEntityType(entity.getTypeName());
 
             if (op.equals(CREATE)) {
@@ -343,6 +354,8 @@ public class EntityGraphMapper {
             }
 
             updateModificationMetadata(vertex);
+
+            RequestContext.get().endMetricRecord(metric);
         }
 
         if (LOG.isDebugEnabled()) {

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/server-api/src/main/java/org/apache/atlas/RequestContext.java
----------------------------------------------------------------------
diff --git a/server-api/src/main/java/org/apache/atlas/RequestContext.java 
b/server-api/src/main/java/org/apache/atlas/RequestContext.java
index 099d713..29f2974 100644
--- a/server-api/src/main/java/org/apache/atlas/RequestContext.java
+++ b/server-api/src/main/java/org/apache/atlas/RequestContext.java
@@ -23,6 +23,8 @@ import org.apache.atlas.model.instance.AtlasEntity;
 import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
 import org.apache.atlas.model.instance.AtlasObjectId;
 import org.apache.atlas.store.DeleteType;
+import org.apache.atlas.utils.AtlasPerfMetrics;
+import org.apache.atlas.utils.AtlasPerfMetrics.MetricRecorder;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -30,10 +32,11 @@ import org.slf4j.LoggerFactory;
 import java.util.*;
 
 public class RequestContext {
-    private static final Logger LOG = 
LoggerFactory.getLogger(RequestContext.class);
+    private static final Logger METRICS = LoggerFactory.getLogger("METRICS");
 
     private static final ThreadLocal<RequestContext> CURRENT_CONTEXT = new 
ThreadLocal<>();
     private static final Set<RequestContext>         ACTIVE_REQUESTS = new 
HashSet<>();
+    private static final boolean                     isMetricsEnabled = 
METRICS.isDebugEnabled();
 
     private final long                                   requestTime         = 
System.currentTimeMillis();
     private final Map<String, AtlasObjectId>             updatedEntities     = 
new HashMap<>();
@@ -42,6 +45,7 @@ public class RequestContext {
     private final Map<String, AtlasEntityWithExtInfo>    entityExtInfoCache  = 
new HashMap<>();
     private final Map<String, List<AtlasClassification>> addedPropagations   = 
new HashMap<>();
     private final Map<String, List<AtlasClassification>> removedPropagations = 
new HashMap<>();
+    private final AtlasPerfMetrics                       metrics             = 
isMetricsEnabled ? new AtlasPerfMetrics() : null;
     private       List<EntityGuidPair>                   entityGuidInRequest = 
null;
 
     private String      user;
@@ -95,6 +99,12 @@ public class RequestContext {
         this.addedPropagations.clear();
         this.removedPropagations.clear();
 
+        if (metrics != null && !metrics.isEmpty()) {
+            METRICS.debug(metrics.toString());
+
+            metrics.clear();
+        }
+
         if (this.entityGuidInRequest != null) {
             this.entityGuidInRequest.clear();
         }
@@ -273,6 +283,16 @@ public class RequestContext {
         return deletedEntities.containsKey(guid);
     }
 
+
+
+    public MetricRecorder startMetricRecord(String name) { return metrics != 
null ? metrics.getMetricRecorder(name) : null; }
+
+    public void endMetricRecord(MetricRecorder recorder) {
+        if (metrics != null && recorder != null) {
+            metrics.recordMetric(recorder);
+        }
+    }
+
     public void recordEntityGuidUpdate(AtlasEntity entity, String 
guidInRequest) {
         if (entityGuidInRequest == null) {
             entityGuidInRequest = new ArrayList<>();

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java
----------------------------------------------------------------------
diff --git 
a/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java
 
b/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java
index e0a60a1..ef9ebab 100644
--- 
a/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java
+++ 
b/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java
@@ -32,6 +32,7 @@ import org.apache.atlas.type.AtlasClassificationType;
 import org.apache.atlas.type.AtlasEntityType;
 import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
 import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.atlas.utils.AtlasPerfMetrics.MetricRecorder;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.configuration.Configuration;
@@ -118,6 +119,8 @@ public class EntityNotificationListenerV2 implements 
EntityChangeListenerV2 {
     }
 
     private void notifyEntityEvents(List<AtlasEntity> entities, OperationType 
operationType) throws AtlasBaseException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityNotification");
+
         List<EntityNotificationV2> messages = new ArrayList<>();
 
         for (AtlasEntity entity : entities) {
@@ -135,6 +138,8 @@ public class EntityNotificationListenerV2 implements 
EntityChangeListenerV2 {
                 throw new 
AtlasBaseException(AtlasErrorCode.ENTITY_NOTIFICATION_FAILED, e, 
operationType.name());
             }
         }
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     private AtlasEntityHeader toNotificationHeader(AtlasEntity entity) {

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
----------------------------------------------------------------------
diff --git 
a/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
 
b/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
index b5e7ed8..520d1ea 100644
--- 
a/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
+++ 
b/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
@@ -19,8 +19,10 @@ package org.apache.atlas.notification;
 
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.atlas.AtlasException;
+import org.apache.atlas.RequestContext;
 import org.apache.atlas.listener.EntityChangeListener;
 import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
+import org.apache.atlas.utils.AtlasPerfMetrics.MetricRecorder;
 import org.apache.atlas.v1.model.instance.Referenceable;
 import org.apache.atlas.v1.model.instance.Struct;
 import org.apache.atlas.v1.model.notification.EntityNotificationV1;
@@ -159,6 +161,8 @@ public class NotificationEntityChangeListener implements 
EntityChangeListener {
     // send notification of entity change
     private void notifyOfEntityEvent(Collection<Referenceable> 
entityDefinitions,
                                      OperationType             operationType) 
throws AtlasException {
+        MetricRecorder metric = 
RequestContext.get().startMetricRecord("entityNotification");
+
         List<EntityNotificationV1> messages = new ArrayList<>();
 
         for (Referenceable entityDefinition : entityDefinitions) {
@@ -186,6 +190,8 @@ public class NotificationEntityChangeListener implements 
EntityChangeListener {
         if (!messages.isEmpty()) {
             notificationSender.send(messages);
         }
+
+        RequestContext.get().endMetricRecord(metric);
     }
 
     private List<String> getNotificationAttributes(String entityType) {

http://git-wip-us.apache.org/repos/asf/atlas/blob/d3feb040/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
----------------------------------------------------------------------
diff --git 
a/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
 
b/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
index b955948..460fa6a 100644
--- 
a/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
+++ 
b/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
@@ -88,6 +88,7 @@ public class NotificationHookConsumer implements Service, 
ActiveStateChangeHandl
     private static final Logger LOG        = 
LoggerFactory.getLogger(NotificationHookConsumer.class);
     private static final Logger PERF_LOG   = 
AtlasPerfTracer.getPerfLogger(NotificationHookConsumer.class);
     private static final Logger FAILED_LOG = LoggerFactory.getLogger("FAILED");
+    private static final Logger LARGE_MESSAGES_LOG = 
LoggerFactory.getLogger("LARGE_MESSAGES");
 
     private static final int    SC_OK          = 200;
     private static final int    SC_BAD_REQUEST = 400;
@@ -121,6 +122,7 @@ public class NotificationHookConsumer implements Service, 
ActiveStateChangeHandl
     private final int                    maxWaitDuration;
     private final boolean                skipHiveColumnLineageHive20633;
     private final int                    
skipHiveColumnLineageHive20633InputsThreshold;
+    private final int                    largeMessageProcessingTimeThresholdMs;
     private final boolean                consumerDisabled;
 
     private NotificationInterface notificationInterface;
@@ -153,6 +155,7 @@ public class NotificationHookConsumer implements Service, 
ActiveStateChangeHandl
         skipHiveColumnLineageHive20633                = 
applicationProperties.getBoolean(CONSUMER_SKIP_HIVE_COLUMN_LINEAGE_HIVE_20633, 
false);
         skipHiveColumnLineageHive20633InputsThreshold = 
applicationProperties.getInt(CONSUMER_SKIP_HIVE_COLUMN_LINEAGE_HIVE_20633_INPUTS_THRESHOLD,
 15); // skip if avg # of inputs is > 15
         consumerDisabled                                                       
  = applicationProperties.getBoolean(CONSUMER_DISABLED, false);
+        largeMessageProcessingTimeThresholdMs         = 
applicationProperties.getInt("atlas.notification.consumer.large.message.processing.time.threshold.ms",
 60 * 1000);  //  60 sec by default
 
         LOG.info("{}={}", CONSUMER_SKIP_HIVE_COLUMN_LINEAGE_HIVE_20633, 
skipHiveColumnLineageHive20633);
         LOG.info("{}={}", 
CONSUMER_SKIP_HIVE_COLUMN_LINEAGE_HIVE_20633_INPUTS_THRESHOLD, 
skipHiveColumnLineageHive20633InputsThreshold);
@@ -592,6 +595,15 @@ public class NotificationHookConsumer implements Service, 
ActiveStateChangeHandl
             } finally {
                 AtlasPerfTracer.log(perf);
 
+                long msgProcessingTime = perf != null ? perf.getElapsedTime() 
: 0;
+
+                if (msgProcessingTime > largeMessageProcessingTimeThresholdMs) 
{
+                    String strMessage = 
AbstractNotification.getMessageJson(message);
+
+                    LOG.warn("msgProcessingTime={}, msgSize={}, 
topicOffset={}}", msgProcessingTime, strMessage.length(), kafkaMsg.getOffset());
+                    
LARGE_MESSAGES_LOG.warn("{\"msgProcessingTime\":{},\"msgSize\":{},\"topicOffset\":{},\"data\":{}}",
 msgProcessingTime, strMessage.length(), kafkaMsg.getOffset(), strMessage);
+                }
+
                 if (auditLog != null) {
                     auditLog.setHttpStatus(isFailedMsg ? SC_BAD_REQUEST : 
SC_OK);
                     auditLog.setTimeTaken(System.currentTimeMillis() - 
startTime);

Reply via email to