This is an automated email from the ASF dual-hosted git repository. czy006 pushed a commit to branch 0.9.x in repository https://gitbox.apache.org/repos/asf/amoro.git
commit 26062c23dd9ca9a560d748d0115b4b2360260d04 Author: Sivakumar Mahalingam <[email protected]> AuthorDate: Tue Sep 29 08:04:03 2026 +0400 [AMORO-4393][AMS] Fix shared state between AMS test classes (#4394) AMS test classes share one AMS instance and Derby database, so state left by one test class broke others depending on execution order. - TestIcebergV3OptimizingPlanning: delete only rows for tables it created instead of truncating the shared database; initialize DerbyPersistence only when run standalone. - TestInternalMixedCatalogService: drop residual tables and test_ns before each test; assert on test_ns presence instead of an empty catalog. - TestInternalIcebergCatalogService: reload through REST and reapply the catalog update if the background catalog scan overwrote it. (cherry picked from commit 9442f8c0fe86f3199c77c21b730e1d9cf6cfe984) --- .../server/TestInternalIcebergCatalogService.java | 26 ++++++--- .../server/TestInternalMixedCatalogService.java | 31 +++++++++-- .../TestIcebergV3OptimizingPlanning.java | 62 ++++++++++++++++++++-- 3 files changed, 102 insertions(+), 17 deletions(-) diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java b/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java index 3ba804d90..1ddae1f79 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java @@ -93,18 +93,28 @@ public class TestInternalIcebergCatalogService extends RestCatalogServiceTestBas CatalogMeta oldMeta = meta.deepCopy(); meta.putToCatalogProperties("cache-enabled", "false"); meta.putToCatalogProperties("cache.expiration-interval-ms", "10000"); - catalogManager.updateCatalog(meta); - // Force a cache reload after invalidation to prevent the background catalog-scan task from - // overwriting the cache with a stale DB snapshot it read before the update completed. - CatalogMeta updatedMeta = catalogManager.getCatalogMeta(catalogName()); - String warehouseInAMS = - updatedMeta.getCatalogProperties().get(CatalogMetaProperties.KEY_WAREHOUSE); Map<String, String> clientSideConfiguration = Maps.newHashMap(); clientSideConfiguration.put("cache-enabled", "true"); - try (RESTCatalog catalog = loadIcebergCatalog(clientSideConfiguration)) { - Map<String, String> finallyConfigs = catalog.properties(); + try { + Map<String, String> finallyConfigs = Maps.newHashMap(); + for (int attempt = 0; attempt < 3; attempt++) { + catalogManager.updateCatalog(meta.deepCopy()); + // A concurrent catalog scan may have read the old metadata before this update and reload + // it afterward. Reload through REST and reapply the update if that race occurred. + try (RESTCatalog catalog = loadIcebergCatalog(clientSideConfiguration)) { + finallyConfigs = catalog.properties(); + } + if ("10000".equals(finallyConfigs.get("cache.expiration-interval-ms"))) { + break; + } + } + String warehouseInAMS = + catalogManager + .getCatalogMeta(catalogName()) + .getCatalogProperties() + .get(CatalogMetaProperties.KEY_WAREHOUSE); // overwrites properties using value from ams Assertions.assertEquals(warehouseInAMS, finallyConfigs.get("warehouse")); // default properties using value from client then properties. diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalMixedCatalogService.java b/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalMixedCatalogService.java index 3e24e42b6..745ae6716 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalMixedCatalogService.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalMixedCatalogService.java @@ -34,6 +34,7 @@ import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; import org.apache.amoro.shade.guava32.com.google.common.collect.Sets; import org.apache.amoro.table.MixedTable; import org.apache.amoro.table.TableBuilder; +import org.apache.amoro.table.TableIdentifier; import org.apache.amoro.table.UnkeyedTable; import org.apache.amoro.utils.CatalogUtil; import org.apache.amoro.utils.MixedTableUtil; @@ -140,6 +141,26 @@ public class TestInternalMixedCatalogService extends RestCatalogServiceTestBase @BeforeEach public void setMixedCatalog() { catalog = loadMixedIcebergCatalog(); + // The AMS instance is shared, so drop anything a previous test left behind. + dropTestDatabaseIfExists(catalog); + } + + private void dropTestDatabaseIfExists(MixedFormatCatalog target) { + try { + if (!target.listDatabases().contains(database)) { + return; + } + for (TableIdentifier identifier : target.listTables(database)) { + try { + target.dropTable(identifier, true); + } catch (Exception e) { + LOG.warn("Failed to drop residual table {}", identifier, e); + } + } + target.dropDatabase(database); + } catch (Exception e) { + LOG.warn("Failed to drop residual database {}", database, e); + } } @Override @@ -165,16 +186,16 @@ public class TestInternalMixedCatalogService extends RestCatalogServiceTestBase MixedFormatCatalog catalog = loadMixedIcebergCatalog(); Assertions.assertEquals( InternalMixedIcebergCatalog.class.getName(), catalog.getClass().getName()); - Assertions.assertTrue(catalog.listDatabases().isEmpty()); + Assertions.assertFalse(catalog.listDatabases().contains(database)); catalog.createDatabase(database); - Assertions.assertEquals(1, catalog.listDatabases().size()); Assertions.assertTrue(catalog.listDatabases().contains(database)); - Assertions.assertEquals(1, nsCatalog.listNamespaces(Namespace.of()).size()); + Assertions.assertTrue( + nsCatalog.listNamespaces(Namespace.of()).contains(Namespace.of(database))); catalog.dropDatabase(database); - Assertions.assertTrue(catalog.listDatabases().isEmpty()); - Assertions.assertTrue(nsCatalog.listNamespaces().isEmpty()); + Assertions.assertFalse(catalog.listDatabases().contains(database)); + Assertions.assertFalse(nsCatalog.listNamespaces().contains(Namespace.of(database))); } } diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestIcebergV3OptimizingPlanning.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestIcebergV3OptimizingPlanning.java index 6bad7f4db..bff12d7fe 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestIcebergV3OptimizingPlanning.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestIcebergV3OptimizingPlanning.java @@ -44,6 +44,7 @@ import org.apache.amoro.process.ProcessStatus; import org.apache.amoro.resource.ResourceGroup; import org.apache.amoro.server.catalog.CatalogManager; import org.apache.amoro.server.persistence.PersistentBase; +import org.apache.amoro.server.persistence.SqlSessionFactoryProvider; import org.apache.amoro.server.persistence.TableRuntimeMeta; import org.apache.amoro.server.persistence.mapper.TableMetaMapper; import org.apache.amoro.server.persistence.mapper.TableProcessMapper; @@ -60,6 +61,7 @@ import org.apache.amoro.table.TableProperties; import org.apache.amoro.table.UnkeyedTable; import org.apache.amoro.utils.SerializationUtil; import org.apache.hadoop.conf.Configuration; +import org.apache.ibatis.session.SqlSession; import org.apache.iceberg.DeleteFile; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -69,23 +71,43 @@ import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; import java.nio.file.Path; +import java.sql.PreparedStatement; +import java.sql.SQLException; import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; +import java.util.Set; import java.util.stream.Collectors; class TestIcebergV3OptimizingPlanning extends PersistentBase { @TempDir private Path temp; + // Preserve standalone execution without loading DerbyPersistence when the shared AMS test + // environment has already initialized persistence. Loading it would truncate the shared DB. + @SuppressWarnings("unused") + private final DerbyPersistence standalonePersistence = initializePersistence(); + private final TestPersistence persistence = new TestPersistence(); private final CatalogManager catalogManager = mock(CatalogManager.class); + private final Set<Long> tableIds = new HashSet<>(); private OptimizingQueue queue; + private static DerbyPersistence initializePersistence() { + try { + SqlSessionFactoryProvider.getInstance().get(); + return null; + } catch (IllegalStateException e) { + return new DerbyPersistence(); + } + } + @AfterEach void cleanUp() { if (queue != null) { queue.dispose(); } - persistence.cleanUp(); + persistence.cleanUp(tableIds); + tableIds.clear(); } @Test @@ -394,6 +416,7 @@ class TestIcebergV3OptimizingPlanning extends PersistentBase { meta.setGroupName("default"); meta.setStatusCode(initialStatus.getCode()); doAs(TableMetaMapper.class, mapper -> mapper.insertTable(identifier)); + tableIds.add(identifier.getId()); meta.setTableId(identifier.getId()); doAs(TableRuntimeMapper.class, mapper -> mapper.insertRuntime(meta)); DefaultTableRuntimeStore store = @@ -437,9 +460,40 @@ class TestIcebergV3OptimizingPlanning extends PersistentBase { }; } - private static class TestPersistence extends DerbyPersistence { - void cleanUp() { - super.after(); + private static class TestPersistence { + private static final String[] TABLES_WITH_TABLE_ID = { + "task_runtime", + "optimizing_task_quota", + "optimizing_process_state", + "table_process_state", + "table_process", + "table_runtime_state", + "table_runtime", + "table_metadata", + "table_identifier" + }; + + void cleanUp(Set<Long> tableIds) { + if (tableIds.isEmpty()) { + return; + } + + try (SqlSession session = SqlSessionFactoryProvider.getInstance().get().openSession(true)) { + for (String table : TABLES_WITH_TABLE_ID) { + try (PreparedStatement statement = + session + .getConnection() + .prepareStatement("DELETE FROM " + table + " WHERE table_id = ?")) { + for (Long tableId : tableIds) { + statement.setLong(1, tableId); + statement.addBatch(); + } + statement.executeBatch(); + } + } + } catch (SQLException e) { + throw new RuntimeException("Failed to clean up optimizing planning test data", e); + } } } }
