Fix triggerTableAdded idempotency and stale runtime recovery

Author: czy006

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

@@ -781,31 +781,27 @@ private void syncTable(ExternalCatalog externalCatalog, TableIdentity tableIdent
 
   private boolean triggerTableAdded(
       ServerCatalog catalog, ServerTableIdentifier serverTableIdentifier) {
-    // Clean up any pre-existing runtime to prevent metric registration conflicts.
-    // This can happen when a table is deleted from DB (e.g., by dispose after filter change)
-    // but its runtime and metrics remain in memory. We search by table name rather than by ID
-    // because re-syncing a table creates a new DB row with a different ID.
-    tableRuntimeMap
-        .values()
-        .removeIf(
-            existing -> {
-              ServerTableIdentifier existingId = existing.getTableIdentifier();
-              if (existingId.getCatalog().equals(serverTableIdentifier.getCatalog())
-                  && existingId.getDatabase().equals(serverTableIdentifier.getDatabase())
-                  && existingId.getTableName().equals(serverTableIdentifier.getTableName())) {
-                LOG.warn(
-                    "Found existing table runtime for {}, disposing before re-adding.",
-                    serverTableIdentifier);
-                try {
-                  existing.dispose();
-                } catch (Exception e) {
-                  LOG.warn(
-                      "Error disposing existing table runtime for {}", serverTableIdentifier, e);
-                }
-                return true;
-              }
-              return false;
-            });
+    cleanupStaleRuntimeWithSameName(serverTableIdentifier);
+
+    TableRuntime existedRuntime = tableRuntimeMap.get(serverTableIdentifier.getId());
+    if (existedRuntime != null) {
+      TableRuntimeMeta existingRuntimeMeta =
+          getAs(
+              TableRuntimeMapper.class,
+              mapper -> mapper.selectRuntime(serverTableIdentifier.getId()));
+      if (existingRuntimeMeta != null) {
+        LOG.info(
+            "Table runtime already exists for {}, skip duplicate add trigger.",
+            serverTableIdentifier);
+        return true;
+      }
+
+      LOG.warn(
+          "Found stale in-memory runtime for {}, runtime metadata is missing, re-creating.",
+          serverTableIdentifier);
+      disposeRuntimeSafely(existedRuntime, serverTableIdentifier, "while repairing stale runtime");
+      tableRuntimeMap.remove(serverTableIdentifier.getId(), existedRuntime);
+    }
 
     AmoroTable<?> table =
         catalog.loadTable(
@@ -901,6 +897,39 @@ private boolean triggerTableAdded(
     return true;
   }
 
+  private void cleanupStaleRuntimeWithSameName(ServerTableIdentifier serverTableIdentifier) {
+    tableRuntimeMap
+        .values()
+        .removeIf(
+            existing -> {
+              ServerTableIdentifier existingId = existing.getTableIdentifier();
+              if (Objects.equal(existingId.getId(), serverTableIdentifier.getId())) {
+                return false;
+              }
+
+              if (existingId.getCatalog().equals(serverTableIdentifier.getCatalog())
+                  && existingId.getDatabase().equals(serverTableIdentifier.getDatabase())
+                  && existingId.getTableName().equals(serverTableIdentifier.getTableName())) {
+                LOG.warn(
+                    "Found stale runtime {} for table {}, disposing before re-adding.",
+                    existingId,
+                    serverTableIdentifier);
+                disposeRuntimeSafely(existing, existingId, "while cleaning stale runtime");
+                return true;
+              }
+              return false;
+            });
+  }
+
+  private void disposeRuntimeSafely(
+      TableRuntime runtime, ServerTableIdentifier tableIdentifier, String operation) {
+    try {
+      runtime.dispose();
+    } catch (Exception e) {
+      LOG.warn("Error disposing runtime for {} {}", tableIdentifier, operation, e);
+    }
+  }
+
   private Optional<TableRuntime> createTableRuntime(
       ServerTableIdentifier identifier,
       TableRuntimeMeta runtimeMeta,

amoro-ams/src/test/java/org/apache/amoro/server/table/TestSyncTableOfExternalCatalog.java

@@ -62,6 +62,7 @@
 import org.mockito.MockitoAnnotations;
 
 import java.io.IOException;
+import java.lang.reflect.Method;
 import java.util.List;
 import java.util.Map;
 
@@ -500,7 +501,6 @@ public void testTriggerTableAddedIdempotent() {
     ServerTableIdentifier tableIdentifier = serverTableIdentifier();
     MetricRegistry globalRegistry = MetricManager.getInstance().getGlobalRegistry();
 
-    // Verify initial state: 1 runtime, metrics registered
     Assert.assertEquals(1, persistency.getTableRuntimeMetas().size());
     int metricCountBefore = globalRegistry.getMetrics().size();
     Assert.assertTrue(metricCountBefore > 0);
@@ -533,6 +533,56 @@ public void testTriggerTableAddedIdempotent() {
     dropDatabase();
   }
 
+  @Test
+  public void testTriggerTableAddedDuplicateSameTableIdIsIdempotent() {
+    ExternalCatalog externalCatalog = initExternalCatalog();
+    createTable();
+    ServerTableIdentifier tableIdentifier = serverTableIdentifier();
+    MetricRegistry globalRegistry = MetricManager.getInstance().getGlobalRegistry();
+
+    Assert.assertEquals(1, persistency.getTableRuntimeMetas().size());
+    int metricCountBefore = globalRegistry.getMetrics().size();
+    Assert.assertTrue(metricCountBefore > 0);
+
+    boolean added = invokeTriggerTableAdded(externalCatalog, tableIdentifier);
+
+    Assert.assertTrue(added);
+    Assert.assertEquals(1, persistency.getTableRuntimeMetas().size());
+    Assert.assertTrue(tableService().contains(tableIdentifier.getId()));
+    Assert.assertEquals(metricCountBefore, globalRegistry.getMetrics().size());
+
+    dropTable();
+    dropDatabase();
+  }
+
+  @Test
+  public void testTriggerTableAddedRepairStaleRuntimeSameTableId() {
+    ExternalCatalog externalCatalog = initExternalCatalog();
+    createTable();
+    ServerTableIdentifier tableIdentifier = serverTableIdentifier();
+    MetricRegistry globalRegistry = MetricManager.getInstance().getGlobalRegistry();
+
+    int metricCountBefore = globalRegistry.getMetrics().size();
+    Assert.assertTrue(metricCountBefore > 0);
+
+    persistency.deleteTableRuntime(tableIdentifier.getId());
+    Assert.assertEquals(0, persistency.getTableRuntimeMetas().size());
+    Assert.assertTrue(tableService().contains(tableIdentifier.getId()));
+
+    boolean added = invokeTriggerTableAdded(externalCatalog, tableIdentifier);
+
+    Assert.assertTrue(added);
+    Assert.assertEquals(1, persistency.getTableRuntimeMetas().size());
+    Assert.assertTrue(tableService().contains(tableIdentifier.getId()));
+    Assert.assertEquals(metricCountBefore, globalRegistry.getMetrics().size());
+
+    tableService().exploreTableRuntimes();
+    Assert.assertEquals(1, persistency.getTableRuntimeMetas().size());
+
+    dropTable();
+    dropDatabase();
+  }
+
   @Mock private DefaultTableRuntime tableRuntimeWithException;
   @Mock private ServerTableIdentifier tableIdentifierWithException;
 
@@ -659,6 +709,19 @@ private void disposeNewCatalogTable(
     catalogManager().dropCatalog(catalogName);
   }
 
+  private boolean invokeTriggerTableAdded(
+      ServerCatalog catalog, ServerTableIdentifier tableIdentifier) {
+    try {
+      Method triggerTableAddedMethod =
+          DefaultTableService.class.getDeclaredMethod(
+              "triggerTableAdded", ServerCatalog.class, ServerTableIdentifier.class);
+      triggerTableAddedMethod.setAccessible(true);
+      return (boolean) triggerTableAddedMethod.invoke(tableService(), catalog, tableIdentifier);
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   private static class Persistency extends PersistentBase {
     public void addTableIdentifier(
         String catalog, String database, String tableName, TableFormat format) {