This is an automated email from the ASF dual-hosted git repository.
mchades pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 1a0a5e9c2e [Cherry-pick to branch-1.3] [#12794] feat(catalog): Support
connection tests with proposed changes (#12798) (#12884)
1a0a5e9c2e is described below
commit 1a0a5e9c2e9ac002e4fed76aec2d34b3c365c19e
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Sep 4 09:43:35 2026 +0800
[Cherry-pick to branch-1.3] [#12794] feat(catalog): Support connection
tests with proposed changes (#12798) (#12884)
**Cherry-pick Information:**
- Original commit: 2325b412a29e40221d8c3b729542f6d5f4e995be
- Original PR: #12798
- Target branch: `branch-1.3`
- Status: ✅ Conflicts resolved manually
**Resolution:**
- Rebuilt the #12798 change instead of retaining the bot-generated
conflict markers.
- Preserved the no-body behavior for stored-configuration probes and
added temporary rename, comment, set-property, and remove-property
changes without persistence.
- Kept property validation, catalog read locking, temporary-wrapper
cleanup, authorization, REST error handling, and Java/Python client
behavior.
- Omitted the main-only `SecretAlterChanges` and `SecretManager` changes
and secret-specific tests because `branch-1.3` does not contain the
Secret API or subsystem.
**Validation:**
- Compiled the affected API, Core, Server, and Java client production
and test sources.
- Ran Core, Server, and Java client unit tests.
- Ran the targeted Hive Docker integration test covering temporary
failure, non-persistence, and subsequent stored-configuration success.
- Python client: 1018 unit tests passed; Black 26.3.1 and Pylint 4.0.5
passed.
- `./gradlew spotlessApply`
- `./gradlew :docs:build :api:javadoc -PskipITs -PskipDockerTests=true`
- Conflict-marker scan and `git diff --check`
Co-authored-by: mchades <[email protected]>
---
.../org/apache/gravitino/SupportsCatalogs.java | 22 ++++++
.../apache/gravitino/client/GravitinoClient.java | 12 ++++
.../apache/gravitino/client/GravitinoMetalake.java | 40 +++++++++++
.../gravitino/client/TestGravitinoAdminClient.java | 30 ++++++++
.../client/integration/test/CatalogIT.java | 13 ++++
.../gravitino/client/gravitino_client.py | 6 +-
.../gravitino/client/gravitino_metalake.py | 17 ++++-
.../client-python/tests/unittests/test_metalake.py | 21 ++++++
.../apache/gravitino/catalog/CatalogManager.java | 71 +++++++++++++++++++
.../catalog/CatalogNormalizeDispatcher.java | 13 ++++
.../apache/gravitino/catalog/SupportsCatalogs.java | 10 +++
.../gravitino/hook/CatalogHookDispatcher.java | 5 ++
.../gravitino/listener/CatalogEventDispatcher.java | 6 ++
.../gravitino/catalog/TestCatalogManager.java | 81 ++++++++++++++++++++++
.../catalog/TestCatalogNormalizeDispatcher.java | 14 ++++
docs/open-api/catalogs.yaml | 20 ++++--
.../server/web/rest/CatalogOperations.java | 14 +++-
.../server/web/rest/TestCatalogOperations.java | 42 +++++++++++
18 files changed, 423 insertions(+), 14 deletions(-)
diff --git a/api/src/main/java/org/apache/gravitino/SupportsCatalogs.java
b/api/src/main/java/org/apache/gravitino/SupportsCatalogs.java
index 650d60fdbb..1009a1f332 100644
--- a/api/src/main/java/org/apache/gravitino/SupportsCatalogs.java
+++ b/api/src/main/java/org/apache/gravitino/SupportsCatalogs.java
@@ -226,4 +226,26 @@ public interface SupportsCatalogs {
throw new UnsupportedOperationException(
String.format("Catalog %s does not support connection testing",
catalogName));
}
+
+ /**
+ * Test the connection of an existing catalog with proposed changes without
persisting them.
+ *
+ * <p>The default implementation preserves the existing connection-test
behavior when no changes
+ * are supplied and rejects non-empty changes. Implementations that support
testing proposed
+ * changes should override this method.
+ *
+ * @param catalogName the name of the existing catalog.
+ * @param changes the proposed changes to apply temporarily.
+ * @throws NoSuchCatalogException if the catalog does not exist.
+ * @throws Exception if the test failed.
+ */
+ default void testConnection(String catalogName, CatalogChange... changes)
throws Exception {
+ if (changes.length == 0) {
+ testConnection(catalogName);
+ return;
+ }
+ throw new UnsupportedOperationException(
+ String.format(
+ "Catalog %s does not support connection testing with proposed
changes", catalogName));
+ }
}
diff --git
a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java
b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java
index 25837678e4..24ca818415 100644
---
a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java
+++
b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java
@@ -547,6 +547,18 @@ public class GravitinoClient extends GravitinoClientBase
getMetalake().testConnection(catalogName);
}
+ /**
+ * Test the connection of an existing catalog with proposed changes without
persisting them.
+ *
+ * @param catalogName the name of the existing catalog.
+ * @param changes the proposed changes to apply temporarily.
+ * @throws Exception if the test failed.
+ */
+ @Override
+ public void testConnection(String catalogName, CatalogChange... changes)
throws Exception {
+ getMetalake().testConnection(catalogName, changes);
+ }
+
@Override
public String[] listTags() throws NoSuchMetalakeException {
return getMetalake().listTags();
diff --git
a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java
b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java
index 70b15c329c..5887abbedf 100644
---
a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java
+++
b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java
@@ -448,6 +448,46 @@ public class GravitinoMetalake extends MetalakeDTO
ErrorHandlers.catalogErrorHandler().accept(resp);
}
+ /**
+ * Test the connection of an existing catalog with proposed changes without
persisting them.
+ *
+ * @param catalogName the name of the existing catalog.
+ * @param changes the proposed changes to apply temporarily.
+ * @throws Exception if the test failed.
+ */
+ @Override
+ public void testConnection(String catalogName, CatalogChange... changes)
throws Exception {
+ Preconditions.checkArgument(changes != null, "changes must not be null");
+ if (changes.length == 0) {
+ testConnection(catalogName);
+ return;
+ }
+
+ List<CatalogUpdateRequest> requests =
+ Arrays.stream(changes)
+ .map(DTOConverters::toCatalogUpdateRequest)
+ .collect(Collectors.toList());
+ CatalogUpdatesRequest updatesRequest = new CatalogUpdatesRequest(requests);
+ updatesRequest.validate();
+
+ ErrorResponse resp =
+ restClient.post(
+ String.format(
+ API_METALAKES_CATALOGS_PATH + "/testConnection",
+ RESTUtils.encodeString(this.name()),
+ RESTUtils.encodeString(catalogName)),
+ updatesRequest,
+ ErrorResponse.class,
+ Collections.emptyMap(),
+ ErrorHandlers.catalogErrorHandler());
+
+ if (resp.getCode() == 0) {
+ return;
+ }
+
+ ErrorHandlers.catalogErrorHandler().accept(resp);
+ }
+
@Override
public SupportsRoles supportsRoles() {
return this;
diff --git
a/clients/client-java/src/test/java/org/apache/gravitino/client/TestGravitinoAdminClient.java
b/clients/client-java/src/test/java/org/apache/gravitino/client/TestGravitinoAdminClient.java
index 0bcf28d0d0..4073bca3be 100644
---
a/clients/client-java/src/test/java/org/apache/gravitino/client/TestGravitinoAdminClient.java
+++
b/clients/client-java/src/test/java/org/apache/gravitino/client/TestGravitinoAdminClient.java
@@ -25,11 +25,14 @@ import java.util.Collections;
import java.util.Map;
import java.util.stream.Collectors;
import org.apache.gravitino.Catalog;
+import org.apache.gravitino.CatalogChange;
import org.apache.gravitino.MetalakeChange;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.dto.AuditDTO;
import org.apache.gravitino.dto.MetalakeDTO;
import org.apache.gravitino.dto.requests.CatalogCreateRequest;
+import org.apache.gravitino.dto.requests.CatalogUpdateRequest;
+import org.apache.gravitino.dto.requests.CatalogUpdatesRequest;
import org.apache.gravitino.dto.requests.MetalakeCreateRequest;
import org.apache.gravitino.dto.requests.MetalakeUpdatesRequest;
import org.apache.gravitino.dto.responses.BaseResponse;
@@ -324,6 +327,33 @@ public class TestGravitinoAdminClient extends TestBase {
HttpStatus.SC_OK);
Assertions.assertDoesNotThrow(() -> metaLake.testConnection("catalog"));
+ buildMockResource(
+ Method.POST,
+ "/api/metalakes/mock/catalogs/catalog/testConnection",
+ null,
+ new BaseResponse(),
+ HttpStatus.SC_OK);
+ Assertions.assertDoesNotThrow(() -> metaLake.testConnection("catalog", new
CatalogChange[0]));
+
+ exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> metaLake.testConnection("catalog", (CatalogChange[]) null));
+ Assertions.assertTrue(exception.getMessage().contains("changes must not be
null"));
+
+ CatalogUpdatesRequest updatesRequest =
+ new CatalogUpdatesRequest(
+ Collections.singletonList(
+ new CatalogUpdateRequest.SetCatalogPropertyRequest("key",
"value")));
+ buildMockResource(
+ Method.POST,
+ "/api/metalakes/mock/catalogs/catalog/testConnection",
+ updatesRequest,
+ new BaseResponse(),
+ HttpStatus.SC_OK);
+ Assertions.assertDoesNotThrow(
+ () -> metaLake.testConnection("catalog",
CatalogChange.setProperty("key", "value")));
+
buildMockResource(
Method.POST,
"/api/metalakes/mock/catalogs/catalog/testConnection",
diff --git
a/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/CatalogIT.java
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/CatalogIT.java
index f3df2f4c9d..21e46b1ebe 100644
---
a/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/CatalogIT.java
+++
b/clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/CatalogIT.java
@@ -40,6 +40,7 @@ import org.apache.gravitino.client.GravitinoMetalake;
import org.apache.gravitino.exceptions.CatalogAlreadyExistsException;
import org.apache.gravitino.exceptions.CatalogInUseException;
import org.apache.gravitino.exceptions.CatalogNotInUseException;
+import org.apache.gravitino.exceptions.ConnectionFailedException;
import org.apache.gravitino.file.Fileset;
import org.apache.gravitino.file.FilesetCatalog;
import org.apache.gravitino.file.FilesetChange;
@@ -125,6 +126,18 @@ public class CatalogIT extends BaseIT {
Assertions.assertTrue(catalog.properties().containsKey("metastore.uris"));
Assertions.assertDoesNotThrow(() -> metalake.testConnection(catalogName));
+ Assertions.assertThrows(
+ ConnectionFailedException.class,
+ () ->
+ metalake.testConnection(
+ catalogName,
+ CatalogChange.updateComment("temporary comment"),
+ CatalogChange.setProperty("metastore.uris",
"thrift://127.0.0.1:1")));
+ Catalog unchangedCatalog = metalake.loadCatalog(catalogName);
+ Assertions.assertEquals("catalog comment", unchangedCatalog.comment());
+ Assertions.assertEquals(hmsUri,
unchangedCatalog.properties().get("metastore.uris"));
+ Assertions.assertDoesNotThrow(() -> metalake.testConnection(catalogName));
+
metalake.dropCatalog(catalogName, true);
}
diff --git a/clients/client-python/gravitino/client/gravitino_client.py
b/clients/client-python/gravitino/client/gravitino_client.py
index 55b384939b..58cc9251d2 100644
--- a/clients/client-python/gravitino/client/gravitino_client.py
+++ b/clients/client-python/gravitino/client/gravitino_client.py
@@ -120,9 +120,9 @@ class GravitinoClient(GravitinoClientBase, SupportsJobs,
TagOperations):
def disable_catalog(self, name: str):
return self.get_metalake().disable_catalog(name)
- def test_connection(self, name: str) -> None:
- """Test an existing catalog connection using its stored
configuration."""
- self.get_metalake().test_connection(name)
+ def test_connection(self, name: str, *changes: CatalogChange) -> None:
+ """Test an existing catalog connection with optional proposed
changes."""
+ self.get_metalake().test_connection(name, *changes)
def list_job_templates(self) -> List[JobTemplate]:
"""Lists all job templates in the current metalake.
diff --git a/clients/client-python/gravitino/client/gravitino_metalake.py
b/clients/client-python/gravitino/client/gravitino_metalake.py
index 3dd2d30462..bdd557e3b3 100644
--- a/clients/client-python/gravitino/client/gravitino_metalake.py
+++ b/clients/client-python/gravitino/client/gravitino_metalake.py
@@ -353,11 +353,12 @@ class GravitinoMetalake(
url, json=catalog_disable_request,
error_handler=CATALOG_ERROR_HANDLER
)
- def test_connection(self, name: str) -> None:
- """Test an existing catalog connection using its stored configuration.
+ def test_connection(self, name: str, *changes: CatalogChange) -> None:
+ """Test an existing catalog connection with optional proposed changes.
Args:
name: The name of the existing catalog.
+ changes: Proposed catalog changes to apply temporarily without
persisting.
Raises:
NoSuchCatalogException: If the catalog does not exist.
@@ -370,7 +371,17 @@ class GravitinoMetalake(
)
+ "/testConnection"
)
- response = self.rest_client.post(url,
error_handler=CATALOG_ERROR_HANDLER)
+ if changes:
+ requests = [
+ DTOConverters.to_catalog_update_request(change) for change in
changes
+ ]
+ updates_request = CatalogUpdatesRequest(requests)
+ updates_request.validate()
+ response = self.rest_client.post(
+ url, json=updates_request, error_handler=CATALOG_ERROR_HANDLER
+ )
+ else:
+ response = self.rest_client.post(url,
error_handler=CATALOG_ERROR_HANDLER)
base_response = BaseResponse.from_json(response.body,
infer_missing=True)
base_response.validate()
if base_response.code() == 0:
diff --git a/clients/client-python/tests/unittests/test_metalake.py
b/clients/client-python/tests/unittests/test_metalake.py
index 674539389c..1a00985b13 100644
--- a/clients/client-python/tests/unittests/test_metalake.py
+++ b/clients/client-python/tests/unittests/test_metalake.py
@@ -18,9 +18,12 @@
import unittest
from unittest.mock import MagicMock
+from gravitino.api.catalog_change import CatalogChange
from gravitino.client.gravitino_metalake import GravitinoMetalake
from gravitino.constants.error import ErrorConstants
from gravitino.dto.metalake_dto import MetalakeDTO
+from gravitino.dto.requests.catalog_update_request import CatalogUpdateRequest
+from gravitino.dto.requests.catalog_updates_request import
CatalogUpdatesRequest
from gravitino.dto.responses.metalake_response import MetalakeResponse
from gravitino.exceptions.base import (
ConnectionFailedException,
@@ -44,6 +47,24 @@ class TestMetalake(unittest.TestCase):
error_handler=CATALOG_ERROR_HANDLER,
)
+ def test_existing_catalog_connection_with_changes(self):
+ rest_client = MagicMock()
+ rest_client.post.return_value.body = b'{"code":0}'
+ metalake = GravitinoMetalake(
+ MetalakeDTO("metalake", None, {}, None), rest_client
+ )
+
+ metalake.test_connection("catalog", CatalogChange.set_property("key",
"value"))
+
+ expected_request = CatalogUpdatesRequest(
+ [CatalogUpdateRequest.SetCatalogPropertyRequest("key", "value")]
+ )
+ rest_client.post.assert_called_once_with(
+ "api/metalakes/metalake/catalogs/catalog/testConnection",
+ json=expected_request,
+ error_handler=CATALOG_ERROR_HANDLER,
+ )
+
def test_existing_catalog_connection_failure(self):
rest_client = MagicMock()
rest_client.post.return_value.body = (
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
index a974dfb486..d7e2739f07 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -702,6 +702,77 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
});
}
+ /**
+ * Test the connection of an existing catalog with proposed changes without
persisting them.
+ *
+ * @param ident The identifier of the existing catalog.
+ * @param changes The proposed changes to apply temporarily.
+ */
+ @Override
+ public void testConnection(NameIdentifier ident, CatalogChange... changes) {
+ Preconditions.checkArgument(changes != null, "changes must not be null");
+ if (changes.length == 0) {
+ testConnection(ident);
+ return;
+ }
+
+ TreeLockUtils.doWithTreeLock(
+ ident,
+ LockType.READ,
+ () -> {
+ CatalogWrapper storedWrapper = loadCatalogAndWrap(ident);
+ BaseCatalog<?> storedCatalog = storedWrapper.catalog();
+ storedCatalog.checkMetalakeAndCatalogInUse();
+ try {
+ storedWrapper.doWithPropertiesMeta(
+ metadata -> {
+ Pair<Map<String, String>, Map<String, String>> alterProperty
=
+ getCatalogAlterProperty(changes);
+ validatePropertyForAlter(
+ metadata.catalogPropertiesMetadata(),
+ alterProperty.getLeft(),
+ alterProperty.getRight());
+ return null;
+ });
+
+ CatalogEntity storedEntity = storedCatalog.entity();
+ Map<String, String> effectiveProperties =
+ storedEntity.getProperties() == null
+ ? new HashMap<>()
+ : new HashMap<>(storedEntity.getProperties());
+ CatalogEntity effectiveEntity =
+ updateEntity(
+ newCatalogBuilder(storedEntity.namespace(),
storedEntity),
+ effectiveProperties,
+ changes)
+ .build();
+ effectiveEntity = convertFilesetCatalogEntity(effectiveEntity);
+
+ CatalogWrapper temporaryWrapper =
createCatalogWrapper(effectiveEntity, null);
+ try {
+ NameIdentifier effectiveIdent = effectiveEntity.nameIdentifier();
+ temporaryWrapper.doWithCatalogOps(
+ operations -> {
+ operations.testConnection(effectiveIdent);
+ return null;
+ });
+ } finally {
+ temporaryWrapper.close();
+ }
+ } catch (UnsupportedOperationException e) {
+ throw e;
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to test existing catalog connection {} with proposed
changes", ident, e);
+ if (e instanceof RuntimeException) {
+ throw (RuntimeException) e;
+ }
+ throw new RuntimeException(e);
+ }
+ return null;
+ });
+ }
+
@Override
public void enableCatalog(NameIdentifier ident)
throws NoSuchCatalogException, CatalogNotInUseException {
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogNormalizeDispatcher.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogNormalizeDispatcher.java
index 85ba013ba2..14f7db422f 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/CatalogNormalizeDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/CatalogNormalizeDispatcher.java
@@ -130,6 +130,19 @@ public class CatalogNormalizeDispatcher implements
CatalogDispatcher {
dispatcher.testConnection(ident);
}
+ @Override
+ public void testConnection(NameIdentifier ident, CatalogChange... changes)
throws Exception {
+ validateCatalogName(ident.name());
+ Arrays.stream(changes)
+ .forEach(
+ change -> {
+ if (change instanceof CatalogChange.RenameCatalog) {
+ validateCatalogName(((CatalogChange.RenameCatalog)
change).getNewName());
+ }
+ });
+ dispatcher.testConnection(ident, changes);
+ }
+
@Override
public void enableCatalog(NameIdentifier ident) throws
NoSuchCatalogException {
dispatcher.enableCatalog(ident);
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/SupportsCatalogs.java
b/core/src/main/java/org/apache/gravitino/catalog/SupportsCatalogs.java
index a6f6ccb11e..84cd066792 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/SupportsCatalogs.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/SupportsCatalogs.java
@@ -185,6 +185,16 @@ public interface SupportsCatalogs {
*/
void testConnection(NameIdentifier ident) throws Exception;
+ /**
+ * Test the connection of an existing catalog with proposed changes without
persisting them.
+ *
+ * @param ident The identifier of the existing catalog.
+ * @param changes The proposed changes to apply temporarily.
+ * @throws NoSuchCatalogException If the catalog does not exist.
+ * @throws Exception If the connection test fails.
+ */
+ void testConnection(NameIdentifier ident, CatalogChange... changes) throws
Exception;
+
/**
* Enable a catalog. If the catalog is already enabled, this method does
nothing.
*
diff --git
a/core/src/main/java/org/apache/gravitino/hook/CatalogHookDispatcher.java
b/core/src/main/java/org/apache/gravitino/hook/CatalogHookDispatcher.java
index 577c71dfd8..5852b18cb1 100644
--- a/core/src/main/java/org/apache/gravitino/hook/CatalogHookDispatcher.java
+++ b/core/src/main/java/org/apache/gravitino/hook/CatalogHookDispatcher.java
@@ -172,6 +172,11 @@ public class CatalogHookDispatcher implements
CatalogDispatcher {
dispatcher.testConnection(ident);
}
+ @Override
+ public void testConnection(NameIdentifier ident, CatalogChange... changes)
throws Exception {
+ dispatcher.testConnection(ident, changes);
+ }
+
@Override
public void enableCatalog(NameIdentifier ident)
throws NoSuchCatalogException, CatalogNotInUseException {
diff --git
a/core/src/main/java/org/apache/gravitino/listener/CatalogEventDispatcher.java
b/core/src/main/java/org/apache/gravitino/listener/CatalogEventDispatcher.java
index bd738380b7..cc64bdf4b7 100644
---
a/core/src/main/java/org/apache/gravitino/listener/CatalogEventDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/listener/CatalogEventDispatcher.java
@@ -209,6 +209,12 @@ public class CatalogEventDispatcher implements
CatalogDispatcher {
dispatcher.testConnection(ident);
}
+ @Override
+ public void testConnection(NameIdentifier ident, CatalogChange... changes)
throws Exception {
+ // TODO(#12566): Support event dispatching for testConnection
+ dispatcher.testConnection(ident, changes);
+ }
+
@Override
public void enableCatalog(NameIdentifier ident)
throws NoSuchCatalogException, CatalogNotInUseException {
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
index f54722c330..4424dc848f 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -57,6 +57,7 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
import org.apache.gravitino.connector.BaseCatalog;
+import org.apache.gravitino.connector.CatalogOperations;
import org.apache.gravitino.connector.capability.Capability;
import org.apache.gravitino.connector.capability.CapabilityResult;
import org.apache.gravitino.exceptions.CatalogAlreadyExistsException;
@@ -80,6 +81,7 @@ import
org.apache.gravitino.storage.relational.SupportsEntityChangeLog;
import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
import org.apache.gravitino.storage.relational.po.cache.OperateType;
import org.apache.gravitino.utils.PrincipalUtils;
+import org.apache.gravitino.utils.ThrowableFunction;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
@@ -1284,6 +1286,85 @@ public class TestCatalogManager {
}
}
+ @Test
+ void testExistingCatalogConnectionWithProposedChanges() throws Exception {
+ NameIdentifier ident = NameIdentifier.of("metalake",
"connection_changes_test");
+ Map<String, String> properties =
+ ImmutableMap.<String, String>builder()
+ .put("provider", "test")
+ .put(PROPERTY_KEY1, "value1")
+ .put(PROPERTY_KEY2, "value2")
+ .put(PROPERTY_KEY5_PREFIX + "1", "value3")
+ .put("removable", "stored")
+ .build();
+ catalogManager.createCatalog(
+ ident, Catalog.Type.RELATIONAL, provider, "stored comment",
properties);
+ CatalogEntity storedBefore = entityStore.get(ident, EntityType.CATALOG,
CatalogEntity.class);
+ CatalogManager.CatalogWrapper cachedBefore =
+ catalogManager.getCatalogCache().getIfPresent(ident);
+
+ CatalogManager.CatalogWrapper temporaryWrapper =
+ Mockito.mock(CatalogManager.CatalogWrapper.class);
+ CatalogOperations temporaryOperations =
Mockito.mock(CatalogOperations.class);
+ AtomicReference<CatalogEntity> effectiveEntity = new AtomicReference<>();
+ Mockito.doAnswer(
+ invocation -> {
+ effectiveEntity.set(invocation.getArgument(0));
+ return temporaryWrapper;
+ })
+ .when(catalogManager)
+ .createCatalogWrapper(any(CatalogEntity.class), eq(null));
+ Mockito.doAnswer(
+ invocation -> {
+ ThrowableFunction<CatalogOperations, Object> operation =
invocation.getArgument(0);
+ return operation.apply(temporaryOperations);
+ })
+ .when(temporaryWrapper)
+ .doWithCatalogOps(any());
+
+ NameIdentifier renamedIdent = NameIdentifier.of("metalake",
"connection_changes_renamed");
+ try {
+ catalogManager.testConnection(
+ ident,
+ CatalogChange.rename(renamedIdent.name()),
+ CatalogChange.updateComment("temporary comment"),
+ CatalogChange.setProperty(PROPERTY_KEY2, "temporary value"),
+ CatalogChange.removeProperty("removable"));
+
+ CatalogEntity effective = effectiveEntity.get();
+ Assertions.assertNotNull(effective);
+ Assertions.assertEquals(renamedIdent.name(), effective.name());
+ Assertions.assertEquals("temporary comment", effective.getComment());
+ Assertions.assertEquals("temporary value",
effective.getProperties().get(PROPERTY_KEY2));
+
Assertions.assertFalse(effective.getProperties().containsKey("removable"));
+ Mockito.verify(temporaryOperations).testConnection(renamedIdent);
+ Mockito.verify(temporaryWrapper).close();
+
+ CatalogEntity storedAfter = entityStore.get(ident, EntityType.CATALOG,
CatalogEntity.class);
+ Assertions.assertEquals(storedBefore.name(), storedAfter.name());
+ Assertions.assertEquals(storedBefore.getComment(),
storedAfter.getComment());
+ Assertions.assertEquals(storedBefore.getProperties(),
storedAfter.getProperties());
+ Assertions.assertFalse(entityStore.exists(renamedIdent,
EntityType.CATALOG));
+ Assertions.assertSame(cachedBefore,
catalogManager.getCatalogCache().getIfPresent(ident));
+
+ Mockito.doThrow(new IOException("probe failed"))
+ .when(temporaryOperations)
+ .testConnection(any(NameIdentifier.class));
+ RuntimeException failure =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () ->
+ catalogManager.testConnection(
+ ident, CatalogChange.setProperty(PROPERTY_KEY2, "another
value")));
+ Assertions.assertInstanceOf(IOException.class, failure.getCause());
+ Mockito.verify(temporaryWrapper, Mockito.times(2)).close();
+ } finally {
+ Mockito.doCallRealMethod()
+ .when(catalogManager)
+ .createCatalogWrapper(any(CatalogEntity.class), eq(null));
+ }
+ }
+
@Test
public void testCatalogCacheRemoveListener() throws IOException {
NameIdentifier ident = NameIdentifier.of(metalake, "catalog");
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogNormalizeDispatcher.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogNormalizeDispatcher.java
index 9babbe7ce6..9a0810b7db 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogNormalizeDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogNormalizeDispatcher.java
@@ -28,6 +28,7 @@ import java.io.IOException;
import java.time.Instant;
import java.util.Map;
import org.apache.gravitino.Catalog;
+import org.apache.gravitino.CatalogChange;
import org.apache.gravitino.Config;
import org.apache.gravitino.Configs;
import org.apache.gravitino.EntityStore;
@@ -169,4 +170,17 @@ public class TestCatalogNormalizeDispatcher {
"The catalog name '" + illegalName + "' is illegal.",
exception.getMessage());
}
}
+
+ @Test
+ void testConnectionChangesValidateRenamedCatalogName() {
+ NameIdentifier catalogIdent = NameIdentifier.of(metalake, "catalog");
+ Mockito.clearInvocations(catalogManager);
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ catalogNormalizeDispatcher.testConnection(
+ catalogIdent, CatalogChange.rename("invalid/name")));
+ Mockito.verify(catalogManager, Mockito.never())
+ .testConnection(Mockito.eq(catalogIdent),
Mockito.any(CatalogChange[].class));
+ }
}
diff --git a/docs/open-api/catalogs.yaml b/docs/open-api/catalogs.yaml
index c9972aa600..6fe23fcdec 100644
--- a/docs/open-api/catalogs.yaml
+++ b/docs/open-api/catalogs.yaml
@@ -151,13 +151,21 @@ paths:
summary: Test an existing catalog connection
description: >-
Runs the provider's smallest meaningful, read-only catalog-level
operation using the
- catalog's stored configuration and effective credentials. This
fail-fast preflight
- surfaces configuration and connectivity failures before regular
catalog operations, but
- does not guarantee that every object-level or mutating operation will
succeed. Fileset
- catalogs test all catalog-level `location` and `location-*` targets.
Model and Generic
- catalogs do not support connection testing. Expected test failures are
returned as
- application error codes in an HTTP 200 response.
+ catalog's stored configuration and effective credentials. Optional
proposed catalog
+ changes are applied to a temporary effective configuration for this
probe and are not
+ persisted. This fail-fast preflight surfaces configuration and
connectivity failures before
+ regular catalog operations, but does not guarantee that every
object-level or mutating
+ operation will succeed. Fileset catalogs test all catalog-level
`location` and `location-*`
+ targets. Model and Generic catalogs do not support connection testing.
Expected test
+ failures are returned as application error codes in an HTTP 200
response.
operationId: testExistingCatalogConnection
+ requestBody:
+ required: false
+ description: Optional catalog changes to apply only to this connection
test
+ content:
+ application/json:
+ schema:
+ $ref: "#/components/schemas/CatalogUpdatesRequest"
responses:
"200":
description: Connection test completed, including expected test
failures
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/rest/CatalogOperations.java
b/server/src/main/java/org/apache/gravitino/server/web/rest/CatalogOperations.java
index 1e026b96f2..43226b156e 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/rest/CatalogOperations.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/rest/CatalogOperations.java
@@ -232,14 +232,24 @@ public class CatalogOperations {
@PathParam("metalake") @AuthorizationMetadata(type =
Entity.EntityType.METALAKE)
String metalake,
@PathParam("catalog") @AuthorizationMetadata(type =
Entity.EntityType.CATALOG)
- String catalogName) {
+ String catalogName,
+ CatalogUpdatesRequest request) {
LOG.info("Received test connection request for existing catalog: {}.{}",
metalake, catalogName);
try {
return Utils.doAs(
httpRequest,
() -> {
NameIdentifier ident = NameIdentifierUtil.ofCatalog(metalake,
catalogName);
- catalogDispatcher.testConnection(ident);
+ if (request == null) {
+ catalogDispatcher.testConnection(ident);
+ } else {
+ request.validate();
+ CatalogChange[] changes =
+ request.getUpdates().stream()
+ .map(CatalogUpdateRequest::catalogChange)
+ .toArray(CatalogChange[]::new);
+ catalogDispatcher.testConnection(ident, changes);
+ }
LOG.info(
"Successfully tested connection for existing catalog: {}.{}",
metalake,
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestCatalogOperations.java
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestCatalogOperations.java
index 1fa527630b..54b91fdad8 100644
---
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestCatalogOperations.java
+++
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestCatalogOperations.java
@@ -47,6 +47,7 @@ import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.Catalog;
+import org.apache.gravitino.CatalogChange;
import org.apache.gravitino.Config;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.NameIdentifier;
@@ -81,6 +82,7 @@ import org.glassfish.jersey.test.TestProperties;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
public class TestCatalogOperations extends BaseOperationsTest {
@@ -399,6 +401,46 @@ public class TestCatalogOperations extends
BaseOperationsTest {
Assertions.assertEquals(Response.Status.OK.getStatusCode(),
response.getStatus());
Assertions.assertEquals(0,
response.readEntity(BaseResponse.class).getCode());
+ CatalogUpdatesRequest proposedChanges =
+ new CatalogUpdatesRequest(
+ ImmutableList.of(
+ new CatalogUpdateRequest.RenameCatalogRequest("catalog2"),
+ new CatalogUpdateRequest.UpdateCatalogCommentRequest("new
comment"),
+ new CatalogUpdateRequest.SetCatalogPropertyRequest("key", "new
value"),
+ new
CatalogUpdateRequest.RemoveCatalogPropertyRequest("old-key")));
+ doNothing().when(manager).testConnection(any(NameIdentifier.class),
any(CatalogChange[].class));
+ Response changedResponse =
+ target("/metalakes/metalake1/catalogs/catalog1/testConnection")
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .post(Entity.entity(proposedChanges,
MediaType.APPLICATION_JSON_TYPE));
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
changedResponse.getStatus());
+ Assertions.assertEquals(0,
changedResponse.readEntity(BaseResponse.class).getCode());
+ ArgumentCaptor<CatalogChange[]> changesCaptor =
ArgumentCaptor.forClass(CatalogChange[].class);
+ Mockito.verify(manager).testConnection(any(NameIdentifier.class),
changesCaptor.capture());
+ Assertions.assertArrayEquals(
+ new CatalogChange[] {
+ CatalogChange.rename("catalog2"),
+ CatalogChange.updateComment("new comment"),
+ CatalogChange.setProperty("key", "new value"),
+ CatalogChange.removeProperty("old-key")
+ },
+ changesCaptor.getValue());
+
+ doThrow(new IllegalArgumentException("invalid proposed change"))
+ .when(manager)
+ .testConnection(any(NameIdentifier.class), any(CatalogChange[].class));
+ Response invalidChangesResponse =
+ target("/metalakes/metalake1/catalogs/catalog1/testConnection")
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .post(Entity.entity(proposedChanges,
MediaType.APPLICATION_JSON_TYPE));
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
invalidChangesResponse.getStatus());
+ ErrorResponse invalidChanges =
invalidChangesResponse.readEntity(ErrorResponse.class);
+ Assertions.assertEquals(ErrorConstants.ILLEGAL_ARGUMENTS_CODE,
invalidChanges.getCode());
+ Assertions.assertEquals("invalid proposed change",
invalidChanges.getMessage());
+ Assertions.assertNull(invalidChanges.getStack());
+
doThrow(new ConnectionFailedException("sanitized failure"))
.when(manager)
.testConnection(any(NameIdentifier.class));
