Apply modifications from test (#1709)

* fix hadoop version issue
* make top hot files num configurable
* fix class loader issue
* avoid checkstorage error for directory

Author: PHILO-HE

conf/smart-default.xml

@@ -159,6 +159,14 @@
     </description>
   </property>
 
+  <property>
+    <name>smart.top.hot.files.num</name>
+    <value>200</value>
+    <description>
+      The number of top hot files displayed on web UI.
+    </description>
+  </property>
+
   <property>
     <name>pd.client.port</name>
     <value>7060</value>

smart-common/src/main/java/org/smartdata/conf/SmartConfKeys.java

@@ -155,6 +155,8 @@ public class SmartConfKeys {
       "smart.storage.info.sampling.intervals";
   public static final String SMART_STORAGE_INFO_SAMPLING_INTERVALS_DEFAULT =
       "60s,60;1hour,60;1day";
+  public static final String SMART_TOP_HOT_FILES_NUM_KEY = "smart.top.hot.files.num";
+  public static final int SMART_TOP_HOT_FILES_NUM_DEFAULT = 200;
 
   public static final String SMART_STATUS_REPORT_PERIOD_KEY = "smart.status.report.period";
   public static final int SMART_STATUS_REPORT_PERIOD_DEFAULT = 10;

smart-engine/src/main/java/org/smartdata/server/engine/StatesManager.java

@@ -162,6 +162,11 @@ public void reportFileAccessEvent(FileAccessEvent event) throws IOException {
   public List<FileAccessInfo> getHotFiles(List<AccessCountTable> tables,
       int topNum) throws IOException {
     try {
+      if (topNum == 0) {
+        topNum = serverContext.getConf().getInt(SmartConfKeys.SMART_TOP_HOT_FILES_NUM_KEY,
+            SmartConfKeys.SMART_TOP_HOT_FILES_NUM_DEFAULT);
+        return serverContext.getMetaStore().getHotFiles(tables, topNum);
+      }
       return serverContext.getMetaStore().getHotFiles(tables, topNum);
     } catch (MetaStoreException e) {
       throw new IOException(e);

smart-hadoop-support/smart-hadoop/src/main/java/org/smartdata/hdfs/CompatibilityHelperLoader.java

@@ -18,15 +18,23 @@
 package org.smartdata.hdfs;
 
 import org.apache.hadoop.util.VersionInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class CompatibilityHelperLoader {
   private static CompatibilityHelper instance;
   private static final String HADOOP_26_HELPER_CLASS = "org.smartdata.hdfs.CompatibilityHelper26";
   private static final String HADOOP_27_HELPER_CLASS = "org.smartdata.hdfs.CompatibilityHelper27";
+  public static final Logger LOG =
+      LoggerFactory.getLogger(CompatibilityHelperLoader.class);
 
   public static CompatibilityHelper getHelper() {
     if (instance == null) {
       String version = VersionInfo.getVersion();
+      if (version == null || version.isEmpty() || version.equalsIgnoreCase("Unknown")) {
+        LOG.error("Cannot get Hadoop version. Use default 2.6.3 version. ");
+        version = "2.6.3";
+      }
       String[] parts = version.split("\\.");
       if (parts.length < 2) {
         throw new RuntimeException("Illegal Hadoop Version: " + version + " (expected A.B.* format)");
@@ -47,7 +55,11 @@ public static CompatibilityHelper getHelper() {
 
   private static CompatibilityHelper create(String classString) {
     try {
-      Class clazz = Thread.currentThread().getContextClassLoader().loadClass(classString);
+      ClassLoader loader = Thread.currentThread().getContextClassLoader();
+      if (loader == null) {
+        loader = ClassLoader.getSystemClassLoader();
+      }
+      Class clazz = loader.loadClass(classString);
       return (CompatibilityHelper) clazz.newInstance();
     } catch (ClassNotFoundException | IllegalAccessException | InstantiationException e) {
       e.printStackTrace();

smart-hadoop-support/smart-hadoop/src/main/java/org/smartdata/hdfs/action/CheckStorageAction.java

@@ -53,6 +53,10 @@ protected void execute() throws Exception {
     if (fileStatus == null) {
       throw new ActionException("File does not exist.");
     }
+    if (fileStatus.isDir()) {
+      appendResult("This is a directory which has no storage result!");
+      return;
+    }
     long length = fileStatus.getLen();
     List<LocatedBlock> locatedBlocks =
         dfsClient.getLocatedBlocks(fileName, 0, length).getLocatedBlocks();

smart-zeppelin/zeppelin-server/src/main/java/org/smartdata/server/rest/ClusterRestApi.java

@@ -77,14 +77,16 @@ public Response cachedFiles() {
     }
   }
 
+  //TODO(Philo): Make topNum settable on web UI.
+  //Currently, topNum=0 means that SSM will use the value configured in smart-default.xml
   @GET
   @Path("/primary/hotfiles")
   public Response hotFiles() {
     try {
       List<AccessCountTable> tables =
           smartEngine.getStatesManager().getTablesInLast(Constants.ONE_HOUR_IN_MILLIS);
       return new JsonResponse<>(Response.Status.OK,
-          smartEngine.getStatesManager().getHotFiles(tables, 20)).build();
+          smartEngine.getStatesManager().getHotFiles(tables, 0)).build();
     } catch (Exception e) {
       logger.error("Exception in ClusterRestApi while listing hot files", e);
       return new JsonResponse<>(Response.Status.INTERNAL_SERVER_ERROR,