This is an automated email from the ASF dual-hosted git repository. diqiu50 pushed a commit to branch dell-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 320489684ee048076101c49510e285b500e51013 Author: diqiu50 <[email protected]> AuthorDate: Mon Aug 24 15:26:50 2026 +0800 [Cherry-pick to branch-1.3] [#12554] improvement(trino-connector): Discover the Iceberg REST server instead of a manual switch Replace gravitino.iceberg.rest-enabled with automatic discovery. The Trino connector already connects to the Gravitino server to load metalakes, so it now also asks that server whether it has an Iceberg REST server running for the connector's metalake, and routes lakehouse-iceberg catalogs through the reported endpoint. When the server reports nothing — the Iceberg REST server is not running, or serves a different metalake — catalogs fall back to translating catalog-backend as before, with no configuration required either way. Server side: AuxiliaryServiceManager gains a query method for whether an auxiliary service is registered, and a new GET /api/system/iceberg-rest endpoint reports the Iceberg REST server's address for a given metalake, falling back to the request's own host when the service binds to a wildcard address. The client gains a matching method following the existing REST client patterns. gravitino.iceberg.rest-uri remains as a manual override for topologies where the discovered address is not reachable. The discovery call is piggybacked on the existing periodic metalake poll, so it adds no new network schedule, and failures (including talking to a Gravitino server older than this endpoint) are swallowed rather than interrupting catalog loading. This also resolves the multi-metalake ambiguity in the previous design: the server only reports an endpoint for the metalake it is configured to serve, so a same-named catalog in a different metalake can no longer be silently misrouted. (cherry picked from commit a8015c9e3fc860bf0059adce70244646ea869492) --- .../gravitino/client/GravitinoClientBase.java | 23 ++++ .../dto/responses/IcebergRESTServiceResponse.java | 56 ++++++++ .../java/org/apache/gravitino/GravitinoEnv.java | 9 ++ .../auxiliary/AuxiliaryServiceManager.java | 10 ++ docs/trino-connector/authentication.md | 3 +- docs/trino-connector/catalog-iceberg.md | 28 ++-- docs/trino-connector/configuration.md | 13 +- .../docker-script/docker-compose.yaml | 2 - .../init/trino/config/catalog/gravitino.properties | 1 - .../docker-script/init/trino/init.sh | 4 - .../integration/test/container/ContainerSuite.java | 23 +--- .../test/container/TrinoITContainers.java | 10 +- .../gravitino/integration/test/util/BaseIT.java | 11 -- .../web/rest/IcebergRESTServiceOperations.java | 105 +++++++++++++++ .../integration/test/TrinoConnectorIT.java | 1 - .../integration/test/TrinoQueryITBase.java | 1 - .../gravitino/trino/connector/GravitinoConfig.java | 84 +++++------- .../connector/catalog/CatalogConnectorManager.java | 18 +++ .../iceberg/IcebergCatalogPropertyConverter.java | 28 ++-- .../catalog/iceberg/IcebergConnectorAdapter.java | 11 +- .../trino/connector/TestGravitinoConfig.java | 55 +++++--- .../TestIcebergCatalogPropertyConverter.java | 148 ++++++++++++++++----- 22 files changed, 453 insertions(+), 191 deletions(-) diff --git a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClientBase.java b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClientBase.java index 996c149591..07c7e9cca6 100644 --- a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClientBase.java +++ b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClientBase.java @@ -29,10 +29,12 @@ import java.net.URI; import java.net.URISyntaxException; import java.util.Collections; import java.util.Map; +import java.util.Optional; import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; import org.apache.gravitino.Version; +import org.apache.gravitino.dto.responses.IcebergRESTServiceResponse; import org.apache.gravitino.dto.responses.MetalakeResponse; import org.apache.gravitino.dto.responses.VersionResponse; import org.apache.gravitino.exceptions.GravitinoRuntimeException; @@ -185,6 +187,27 @@ public abstract class GravitinoClientBase implements Closeable { return new GravitinoVersion(resp.getVersion()); } + /** + * Retrieves the endpoint of the Gravitino Iceberg REST server, if it is running and serves the + * given metalake. + * + * @param metalakeName the metalake the caller intends to route through the Iceberg REST server + * @return the Iceberg REST server endpoint, or empty if it is not running or serves a different + * metalake + */ + public Optional<String> icebergRestServiceUri(String metalakeName) { + IcebergRESTServiceResponse resp = + restClient.get( + "api/system/iceberg-rest", + ImmutableMap.of("metalake", metalakeName), + IcebergRESTServiceResponse.class, + Collections.emptyMap(), + ErrorHandlers.restErrorHandler()); + resp.validate(); + + return Optional.ofNullable(resp.getUri()); + } + /** Closes the GravitinoClient and releases any underlying resources. */ @Override public void close() { diff --git a/common/src/main/java/org/apache/gravitino/dto/responses/IcebergRESTServiceResponse.java b/common/src/main/java/org/apache/gravitino/dto/responses/IcebergRESTServiceResponse.java new file mode 100644 index 0000000000..89c071dfcc --- /dev/null +++ b/common/src/main/java/org/apache/gravitino/dto/responses/IcebergRESTServiceResponse.java @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.dto.responses; + +import com.fasterxml.jackson.annotation.JsonProperty; +import javax.annotation.Nullable; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.ToString; + +/** + * Represents a response reporting the endpoint of the Gravitino Iceberg REST server, if it is + * running and serves the requested metalake. + */ +@Getter +@EqualsAndHashCode(callSuper = true) +@ToString +public class IcebergRESTServiceResponse extends BaseResponse { + + @Nullable + @JsonProperty("uri") + private final String uri; + + /** + * Constructor for IcebergRESTServiceResponse. + * + * @param uri the Iceberg REST server endpoint, or {@code null} when it is not running or does not + * serve the requested metalake + */ + public IcebergRESTServiceResponse(@Nullable String uri) { + super(0); + this.uri = uri; + } + + /** Default constructor for IcebergRESTServiceResponse. (Used for Jackson deserialization.) */ + public IcebergRESTServiceResponse() { + super(); + this.uri = null; + } +} diff --git a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java index 4f42989045..fc69a7cbde 100644 --- a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java +++ b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java @@ -235,6 +235,15 @@ public class GravitinoEnv { return config; } + /** + * Get the auxiliary service manager associated with the Gravitino environment. + * + * @return The auxiliary service manager instance. + */ + public AuxiliaryServiceManager auxServiceManager() { + return auxServiceManager; + } + /** * Get the EntityStore associated with the Gravitino environment. * diff --git a/core/src/main/java/org/apache/gravitino/auxiliary/AuxiliaryServiceManager.java b/core/src/main/java/org/apache/gravitino/auxiliary/AuxiliaryServiceManager.java index 08444abb3f..accb23f799 100644 --- a/core/src/main/java/org/apache/gravitino/auxiliary/AuxiliaryServiceManager.java +++ b/core/src/main/java/org/apache/gravitino/auxiliary/AuxiliaryServiceManager.java @@ -95,6 +95,16 @@ public class AuxiliaryServiceManager { } @VisibleForTesting + /** + * Returns whether the given auxiliary service was configured and registered. + * + * @param auxServiceName the auxiliary service's short name, e.g. {@code iceberg-rest} + * @return true if the service is registered + */ + public boolean isAuxServiceRegistered(String auxServiceName) { + return auxServices.containsKey(auxServiceName); + } + public IsolatedClassLoader getIsolatedClassLoader(List<String> classPaths) { return IsolatedClassLoader.buildClassLoader(classPaths); } diff --git a/docs/trino-connector/authentication.md b/docs/trino-connector/authentication.md index 322eef9f5d..42c7932f6b 100644 --- a/docs/trino-connector/authentication.md +++ b/docs/trino-connector/authentication.md @@ -194,13 +194,12 @@ Whether the coordinator can populate this extra-credential depends on the Trino The `gravitino.client.oauth2.*` properties above still configure the shared bootstrap/admin client used for catalog discovery — they are unrelated to the per-user forwarded token. For an Iceberg catalog reached through the Gravitino Iceberg REST server (IRC) — every -`lakehouse-iceberg` catalog, unless `gravitino.iceberg.rest-enabled=false`; see [Iceberg +`lakehouse-iceberg` catalog for which the Gravitino server reports a running IRC; see [Iceberg catalog](./catalog-iceberg.md#how-trino-reaches-the-catalog) — the IRC's own authentication is configured once per Trino cluster with the `gravitino.iceberg.rest-catalog.` prefix, and `iceberg.rest-catalog.session=USER` is set automatically when `forwardUser=true`: ```properties -gravitino.iceberg.rest-uri=http://gravitino-host:9001/iceberg gravitino.iceberg.rest-catalog.security=OAUTH2 gravitino.iceberg.rest-catalog.oauth2.credential=service-account-id:service-account-secret gravitino.iceberg.rest-catalog.oauth2.server-uri=http://your-idp/realms/gravitino/protocol/openid-connect/token diff --git a/docs/trino-connector/catalog-iceberg.md b/docs/trino-connector/catalog-iceberg.md index dd00d8cab1..02779e8041 100644 --- a/docs/trino-connector/catalog-iceberg.md +++ b/docs/trino-connector/catalog-iceberg.md @@ -33,7 +33,17 @@ nowhere to put the session token of an STS temporary credential — so a catalog IRC means every table access gets a freshly issued temporary credential over the Iceberg REST protocol. -Configure the IRC endpoint once per Trino cluster, in `etc/catalog/gravitino.properties`: +The connector already connects to the Gravitino server (it is how catalogs are discovered in the +first place), so it also asks that server whether it has an Iceberg REST server running for the +connector's metalake, and uses the endpoint the server reports. Nothing needs to be configured for +this: `etc/catalog/gravitino.properties` only needs the usual `gravitino.uri` and +`gravitino.metalake`. If the server reports no endpoint — the IRC is not running, or it serves a +different metalake — `lakehouse-iceberg` catalogs fall back to translating `catalog-backend` as +before, and credential vending does not work. + +Set `gravitino.iceberg.rest-uri` only to override the discovered endpoint, for example when the IRC +is not reachable at the address the Gravitino server itself reports (a split network, or the IRC +deployed independently of the main server): ```properties connector.name=gravitino @@ -43,10 +53,6 @@ gravitino.uri=http://gravitino-host:8090 gravitino.iceberg.rest-uri=http://gravitino-host:9001/iceberg ``` -The endpoint is deployment topology, not a property of the data source, so it does not belong on the -catalog: it would have to be repeated on every new catalog, moving the IRC would mean editing all of -them, and one catalog could not serve two Trino clusters that reach the IRC by different hostnames. - The connector derives everything else from the catalog itself. The Gravitino catalog name is passed as both `iceberg.rest-catalog.warehouse` and `iceberg.rest-catalog.prefix` — the Iceberg client selects the catalog twice over, first as the query parameter of the `GET /v1/config` call that @@ -106,14 +112,14 @@ keeping per-user credential vending and per-user authorization intact. Set ### Limitations - One IRC serves exactly one metalake, fixed at startup by - `gravitino.iceberg-rest.gravitino-metalake`. In multi-metalake mode - (`gravitino.use-single-metalake=false`), only catalogs in that metalake are reachable; the others - fail on the IRC side with `NoSuchCatalogException`. + `gravitino.iceberg-rest.gravitino-metalake`. The Gravitino server only reports the IRC's endpoint + for that metalake; in multi-metalake mode (`gravitino.use-single-metalake=false`), catalogs in any + other metalake fall back to translating `catalog-backend` instead of failing outright. - A catalog created with `catalog-backend=rest` keeps pointing at its own configured `uri` and is not re-routed, since it already reaches an Iceberg REST catalog directly. -- To disable this routing entirely — for a deployment that does not run the IRC — set - `gravitino.iceberg.rest-enabled=false`. Iceberg catalogs are then translated into Trino's `jdbc` - or `hive_metastore` catalog types as before, and credential vending does not work. +- A deployment that does not run the IRC needs no configuration at all: the server reports no + endpoint, and `lakehouse-iceberg` catalogs are translated into Trino's `jdbc` or `hive_metastore` + catalog types as before. ## Schema Operations diff --git a/docs/trino-connector/configuration.md b/docs/trino-connector/configuration.md index 6d225a6cec..9e6747ef67 100644 --- a/docs/trino-connector/configuration.md +++ b/docs/trino-connector/configuration.md @@ -29,21 +29,16 @@ license: "This software is licensed under the Apache License version 2." | gravitino.client. | string | (none) | The configuration key prefix for the Gravitino client config. | No | | gravitino.trino.skip-catalog-patterns | string | (none) | The `gravitino.trino.skip-catalog-patterns` defines a comma-separated list of catalog name regex patterns that should be excluded from loading. For example, `test_.*, .*_tmp` excludes all catalogs starting with `test_` or ending with `_tmp`. | No | | gravitino.use-single-metalake | boolean | true | If `true`, only one metalake is used and catalogs are identified by `<catalog_name>`. If `false`, multi-metalake mode is enabled and catalogs are identified by `<metalake_name>.<catalog_name>`. | No | -| gravitino.iceberg.rest-enabled | boolean | true | If `true`, `lakehouse-iceberg` catalogs are loaded through the Gravitino Iceberg REST server (IRC) instead of being translated into Trino's `jdbc` or `hive_metastore` Iceberg catalog type. This is what makes credential vending work. Requires `gravitino.iceberg.rest-uri`. Set to `false` for a deployment that does not run the IRC. | No | -| gravitino.iceberg.rest-uri | string | (none) | The endpoint of the Gravitino Iceberg REST server, for example `http://gravitino-host:9001/iceberg`. Required when `gravitino.iceberg.rest-enabled` is `true` and the metalake contains `lakehouse-iceberg` catalogs whose `catalog-backend` is not `rest`. | No | +| gravitino.iceberg.rest-uri | string | (none) | The endpoint of the Gravitino Iceberg REST server (IRC). Discovered automatically from the Gravitino server, which is asked whether it has an IRC running for this connector's metalake; set this only to override the discovered value. When an endpoint is available (discovered or configured), `lakehouse-iceberg` catalogs whose `catalog-backend` is not `rest` are loaded through it instead of being translated in [...] | gravitino.iceberg.rest-catalog. | string | (none) | The configuration key prefix for properties passed through to the internal Trino Iceberg REST catalog. The prefix is rewritten to `iceberg.rest-catalog.`, so `gravitino.iceberg.rest-catalog.security=OAUTH2` becomes `iceberg.rest-catalog.security=OAUTH2`. The `uri`, `warehouse` and `prefix` keys are reserved and always derived by the connector. | No | To configure the Gravitino client, use properties prefixed with `gravitino.client.`. These properties will directly passed to the Gravitino client. **Note:** Invalid configuration properties will result in exceptions. Please see [Gravitino Java client configurations](../how-to-use-gravitino-client.md#java-client-configuration) for more support client configuration. -:::caution -`gravitino.iceberg.rest-enabled` defaults to `true`. When upgrading an existing deployment whose -metalake contains `lakehouse-iceberg` catalogs with `catalog-backend=jdbc` or `hive`, those catalogs -fail to load unless you either set `gravitino.iceberg.rest-uri` to your Iceberg REST server endpoint, -or set `gravitino.iceberg.rest-enabled=false` to keep the previous behavior. See -[Iceberg catalog](./catalog-iceberg.md#how-trino-reaches-the-catalog). -::: +Upgrading an existing deployment needs no action: a Gravitino server without the Iceberg REST server +running reports no endpoint, so `lakehouse-iceberg` catalogs keep translating `catalog-backend` as +before. See [Iceberg catalog](./catalog-iceberg.md#how-trino-reaches-the-catalog). Multi-metalake mode (`gravitino.use-single-metalake=false`) is supported on Trino connector versions 440-445 and 469-478. On versions 446-468, a warning is logged and the connector initializes, but the mode is not fully supported and some operations may fail. diff --git a/integration-test-common/docker-script/docker-compose.yaml b/integration-test-common/docker-script/docker-compose.yaml index 1dfb8f5a7d..3ed57d07ca 100644 --- a/integration-test-common/docker-script/docker-compose.yaml +++ b/integration-test-common/docker-script/docker-compose.yaml @@ -87,7 +87,6 @@ services: - HADOOP_USER_NAME=anonymous - GRAVITINO_HOST_IP=host.docker.internal - GRAVITINO_HOST_PORT=${GRAVITINO_SERVER_PORT:-8090} - - GRAVITINO_ICEBERG_REST_PORT=${GRAVITINO_ICEBERG_REST_PORT:-9001} - GRAVITINO_METALAKE_NAME=test - HIVE_HOST_IP=hive - TRINO_WORKER_NUM=${TRINO_WORKER_NUM:-0} @@ -123,7 +122,6 @@ services: - HADOOP_USER_NAME=anonymous - GRAVITINO_HOST_IP=host.docker.internal - GRAVITINO_HOST_PORT=${GRAVITINO_SERVER_PORT:-8090} - - GRAVITINO_ICEBERG_REST_PORT=${GRAVITINO_ICEBERG_REST_PORT:-9001} - GRAVITINO_METALAKE_NAME=test - HIVE_HOST_IP=hive - TRINO_ROLE=worker diff --git a/integration-test-common/docker-script/init/trino/config/catalog/gravitino.properties b/integration-test-common/docker-script/init/trino/config/catalog/gravitino.properties index d38c18c458..08789c84c0 100644 --- a/integration-test-common/docker-script/init/trino/config/catalog/gravitino.properties +++ b/integration-test-common/docker-script/init/trino/config/catalog/gravitino.properties @@ -20,7 +20,6 @@ connector.name = gravitino gravitino.uri = http://GRAVITINO_HOST_IP:GRAVITINO_HOST_PORT gravitino.metalake = GRAVITINO_METALAKE_NAME -gravitino.iceberg.rest-uri = http://GRAVITINO_ICEBERG_REST_HOST:GRAVITINO_ICEBERG_REST_PORT/iceberg gravitino.trino.skip-version-validation=true gravitino.client.authType = simple gravitino.client.session.forwardUser = true diff --git a/integration-test-common/docker-script/init/trino/init.sh b/integration-test-common/docker-script/init/trino/init.sh index a65ec5ab90..077c6f77d0 100644 --- a/integration-test-common/docker-script/init/trino/init.sh +++ b/integration-test-common/docker-script/init/trino/init.sh @@ -34,10 +34,6 @@ cp /usr/lib/trino/plugin/mysql/*mysql-connector-j-*.jar /usr/lib/trino/plugin/ic sed -i "s/GRAVITINO_HOST_IP:GRAVITINO_HOST_PORT/${GRAVITINO_HOST_IP}:${GRAVITINO_HOST_PORT}/g" /etc/trino/catalog/gravitino.properties # Update `gravitino.metalake = GRAVITINO_METALAKE_NAME` in the `conf/catalog/gravitino.properties` sed -i "s/GRAVITINO_METALAKE_NAME/${GRAVITINO_METALAKE_NAME}/g" /etc/trino/catalog/gravitino.properties -# Update `gravitino.iceberg.rest-uri` in the `conf/catalog/gravitino.properties`. The Iceberg REST -# server runs on the same host as the Gravitino server but on a different port, so it needs its own -# placeholder pair — reusing GRAVITINO_HOST_IP:GRAVITINO_HOST_PORT would substitute the server port. -sed -i "s/GRAVITINO_ICEBERG_REST_HOST:GRAVITINO_ICEBERG_REST_PORT/${GRAVITINO_HOST_IP}:${GRAVITINO_ICEBERG_REST_PORT}/g" /etc/trino/catalog/gravitino.properties # Update `node.id=NODE_ID` in the `/conf/node.properties` sed -i "s/NODE_ID/${RANDOM}-${RANDOM}-${RANDOM}-${RANDOM}-${RANDOM}/g" /etc/trino/node.properties diff --git a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java index ad35692dbb..b2b0ae50da 100644 --- a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java +++ b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java @@ -288,7 +288,6 @@ public class ContainerSuite implements Closeable { String trinoConfDir, String trinoConnectorLibDir, int gravitinoServerPort, - int icebergRestServerPort, String metalakeName) { ITUtils.cleanDisk(); if (trinoContainer == null) { @@ -297,23 +296,15 @@ public class ContainerSuite implements Closeable { initIfNecessary(); // Start Trino container String hiveContainerIp = hiveContainer.getContainerIpAddress(); - ImmutableMap.Builder<String, String> envVars = - ImmutableMap.<String, String>builder() - .put("HADOOP_USER_NAME", "anonymous") - .put("GRAVITINO_HOST_IP", "host.docker.internal") - .put("GRAVITINO_HOST_PORT", String.valueOf(gravitinoServerPort)) - .put("GRAVITINO_METALAKE_NAME", metalakeName); - if (icebergRestServerPort > 0) { - envVars.put("GRAVITINO_ICEBERG_REST_PORT", String.valueOf(icebergRestServerPort)); - } else { - LOG.info( - "No Iceberg REST server port supplied; Trino falls back to the compose default. " - + "Catalogs routed through the Iceberg REST server will not work in this " - + "container."); - } TrinoContainer.Builder trinoBuilder = TrinoContainer.builder() - .withEnvVars(envVars.build()) + .withEnvVars( + ImmutableMap.<String, String>builder() + .put("HADOOP_USER_NAME", "anonymous") + .put("GRAVITINO_HOST_IP", "host.docker.internal") + .put("GRAVITINO_HOST_PORT", String.valueOf(gravitinoServerPort)) + .put("GRAVITINO_METALAKE_NAME", metalakeName) + .build()) .withNetwork(getNetwork()) .withExtraHosts( ImmutableMap.<String, String>builder() diff --git a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TrinoITContainers.java b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TrinoITContainers.java index 572b948d55..a02e12de6a 100644 --- a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TrinoITContainers.java +++ b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TrinoITContainers.java @@ -50,12 +50,11 @@ public class TrinoITContainers implements AutoCloseable { } public void launch(int gravitinoServerPort) throws Exception { - launch(gravitinoServerPort, 0, "hive2", false, null, null, null); + launch(gravitinoServerPort, "hive2", false, null, null, null); } public void launch( int gravitinoServerPort, - int icebergRestServerPort, String hiveRuntimeVersion, boolean isTrinoConnectorTest, Integer trinoWorkerNum, @@ -82,13 +81,6 @@ public class TrinoITContainers implements AutoCloseable { } env.put("GRAVITINO_SERVER_PORT", String.valueOf(gravitinoServerPort)); env.put("HIVE_RUNTIME_VERSION", hiveRuntimeVersion); - if (icebergRestServerPort > 0) { - env.put("GRAVITINO_ICEBERG_REST_PORT", String.valueOf(icebergRestServerPort)); - } else { - LOG.info( - "No Iceberg REST server port supplied; Trino falls back to the compose default. " - + "Catalogs routed through the Iceberg REST server will not work in this container."); - } env.put("TRINO_CONNECTOR_TEST", String.valueOf(isTrinoConnectorTest)); if (System.getProperty("gravitino.log.path") != null) { env.put("GRAVITINO_LOG_PATH", System.getProperty("gravitino.log.path")); diff --git a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java index 0c247a22f1..ef45535de9 100644 --- a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java +++ b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java @@ -30,7 +30,6 @@ import com.google.common.collect.ImmutableMap; import java.io.File; import java.io.IOException; import java.lang.reflect.Field; -import java.net.URI; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; @@ -687,16 +686,6 @@ public class BaseIT { this.customConfigs.putAll(icebergRestConfigs); } - /** - * Returns the port the Iceberg REST auxiliary service listens on. Only valid once {@link - * #startIntegrationTest()} has run, since it reads the started server's config. - * - * @return the Iceberg REST service port - */ - public int getIcebergRestServicePort() { - return URI.create(getIcebergRestServiceUri()).getPort(); - } - protected String getIcebergRestServiceUri() { JettyServerConfig jettyServerConfig = JettyServerConfig.fromConfig(serverConfig, String.format("gravitino.iceberg-rest.")); diff --git a/server/src/main/java/org/apache/gravitino/server/web/rest/IcebergRESTServiceOperations.java b/server/src/main/java/org/apache/gravitino/server/web/rest/IcebergRESTServiceOperations.java new file mode 100644 index 0000000000..0ea698a6a4 --- /dev/null +++ b/server/src/main/java/org/apache/gravitino/server/web/rest/IcebergRESTServiceOperations.java @@ -0,0 +1,105 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.server.web.rest; + +import com.codahale.metrics.annotation.ResponseMetered; +import com.codahale.metrics.annotation.Timed; +import javax.servlet.http.HttpServletRequest; +import javax.ws.rs.Consumes; +import javax.ws.rs.GET; +import javax.ws.rs.Path; +import javax.ws.rs.Produces; +import javax.ws.rs.QueryParam; +import javax.ws.rs.core.Context; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.Config; +import org.apache.gravitino.GravitinoEnv; +import org.apache.gravitino.dto.responses.IcebergRESTServiceResponse; +import org.apache.gravitino.metrics.MetricNames; +import org.apache.gravitino.server.web.JettyServerConfig; +import org.apache.gravitino.server.web.Utils; + +/** + * Reports the endpoint of the Gravitino Iceberg REST server, so that clients which already connect + * to this Gravitino server can discover it instead of requiring it to be configured separately. + */ +@Path("/system/iceberg-rest") +@Consumes(MediaType.APPLICATION_JSON) +@Produces(MediaType.APPLICATION_JSON) +public class IcebergRESTServiceOperations { + + // Matches gravitino.auxService.names / AuxiliaryServiceManager's registration key. + private static final String AUX_SERVICE_NAME = "iceberg-rest"; + private static final String CONFIG_PREFIX = "gravitino.iceberg-rest."; + // The post-strip key used by the Iceberg REST server itself; see + // IcebergConstants.GRAVITINO_METALAKE and DynamicIcebergConfigProvider. + private static final String SERVED_METALAKE_KEY = CONFIG_PREFIX + "gravitino-metalake"; + + @Context private HttpServletRequest httpRequest; + + /** + * Reports the Iceberg REST server's endpoint for the requested metalake. + * + * @param metalake the metalake the caller intends to route through the Iceberg REST server; may + * be blank, in which case the endpoint is reported regardless of which metalake it serves + * @return a response whose {@code uri} is {@code null} when the Iceberg REST server is not + * running, or does not serve the requested metalake + */ + @GET + @Timed(name = "iceberg-rest-service." + MetricNames.HTTP_PROCESS_DURATION, absolute = true) + @ResponseMetered(name = "iceberg-rest-service", absolute = true) + public Response getIcebergRestServiceUri(@QueryParam("metalake") String metalake) { + return Utils.ok(new IcebergRESTServiceResponse(resolveUri(metalake))); + } + + private String resolveUri(String metalake) { + if (!GravitinoEnv.getInstance().auxServiceManager().isAuxServiceRegistered(AUX_SERVICE_NAME)) { + return null; + } + + Config config = GravitinoEnv.getInstance().config(); + String servedMetalake = config.getRawString(SERVED_METALAKE_KEY, ""); + if (StringUtils.isNotBlank(metalake) + && StringUtils.isNotBlank(servedMetalake) + && !servedMetalake.equals(metalake)) { + // The Iceberg REST server serves exactly one metalake. Routing a different metalake's + // catalogs at it would 404 on every request, so report it as unavailable instead. + return null; + } + + JettyServerConfig icebergRestConfig = JettyServerConfig.fromConfig(config, CONFIG_PREFIX); + String host = icebergRestConfig.getHost(); + if (isWildcardHost(host)) { + // The Iceberg REST server binds to all interfaces, so it has no single externally + // reachable address of its own. The caller already reached this Gravitino server at some + // resolvable host, so reuse it — this holds whenever both services share a host, which is + // the common case, and callers with a genuinely split topology can still set + // gravitino.iceberg.rest-uri manually. + host = httpRequest.getServerName(); + } + String scheme = icebergRestConfig.isEnableHttps() ? "https" : "http"; + return String.format("%s://%s:%d/iceberg", scheme, host, icebergRestConfig.getHttpPort()); + } + + private static boolean isWildcardHost(String host) { + return StringUtils.isBlank(host) || "0.0.0.0".equals(host) || "::".equals(host); + } +} diff --git a/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoConnectorIT.java b/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoConnectorIT.java index 1be5ef43bc..968e26fb30 100644 --- a/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoConnectorIT.java +++ b/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoConnectorIT.java @@ -131,7 +131,6 @@ public class TrinoConnectorIT extends BaseIT { trinoConfDir, System.getenv("GRAVITINO_ROOT_DIR") + "/trino-connector/build/libs", getGravitinoServerPort(), - getIcebergRestServicePort(), metalakeName); Assertions.assertTrue( containerSuite.getTrinoContainer().checkSyncCatalogFromGravitino(5, catalogName), diff --git a/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoQueryITBase.java b/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoQueryITBase.java index 24d33cbf2e..e5be0c11c4 100644 --- a/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoQueryITBase.java +++ b/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoQueryITBase.java @@ -120,7 +120,6 @@ public class TrinoQueryITBase { trinoITContainers.launch( baseIT.getGravitinoServerPort(), - baseIT.getIcebergRestServicePort(), hiveRuntimeVersion, isTrinoConnectorTest, trinoWorkerNum, diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConfig.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConfig.java index 50af268066..2c1f795afb 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConfig.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConfig.java @@ -31,18 +31,15 @@ import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Properties; +import java.util.concurrent.ConcurrentHashMap; import java.util.regex.Pattern; import java.util.stream.Collectors; import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.trino.connector.security.GravitinoAuthProvider; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** Gravitino config. */ public class GravitinoConfig { - private static final Logger LOG = LoggerFactory.getLogger(GravitinoConfig.class); - // Trino config keys /** The Trino discovery URI. */ private static final String TRINO_DISCOVERY_URI = "discovery.uri"; @@ -86,6 +83,9 @@ public class GravitinoConfig { private static final Map<String, ConfigEntry> CONFIG_DEFINITIONS = new HashMap<>(); private final Map<String, String> config; private final List<Pattern> skipCatalogPatternList; + // Iceberg REST server endpoints discovered from the Gravitino server, keyed by metalake. Written + // by the catalog connector manager's periodic poll and read on every Iceberg catalog load. + private final Map<String, String> discoveredIcebergRestUriByMetalake = new ConcurrentHashMap<>(); // Gravitino config entity private static final ConfigEntry GRAVITINO_URI = @@ -270,20 +270,13 @@ public class GravitinoConfig { "3600", false); - private static final ConfigEntry GRAVITINO_ICEBERG_REST_ENABLED = - new ConfigEntry( - "gravitino.iceberg.rest-enabled", - "When true, lakehouse-iceberg catalogs are loaded through the Gravitino Iceberg REST " - + "server instead of being translated into a Trino JDBC or Hive metastore Iceberg " - + "catalog. Requires gravitino.iceberg.rest-uri.", - "true", - false); - private static final ConfigEntry GRAVITINO_ICEBERG_REST_URI = new ConfigEntry( "gravitino.iceberg.rest-uri", - "The endpoint of the Gravitino Iceberg REST server, for example " - + "http://localhost:9001/iceberg.";, + "The endpoint of the Gravitino Iceberg REST server. Discovered automatically from the " + + "Gravitino server by default; set this only to override the discovered value, " + + "for example when the Iceberg REST server is not reachable at the address the " + + "Gravitino server itself reports.", "", false); @@ -325,28 +318,6 @@ public class GravitinoConfig { "Config `gravitino.trino.skip-catalog-patterns` is invalid because it contains an illegal regular expression", e); } - warnOnMissingIcebergRestUri(); - } - - /** - * Warns at startup when the Iceberg REST routing is on but has no endpoint. Without this, the - * failure only surfaces once a lakehouse-iceberg catalog is loaded, and that error is swallowed - * by the catalog refresh loop, leaving the user with an unexplained missing catalog. - */ - private void warnOnMissingIcebergRestUri() { - // Only the statically configured connector warns; the dynamic per-catalog ones would repeat it. - if (isDynamicConnector() - || !isIcebergRestEnabled() - || StringUtils.isNotBlank(getIcebergRestUri())) { - return; - } - LOG.warn( - "'{}' is enabled but '{}' is not set, so every lakehouse-iceberg catalog will fail to " - + "load. Set the Iceberg REST server endpoint, or set '{}=false' to load Iceberg " - + "catalogs through their catalog backend instead.", - GRAVITINO_ICEBERG_REST_ENABLED.key, - GRAVITINO_ICEBERG_REST_URI.key, - GRAVITINO_ICEBERG_REST_ENABLED.key); } /** @@ -738,25 +709,40 @@ public class GravitinoConfig { } /** - * Returns whether lakehouse-iceberg catalogs are routed through the Gravitino Iceberg REST - * server. + * Sets the Iceberg REST server endpoint discovered from the Gravitino server for the given + * metalake. Called by the catalog connector manager's periodic metalake poll; ignored for a + * metalake where {@code gravitino.iceberg.rest-uri} is explicitly configured, which always takes + * precedence. * - * @return true if the Iceberg REST routing is enabled + * @param metalake the metalake the endpoint was discovered for + * @param uri the discovered endpoint, or {@code null} when the Iceberg REST server is not running + * or does not serve this metalake */ - public boolean isIcebergRestEnabled() { - return Boolean.parseBoolean( - config.getOrDefault( - GRAVITINO_ICEBERG_REST_ENABLED.key, GRAVITINO_ICEBERG_REST_ENABLED.defaultValue)); + public void setDiscoveredIcebergRestUri(String metalake, String uri) { + if (StringUtils.isBlank(uri)) { + discoveredIcebergRestUriByMetalake.remove(metalake); + } else { + discoveredIcebergRestUriByMetalake.put(metalake, uri); + } } /** - * Retrieves the endpoint of the Gravitino Iceberg REST server. + * Retrieves the Iceberg REST server endpoint to route the given metalake's lakehouse-iceberg + * catalogs through. Prefers an explicitly configured {@code gravitino.iceberg.rest-uri}; + * otherwise falls back to the endpoint discovered from the Gravitino server for this metalake, if + * any. * - * @return the Iceberg REST server endpoint, or an empty string if not configured + * @param metalake the metalake to resolve the endpoint for + * @return the Iceberg REST server endpoint, or an empty string when none is available */ - public String getIcebergRestUri() { - return config.getOrDefault( - GRAVITINO_ICEBERG_REST_URI.key, GRAVITINO_ICEBERG_REST_URI.defaultValue); + public String getIcebergRestUri(String metalake) { + String configured = + config.getOrDefault( + GRAVITINO_ICEBERG_REST_URI.key, GRAVITINO_ICEBERG_REST_URI.defaultValue); + if (StringUtils.isNotBlank(configured)) { + return configured; + } + return discoveredIcebergRestUriByMetalake.getOrDefault(metalake, ""); } /** diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java index 34872d2e2f..e11c16b30b 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java @@ -197,6 +197,7 @@ public class CatalogConnectorManager { try { GravitinoMetalake metalake = metalakes.get(usedMetalake); LOG.debug("Load metalake: {}", usedMetalake); + refreshIcebergRestUri(usedMetalake); loadCatalogs(metalake); } catch (Exception e) { LOG.error("Load Metalake {} failed.", usedMetalake, e); @@ -207,6 +208,23 @@ public class CatalogConnectorManager { } } + /** + * Asks the Gravitino server whether it has an Iceberg REST server running for this metalake, and + * caches the answer on the shared {@link GravitinoConfig} for {@code IcebergConnectorAdapter} to + * read on the next catalog load. Failures — including talking to a Gravitino server older than + * this endpoint — must not interrupt catalog loading, so they are swallowed here; Iceberg + * catalogs simply keep their last known routing decision until the next successful poll. + */ + private void refreshIcebergRestUri(String metalakeName) { + try { + config.setDiscoveredIcebergRestUri( + metalakeName, gravitinoClient.icebergRestServiceUri(metalakeName).orElse(null)); + } catch (Exception e) { + LOG.debug( + "Failed to query the Iceberg REST service endpoint for metalake {}.", metalakeName, e); + } + } + /** * Retrieves a metalake by its name. * diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergCatalogPropertyConverter.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergCatalogPropertyConverter.java index dd8a492e1d..5d81fd02ab 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergCatalogPropertyConverter.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergCatalogPropertyConverter.java @@ -151,18 +151,21 @@ public class IcebergCatalogPropertyConverter extends CatalogPropertyConverter { * @param catalog the Gravitino catalog to load * @param gravitinoConfig the connector configuration holding the Iceberg REST server endpoint * @return the Trino Iceberg connector config - * @throws TrinoException if {@code gravitino.iceberg.rest-uri} is not configured + * @throws TrinoException if no Iceberg REST server endpoint is configured or discovered for this + * catalog's metalake */ public Map<String, String> buildIcebergRestProperties( GravitinoCatalog catalog, GravitinoConfig gravitinoConfig) { - String restUri = gravitinoConfig.getIcebergRestUri(); + String restUri = gravitinoConfig.getIcebergRestUri(catalog.getMetalake()); if (StringUtils.isBlank(restUri)) { + // The caller only reaches this method once it has already confirmed a non-blank URI, so + // this is defensive: it can only fire if that URI disappeared between the two calls. throw new TrinoException( GravitinoErrorCode.GRAVITINO_MISSING_CONFIG, - "Missing required config 'gravitino.iceberg.rest-uri'. Set it to the Gravitino Iceberg " - + "REST server endpoint, for example http://localhost:9001/iceberg, or set " - + "'gravitino.iceberg.rest-enabled=false' to load Iceberg catalogs through their " - + "catalog backend instead."); + "No Iceberg REST server endpoint is available for metalake '" + + catalog.getMetalake() + + "'. Set 'gravitino.iceberg.rest-uri' to override the address the Gravitino " + + "server reports."); } Map<String, String> config = new HashMap<>(); @@ -189,19 +192,6 @@ public class IcebergCatalogPropertyConverter extends CatalogPropertyConverter { // The server echoes `prefix` back as a config default, so setting it here is belt-and-braces. config.put(TRINO_ICEBERG_REST_WAREHOUSE, catalog.getName()); config.put(TRINO_ICEBERG_REST_PREFIX, catalog.getName()); - - if (!gravitinoConfig.singleMetalakeMode()) { - // The IRC serves exactly one metalake, and the prefix carries only the catalog name, so two - // metalakes holding a same-named catalog both resolve to whichever one the IRC was started - // with. - LOG.warn( - "Catalog '{}' in metalake '{}' is routed to the Iceberg REST server using the catalog " - + "name alone. In multi-metalake mode this is ambiguous: the Iceberg REST server " - + "serves a single metalake, so a same-named catalog in another metalake resolves to " - + "the same prefix.", - catalog.getName(), - catalog.getMetalake()); - } return config; } diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergConnectorAdapter.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergConnectorAdapter.java index 1480dc13d4..4cdd9cd1c0 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergConnectorAdapter.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergConnectorAdapter.java @@ -24,6 +24,7 @@ import io.trino.spi.session.PropertyMetadata; import java.util.HashMap; import java.util.List; import java.util.Map; +import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; import org.apache.gravitino.credential.Credential; import org.apache.gravitino.trino.connector.GravitinoConfig; @@ -64,10 +65,12 @@ public class IcebergConnectorAdapter implements CatalogConnectorAdapter { public Map<String, String> buildInternalConnectorConfig( GravitinoCatalog catalog, Credential[] credentials) throws Exception { // The catalog backend describes how Gravitino stores the metadata; it does not decide how - // Trino reaches the data. Unless the routing is disabled, every catalog is loaded through the - // Gravitino Iceberg REST server, the only path that supports temporary credentials. - // A catalog that already has a REST backend keeps pointing at its own configured endpoint. - if (config.isIcebergRestEnabled() + // Trino reaches the data. Whenever the Gravitino server reports a running Iceberg REST + // server for this catalog's metalake, the catalog is loaded through it, the only path that + // supports temporary credentials. A catalog that already has a REST backend keeps pointing at + // its own configured endpoint. If the server reports nothing, this falls back to translating + // catalog-backend as before — nothing to configure either way. + if (StringUtils.isNotBlank(config.getIcebergRestUri(catalog.getMetalake())) && !REST_CATALOG_BACKEND.equalsIgnoreCase( catalog.getProperty(IcebergConstants.CATALOG_BACKEND, null))) { // `credentials` is intentionally unused here: with vended credentials enabled, Trino obtains diff --git a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConfig.java b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConfig.java index 357280cf90..0528006823 100644 --- a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConfig.java +++ b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConfig.java @@ -304,8 +304,8 @@ public class TestGravitinoConfig { public void testIcebergRestConfigDefaults() { GravitinoConfig config = new GravitinoConfig(ImmutableMap.of("gravitino.metalake", "user_001")); - assertTrue(config.isIcebergRestEnabled()); - assertEquals("", config.getIcebergRestUri()); + // Nothing configured, and nothing discovered yet. + assertEquals("", config.getIcebergRestUri("user_001")); assertTrue(config.getIcebergRestCatalogConfig().isEmpty()); } @@ -315,8 +315,6 @@ public class TestGravitinoConfig { ImmutableMap.of( "gravitino.metalake", "user_001", - "gravitino.iceberg.rest-enabled", - "false", "gravitino.iceberg.rest-uri", "http://127.0.0.1:9001/iceberg";, "gravitino.iceberg.rest-catalog.security", @@ -325,8 +323,9 @@ public class TestGravitinoConfig { "client_id:client_secret"); GravitinoConfig config = new GravitinoConfig(configMap); - assertFalse(config.isIcebergRestEnabled()); - assertEquals("http://127.0.0.1:9001/iceberg";, config.getIcebergRestUri()); + // A manually configured URI applies regardless of metalake. + assertEquals("http://127.0.0.1:9001/iceberg";, config.getIcebergRestUri("user_001")); + assertEquals("http://127.0.0.1:9001/iceberg";, config.getIcebergRestUri("other_metalake")); Map<String, String> restCatalogConfig = config.getIcebergRestCatalogConfig(); assertEquals(2, restCatalogConfig.size()); @@ -335,6 +334,38 @@ public class TestGravitinoConfig { "client_id:client_secret", restCatalogConfig.get("iceberg.rest-catalog.oauth2.credential")); } + @Test + public void testDiscoveredIcebergRestUriIsPerMetalake() { + GravitinoConfig config = new GravitinoConfig(ImmutableMap.of("gravitino.metalake", "user_001")); + + config.setDiscoveredIcebergRestUri("metalake_a", "http://irc-a:9001/iceberg";); + config.setDiscoveredIcebergRestUri("metalake_b", "http://irc-b:9001/iceberg";); + + assertEquals("http://irc-a:9001/iceberg";, config.getIcebergRestUri("metalake_a")); + assertEquals("http://irc-b:9001/iceberg";, config.getIcebergRestUri("metalake_b")); + // Unknown/unmatched metalakes fall back to nothing. + assertEquals("", config.getIcebergRestUri("metalake_c")); + + // Clearing (e.g. the server stopped reporting an endpoint) removes it again. + config.setDiscoveredIcebergRestUri("metalake_a", null); + assertEquals("", config.getIcebergRestUri("metalake_a")); + } + + @Test + public void testManualIcebergRestUriOverridesDiscovery() { + GravitinoConfig config = + new GravitinoConfig( + ImmutableMap.of( + "gravitino.metalake", + "user_001", + "gravitino.iceberg.rest-uri", + "http://manual:9001/iceberg";)); + + config.setDiscoveredIcebergRestUri("user_001", "http://discovered:9001/iceberg";); + + assertEquals("http://manual:9001/iceberg";, config.getIcebergRestUri("user_001")); + } + @Test public void testToCatalogConfigWithIcebergRestProperties() { ImmutableMap<String, String> configMap = @@ -353,18 +384,6 @@ public class TestGravitinoConfig { assertTrue(catalogConfig.contains("\"gravitino.iceberg.rest-catalog.security\"='OAUTH2'")); } - @Test - public void testToCatalogConfigPropagatesIcebergRestEnabled() { - // The switch rides the exact-key loop rather than the prefix filter; if it fails to propagate, - // the coordinator and the workers build different configs for the same catalog. - GravitinoConfig config = - new GravitinoConfig( - ImmutableMap.of( - "gravitino.metalake", "user_001", "gravitino.iceberg.rest-enabled", "false")); - - assertTrue(config.toCatalogConfig().contains("\"gravitino.iceberg.rest-enabled\"='false'")); - } - private static boolean skipCatalog(String catalogName, GravitinoConfig config) { for (Pattern pattern : config.getSkipCatalogPatterns()) { if (pattern.matcher(catalogName).matches()) { diff --git a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/iceberg/TestIcebergCatalogPropertyConverter.java b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/iceberg/TestIcebergCatalogPropertyConverter.java index f09e4d7469..8ad738188f 100644 --- a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/iceberg/TestIcebergCatalogPropertyConverter.java +++ b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/iceberg/TestIcebergCatalogPropertyConverter.java @@ -138,7 +138,7 @@ public class TestIcebergCatalogPropertyConverter { Catalog mockCatalog = TestGravitinoCatalog.mockCatalog( name, "lakehouse-iceberg", "test catalog", Catalog.Type.RELATIONAL, properties); - IcebergConnectorAdapter adapter = new IcebergConnectorAdapter(icebergRestDisabledConfig()); + IcebergConnectorAdapter adapter = new IcebergConnectorAdapter(icebergRestUnavailableConfig()); Map<String, String> config = adapter.buildInternalConnectorConfig( @@ -178,7 +178,7 @@ public class TestIcebergCatalogPropertyConverter { Catalog mockCatalog = TestGravitinoCatalog.mockCatalog( name, "lakehouse-iceberg", "test catalog", Catalog.Type.RELATIONAL, properties); - IcebergConnectorAdapter adapter = new IcebergConnectorAdapter(icebergRestDisabledConfig()); + IcebergConnectorAdapter adapter = new IcebergConnectorAdapter(icebergRestUnavailableConfig()); Map<String, String> config = adapter.buildInternalConnectorConfig( @@ -215,7 +215,7 @@ public class TestIcebergCatalogPropertyConverter { Catalog mockCatalog = TestGravitinoCatalog.mockCatalog( name, "lakehouse-iceberg", "test catalog", Catalog.Type.RELATIONAL, properties); - IcebergConnectorAdapter adapter = new IcebergConnectorAdapter(icebergRestDisabledConfig()); + IcebergConnectorAdapter adapter = new IcebergConnectorAdapter(icebergRestUnavailableConfig()); Map<String, String> config = adapter.buildInternalConnectorConfig( @@ -242,7 +242,8 @@ public class TestIcebergCatalogPropertyConverter { .build(); Map<String, String> config = - buildConnectorConfig("catalog1", properties, icebergRestEnabledConfig(ImmutableMap.of())); + buildConnectorConfig( + "catalog1", properties, icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals("rest", config.get("iceberg.catalog.type")); Assertions.assertEquals( @@ -268,7 +269,8 @@ public class TestIcebergCatalogPropertyConverter { .build(); Map<String, String> config = - buildConnectorConfig("catalog1", properties, icebergRestEnabledConfig(ImmutableMap.of())); + buildConnectorConfig( + "catalog1", properties, icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals("rest", config.get("iceberg.catalog.type")); Assertions.assertEquals("catalog1", config.get("iceberg.rest-catalog.prefix")); @@ -285,7 +287,8 @@ public class TestIcebergCatalogPropertyConverter { .build(); Map<String, String> config = - buildConnectorConfig("catalog1", properties, icebergRestEnabledConfig(ImmutableMap.of())); + buildConnectorConfig( + "catalog1", properties, icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals("rest", config.get("iceberg.catalog.type")); Assertions.assertEquals( @@ -305,7 +308,7 @@ public class TestIcebergCatalogPropertyConverter { .build(); Map<String, String> config = - buildConnectorConfig("catalog1", properties, icebergRestDisabledConfig()); + buildConnectorConfig("catalog1", properties, icebergRestUnavailableConfig()); Assertions.assertEquals("jdbc", config.get("iceberg.catalog.type")); Assertions.assertEquals( @@ -315,18 +318,85 @@ public class TestIcebergCatalogPropertyConverter { } @Test - public void testBuildConnectorPropertiesMissingIcebergRestUri() { + public void testBuildConnectorPropertiesFallsBackWhenNoIcebergRestUriIsAvailable() + throws Exception { Map<String, String> properties = ImmutableMap.<String, String>builder() .put("uri", "jdbc:postgresql://localhost:5432/iceberg") .put("catalog-backend", "jdbc") .put("jdbc-driver", "org.postgresql.Driver") .build(); - GravitinoConfig gravitinoConfig = - new GravitinoConfig(ImmutableMap.of("gravitino.metalake", "test")); + + // Neither a manual gravitino.iceberg.rest-uri nor a discovered one for this metalake: the + // routing gate never fires, so this is the ordinary backend-translation path, not an error. + Map<String, String> config = + buildConnectorConfig("catalog1", properties, icebergRestUnavailableConfig()); + + Assertions.assertEquals("jdbc", config.get("iceberg.catalog.type")); + Assertions.assertNull(config.get("iceberg.rest-catalog.uri")); + } + + @Test + public void testBuildIcebergRestPropertiesThrowsOnBlankUri() { + // buildIcebergRestProperties keeps its own defensive check even though + // IcebergConnectorAdapter's routing gate means callers can no longer reach it with a blank + // URI under normal operation. + Catalog mockCatalog = + TestGravitinoCatalog.mockCatalog( + "catalog1", + "lakehouse-iceberg", + "test catalog", + Catalog.Type.RELATIONAL, + ImmutableMap.of("catalog-backend", "jdbc")); + GravitinoCatalog catalog = new GravitinoCatalog("test", mockCatalog); + IcebergCatalogPropertyConverter converter = new IcebergCatalogPropertyConverter(); Assertions.assertThrows( - TrinoException.class, () -> buildConnectorConfig("catalog1", properties, gravitinoConfig)); + TrinoException.class, + () -> converter.buildIcebergRestProperties(catalog, icebergRestUnavailableConfig())); + } + + @Test + public void testBuildConnectorPropertiesRoutesThroughDiscoveredIcebergRestUri() throws Exception { + Map<String, String> properties = + ImmutableMap.<String, String>builder() + .put("catalog-backend", "jdbc") + .put("uri", "jdbc:postgresql://localhost:5432/iceberg") + .put("jdbc-driver", "org.postgresql.Driver") + .build(); + + // No manual gravitino.iceberg.rest-uri: only a per-metalake value discovered from the + // Gravitino server, exactly as CatalogConnectorManager's periodic poll would set it. + Map<String, String> config = + buildConnectorConfig( + "catalog1", + properties, + icebergRestDiscoveredConfig("test", "http://discovered:9001/iceberg";)); + + Assertions.assertEquals("rest", config.get("iceberg.catalog.type")); + Assertions.assertEquals( + "http://discovered:9001/iceberg";, config.get("iceberg.rest-catalog.uri")); + } + + @Test + public void testBuildConnectorPropertiesIgnoresDiscoveryForOtherMetalakes() throws Exception { + Map<String, String> properties = + ImmutableMap.<String, String>builder() + .put("catalog-backend", "jdbc") + .put("uri", "jdbc:postgresql://localhost:5432/iceberg") + .put("jdbc-driver", "org.postgresql.Driver") + .build(); + + // Discovery reported an endpoint for a different metalake than the one this catalog belongs + // to (the mock catalog's metalake is "test"); it must not leak across metalakes. + Map<String, String> config = + buildConnectorConfig( + "catalog1", + properties, + icebergRestDiscoveredConfig("other_metalake", "http://other:9001/iceberg";)); + + Assertions.assertEquals("jdbc", config.get("iceberg.catalog.type")); + Assertions.assertNull(config.get("iceberg.rest-catalog.uri")); } @Test @@ -339,7 +409,7 @@ public class TestIcebergCatalogPropertyConverter { .put("warehouse", "s3://bucket/warehouse/") .build(); GravitinoConfig gravitinoConfig = - icebergRestEnabledConfig( + icebergRestConfiguredConfig( ImmutableMap.of( "gravitino.iceberg.rest-catalog.security", "OAUTH2", "gravitino.iceberg.rest-catalog.oauth2.credential", "client_id:client_secret", @@ -368,7 +438,7 @@ public class TestIcebergCatalogPropertyConverter { buildConnectorConfig( "catalog1", properties, - icebergRestEnabledConfig( + icebergRestConfiguredConfig( ImmutableMap.of("gravitino.client.session.forwardUser", "true"))); Assertions.assertEquals("USER", config.get("iceberg.rest-catalog.session")); @@ -376,14 +446,15 @@ public class TestIcebergCatalogPropertyConverter { buildConnectorConfig( "catalog1", properties, - icebergRestEnabledConfig( + icebergRestConfiguredConfig( ImmutableMap.of( "gravitino.client.session.forwardUser", "true", "gravitino.iceberg.rest-catalog.session", "NONE"))); Assertions.assertEquals("NONE", explicitConfig.get("iceberg.rest-catalog.session")); Map<String, String> defaultConfig = - buildConnectorConfig("catalog1", properties, icebergRestEnabledConfig(ImmutableMap.of())); + buildConnectorConfig( + "catalog1", properties, icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertNull(defaultConfig.get("iceberg.rest-catalog.session")); } @@ -397,7 +468,7 @@ public class TestIcebergCatalogPropertyConverter { "warehouse", "s3://bucket/warehouse/", "s3-region", "us-east-1", "s3-endpoint", "http://minio:9000";), - icebergRestEnabledConfig(ImmutableMap.of())); + icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals("true", s3Config.get("fs.native-s3.enabled")); Assertions.assertEquals("us-east-1", s3Config.get("s3.region")); Assertions.assertEquals("http://minio:9000";, s3Config.get("s3.endpoint")); @@ -407,7 +478,7 @@ public class TestIcebergCatalogPropertyConverter { buildConnectorConfig( "catalog1", ImmutableMap.of("catalog-backend", "jdbc", "warehouse", "gs://bucket/warehouse/"), - icebergRestEnabledConfig(ImmutableMap.of())); + icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals("true", gcsConfig.get("fs.native-gcs.enabled")); Assertions.assertNull(gcsConfig.get("fs.native-s3.enabled")); @@ -416,14 +487,14 @@ public class TestIcebergCatalogPropertyConverter { "catalog1", ImmutableMap.of( "catalog-backend", "jdbc", "warehouse", "abfss://container@account/warehouse/"), - icebergRestEnabledConfig(ImmutableMap.of())); + icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals("true", azureConfig.get("fs.native-azure.enabled")); Map<String, String> hdfsConfig = buildConnectorConfig( "catalog1", ImmutableMap.of("catalog-backend", "jdbc", "warehouse", "hdfs://namenode:9000/wh"), - icebergRestEnabledConfig(ImmutableMap.of())); + icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals("true", hdfsConfig.get("fs.hadoop.enabled")); Assertions.assertNull(hdfsConfig.get("fs.native-s3.enabled")); Assertions.assertNull(hdfsConfig.get("fs.native-gcs.enabled")); @@ -441,7 +512,8 @@ public class TestIcebergCatalogPropertyConverter { .build(); Map<String, String> config = - buildConnectorConfig("catalog1", properties, icebergRestEnabledConfig(ImmutableMap.of())); + buildConnectorConfig( + "catalog1", properties, icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals("true", config.get("iceberg.table-statistics-enabled")); // A catalog can override a derived default, so a Trino release renaming it is not a blocker. @@ -470,7 +542,7 @@ public class TestIcebergCatalogPropertyConverter { Map<String, String> config = buildConnectorConfig( - "catalog1", properties, icebergRestEnabledConfig(ImmutableMap.of()), credentials); + "catalog1", properties, icebergRestConfiguredConfig(ImmutableMap.of()), credentials); // The REST protocol vends a fresh credential per table access, so the catalog-level snapshot // must not be pinned into the connector config. @@ -482,7 +554,7 @@ public class TestIcebergCatalogPropertyConverter { // The backend path still applies them. Map<String, String> legacyConfig = - buildConnectorConfig("catalog1", properties, icebergRestDisabledConfig(), credentials); + buildConnectorConfig("catalog1", properties, icebergRestUnavailableConfig(), credentials); Assertions.assertEquals("root", legacyConfig.get("iceberg.jdbc-catalog.connection-user")); Assertions.assertEquals("AKIAEXAMPLE", legacyConfig.get("hive.s3.aws-access-key")); } @@ -500,7 +572,8 @@ public class TestIcebergCatalogPropertyConverter { .build(); Map<String, String> config = - buildConnectorConfig("catalog1", properties, icebergRestEnabledConfig(ImmutableMap.of())); + buildConnectorConfig( + "catalog1", properties, icebergRestConfiguredConfig(ImmutableMap.of())); // A catalog property must never redirect the connector at another endpoint or catalog. Assertions.assertEquals("rest", config.get("iceberg.catalog.type")); @@ -523,7 +596,7 @@ public class TestIcebergCatalogPropertyConverter { buildConnectorConfig( "catalog1", properties, - icebergRestEnabledConfig( + icebergRestConfiguredConfig( ImmutableMap.of("gravitino.iceberg.rest-catalog.security", "OAUTH2"))); // Cluster-level operational settings outrank per-catalog properties. @@ -538,7 +611,7 @@ public class TestIcebergCatalogPropertyConverter { buildConnectorConfig( "catalog1", ImmutableMap.of("catalog-backend", "jdbc", "warehouse", warehouse), - icebergRestEnabledConfig(ImmutableMap.of())); + icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals( "true", config.get("fs.native-s3.enabled"), "warehouse: " + warehouse); } @@ -548,7 +621,7 @@ public class TestIcebergCatalogPropertyConverter { buildConnectorConfig( "catalog1", ImmutableMap.of("catalog-backend", "jdbc", "warehouse", "gt_iceberg_rest"), - icebergRestEnabledConfig(ImmutableMap.of())); + icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals("true", schemeless.get("fs.hadoop.enabled")); Assertions.assertNull(schemeless.get("fs.native-s3.enabled")); @@ -557,7 +630,7 @@ public class TestIcebergCatalogPropertyConverter { buildConnectorConfig( "catalog1", ImmutableMap.of("catalog-backend", "jdbc"), - icebergRestEnabledConfig(ImmutableMap.of())); + icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals("true", noWarehouse.get("fs.hadoop.enabled")); Assertions.assertNull(noWarehouse.get("fs.native-s3.enabled")); Assertions.assertNull(noWarehouse.get("fs.native-gcs.enabled")); @@ -571,7 +644,7 @@ public class TestIcebergCatalogPropertyConverter { "catalog-backend", "jdbc", "warehouse", "oss://bucket/wh", "credential-providers", "oss-token"), - icebergRestEnabledConfig(ImmutableMap.of())); + icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals("true", oss.get("fs.hadoop.enabled")); Assertions.assertNull(oss.get("fs.native-s3.enabled")); } @@ -586,7 +659,7 @@ public class TestIcebergCatalogPropertyConverter { "warehouse", "s3://bucket/wh", "s3-endpoint", "http://minio:9000";, "s3-path-style-access", "true"), - icebergRestEnabledConfig(ImmutableMap.of())); + icebergRestConfiguredConfig(ImmutableMap.of())); // S3-compatible stores such as MinIO need path-style addressing to resolve the bucket. Assertions.assertEquals("true", config.get("s3.path-style-access")); @@ -599,7 +672,7 @@ public class TestIcebergCatalogPropertyConverter { buildConnectorConfig( "catalog1", ImmutableMap.of("catalog-backend", "REST", "uri", "http://other-irc:9001/iceberg";), - icebergRestEnabledConfig(ImmutableMap.of())); + icebergRestConfiguredConfig(ImmutableMap.of())); Assertions.assertEquals( "http://other-irc:9001/iceberg";, config.get("iceberg.rest-catalog.uri")); @@ -619,12 +692,13 @@ public class TestIcebergCatalogPropertyConverter { .buildInternalConnectorConfig(new GravitinoCatalog("test", mockCatalog), credentials); } - private static GravitinoConfig icebergRestDisabledConfig() { - return new GravitinoConfig( - ImmutableMap.of("gravitino.metalake", "test", "gravitino.iceberg.rest-enabled", "false")); + private static GravitinoConfig icebergRestUnavailableConfig() { + // No manual gravitino.iceberg.rest-uri, and nothing discovered for this metalake: the + // connector has no endpoint to route through, so catalogs fall back to their own backend. + return new GravitinoConfig(ImmutableMap.of("gravitino.metalake", "test")); } - private static GravitinoConfig icebergRestEnabledConfig(Map<String, String> extraConfig) { + private static GravitinoConfig icebergRestConfiguredConfig(Map<String, String> extraConfig) { return new GravitinoConfig( ImmutableMap.<String, String>builder() .put("gravitino.metalake", "test") @@ -632,4 +706,10 @@ public class TestIcebergCatalogPropertyConverter { .putAll(extraConfig) .build()); } + + private static GravitinoConfig icebergRestDiscoveredConfig(String metalake, String uri) { + GravitinoConfig config = new GravitinoConfig(ImmutableMap.of("gravitino.metalake", metalake)); + config.setDiscoveredIcebergRestUri(metalake, uri); + return config; + } }
