Update MetastoreAuditProcessor to reduce storage, also refactor some code

backend-service/app/actors/KafkaConsumerMaster.java

@@ -29,11 +29,11 @@ import metadata.etl.models.EtlJobName;
 import models.daos.ClusterDao;
 import models.daos.EtlJobDao;
 
-import msgs.KafkaResponseMsg;
+import msgs.KafkaCommMsg;
 import play.Logger;
 import play.Play;
-import utils.KafkaConfig;
-import utils.KafkaConfig.Topic;
+import models.kafka.KafkaConfig;
+import models.kafka.KafkaConfig.Topic;
 import wherehows.common.kafka.schemaregistry.client.CachedSchemaRegistryClient;
 import wherehows.common.kafka.schemaregistry.client.SchemaRegistryClient;
 import wherehows.common.utils.ClusterUtil;
@@ -132,9 +132,9 @@ public class KafkaConsumerMaster extends UntypedActor {
   @Override
   public void onReceive(Object message)
       throws Exception {
-    if (message instanceof KafkaResponseMsg) {
-      final KafkaResponseMsg kafkaMsg = (KafkaResponseMsg) message;
-      Logger.debug("Got kafka response message: " + kafkaMsg.toString());
+    if (message instanceof KafkaCommMsg) {
+      final KafkaCommMsg kafkaCommMsg = (KafkaCommMsg) message;
+      Logger.debug(kafkaCommMsg.toString());
     } else {
       unhandled(message);
     }

backend-service/app/actors/KafkaConsumerWorker.java

@@ -24,7 +24,7 @@ import kafka.consumer.KafkaStream;
 import kafka.message.MessageAndMetadata;
 
 import akka.actor.UntypedActor;
-import msgs.KafkaResponseMsg;
+import msgs.KafkaCommMsg;
 import play.Logger;
 import wherehows.common.kafka.schemaregistry.client.SchemaRegistryClient;
 import wherehows.common.kafka.serializers.KafkaAvroDeserializer;

backend-service/app/controllers/DatasetController.java

@@ -20,7 +20,7 @@ import java.util.Map;
 import java.util.List;
 import models.daos.DatasetDao;
 import models.daos.UserDao;
-import models.utils.Urn;
+import utils.Urn;
 import org.springframework.dao.EmptyResultDataAccessException;
 import play.Logger;
 import play.libs.Json;

backend-service/app/controllers/DatasetInfoController.java

@@ -19,7 +19,7 @@ import java.sql.SQLException;
 import java.util.List;
 import java.util.Map;
 import models.daos.DatasetInfoDao;
-import models.utils.Urn;
+import utils.Urn;
 import org.springframework.dao.EmptyResultDataAccessException;
 import play.Logger;
 import play.libs.Json;

backend-service/app/controllers/LineageController.java

@@ -19,7 +19,7 @@ import java.sql.SQLException;
 import java.util.List;
 import java.util.Map;
 import models.daos.LineageDao;
-import models.utils.Urn;
+import utils.Urn;
 import play.libs.Json;
 import play.mvc.BodyParser;
 import play.mvc.Controller;

backend-service/app/models/daos/DatasetInfoDao.java

@@ -22,7 +22,7 @@ import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import models.utils.Urn;
+import utils.Urn;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.EmptyResultDataAccessException;
 import play.Logger;

backend-service/app/models/daos/LineageDao.java

@@ -20,7 +20,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
-import models.utils.Urn;
+import utils.Urn;
 import utils.JdbcUtil;
 import wherehows.common.schemas.LineageRecord;
 import wherehows.common.utils.PartitionPatternMatcher;

backend-service/app/models/kafka/GobblinTrackingAuditProcessor.java

@@ -41,8 +41,8 @@ public class GobblinTrackingAuditProcessor extends KafkaConsumerProcessor {
   public Record process(GenericData.Record record, String topic)
       throws Exception {
 
-    if (record != null) {
-      String name = (String) record.get("name");
+    if (record != null && record.get("name") != null) {
+      final String name = record.get("name").toString();
       // only handle "DaliLimitedRetentionAuditor","DaliAutoPurgeAuditor" and "DsIgnoreIDPCAuditor"
       if (name.equals(DALI_LIMITED_RETENTION_AUDITOR)
           || name.equals(DALI_AUTOPURGED_AUDITOR)
@@ -52,10 +52,10 @@ public class GobblinTrackingAuditProcessor extends KafkaConsumerProcessor {
 
         String hasError = metadata.get("HasError");
         if (!hasError.equalsIgnoreCase("true")) {
-          String datasetUrn = metadata.get("DatasetPath");
+          String datasetPath = metadata.get("DatasetPath");
+          String datasetUrn = DATASET_URN_PREFIX + (datasetPath.startsWith("/") ? "" : "/") + datasetPath;
           String ownerUrns = metadata.get("OwnerURNs");
-          DatasetInfoDao.updateKafkaDatasetOwner(DATASET_URN_PREFIX + datasetUrn, ownerUrns, DATASET_OWNER_SOURCE,
-              timestamp);
+          DatasetInfoDao.updateKafkaDatasetOwner(datasetUrn, ownerUrns, DATASET_OWNER_SOURCE, timestamp);
         }
       }
     }

backend-service/app/models/kafka/KafkaConfig.java

@@ -11,7 +11,7 @@
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  */
-package utils;
+package models.kafka;
 
 import java.lang.reflect.Method;
 import java.util.HashMap;
@@ -22,6 +22,7 @@ import metadata.etl.models.EtlJobName;
 import models.daos.EtlJobPropertyDao;
 import org.apache.avro.generic.GenericData;
 import play.Logger;
+import utils.JdbcUtil;
 import wherehows.common.kafka.schemaregistry.client.SchemaRegistryClient;
 import wherehows.common.writers.DatabaseWriter;
 

backend-service/app/models/kafka/MetastoreAuditProcessor.java

@@ -86,8 +86,8 @@ public class MetastoreAuditProcessor extends KafkaConsumerProcessor {
       eventRecord = new MetastoreAuditRecord(server, instance, appName, eventName, eventType, timestamp);
       eventRecord.setEventInfo(metastoreThriftUri, metastoreVersion, isSuccessful, isDataDeleted);
       eventRecord.setTableInfo(dbName, tableName, partition, location, owner, createTime, lastAccessTime);
-      eventRecord.setOldInfo(StringUtil.toStringReplaceNull(oldInfo, null));
-      eventRecord.setNewInfo(StringUtil.toStringReplaceNull(newInfo, null));
+      // eventRecord.setOldInfo(StringUtil.toStringReplaceNull(oldInfo, null));
+      // eventRecord.setNewInfo(StringUtil.toStringReplaceNull(newInfo, null));
     }
     return eventRecord;
   }

backend-service/app/msgs/KafkaCommMsg.java

@@ -20,15 +20,21 @@ import java.util.Map;
 /**
  * Generic communication message between KafkaConsumerMaster and KafkaConsumerWorker
  */
-public class KafkaResponseMsg {
+public class KafkaCommMsg {
 
+  // Message type: AUDIT, AUDIT_RESPONSE, FLUSH, FLUSH_RESPONSE, HEARTBEAT, etc
   private String msgType;
+  // Kafka topic of the worker
   private String topic;
+  // Kafka worker thread number
+  private int thread;
+  // Message content
   private Map<String, Object> content;
 
-  public KafkaResponseMsg(String msgType, String topic) {
+  public KafkaCommMsg(String msgType, String topic, int thread) {
     this.msgType = msgType;
     this.topic = topic;
+    this.thread = thread;
     this.content = new HashMap<>();
   }
 
@@ -48,6 +54,14 @@ public class KafkaResponseMsg {
     this.topic = topic;
   }
 
+  public int getThread() {
+    return thread;
+  }
+
+  public void setThread(int thread) {
+    this.thread = thread;
+  }
+
   public Map<String, Object> getContent() {
     return content;
   }
@@ -56,8 +70,13 @@ public class KafkaResponseMsg {
     this.content = content;
   }
 
+  public void putContent(String key, Object value) {
+    this.content.put(key, value);
+  }
+
   @Override
   public String toString() {
-    return "KafkaResponseMsg [type=" + msgType + ", topic=" + topic + ", " + content.toString() + "]";
+    return "KafkaCommMsg [type=" + msgType + ", topic=" + topic + ", thread=" + thread + ", content="
+        + content.toString() + "]";
   }
 }

backend-service/app/utils/Urn.java

@@ -11,7 +11,7 @@
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  */
-package models.utils;
+package utils;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Set;
@@ -19,7 +19,7 @@ import play.Logger;
 
 
 /**
- * Urn class used for urn convertion
+ * Urn class used for urn conversion
  * Created by zsun on 1/15/15.
  */
 public class Urn {