Merge branch 'develop' into release/1.6.0_arenadata1

conf/smart-default.xml

@@ -405,6 +405,8 @@
     <value>48</value>
     <description>
       The max number of access count per hour tables in the Metastore.
+      It should be at least 24 to cover the full day to be successfully
+      aggregated into a per day access count table.
     </description>
   </property>
 
@@ -413,6 +415,8 @@
     <value>120</value>
     <description>
       The max number of access count per minute tables in the Metastore.
+      It should be at least 60 to cover the full hour to be successfully
+      aggregated into a per hour access count table.
     </description>
   </property>
 
@@ -421,6 +425,8 @@
     <value>30</value>
     <description>
       The max number of access count per second tables in the Metastore.
+      It should be at least 60000/'smart.access.count.aggregation.interval.ms'
+      to cover the full minute to be successfully aggregated into a per minute access count table.
     </description>
   </property>
 
@@ -471,4 +477,23 @@
       The max time in milliseconds to wait an answer from the SmartAgent master actor during action submission.
     </description>
   </property>
+
+  <property>
+    <name>smart.access.count.aggregation.interval.ms</name>
+    <value>5000</value>
+    <description>
+      The interval in milliseconds that is covered by single second-granularity access count table.
+    </description>
+  </property>
+
+  <property>
+    <name>smart.sync.schedule.strategy</name>
+    <value>UNORDERED</value>
+    <description>
+      Strategy of copying files during 'sync' rule. Possible values:
+        FIFO - the files created/modified first will be scheduled for transfer first
+        LIFO - the files created/modified last will be scheduled for transfer first
+        UNORDERED - no guarantees of the file scheduling order
+    </description>
+  </property>
 </configuration>

smart-action/src/main/java/org/smartdata/action/SmartAction.java

