Added more metrics for kafka consumption call

Author: hr2904

Diff for: notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java

@@ -117,7 +117,7 @@ public KafkaNotification(Configuration applicationProperties) throws AtlasExcept
         Configuration kafkaConf = ApplicationProperties.getSubsetConfiguration(applicationProperties, PROPERTY_PREFIX);
 
         properties             = ConfigurationConverter.getProperties(kafkaConf);
-        pollTimeOutMs          = kafkaConf.getLong("poll.timeout.ms", 1000);
+        pollTimeOutMs          = 200L;
         consumerClosedErrorMsg = kafkaConf.getString("error.message.consumer_closed", DEFAULT_CONSUMER_CLOSED_ERROR_MESSAGE);
 
         //Override default configs
@@ -156,7 +156,7 @@ protected KafkaNotification(Properties properties) {
         LOG.info("==> KafkaNotification()");
 
         this.properties    = properties;
-        this.pollTimeOutMs = 1000L;
+        this.pollTimeOutMs = 200L;
 
         LOG.info("<== KafkaNotification()");
     }

Diff for: webapp/src/main/java/org/apache/atlas/notification/ObjectPropEventConsumer.java

@@ -343,6 +343,7 @@ public ObjectPropConsumer(NotificationConsumer<ObjectPropEvent> consumer) {
 
         @Override
         public void doWork() {
+            // INC kafka batch size -> reduce kafka calls
             long lineStart = System.currentTimeMillis();
 
             LOG.info("ObjectPropConsumer::doWork() [Line 1] ==> Entered doWork()");
@@ -387,7 +388,7 @@ public void doWork() {
                         lineStart = System.currentTimeMillis();
 
                         for (AtlasKafkaMessage<ObjectPropEvent> msg : messages) {
-                            long msgStart = System.currentTimeMillis(); // track each message individually
+                            long msgStart = System.currentTimeMillis();
 
                             LOG.info("ObjectPropConsumer::doWork() -> Msg consumed on offset : {} with value : {}",
                                     msg.getOffset(), msg.toString());
@@ -401,7 +402,7 @@ public void doWork() {
                             msgStart = System.currentTimeMillis();
 
                             if (res) {
-                                long commitOffset = msg.getOffset() + 1;
+                                long commitOffset = msg.getOffset() + 1; // Reduce this to a single batch commit
                                 consumer.commit(msg.getTopicPartition(), commitOffset);
                                 subTaskSuccess++;
                                 LOG.info("ObjectPropConsumer::doWork() [Line 7-c] => commit offset done in {} ms",
@@ -431,6 +432,7 @@ public void doWork() {
                             LOG.info("ObjectPropConsumer::doWork() [Line 9] => incremented ASSETS_PROPAGATION_FAILED_COUNT in {} ms",
                                     (System.currentTimeMillis() - lineStart));
                         }
+                        // move redis above than kafka
                         lineStart = System.currentTimeMillis();
 
                     } catch (IllegalStateException ex) {