This is an automated email from the ASF dual-hosted git repository.
j1wonpark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 9442f8c0f [AMORO-4393][AMS] Fix shared state between AMS test classes
(#4394)
9442f8c0f is described below
commit 9442f8c0fe86f3199c77c21b730e1d9cf6cfe984
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.
---
.../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);
+ }
}
}
}
