[#13176] fix(core): Report unsupported conditional catalog drops and …

…cover the manager path - Keep UnsupportedOperationException unwrapped in dropCatalog so REST reports an unsupported operation instead of an internal error, with an actionable message. - Add CatalogManager tests for the allowlist delete, a schema created after classification, and a store without SupportsConditionalCatalogDelete. - Make the in-memory test store drop allowed schemas with the catalog.
yuqi1129 · Sep 24, 2026 · Sep 24, 2026 · Sep 24, 2026 · Sep 28, 2026 · Sep 30, 2026
commit 520d9ecd70be142a01baa5da70ea9590d060bf86
diff --git a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -1426,8 +1426,17 @@ public boolean dropCatalog(NameIdentifier ident, boolean force)
                   Set<Long> allowedSchemaIds =
                       schemaEntities.stream().map(SchemaEntity::id).collect(Collectors.toSet());
                   if (!(store instanceof SupportsConditionalCatalogDelete)) {
+                    // Fail closed: an unconditional cascade could delete a schema created after
+                    // the classification above.
                     throw new UnsupportedOperationException(
-                        "Atomic catalog delete with allowed schemas is not supported by this store");
+                        String.format(
+                            "Catalog %s still has built-in, imported, or externally removed "
+                                + "schemas, and entity store %s cannot delete it atomically with "
+                                + "them. Use the force option, or use an entity store that "
+                                + "implements %s",
+                            ident,
+                            store.getClass().getName(),
+                            SupportsConditionalCatalogDelete.class.getSimpleName()));
                   }
                   deleted =
                       ((SupportsConditionalCatalogDelete) store)
@@ -1471,7 +1480,9 @@ public boolean dropCatalog(NameIdentifier ident, boolean force)
             // reaches here: it maps a missing entity to false on its own.
             catalogCache.invalidate(ident);
             return false;
