persist decommission type and load check lost cluster property

fe/src/com/baidu/palo/alter/DecommissionBackendJob.java

@@ -159,7 +159,12 @@ public synchronized boolean sendTasks() {
         final Map<String, Integer> clusterAvailBeMap = Maps.newHashMap();
         for (String cluster : clusterBackendsMap.keySet()) {
             final Map<Long, Backend> backends = clusterBackendsMap.get(cluster);
-            int aliveBackendNum = clusterInfo.getClusterBackends(cluster, true).size();
+            final List<Backend> clusterAllLiveBackends = clusterInfo.getClusterBackends(cluster, true);
+            if (clusterAllLiveBackends == null) {
+                LOG.warn("does not belong to any cluster.");
+                return true;
+            }   
+            int aliveBackendNum = clusterAllLiveBackends.size();
             int availableBackendNum = aliveBackendNum - backends.keySet().size();
             if (availableBackendNum <= 0) {
                 // do nothing, just log
@@ -562,6 +567,10 @@ public void readFields(DataInput in) throws IOException {
             }
             clusterBackendsMap.put(SystemInfoService.DEFAULT_CLUSTER, backends);
         }
+
+        if (Catalog.getCurrentCatalogJournalVersion() >= FeMetaVersion.VERSION_33) {
+            decomissionType = DecomissionType.valueOf(Text.readString(in));
+        }
     }
 
     @Override
@@ -577,6 +586,7 @@ public void write(DataOutput out) throws IOException {
                 out.writeLong(id);
             }
         }
+        Text.writeString(out, decomissionType.toString());
     }
 
     public static DecommissionBackendJob read(DataInput in) throws IOException {

fe/src/com/baidu/palo/analysis/LoadStmt.java

@@ -127,6 +127,7 @@ public static void checkProperties(Map<String, String> properties) throws DdlExc
         propertySet.add(LoadStmt.TIMEOUT_PROPERTY);
         propertySet.add(LoadStmt.MAX_FILTER_RATIO_PROPERTY);
         propertySet.add(LoadStmt.LOAD_DELETE_FLAG_PROPERTY);
+        propertySet.add(LoadStmt.CLUSTER_PROPERTY);
 
         for (Entry<String, String> entry : properties.entrySet()) {
             if (!propertySet.contains(entry.getKey())) {

fe/src/com/baidu/palo/common/FeConstants.java

@@ -38,5 +38,5 @@ public class FeConstants {
 
     // general model
     // Current meta data version. Use this version to write journals and image
-    public static int meta_version = FeMetaVersion.VERSION_32;
+    public static int meta_version = FeMetaVersion.VERSION_33;
 }

fe/src/com/baidu/palo/common/FeMetaVersion.java

@@ -62,4 +62,7 @@ public final class FeMetaVersion {
 
     // Palo3.2
     public static final int VERSION_32 = 32;
+
+    // persist decommission type
+    public static final int VERSION_33 = 33;
 }