@@ -35,16 +35,16 @@
  * are also meant to extend this.
  */
 public abstract class SmartAction {
-  static final Logger LOG = LoggerFactory.getLogger(SmartAction.class);
+  private static final Logger LOG = LoggerFactory.getLogger(SmartAction.class);
   private long cmdletId;
   private boolean lastAction;
   private long actionId;
   private Map<String, String> actionArgs;
   private SmartContext context;
-  private ByteArrayOutputStream resultOs;
-  private PrintStream psResultOs;
-  private ByteArrayOutputStream logOs;
-  private PrintStream psLogOs;
+  private final ByteArrayOutputStream resultOutputStream;
+  private final PrintStream resultPrintStream;
+  private final ByteArrayOutputStream logOutputStream;
+  private final PrintStream logPrintStream;
   private volatile boolean successful;
   protected String name;
   private long startTime;
@@ -55,10 +55,10 @@ public abstract class SmartAction {
   public SmartAction() {
     this.successful = false;
     //Todo: extract the print stream out of this class
-    this.resultOs = new ByteArrayOutputStream(64 * 1024);
-    this.psResultOs = new PrintStream(resultOs, false);
-    this.logOs = new ByteArrayOutputStream(64 * 1024);
-    this.psLogOs = new PrintStream(logOs, false);
+    this.resultOutputStream = new ByteArrayOutputStream(64 * 1024);
+    this.resultPrintStream = new PrintStream(resultOutputStream, false);
+    this.logOutputStream = new ByteArrayOutputStream(64 * 1024);
+    this.logPrintStream = new PrintStream(logOutputStream, false);
   }
 
   public String getName() {
@@ -155,20 +155,20 @@ private void setFinishTime() {
 
   // The result will be shown in each action's summary page.
   protected void appendResult(String result) {
-    psResultOs.println(result);
+    resultPrintStream.println(result);
   }
 
   // The log will be shown in action's submission section and summary page.
   protected void appendLog(String log) {
-    psLogOs.println(log);
+    logPrintStream.println(log);
   }
 
-  public PrintStream getResultOs() {
-    return psResultOs;
+  public PrintStream getResultOutputStream() {
+    return resultPrintStream;
   }
 
-  public PrintStream getLogOs() {
-    return psLogOs;
+  public PrintStream getLogPrintStream() {
+    return logPrintStream;
   }
 
   public float getProgress() {
@@ -184,17 +184,17 @@ public ActionStatus getActionStatus() throws UnsupportedEncodingException {
         lastAction,
         actionId,
         getProgress(),
-        resultOs.toString("UTF-8"),
-        logOs.toString("UTF-8"),
+        resultOutputStream.toString("UTF-8"),
+        logOutputStream.toString("UTF-8"),
         startTime,
         finishTime,
         throwable,
         finished);
   }
 
   private void stop() {
-    psLogOs.close();
-    psResultOs.close();
+    logPrintStream.close();
+    resultPrintStream.close();
   }
 
   public boolean isSuccessful() {

smart-action/src/main/java/org/smartdata/action/SyncAction.java

@@ -25,13 +25,14 @@
  * dest path, e.g., "sync -src /test/1 -dest hdfs:/remoteIP:port/test/1"
  */
 @ActionSignature(
-    actionId = "sync",
+    actionId = SyncAction.NAME,
     displayName = "sync",
     usage = SyncAction.SRC + " $src "
         + SyncAction.DEST + " $dest "
         + SyncAction.PRESERVE + " $attributes"
 )
 public class SyncAction extends SmartAction {
+  public static final String NAME = "sync";
   // related to fileDiff.src
   public static final String SRC = "-src";
   // related to remote cluster and fileDiff.src

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

@@ -90,17 +90,23 @@ public class SmartConfKeys {
       "smart.metastore.mysql.legacy.enabled";
   public static final boolean SMART_METASTORE_LEGACY_MYSQL_SUPPORT_DEFAULT = false;
 
+  public static final String SMART_ACCESS_COUNT_AGGREGATION_INTERVAL_MS =
+      "smart.access.count.aggregation.interval.ms";
+  public static final int SMART_ACCESS_COUNT_AGGREGATION_INTERVAL_MS_DEFAULT = 5000;
+
   public static final String SMART_NUM_DAY_TABLES_TO_KEEP_KEY =
       "smart.access.count.day.tables.num";
   public static final int SMART_NUM_DAY_TABLES_TO_KEEP_DEFAULT = 30;
 
   public static final String SMART_NUM_HOUR_TABLES_TO_KEEP_KEY =
       "smart.access.count.hour.tables.num";
   public static final int SMART_NUM_HOUR_TABLES_TO_KEEP_DEFAULT = 48;
+  public static final int SMART_NUM_HOUR_TABLES_TO_KEEP_MIN = 24;
 
   public static final String SMART_NUM_MINUTE_TABLES_TO_KEEP_KEY =
       "smart.access.count.minute.tables.num";
   public static final int SMART_NUM_MINUTE_TABLES_TO_KEEP_DEFAULT = 120;
+  public static final int SMART_NUM_MINUTE_TABLES_TO_KEEP_MIN = 60;
 
   public static final String SMART_NUM_SECOND_TABLES_TO_KEEP_KEY =
       "smart.access.count.second.tables.num";
@@ -130,6 +136,9 @@ public class SmartConfKeys {
       "smart.dispatch.cmdlets.extra.num";
   public static final int SMART_DISPATCH_CMDLETS_EXTRA_NUM_DEFAULT = 10;
 
+  public static final String SMART_SYNC_SCHEDULE_STRATEGY_KEY = "smart.sync.schedule.strategy";
+  public static final String SMART_SYNC_SCHEDULE_STRATEGY_DEFAULT = "UNORDERED";
+
   // Keep it only for test
   public static final String SMART_ENABLE_ZEPPELIN_WEB = "smart.zeppelin.web.enable";
   public static final boolean SMART_ENABLE_ZEPPELIN_WEB_DEFAULT = true;

smart-common/src/main/java/org/smartdata/model/CmdletDescriptor.java

@@ -38,9 +38,9 @@ public class CmdletDescriptor {
   public static final String RULE_ID = "-ruleId";
   public static final String HDFS_FILE_PATH = "-file";
 
-  private Map<String, String> actionCommon = new HashMap<>();
-  private List<String> actionNames = new ArrayList<>();
-  private List<Map<String, String>> actionArgs = new ArrayList<>();
+  private final Map<String, String> actionCommon = new HashMap<>();
+  private final List<String> actionNames = new ArrayList<>();
+  private final List<Map<String, String>> actionArgs = new ArrayList<>();
   private String cmdletString = null;
   private long deferIntervalMs = 0L;  // Not persist into DB now
 

smart-common/src/main/java/org/smartdata/model/FileDiff.java

@@ -91,6 +91,10 @@ public void setParameters(Map<String, String>  parameters) {
     this.parameters = parameters;
   }
 
+  public void setParameter(String key, String value) {
+    parameters.put(key, value);
+  }
+
   public String getParametersJsonString() {
     Gson gson = new Gson();
     return gson.toJson(parameters);

smart-common/src/main/java/org/smartdata/model/PathChecker.java

@@ -40,7 +40,7 @@
 public class PathChecker {
   private static final String IGNORED_PATH_TEMPLATES_DELIMITER = ",";
 
-  private final Matcher patternMatcher;
+  private final ThreadLocal<Matcher> patternMatcherThreadLocal;
   private final List<String> coverDirs;
 
   public PathChecker(SmartConf configuration) {
@@ -52,13 +52,15 @@ public PathChecker(List<String> ignoredPathPatterns, List<String> coverDirs) {
     ignoredPathPatterns.forEach(patternBuilder::add);
 
     Pattern pattern = Pattern.compile(patternBuilder.toString());
-    this.patternMatcher = pattern.matcher("");
+    this.patternMatcherThreadLocal =
+        ThreadLocal.withInitial(() -> pattern.matcher(""));
     this.coverDirs = coverDirs;
   }
 
   public boolean isIgnored(String absolutePath) {
-    patternMatcher.reset(absolutePath);
-    return patternMatcher.find();
+    return patternMatcherThreadLocal.get()
+        .reset(absolutePath)
+        .find();
   }
 
   public boolean isCovered(String absolutePath) {

smart-common/src/main/java/org/smartdata/model/RuleState.java

@@ -22,7 +22,7 @@
  */
 public enum RuleState {
   ACTIVE(0),      // functioning
-  DRYRUN(1),      // without execute the rule cmdlets
+  NEW(1),      // without execute the rule cmdlets
   DISABLED(2),    // stop maintain info for the rule
   FINISHED(3),    // for one-shot rule
   DELETED(4);

smart-common/src/main/java/org/smartdata/model/rule/RuleExecutorPlugin.java

@@ -31,7 +31,7 @@ public interface RuleExecutorPlugin {
    * @param ruleInfo
    * @param tResult
    */
-  void onNewRuleExecutor(RuleInfo ruleInfo, TranslateResult tResult);
+  void onNewRuleExecutor(RuleInfo ruleInfo, RuleTranslationResult tResult);
 
   /**
    * Called just before rule executor begin to execute rule.
@@ -40,7 +40,7 @@ public interface RuleExecutorPlugin {
    * @param tResult
    * @return continue this execution if true.
    */
-  boolean preExecution(RuleInfo ruleInfo, TranslateResult tResult);
+  boolean preExecution(RuleInfo ruleInfo, RuleTranslationResult tResult);
 
   /**
    * Called after rule condition checked.
@@ -56,7 +56,7 @@ public interface RuleExecutorPlugin {
    * @param descriptor
    * @return the descriptor that will be used to submit to CmdletManager
    */
-  CmdletDescriptor preSubmitCmdletDescriptor(RuleInfo ruleInfo, TranslateResult tResult,
+  CmdletDescriptor preSubmitCmdletDescriptor(RuleInfo ruleInfo, RuleTranslationResult tResult,
       CmdletDescriptor descriptor);
 
   /**

smart-common/src/main/java/org/smartdata/model/rule/RulePlugin.java

@@ -30,13 +30,13 @@ public interface RulePlugin {
    * @param tr
    * @throws IOException the rule won't be added if exception.
    */
-  void onAddingNewRule(RuleInfo ruleInfo, TranslateResult tr) throws IOException;
+  void onAddingNewRule(RuleInfo ruleInfo, RuleTranslationResult tr) throws IOException;
 
   /**
    * Called when new rule has been added into SSM.
    *
    * @param ruleInfo
    * @param tr
    */
-  void onNewRuleAdded(RuleInfo ruleInfo, TranslateResult tr);
+  void onNewRuleAdded(RuleInfo ruleInfo, RuleTranslationResult tr);
 }

smart-common/src/main/java/org/smartdata/model/rule/RulePluginManager.java

@@ -53,14 +53,14 @@ public static synchronized List<RulePlugin> getPlugins() {
     return copy;
   }
 
-  public static void onAddingNewRule(RuleInfo ruleInfo, TranslateResult tr)
+  public static void onAddingNewRule(RuleInfo ruleInfo, RuleTranslationResult tr)
       throws IOException {
     for (RulePlugin plugin : getPlugins()) {
       plugin.onAddingNewRule(ruleInfo, tr);
     }
   }
 
-  public static void onNewRuleAdded(RuleInfo ruleInfo, TranslateResult tr) {
+  public static void onNewRuleAdded(RuleInfo ruleInfo, RuleTranslationResult tr) {
     for (RulePlugin plugin : getPlugins()) {
       plugin.onNewRuleAdded(ruleInfo, tr);
     }