-          } catch (GravitinoRuntimeException e) {
+          } catch (GravitinoRuntimeException | UnsupportedOperationException e) {
+            // Keep UnsupportedOperationException unwrapped so REST reports an unsupported
+            // operation instead of an internal error.
             throw e;
           } catch (Exception e) {
             throw new RuntimeException(e);

diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -28,6 +28,7 @@
 import static org.apache.gravitino.TestCatalog.PROPERTY_KEY6_PREFIX;
 import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.eq;
 
 import com.google.common.collect.ImmutableMap;
@@ -74,6 +75,7 @@
 import org.apache.gravitino.exceptions.NoSuchEntityException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
 import org.apache.gravitino.exceptions.NoSuchSchemaException;
+import org.apache.gravitino.exceptions.NonEmptyCatalogException;
 import org.apache.gravitino.lock.LockManager;
 import org.apache.gravitino.lock.LockType;
 import org.apache.gravitino.lock.TreeLockUtils;
@@ -107,6 +109,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.AdditionalAnswers;
 import org.mockito.Mockito;
 import org.mockito.stubbing.Answer;
 
@@ -1412,6 +1415,125 @@ void testDropCatalogInvalidatesCacheAfterStoreDelete() throws Exception {
     Mockito.verify((CatalogDropAware) operations).onCatalogDropped();
   }
 
+  @Test
+  void testNonForceDropCatalogDeletesClassifiedSchemasAtomically() throws Exception {
+    InMemoryEntityStore store = Mockito.spy(newMetalakeStore());
+    SecretManager secretManager = Mockito.spy(new SecretManager(config));
+    try (CatalogManager manager =
+        new CatalogManager(config, store, new RandomIdGenerator(), secretManager)) {
+      NameIdentifier ident = NameIdentifier.of(metalake, "classified_schema_drop");
+      SchemaEntity removedExternally =
+          createDisabledCatalogWithStoreOnlySchema(manager, store, ident);
+
+      Assertions.assertTrue(manager.dropCatalog(ident));
+
+      Mockito.verify(store).deleteCatalogWithAllowedSchemas(ident, Set.of(removedExternally.id()));
+      Assertions.assertFalse(store.exists(ident, EntityType.CATALOG));
+      Assertions.assertFalse(store.exists(removedExternally.nameIdentifier(), EntityType.SCHEMA));
+    } finally {
+      store.close();
+    }
+  }
+
+  @Test
+  void testNonForceDropCatalogRejectsSchemaCreatedAfterClassification() throws Exception {
+    InMemoryEntityStore store = Mockito.spy(newMetalakeStore());
+    SecretManager secretManager = Mockito.spy(new SecretManager(config));
+    try (CatalogManager manager =
+        new CatalogManager(config, store, new RandomIdGenerator(), secretManager)) {
+      NameIdentifier ident = NameIdentifier.of(metalake, "late_schema_drop");
+      SchemaEntity removedExternally =
+          createDisabledCatalogWithStoreOnlySchema(manager, store, ident);
+      SchemaEntity lateSchema = schemaEntity(ident, "late_schema");
+      // Another server creates a schema after the manager classified the existing ones.
+      Mockito.doAnswer(
+              invocation -> {
+                store.put(lateSchema, false);
+                return invocation.callRealMethod();
+              })
+          .when(store)
+          .deleteCatalogWithAllowedSchemas(ident, Set.of(removedExternally.id()));
+
+      Assertions.assertThrows(NonEmptyCatalogException.class, () -> manager.dropCatalog(ident));
+
+      Assertions.assertTrue(store.exists(ident, EntityType.CATALOG));
+      Assertions.assertTrue(store.exists(removedExternally.nameIdentifier(), EntityType.SCHEMA));
+      Assertions.assertTrue(store.exists(lateSchema.nameIdentifier(), EntityType.SCHEMA));
+      Mockito.verify(secretManager, Mockito.never()).deleteSecretsFromProperties(any());
+    } finally {
+      store.close();
+    }
+  }
+
+  @Test
+  void testNonForceDropCatalogFailsClosedWithoutConditionalDeleteStore() throws Exception {
+    InMemoryEntityStore backingStore = newMetalakeStore();
+    // A custom entity store that does not implement SupportsConditionalCatalogDelete.
+    EntityStore store =
+        Mockito.mock(EntityStore.class, AdditionalAnswers.delegatesTo(backingStore));
+    SecretManager secretManager = Mockito.spy(new SecretManager(config));
+    try (CatalogManager manager =
+        new CatalogManager(config, store, new RandomIdGenerator(), secretManager)) {
+      NameIdentifier ident = NameIdentifier.of(metalake, "unsupported_store_drop");
+      SchemaEntity removedExternally =
+          createDisabledCatalogWithStoreOnlySchema(manager, backingStore, ident);
+
+      UnsupportedOperationException e =
+          Assertions.assertThrows(
+              UnsupportedOperationException.class, () -> manager.dropCatalog(ident));
+
+      Assertions.assertTrue(e.getMessage().contains("Use the force option"), e.getMessage());
+      Mockito.verify(store, Mockito.never())
+          .delete(eq(ident), eq(EntityType.CATALOG), anyBoolean());
+      Assertions.assertTrue(backingStore.exists(ident, EntityType.CATALOG));
+      Assertions.assertTrue(
+          backingStore.exists(removedExternally.nameIdentifier(), EntityType.SCHEMA));
+      Mockito.verify(secretManager, Mockito.never()).deleteSecretsFromProperties(any());
+    } finally {
+      backingStore.close();
+    }
+  }
+
+  private static InMemoryEntityStore newMetalakeStore() throws IOException {
+    InMemoryEntityStore store = new InMemoryEntityStore();
+    store.initialize(config);
+    store.put(metalakeEntity, true);
+    return store;
+  }
+
+  /**
+   * Creates a disabled catalog with one schema that exists only in the entity store, so the manager
+   * classifies it as externally removed and lets a non-force drop discard it.
+   */
+  private static SchemaEntity createDisabledCatalogWithStoreOnlySchema(
+      CatalogManager manager, EntityStore store, NameIdentifier ident) throws IOException {
+    Map<String, String> props =
+        ImmutableMap.of(
+            "provider",
+            "test",
+            PROPERTY_KEY1,
+            "value1",
+            PROPERTY_KEY2,
+            "value2",
+            PROPERTY_KEY5_PREFIX + "1",
+            "value3");
+    manager.createCatalog(ident, Catalog.Type.RELATIONAL, provider, "comment", props);
+    manager.disableCatalog(ident);
+    SchemaEntity schema = schemaEntity(ident, "removed_externally");
+    store.put(schema, false);
+    return schema;
+  }
+
+  private static SchemaEntity schemaEntity(NameIdentifier catalogIdent, String name) {
+    return SchemaEntity.builder()
+        .withId(RandomIdGenerator.INSTANCE.nextId())
+        .withName(name)
+        .withNamespace(Namespace.of(metalake, catalogIdent.name()))
+        .withAuditInfo(
+            AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+        .build();
+  }
+
   @Test
   void testDropCatalogReloadsClosedCachedWrapper() throws Exception {
     NameIdentifier ident = NameIdentifier.of("metalake", "closed_cache_drop_test");

diff --git a/core/src/test/java/org/apache/gravitino/storage/memory/TestMemoryEntityStore.java b/core/src/test/java/org/apache/gravitino/storage/memory/TestMemoryEntityStore.java
@@ -191,6 +191,10 @@ public boolean deleteCatalogWithAllowedSchemas(NameIdentifier ident, Set<Long> a
               throw new NonEmptyEntityException(
                   "Entity %s has sub-entities, you should remove sub-entities first", ident);
             }
+            // Mirror the relational store, which removes the allowed schemas with the catalog.
+            for (SchemaEntity schema : schemas) {
+              delete(schema.nameIdentifier(), EntityType.SCHEMA, true);
+            }
             return delete(ident, EntityType.CATALOG, true);
           });
     }