From 916e2a940d546ac5dde3884e82dcb8d88c944b0a Mon Sep 17 00:00:00 2001
From: "Yi (Alan) Wang" <ywang5@linkedin.com>
Date: Wed, 18 Oct 2017 15:01:19 -0700
Subject: [PATCH] Modify BaseDao, add dataset name length check (#806)

---
 .../java/wherehows/dao/table/BaseDao.java     | 132 +++++++++++-------
 .../java/wherehows/dao/view/BaseViewDao.java  |  80 ++++++-----
 .../processors/MetadataChangeProcessor.java   |   8 ++
 3 files changed, 139 insertions(+), 81 deletions(-)

diff --git a/wherehows-dao/src/main/java/wherehows/dao/table/BaseDao.java b/wherehows-dao/src/main/java/wherehows/dao/table/BaseDao.java
index 1af600f358..b502a350af 100644
--- a/wherehows-dao/src/main/java/wherehows/dao/table/BaseDao.java
+++ b/wherehows-dao/src/main/java/wherehows/dao/table/BaseDao.java
@@ -45,11 +45,15 @@ public class BaseDao {
   @SneakyThrows
   @SuppressWarnings("unchecked")
   public <T> T find(@Nonnull Class<T> entityClass, @Nonnull Object primaryKey) {
-    EntityManager entityManager = entityManagerFactory.createEntityManager();
+    EntityManager entityManager = null;
     try {
-      return (T) entityManager.find(entityClass, primaryKey);
+      entityManager = entityManagerFactory.createEntityManager();
+
+      return entityManager.find(entityClass, primaryKey);
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 
@@ -64,16 +68,21 @@ public class BaseDao {
   @SneakyThrows
   @SuppressWarnings("unchecked")
   public <T> T findBy(@Nonnull Class<T> entityClass, @Nonnull String criteriaKey, @Nonnull Object criteriaValue) {
-    EntityManager entityManager = entityManagerFactory.createEntityManager();
-    CriteriaBuilder cb = entityManager.getCriteriaBuilder();
-    CriteriaQuery<T> criteria = cb.createQuery(entityClass);
-    Root<T> entityRoot = criteria.from(entityClass);
-    criteria.select(entityRoot);
-    criteria.where(cb.equal(entityRoot.get(criteriaKey), criteriaValue));
+    EntityManager entityManager = null;
     try {
+      entityManager = entityManagerFactory.createEntityManager();
+
+      CriteriaBuilder cb = entityManager.getCriteriaBuilder();
+      CriteriaQuery<T> criteria = cb.createQuery(entityClass);
+      Root<T> entityRoot = criteria.from(entityClass);
+      criteria.select(entityRoot);
+      criteria.where(cb.equal(entityRoot.get(criteriaKey), criteriaValue));
+
       return entityManager.createQuery(criteria).getSingleResult();
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 
@@ -89,16 +98,21 @@ public class BaseDao {
   @SuppressWarnings("unchecked")
   public <T> List<T> findListBy(@Nonnull Class<T> entityClass, @Nonnull String criteriaKey,
       @Nonnull Object criteriaValue) {
-    EntityManager entityManager = entityManagerFactory.createEntityManager();
-    CriteriaBuilder cb = entityManager.getCriteriaBuilder();
-    CriteriaQuery<T> criteria = cb.createQuery(entityClass);
-    Root<T> entityRoot = criteria.from(entityClass);
-    criteria.select(entityRoot);
-    criteria.where(cb.equal(entityRoot.get(criteriaKey), criteriaValue));
+    EntityManager entityManager = null;
     try {
+      entityManager = entityManagerFactory.createEntityManager();
+
+      CriteriaBuilder cb = entityManager.getCriteriaBuilder();
+      CriteriaQuery<T> criteria = cb.createQuery(entityClass);
+      Root<T> entityRoot = criteria.from(entityClass);
+      criteria.select(entityRoot);
+      criteria.where(cb.equal(entityRoot.get(criteriaKey), criteriaValue));
+
       return entityManager.createQuery(criteria).getResultList();
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 
@@ -112,23 +126,27 @@ public class BaseDao {
   @SneakyThrows
   @SuppressWarnings("unchecked")
   public <T> List<T> findListBy(@Nonnull Class<T> entityClass, @Nonnull Map<String, ? extends Object> params) {
-    EntityManager entityManager = entityManagerFactory.createEntityManager();
-    CriteriaBuilder cb = entityManager.getCriteriaBuilder();
-    CriteriaQuery<T> criteria = cb.createQuery(entityClass);
-    Root<T> entityRoot = criteria.from(entityClass);
-
-    //Constructing list of parameters
-    List<Predicate> predicates = new ArrayList<Predicate>();
-    for (Map.Entry<String, ? extends Object> entry : params.entrySet()) {
-      predicates.add(cb.equal(entityRoot.get(entry.getKey()), entry.getValue()));
-    }
-
-    criteria.select(entityRoot);
-    criteria.where(predicates.toArray(new Predicate[]{}));
+    EntityManager entityManager = null;
     try {
+      entityManager = entityManagerFactory.createEntityManager();
+
+      CriteriaBuilder cb = entityManager.getCriteriaBuilder();
+      CriteriaQuery<T> criteria = cb.createQuery(entityClass);
+      Root<T> entityRoot = criteria.from(entityClass);
+
+      //Constructing list of parameters
+      List<Predicate> predicates = new ArrayList<Predicate>();
+      for (Map.Entry<String, ? extends Object> entry : params.entrySet()) {
+        predicates.add(cb.equal(entityRoot.get(entry.getKey()), entry.getValue()));
+      }
+
+      criteria.select(entityRoot).where(predicates.toArray(new Predicate[]{}));
+
       return entityManager.createQuery(criteria).getResultList();
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 
@@ -139,14 +157,18 @@ public class BaseDao {
    */
   @SneakyThrows
   public Object update(@Nonnull Object record) {
-    EntityManager entityManager = entityManagerFactory.createEntityManager();
-    entityManager.getTransaction().begin();
+    EntityManager entityManager = null;
     try {
-      record = entityManager.merge(record);
+      entityManager = entityManagerFactory.createEntityManager();
+
+      entityManager.getTransaction().begin();
+      Object managedRecord = entityManager.merge(record);
       entityManager.getTransaction().commit();
-      return record;
+      return managedRecord;
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 
@@ -156,16 +178,20 @@ public class BaseDao {
    */
   @SneakyThrows
   public void updateList(@Nonnull List<? extends Object> records) {
-    EntityManager entityManager = entityManagerFactory.createEntityManager();
-    entityManager.getTransaction().begin();
+    EntityManager entityManager = null;
     try {
+      entityManager = entityManagerFactory.createEntityManager();
+      entityManager.getTransaction().begin();
+
       for (Object record : records) {
         entityManager.merge(record);
         entityManager.flush();
       }
       entityManager.getTransaction().commit();
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 
@@ -175,13 +201,17 @@ public class BaseDao {
    */
   @SneakyThrows
   public void remove(@Nonnull Object record) {
-    EntityManager entityManager = entityManagerFactory.createEntityManager();
-    entityManager.getTransaction().begin();
+    EntityManager entityManager = null;
     try {
+      entityManager = entityManagerFactory.createEntityManager();
+
+      entityManager.getTransaction().begin();
       entityManager.remove(entityManager.contains(record) ? record : entityManager.merge(record));
       entityManager.getTransaction().commit();
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 
@@ -191,16 +221,20 @@ public class BaseDao {
    */
   @SneakyThrows
   public void removeList(@Nonnull List<? extends Object> records) {
-    EntityManager entityManager = entityManagerFactory.createEntityManager();
-    entityManager.getTransaction().begin();
+    EntityManager entityManager = null;
     try {
+      entityManager = entityManagerFactory.createEntityManager();
+      entityManager.getTransaction().begin();
+
       for (Object record : records) {
         entityManager.remove(entityManager.contains(record) ? record : entityManager.merge(record));
         entityManager.flush();
       }
       entityManager.getTransaction().commit();
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 
@@ -211,8 +245,10 @@ public class BaseDao {
    */
   @SneakyThrows
   public void executeUpdate(@Nonnull String queryStr, @Nonnull Map<String, Object> params) {
-    EntityManager entityManager = entityManagerFactory.createEntityManager();
+    EntityManager entityManager = null;
     try {
+      entityManager = entityManagerFactory.createEntityManager();
+
       Query query = entityManager.createQuery(queryStr);
       for (Map.Entry<String, Object> param : params.entrySet()) {
         query.setParameter(param.getKey(), param.getValue());
@@ -220,7 +256,9 @@ public class BaseDao {
 
       query.executeUpdate();
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 }
diff --git a/wherehows-dao/src/main/java/wherehows/dao/view/BaseViewDao.java b/wherehows-dao/src/main/java/wherehows/dao/view/BaseViewDao.java
index baf9d1e2f0..b919661c8a 100644
--- a/wherehows-dao/src/main/java/wherehows/dao/view/BaseViewDao.java
+++ b/wherehows-dao/src/main/java/wherehows/dao/view/BaseViewDao.java
@@ -19,8 +19,6 @@ import javax.persistence.EntityManager;
 import javax.persistence.EntityManagerFactory;
 import javax.persistence.Query;
 import lombok.SneakyThrows;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 
 /**
@@ -29,8 +27,6 @@ import org.slf4j.LoggerFactory;
  */
 public class BaseViewDao {
 
-  private static final Logger log = LoggerFactory.getLogger(BaseViewDao.class);
-
   private final EntityManagerFactory _emFactory;
 
   public BaseViewDao(EntityManagerFactory factory) {
@@ -48,16 +44,20 @@ public class BaseViewDao {
   @SneakyThrows
   @SuppressWarnings("unchecked")
   protected <T> List<T> getEntityListBy(String sqlQuery, Class<T> classType, Map<String, Object> params) {
-    EntityManager entityManager = _emFactory.createEntityManager();
-    Query query = entityManager.createNativeQuery(sqlQuery, classType);
-    for (Map.Entry<String, Object> param : params.entrySet()) {
-      query.setParameter(param.getKey(), param.getValue());
-    }
-
+    EntityManager entityManager = null;
     try {
+      entityManager = _emFactory.createEntityManager();
+
+      Query query = entityManager.createNativeQuery(sqlQuery, classType);
+      for (Map.Entry<String, Object> param : params.entrySet()) {
+        query.setParameter(param.getKey(), param.getValue());
+      }
+
       return (List<T>) query.getResultList();
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 
@@ -72,16 +72,20 @@ public class BaseViewDao {
   @SneakyThrows
   @SuppressWarnings("unchecked")
   protected <T> T getEntityBy(String sqlQuery, Class<T> classType, Map<String, Object> params) {
-    EntityManager entityManager = _emFactory.createEntityManager();
-    Query query = entityManager.createNativeQuery(sqlQuery, classType);
-    for (Map.Entry<String, Object> param : params.entrySet()) {
-      query.setParameter(param.getKey(), param.getValue());
-    }
-
+    EntityManager entityManager = null;
     try {
+      entityManager = _emFactory.createEntityManager();
+
+      Query query = entityManager.createNativeQuery(sqlQuery, classType);
+      for (Map.Entry<String, Object> param : params.entrySet()) {
+        query.setParameter(param.getKey(), param.getValue());
+      }
+
       return (T) query.getSingleResult();
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 
@@ -94,16 +98,20 @@ public class BaseViewDao {
   @SneakyThrows
   @SuppressWarnings("unchecked")
   protected List<Object[]> getObjectArrayListBy(String sqlQuery, Map<String, Object> params) {
-    EntityManager entityManager = _emFactory.createEntityManager();
-    Query query = entityManager.createNativeQuery(sqlQuery);
-    for (Map.Entry<String, Object> param : params.entrySet()) {
-      query.setParameter(param.getKey(), param.getValue());
-    }
-
+    EntityManager entityManager = null;
     try {
-      return query.getResultList();
+      entityManager = _emFactory.createEntityManager();
+
+      Query query = entityManager.createNativeQuery(sqlQuery);
+      for (Map.Entry<String, Object> param : params.entrySet()) {
+        query.setParameter(param.getKey(), param.getValue());
+      }
+
+      return (List<Object[]>) query.getResultList();
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 
@@ -116,16 +124,20 @@ public class BaseViewDao {
   @SneakyThrows
   @SuppressWarnings("unchecked")
   protected List<Object> getObjectListBy(String sqlQuery, Map<String, Object> params) {
-    EntityManager entityManager = _emFactory.createEntityManager();
-    Query query = entityManager.createNativeQuery(sqlQuery);
-    for (Map.Entry<String, Object> param : params.entrySet()) {
-      query.setParameter(param.getKey(), param.getValue());
-    }
-
+    EntityManager entityManager = null;
     try {
-      return query.getResultList();
+      entityManager = _emFactory.createEntityManager();
+
+      Query query = entityManager.createNativeQuery(sqlQuery);
+      for (Map.Entry<String, Object> param : params.entrySet()) {
+        query.setParameter(param.getKey(), param.getValue());
+      }
+
+      return (List<Object>) query.getResultList();
     } finally {
-      entityManager.close();
+      if (entityManager != null) {
+        entityManager.close();
+      }
     }
   }
 }
diff --git a/wherehows-kafka/src/main/java/wherehows/processors/MetadataChangeProcessor.java b/wherehows-kafka/src/main/java/wherehows/processors/MetadataChangeProcessor.java
index e4c02b4429..e6126408b6 100644
--- a/wherehows-kafka/src/main/java/wherehows/processors/MetadataChangeProcessor.java
+++ b/wherehows-kafka/src/main/java/wherehows/processors/MetadataChangeProcessor.java
@@ -42,6 +42,8 @@ public class MetadataChangeProcessor extends KafkaMessageProcessor {
 
   private final DatasetComplianceDao _complianceDao = DAO_FACTORY.getDatasetComplianceDao();
 
+  private final int _maxDatasetNameLength = 500;
+
   public MetadataChangeProcessor(DaoFactory daoFactory, KafkaProducer<String, IndexedRecord> producer) {
     super(daoFactory, producer);
   }
@@ -78,6 +80,12 @@ public class MetadataChangeProcessor extends KafkaMessageProcessor {
       return;
     }
 
+    // check dataset name length to be within limit. Otherwise, save to DB will fail.
+    if (identifier.nativeName.length() > _maxDatasetNameLength) {
+      log.error("Dataset name too long: " + identifier.nativeName.length(), identifier);
+      return;
+    }
+
     // create or update dataset
     DictDataset ds =
         _dictDatasetDao.insertUpdateDataset(identifier, changeAuditStamp, record.datasetProperty, record.schema,