check

Author: zhangyongxiang.alpha

amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java

@@ -32,6 +32,7 @@
 import org.apache.amoro.process.RecoverProcessFailedException;
 import org.apache.amoro.process.TableProcess;
 import org.apache.amoro.process.TableProcessStore;
+import org.apache.amoro.server.table.DefaultTableRuntime;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
 import org.apache.commons.lang3.tuple.Pair;
@@ -115,7 +116,14 @@ public void open(Map<String, String> properties) {
   }
 
   private Optional<TableProcess> triggerExpireSnapshot(TableRuntime tableRuntime) {
-    if (localEngine == null) {
+    if (localEngine == null || !tableRuntime.getTableConfiguration().isExpireSnapshotEnabled()) {
+      return Optional.empty();
+    }
+
+    long lastExecuteTime =
+        tableRuntime.getState(DefaultTableRuntime.CLEANUP_STATE_KEY).getLastSnapshotsExpiringTime();
+    ProcessTriggerStrategy strategy = actions.get(IcebergActions.EXPIRE_SNAPSHOTS);
+    if (System.currentTimeMillis() - lastExecuteTime < strategy.getTriggerInterval().toMillis()) {
       return Optional.empty();
     }
 

amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/SnapshotsExpiringProcess.java

@@ -28,7 +28,6 @@
 import org.apache.amoro.process.TableProcess;
 import org.apache.amoro.server.optimizing.maintainer.TableMaintainers;
 import org.apache.amoro.server.table.DefaultTableRuntime;
-import org.apache.amoro.server.table.cleanup.CleanupOperation;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -58,10 +57,9 @@ public void run() {
     } catch (Throwable t) {
       LOG.error("unexpected expire error of table {} ", tableRuntime.getTableIdentifier(), t);
     } finally {
-      if (tableRuntime instanceof DefaultTableRuntime) {
-        ((DefaultTableRuntime) tableRuntime)
-            .updateLastCleanTime(CleanupOperation.SNAPSHOTS_EXPIRING, System.currentTimeMillis());
-      }
+      tableRuntime.updateState(
+          DefaultTableRuntime.CLEANUP_STATE_KEY,
+          cleanUp -> cleanUp.setLastSnapshotsExpiringTime(System.currentTimeMillis()));
     }
   }
 

amoro-ams/src/main/java/org/apache/amoro/server/table/AbstractTableRuntime.java

@@ -25,12 +25,14 @@
 import org.apache.amoro.server.persistence.PersistentBase;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
 import org.apache.amoro.shade.zookeeper3.org.apache.curator.shaded.com.google.common.collect.Maps;
+import org.apache.amoro.table.StateKey;
 import org.apache.amoro.table.TableRuntimeStore;
 
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 public abstract class AbstractTableRuntime extends PersistentBase
@@ -62,6 +64,16 @@ public Map<String, String> getTableConfig() {
     return store().getTableConfig();
   }
 
+  @Override
+  public <T> T getState(StateKey<T> key) {
+    return store().getState(key);
+  }
+
+  @Override
+  public <T> void updateState(StateKey<T> key, Function<T, T> updater) {
+    store().begin().updateState(key, updater).commit();
+  }
+
   @Override
   public List<TableProcessStore> getProcessStates() {
     return processContainerMap.values().stream()

amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java

@@ -74,7 +74,7 @@ public class DefaultTableRuntime extends AbstractTableRuntime {
           .jsonType(AbstractOptimizingEvaluator.PendingInput.class)
           .defaultValue(new AbstractOptimizingEvaluator.PendingInput());
 
-  private static final StateKey<TableRuntimeCleanupState> CLEANUP_STATE_KEY =
+  public static final StateKey<TableRuntimeCleanupState> CLEANUP_STATE_KEY =
       StateKey.stateKey("cleanup_state")
           .jsonType(TableRuntimeCleanupState.class)
           .defaultValue(new TableRuntimeCleanupState());

amoro-ams/src/main/java/org/apache/amoro/server/table/cleanup/TableRuntimeCleanupState.java

@@ -52,7 +52,8 @@ public long getLastSnapshotsExpiringTime() {
     return lastSnapshotsExpiringTime;
   }
 
-  public void setLastSnapshotsExpiringTime(long lastSnapshotsExpiringTime) {
+  public TableRuntimeCleanupState setLastSnapshotsExpiringTime(long lastSnapshotsExpiringTime) {
     this.lastSnapshotsExpiringTime = lastSnapshotsExpiringTime;
+    return this;
   }
 }

amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.process.ProcessFactory

@@ -1 +1,19 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
 org.apache.amoro.server.process.iceberg.IcebergProcessFactory

amoro-ams/src/main/resources/META-INF/services/org.apache.amoro.server.process.executor.ExecuteEngine

@@ -22,9 +22,11 @@
 import org.apache.amoro.metrics.MetricRegistry;
 import org.apache.amoro.process.ProcessFactory;
 import org.apache.amoro.process.TableProcessStore;
+import org.apache.amoro.table.StateKey;
 
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
 
 /**
  * TableRuntime is the key interface for the AMS framework to interact with the table. Typically, it
@@ -81,6 +83,24 @@ public interface TableRuntime {
    */
   Map<String, String> getTableConfig();
 
+  /**
+   * Get the value of table-runtime state
+   *
+   * @param key the state key
+   * @return value of the state
+   * @param <T> state value type
+   */
+  <T> T getState(StateKey<T> key);
+
+  /**
+   * Update the state
+   *
+   * @param key key of state
+   * @param updater value updater of state
+   * @param <T> value type of state.
+   */
+  <T> void updateState(StateKey<T> key, Function<T, T> updater);
+
   /**
    * Register the metric of the table runtime.
    *

amoro-common/src/main/java/org/apache/amoro/TableRuntime.java

@@ -0,0 +1,19 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+org.apache.amoro.process.LocalExecutionEngine