[AMS-Refactor] fix runtime bug and add unit test

ams/optimizer/src/main/java/com/netease/arctic/optimizer/LocalOptimizerContainer.java

@@ -58,10 +58,19 @@ protected String buildOptimizerStartupArgsString(Resource resource) {
   public void releaseOptimizer(Resource resource) {
     long jobId = Long.parseLong(PropertyUtil.checkAndGetProperty(resource.getProperties(),
         JOB_ID_PROPERTY));
-    String cmd = "kill -9 " + jobId;
-    Runtime runtime = Runtime.getRuntime();
+
+    String os = System.getProperty("os.name").toLowerCase();
+    String cmd;
+    String[] finalCmd;
+    if (os.contains("win")) {
+      cmd = "taskkill /PID " + jobId + " /F ";
+      finalCmd = new String[] {"cmd", "/c", cmd};
+    } else {
+      cmd = "kill -9 " + jobId;
+      finalCmd = new String[] {"/bin/sh", "-c", cmd};
+    }
     try {
-      String[] finalCmd = {"/bin/sh", "-c", cmd};
+      Runtime runtime = Runtime.getRuntime();
       LOG.info("Stopping optimizer using command:" + cmd);
       runtime.exec(finalCmd);
     } catch (Exception e) {

ams/optimizer/src/main/java/com/netease/arctic/optimizer/OptimizerToucher.java

@@ -16,9 +16,11 @@ public class OptimizerToucher extends AbstractOptimizerOperator {
 
   private OptimizerToucher.TokenChangeListener tokenChangeListener;
   private final Map<String, String> registerProperties = Maps.newHashMap();
+  private long startTime;
 
   public OptimizerToucher(OptimizerConfig config) {
     super(config);
+    this.startTime = System.currentTimeMillis();
   }
 
   public OptimizerToucher withTokenChangeListener(OptimizerToucher.TokenChangeListener tokenChangeListener) {
@@ -57,6 +59,7 @@ private boolean checkToken() {
           registerInfo.setGroupName(getConfig().getGroupName());
           registerInfo.setProperties(registerProperties);
           registerInfo.setResourceId(getConfig().getResourceId());
+          registerInfo.setStartTime(startTime);
           return client.authenticate(registerInfo);
         });
         setToken(token);

ams/server/pom.xml

@@ -35,12 +35,13 @@
 
     <properties>
         <powermock.version>2.0.2</powermock.version>
-        <fastjson.version>1.2.75</fastjson.version>
+        <fastjson.version>1.2.83</fastjson.version>
         <netty.version>4.1.56.Final</netty.version>
         <websocket.version>9.4.46.v20220331</websocket.version>
         <jetty.verison>9.4.46.v20220331</jetty.verison>
         <kotlin.version>1.5.32</kotlin.version>
         <spark.version>3.1.1</spark.version>
+        <arctic.junit.version>5.9.1</arctic.junit.version>
     </properties>
 
     <dependencies>
@@ -204,13 +205,17 @@
                     <groupId>org.apache.thrift</groupId>
                     <artifactId>libthrift</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>org.apache.velocity</groupId>
+                    <artifactId>velocity</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
 
         <dependency>
             <groupId>org.mybatis</groupId>
             <artifactId>mybatis</artifactId>
-            <version>3.5.5</version>
+            <version>3.5.6</version>
         </dependency>
 
         <dependency>
@@ -247,13 +252,6 @@
             <classifier>tests</classifier>
         </dependency>
 
-        <dependency>
-            <groupId>org.powermock</groupId>
-            <artifactId>powermock-module-junit4</artifactId>
-            <version>${powermock.version}</version>
-            <scope>test</scope>
-        </dependency>
-
         <dependency>
             <groupId>org.powermock</groupId>
             <artifactId>powermock-api-mockito2</artifactId>
@@ -399,12 +397,29 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.junit.platform</groupId>
+            <artifactId>junit-platform-launcher</artifactId>
+            <version>1.9.1</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter</artifactId>
-            <version>5.9.1</version>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <version>${arctic.junit.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.junit.vintage</groupId>
+            <artifactId>junit-vintage-engine</artifactId>
+            <version>${arctic.junit.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <version>${arctic.junit.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.javalin</groupId>
             <artifactId>javalin-testtools</artifactId>
@@ -475,6 +490,12 @@
             <artifactId>arctic-hive</artifactId>
             <version>${project.version}</version>
             <scope>compile</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>junit</groupId>
+                    <artifactId>junit</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>
@@ -510,6 +531,10 @@
                     <groupId>com.google.guava</groupId>
                     <artifactId>guava</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>org.junit.jupiter</groupId>
+                    <artifactId>junit-jupiter-api</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
 

ams/server/src/main/java/com/netease/arctic/server/ArcticServiceContainer.java

@@ -20,6 +20,7 @@
 
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
+import com.google.common.annotations.VisibleForTesting;
 import com.netease.arctic.ams.api.ArcticTableMetastore;
 import com.netease.arctic.ams.api.Constants;
 import com.netease.arctic.ams.api.Environments;
@@ -33,7 +34,9 @@
 import com.netease.arctic.server.resource.ContainerMetadata;
 import com.netease.arctic.server.resource.ResourceContainers;
 import com.netease.arctic.server.table.DefaultTableService;
+import com.netease.arctic.server.table.TableService;
 import com.netease.arctic.server.table.executor.AsyncTableExecutors;
+import com.netease.arctic.server.utils.ConfigOption;
 import com.netease.arctic.server.utils.Configurations;
 import org.apache.flink.configuration.IllegalConfigurationException;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -72,7 +75,7 @@ public class ArcticServiceContainer {
   private TServer server;
   private DashboardServer dashboardServer;
 
-  private ArcticServiceContainer() throws Exception {
+  public ArcticServiceContainer() throws Exception {
     initConfig();
     haContainer = new HighAvailabilityContainer(serviceConfig);
   }
@@ -85,6 +88,8 @@ public static void main(String[] args) {
           service.waitLeaderShip();
           service.startService();
           service.waitFollowerShip();
+        } catch (Exception e) {
+          LOG.error("AMS start error", e);
         } finally {
           service.dispose();
         }
@@ -120,6 +125,25 @@ public void dispose() {
     optimizingService = null;
   }
 
+  public boolean isStarted() {
+    return server != null && server.isServing();
+  }
+
+  @VisibleForTesting
+  void setConfig(ConfigOption option, Object value) {
+    serviceConfig.set(option, value);
+  }
+
+  @VisibleForTesting
+  TableService getTableService() {
+    return this.tableService;
+  }
+
+  @VisibleForTesting
+  DefaultOptimizingService getOptimizingService() {
+    return this.optimizingService;
+  }
+
   private void initConfig() throws IllegalConfigurationException, FileNotFoundException {
     LOG.info("initializing configurations...");
     new ConfigurationHelper().init();

ams/server/src/main/java/com/netease/arctic/server/DefaultOptimizingService.java

@@ -160,6 +160,11 @@ public List<OptimizerInstance> listOptimizers(String group) {
     return getQueueByGroup(group).getOptimizers();
   }
 
+  @Override
+  public void deleteOptimizer(String group, String resourceId) {
+    getQueueByGroup(group).removeOptimizer(resourceId);
+  }
+
   private class TableRuntimeHandlerImpl extends TableRuntimeHandler {
 
     @Override

ams/server/src/main/java/com/netease/arctic/server/dashboard/ServerTableDescriptor.java

@@ -162,7 +162,7 @@ public List<DDLInfo> getTableOperations(ServerTableIdentifier tableIdentifier) {
   public List<BaseMajorCompactRecord> getOptimizeInfo(String catalog, String db, String table) {
     List<TableOptimizingProcess> tableOptimizingProcesses = getAs(
         OptimizingMapper.class,
-        mapper -> mapper.selectOptimizingProcessesByTable(catalog, db, table));
+        mapper -> mapper.selectSuccessOptimizingProcesses(catalog, db, table));
     return tableOptimizingProcesses.stream().map(e -> {
       BaseMajorCompactRecord record = new BaseMajorCompactRecord();
       List<TaskRuntime> taskRuntimes = getAs(

ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/CatalogController.java

@@ -316,7 +316,7 @@ public void getCatalogDetail(Context ctx) {
         }
       }
       info.setTableFormatList(Arrays.asList(tableFormat.split(",")));
-      info.setProperties(catalogMeta.getCatalogProperties());
+      info.setProperties(Maps.newHashMap(catalogMeta.getCatalogProperties()));
       info.getProperties().remove(CatalogMetaProperties.TABLE_FORMATS);
       ctx.json(OkResponse.of(info));
       return;

ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/OptimizerController.java

@@ -212,6 +212,7 @@ public void releaseOptimizer(Context ctx) {
       resource.getProperties().putAll(optimizerInstances.get(0).getProperties());
       ResourceContainers.get(resource.getContainerName()).releaseOptimizer(resource);
       optimizerManager.deleteResource(resourceId);
+      optimizerManager.deleteOptimizer(resource.getGroupName(), resourceId);
       ctx.json(OkResponse.of("Success to release optimizer"));
     } catch (Exception e) {
       LOG.error("Failed to release optimizer", e);

ams/server/src/main/java/com/netease/arctic/server/dashboard/model/TableOptimizingProcess.java

@@ -1,5 +1,6 @@
 package com.netease.arctic.server.dashboard.model;
 
+import com.netease.arctic.server.optimizing.MetricsSummary;
 import com.netease.arctic.server.optimizing.OptimizingType;
 
 public class TableOptimizingProcess {
@@ -15,7 +16,7 @@ public class TableOptimizingProcess {
   private long planTime;
   private long endTime;
   private String failReason;
-  private String summary;
+  private MetricsSummary summary;
 
   public TableOptimizingProcess() {
   }
@@ -108,11 +109,11 @@ public void setFailReason(String failReason) {
     this.failReason = failReason;
   }
 
-  public String getSummary() {
+  public MetricsSummary getSummary() {
     return summary;
   }
 
-  public void setSummary(String summary) {
+  public void setSummary(MetricsSummary summary) {
     this.summary = summary;
   }
 }

ams/server/src/main/java/com/netease/arctic/server/dashboard/utils/AmsUtil.java

@@ -161,9 +161,9 @@ public static InetAddress lookForBindHost(String prefix) {
 
   public static String getAMSThriftAddress(Configurations conf) {
     if (conf.getBoolean(HA_ENABLE)) {
-      return "zookeeper://" + conf.getString(HA_ZOOKEEPER_ADDRESS) + "/" + conf.getString(HA_CLUSTER_NAME) + "/";
+      return "zookeeper://" + conf.getString(HA_ZOOKEEPER_ADDRESS) + "/" + conf.getString(HA_CLUSTER_NAME);
     } else {
-      return "thrift://" + conf.getString(SERVER_EXPOSE_HOST) + ":" + conf.getInteger(THRIFT_BIND_PORT) + "/";
+      return "thrift://" + conf.getString(SERVER_EXPOSE_HOST) + ":" + conf.getInteger(THRIFT_BIND_PORT);
     }
   }
 

ams/server/src/main/java/com/netease/arctic/server/optimizing/OptimizingQueue.java

@@ -99,6 +99,10 @@ public List<OptimizerInstance> getOptimizers() {
     return ImmutableList.copyOf(authOptimizers.values());
   }
 
+  public void removeOptimizer(String resourceId) {
+    authOptimizers.values().removeIf(op -> op.getResourceId().equals(resourceId));
+  }
+
   private void clearTasks(TableOptimizingProcess optimizingProcess) {
     retryQueue.removeIf(taskRuntime -> taskRuntime.getProcessId() == optimizingProcess.getProcessId());
     taskQueue.removeIf(taskRuntime -> taskRuntime.getProcessId() == optimizingProcess.getProcessId());
@@ -279,8 +283,8 @@ public TableOptimizingProcess(TableRuntimeMeta tableRuntimeMeta) {
       optimizingType = tableRuntimeMeta.getOptimizingType();
       targetSnapshotId = tableRuntimeMeta.getTargetSnapshotId();
       planTime = tableRuntimeMeta.getPlanTime();
-      metricsSummary = new MetricsSummary(taskMap.values());
       loadTaskRuntimes();
+      metricsSummary = new MetricsSummary(taskMap.values());
       tableRuntimeMeta.constructTableRuntime(this);
     }
 
@@ -331,6 +335,8 @@ public void acceptResult(TaskRuntime taskRuntime) {
             persistProcessCompleted(false);
           }
         }
+      } catch (Exception e) {
+        LOG.error("accept result error:", e);
       } finally {
         tableRuntime.addTaskQuota(taskRuntime.getCurrentQuota());
         lock.unlock();
@@ -454,13 +460,15 @@ private void persistProcessCompleted(boolean success) {
         doAsTransaction(
             () -> taskMap.values().forEach(TaskRuntime::tryCanceling),
             () -> doAs(OptimizingMapper.class, mapper ->
-                mapper.updateOptimizingProcess(tableRuntime.getTableIdentifier().getId(), processId, status, endTime)),
+                mapper.updateOptimizingProcess(tableRuntime.getTableIdentifier().getId(), processId, status, endTime,
+                    new MetricsSummary(taskMap.values()))),
             () -> tableRuntime.completeProcess(true)
         );
       } else {
         doAsTransaction(
             () -> doAs(OptimizingMapper.class, mapper ->
-                mapper.updateOptimizingProcess(tableRuntime.getTableIdentifier().getId(), processId, status, endTime)),
+                mapper.updateOptimizingProcess(tableRuntime.getTableIdentifier().getId(), processId, status, endTime,
+                    new MetricsSummary(taskMap.values()))),
             () -> tableRuntime.completeProcess(true)
         );
       }

ams/server/src/main/java/com/netease/arctic/server/optimizing/TaskRuntime.java

@@ -426,6 +426,7 @@ public TaskQuota(TaskRuntime task) {
       this.processId = task.getTaskId().getProcessId();
       this.taskId = task.getTaskId().getTaskId();
       this.tableId = task.getTableId();
+      this.retryNum = task.getRetry();
     }
 
     public long getStartTime() {

ams/server/src/main/java/com/netease/arctic/server/persistence/mapper/OptimizerMapper.java

@@ -32,11 +32,11 @@
 public interface OptimizerMapper {
 
   @Insert("INSERT INTO optimizer (token, resource_id, group_name, start_time, touch_time, thread_count, total_memory," +
-      " properties) VALUES (#{optimizer.token}, #{optimizer.resourceId},  #{optimizer.groupName}," +
+      " properties) VALUES (#{optimizer.token}, #{optimizer.resourceId, jdbcType=VARCHAR},  #{optimizer.groupName}," +
       " #{optimizer.startTime, typeHandler=com.netease.arctic.server.persistence.converter.Long2TsConverter}," +
       " #{optimizer.touchTime, typeHandler=com.netease.arctic.server.persistence.converter.Long2TsConverter}," +
       " #{optimizer.threadCount}, #{optimizer.memoryMb}," +
-      " #{optimizer.properties, typeHandler=com.netease.arctic.server.persistence.converter.JsonSummaryConverter})")
+      " #{optimizer.properties, typeHandler=com.netease.arctic.server.persistence.converter.Map2StringConverter})")
   void insertOptimizer(@Param("optimizer") OptimizerInstance optimizer);
 
   @Update("UPDATE optimizer SET touch_time = CURRENT_TIMESTAMP WHERE token = #{token}")