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 8e0ac683103b4b7cf25864b87acccd4322810f17 Author: diqiu50 <[email protected]> AuthorDate: Mon Aug 24 15:57:54 2026 +0800 [Cherry-pick to branch-1.3] [#12554] improvement(trino-connector): Fix IRC discovery port default, worker propagation, and other review findings - Reported endpoint defaulted to the Gravitino server's own port (8090) whenever gravitino.iceberg-rest.httpPort was not set explicitly, because ServerConfig does not implement OverwriteDefaultConfig the way IcebergConfig does, so JettyServerConfig fell back to the Gravitino webserver's default instead of the Iceberg REST server's documented default (9001). Read the raw config with the correct defaults instead of going through JettyServerConfig. - The HTTPS branch used the HTTP port. Use the HTTPS port. - The discovered endpoint never reached Trino workers: only the coordinator runs the periodic discovery poll, and the discovered map lived only in that node's own GravitinoConfig. The coordinator now embeds the resolved endpoint into the catalog's own properties before registering it, so it travels to every node through the CREATE CATALOG statement Trino already replicates cluster-wide, the same way GravitinoCatalog.toJson already carries every other catalog property. IcebergConnectorAdapter reads it from the catalog instead of GravitinoConfig's discovered map, which only the manual override (plain local config, identical everywhere already) still uses. - Fixed an AuxiliaryServiceManager edit that had left @VisibleForTesting attached to the wrong method and a Javadoc block orphaned after an annotation instead of before the declaration it was meant to document. - Added Cache-Control: no-store to the discovery response, since the reported host can depend on the caller's own Host header, and broadened the wildcard-host match to cover [::] and the expanded IPv6 form. - Documented that discovery only affects a catalog at its next registration or reload, not retroactively, and that it applies to auxiliary-service IRC only, not standalone IRC. - Added unit tests: AuxiliaryServiceManager.isAuxServiceRegistered, and a full suite for IcebergRESTServiceOperations covering the port regression, https, explicit ports, wildcard/blank host fallback, metalake matching, and cache headers. (cherry picked from commit 26520805dd9c89ca9d76a5b5202b381f48744ff8) --- .../auxiliary/AuxiliaryServiceManager.java | 2 +- .../auxiliary/TestAuxiliaryServiceManager.java | 27 ++++ docs/trino-connector/catalog-iceberg.md | 35 +++-- .../web/rest/IcebergRESTServiceOperations.java | 70 +++++++-- .../web/rest/TestIcebergRESTServiceOperations.java | 174 +++++++++++++++++++++ .../gravitino/trino/connector/GravitinoConfig.java | 38 +++-- .../trino/connector/catalog/CatalogRegister.java | 9 +- .../iceberg/IcebergCatalogPropertyConverter.java | 10 +- .../catalog/iceberg/IcebergConnectorAdapter.java | 71 ++++++++- .../trino/connector/TestGravitinoConfig.java | 31 +--- .../TestIcebergCatalogPropertyConverter.java | 61 +++++++- 11 files changed, 451 insertions(+), 77 deletions(-) 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 accb23f799..c1c0c14235 100644 --- a/core/src/main/java/org/apache/gravitino/auxiliary/AuxiliaryServiceManager.java +++ b/core/src/main/java/org/apache/gravitino/auxiliary/AuxiliaryServiceManager.java @@ -94,7 +94,6 @@ public class AuxiliaryServiceManager { }); } - @VisibleForTesting /** * Returns whether the given auxiliary service was configured and registered. * @@ -105,6 +104,7 @@ public class AuxiliaryServiceManager { return auxServices.containsKey(auxServiceName); } + @VisibleForTesting public IsolatedClassLoader getIsolatedClassLoader(List<String> classPaths) { return IsolatedClassLoader.buildClassLoader(classPaths); } diff --git a/core/src/test/java/org/apache/gravitino/auxiliary/TestAuxiliaryServiceManager.java b/core/src/test/java/org/apache/gravitino/auxiliary/TestAuxiliaryServiceManager.java index d5ff499fc8..5df4b897f2 100644 --- a/core/src/test/java/org/apache/gravitino/auxiliary/TestAuxiliaryServiceManager.java +++ b/core/src/test/java/org/apache/gravitino/auxiliary/TestAuxiliaryServiceManager.java @@ -112,6 +112,33 @@ public class TestAuxiliaryServiceManager { verify(auxService2, times(1)).serviceStop(); } + @Test + public void testIsAuxServiceRegistered() throws Exception { + GravitinoAuxiliaryService auxService = mock(GravitinoAuxiliaryService.class); + IsolatedClassLoader isolatedClassLoader = + new IsolatedClassLoader( + Collections.emptyList(), Collections.emptyList(), Collections.emptyList()); + + AuxiliaryServiceManager auxServiceManager = new AuxiliaryServiceManager(); + AuxiliaryServiceManager spyAuxManager = spy(auxServiceManager); + doReturn(isolatedClassLoader).when(spyAuxManager).getIsolatedClassLoader(anyList()); + doReturn(auxService).when(spyAuxManager).loadAuxService("iceberg-rest", isolatedClassLoader); + + Assertions.assertFalse(spyAuxManager.isAuxServiceRegistered("iceberg-rest")); + + spyAuxManager.serviceInit( + DummyConfig.of( + ImmutableMap.of( + AuxiliaryServiceManager.GRAVITINO_AUX_SERVICE_PREFIX + + AuxiliaryServiceManager.AUX_SERVICE_NAMES, + "iceberg-rest", + "gravitino.iceberg-rest." + AuxiliaryServiceManager.AUX_SERVICE_CLASSPATH, + "/tmp"))); + + Assertions.assertTrue(spyAuxManager.isAuxServiceRegistered("iceberg-rest")); + Assertions.assertFalse(spyAuxManager.isAuxServiceRegistered("lance-rest")); + } + @Test void testAuxiliaryServiceConfigs() { Map<String, String> m = diff --git a/docs/trino-connector/catalog-iceberg.md b/docs/trino-connector/catalog-iceberg.md index 02779e8041..5a214fc624 100644 --- a/docs/trino-connector/catalog-iceberg.md +++ b/docs/trino-connector/catalog-iceberg.md @@ -34,16 +34,25 @@ IRC means every table access gets a freshly issued temporary credential over the protocol. 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): +first place), so it also asks that server whether it has an Iceberg REST server running as an +[auxiliary service](../iceberg-rest-service.md) 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 as an auxiliary service, or it serves a different metalake — `lakehouse-iceberg` +catalogs fall back to translating `catalog-backend` as before, and credential vending does not work. + +Only the coordinator polls the Gravitino server, so the coordinator resolves the endpoint once, when +a catalog is first registered or reloaded, and hands it to every node (coordinator and workers alike) +as part of that catalog's own definition — the same way Trino replicates any other catalog property +cluster-wide. A practical consequence: a catalog registered before the IRC started keeps its existing +routing until Gravitino reports a change *and* the catalog itself is reloaded (its metadata changes, +or Trino restarts) — starting the IRC alone does not retroactively re-route an already-registered +catalog. + +Set `gravitino.iceberg.rest-uri` to override the discovered endpoint, and it is required — not just +an override — for a standalone IRC (its own process, not the Gravitino server's auxiliary service): +the Gravitino server has no way to know a standalone IRC exists, so discovery never finds one. See +[Limitations](#limitations). ```properties connector.name=gravitino @@ -120,6 +129,12 @@ keeping per-user credential vending and per-user authorization intact. Set - 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. +- Discovery only works for an IRC running as a Gravitino auxiliary service + (`gravitino.auxService.names=iceberg-rest`), embedded in the same process as the Gravitino server. + A standalone IRC — its own process, started with `GravitinoIcebergRESTServer` and its own + `gravitino-iceberg-rest-server.conf` — never registers with the Gravitino server, so the server + has no way to know it exists; the Gravitino server reports no endpoint even while a standalone IRC + is running. Set `gravitino.iceberg.rest-uri` manually in this case. ## Schema Operations 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 index 0ea698a6a4..f296e068db 100644 --- 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 @@ -32,9 +32,9 @@ 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.auxiliary.AuxiliaryServiceManager; 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; /** @@ -52,6 +52,17 @@ public class IcebergRESTServiceOperations { // 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"; + private static final String HOST_KEY = CONFIG_PREFIX + "host"; + private static final String HTTP_PORT_KEY = CONFIG_PREFIX + "httpPort"; + private static final String HTTPS_PORT_KEY = CONFIG_PREFIX + "httpsPort"; + private static final String ENABLE_HTTPS_KEY = CONFIG_PREFIX + "enableHttps"; + // Match IcebergConfig.DEFAULT_ICEBERG_REST_SERVICE_HTTP_PORT/HTTPS_PORT: the server module + // cannot depend on iceberg-common, and JettyServerConfig's own defaults are the Gravitino + // server's (8090/8433), not the Iceberg REST server's — reading raw values with these + // defaults avoids silently reporting the wrong port when httpPort is not set explicitly. + private static final int DEFAULT_HTTP_PORT = 9001; + private static final int DEFAULT_HTTPS_PORT = 9433; + private static final String DEFAULT_HOST = "0.0.0.0"; @Context private HttpServletRequest httpRequest; @@ -67,15 +78,33 @@ public class IcebergRESTServiceOperations { @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))); + // The reported host can depend on the caller's own Host header (see resolveUri), so this + // response must never be cached and replayed to a different caller. + return Response.fromResponse(Utils.ok(new IcebergRESTServiceResponse(resolveUri(metalake)))) + .header("Cache-Control", "no-store") + .build(); + } + + // Overridable so tests can inject a fixture without bootstrapping GravitinoEnv, matching + // HealthOperations's testing pattern. + AuxiliaryServiceManager getAuxServiceManager() { + return GravitinoEnv.getInstance().auxServiceManager(); + } + + Config getConfig() { + return GravitinoEnv.getInstance().config(); + } + + HttpServletRequest getHttpRequest() { + return httpRequest; } private String resolveUri(String metalake) { - if (!GravitinoEnv.getInstance().auxServiceManager().isAuxServiceRegistered(AUX_SERVICE_NAME)) { + if (!getAuxServiceManager().isAuxServiceRegistered(AUX_SERVICE_NAME)) { return null; } - Config config = GravitinoEnv.getInstance().config(); + Config config = getConfig(); String servedMetalake = config.getRawString(SERVED_METALAKE_KEY, ""); if (StringUtils.isNotBlank(metalake) && StringUtils.isNotBlank(servedMetalake) @@ -85,21 +114,42 @@ public class IcebergRESTServiceOperations { return null; } - JettyServerConfig icebergRestConfig = JettyServerConfig.fromConfig(config, CONFIG_PREFIX); - String host = icebergRestConfig.getHost(); + String host = config.getRawString(HOST_KEY, DEFAULT_HOST); 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(); + host = getHttpRequest().getServerName(); + } + boolean enableHttps = Boolean.parseBoolean(config.getRawString(ENABLE_HTTPS_KEY, "false")); + String scheme = enableHttps ? "https" : "http"; + int port = + parsePort( + config, + enableHttps ? HTTPS_PORT_KEY : HTTP_PORT_KEY, + enableHttps ? DEFAULT_HTTPS_PORT : DEFAULT_HTTP_PORT); + return String.format("%s://%s:%d/iceberg", scheme, host, port); + } + + private static int parsePort(Config config, String key, int defaultPort) { + String value = config.getRawString(key, ""); + if (StringUtils.isBlank(value)) { + return defaultPort; + } + try { + return Integer.parseInt(value.trim()); + } catch (NumberFormatException e) { + return defaultPort; } - 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); + return StringUtils.isBlank(host) + || "0.0.0.0".equals(host) + || "::".equals(host) + || "[::]".equals(host) + || "0:0:0:0:0:0:0:0".equals(host); } } diff --git a/server/src/test/java/org/apache/gravitino/server/web/rest/TestIcebergRESTServiceOperations.java b/server/src/test/java/org/apache/gravitino/server/web/rest/TestIcebergRESTServiceOperations.java new file mode 100644 index 0000000000..d295f75ada --- /dev/null +++ b/server/src/test/java/org/apache/gravitino/server/web/rest/TestIcebergRESTServiceOperations.java @@ -0,0 +1,174 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.google.common.collect.ImmutableMap; +import java.util.Map; +import javax.servlet.http.HttpServletRequest; +import javax.ws.rs.core.Response; +import org.apache.gravitino.Config; +import org.apache.gravitino.auxiliary.AuxiliaryServiceManager; +import org.apache.gravitino.dto.responses.IcebergRESTServiceResponse; +import org.junit.jupiter.api.Test; + +public class TestIcebergRESTServiceOperations { + + private static class DummyConfig extends Config { + static DummyConfig of(Map<String, String> m) { + DummyConfig config = new DummyConfig(); + config.loadFromMap(m, k -> true); + return config; + } + } + + private IcebergRESTServiceOperations newOps( + boolean registered, Map<String, String> icebergConfig, String requestServerName) { + AuxiliaryServiceManager auxServiceManager = mock(AuxiliaryServiceManager.class); + when(auxServiceManager.isAuxServiceRegistered("iceberg-rest")).thenReturn(registered); + Config config = DummyConfig.of(icebergConfig); + HttpServletRequest request = mock(HttpServletRequest.class); + when(request.getServerName()).thenReturn(requestServerName); + + return new IcebergRESTServiceOperations() { + @Override + AuxiliaryServiceManager getAuxServiceManager() { + return auxServiceManager; + } + + @Override + Config getConfig() { + return config; + } + + @Override + HttpServletRequest getHttpRequest() { + return request; + } + }; + } + + private String uriOf(Response response) { + return ((IcebergRESTServiceResponse) response.getEntity()).getUri(); + } + + @Test + public void testReturnsNullWhenAuxServiceNotRegistered() { + IcebergRESTServiceOperations ops = newOps(false, ImmutableMap.of(), "gravitino-host"); + assertNull(uriOf(ops.getIcebergRestServiceUri("test"))); + } + + @Test + public void testDefaultPortIsTheIcebergRestDefaultNotTheGravitinoServerDefault() { + // Regression test: without an explicit gravitino.iceberg-rest.httpPort, the reported port + // must be the Iceberg REST server's own default (9001), not the Gravitino webserver's + // default (8090) that JettyServerConfig would otherwise fall back to. + IcebergRESTServiceOperations ops = + newOps(true, ImmutableMap.of("gravitino.iceberg-rest.host", "irc-host"), "gravitino-host"); + assertEquals("http://irc-host:9001/iceberg";, uriOf(ops.getIcebergRestServiceUri(""))); + } + + @Test + public void testHttpsUsesTheHttpsPortNotTheHttpPort() { + IcebergRESTServiceOperations ops = + newOps( + true, + ImmutableMap.of( + "gravitino.iceberg-rest.host", "irc-host", + "gravitino.iceberg-rest.enableHttps", "true"), + "gravitino-host"); + assertEquals("https://irc-host:9433/iceberg";, uriOf(ops.getIcebergRestServiceUri(""))); + } + + @Test + public void testExplicitPortIsHonored() { + IcebergRESTServiceOperations ops = + newOps( + true, + ImmutableMap.of( + "gravitino.iceberg-rest.host", "irc-host", + "gravitino.iceberg-rest.httpPort", "19001"), + "gravitino-host"); + assertEquals("http://irc-host:19001/iceberg";, uriOf(ops.getIcebergRestServiceUri(""))); + } + + @Test + public void testWildcardHostFallsBackToRequestServerName() { + IcebergRESTServiceOperations ops = + newOps( + true, + ImmutableMap.of("gravitino.iceberg-rest.host", "0.0.0.0"), + "host.docker.internal"); + assertEquals( + "http://host.docker.internal:9001/iceberg";, uriOf(ops.getIcebergRestServiceUri(""))); + } + + @Test + public void testBlankHostIsTreatedAsWildcard() { + IcebergRESTServiceOperations ops = newOps(true, ImmutableMap.of(), "gravitino-host"); + assertEquals("http://gravitino-host:9001/iceberg";, uriOf(ops.getIcebergRestServiceUri(""))); + } + + @Test + public void testMismatchedMetalakeReturnsNull() { + IcebergRESTServiceOperations ops = + newOps( + true, + ImmutableMap.of( + "gravitino.iceberg-rest.host", "irc-host", + "gravitino.iceberg-rest.gravitino-metalake", "prod"), + "gravitino-host"); + assertNull(uriOf(ops.getIcebergRestServiceUri("test"))); + } + + @Test + public void testMatchingMetalakeIsReported() { + IcebergRESTServiceOperations ops = + newOps( + true, + ImmutableMap.of( + "gravitino.iceberg-rest.host", "irc-host", + "gravitino.iceberg-rest.gravitino-metalake", "test"), + "gravitino-host"); + assertEquals("http://irc-host:9001/iceberg";, uriOf(ops.getIcebergRestServiceUri("test"))); + } + + @Test + public void testBlankRequestedMetalakeSkipsTheMetalakeCheck() { + IcebergRESTServiceOperations ops = + newOps( + true, + ImmutableMap.of( + "gravitino.iceberg-rest.host", "irc-host", + "gravitino.iceberg-rest.gravitino-metalake", "prod"), + "gravitino-host"); + assertEquals("http://irc-host:9001/iceberg";, uriOf(ops.getIcebergRestServiceUri(""))); + } + + @Test + public void testResponseIsNotCacheable() { + IcebergRESTServiceOperations ops = newOps(true, ImmutableMap.of(), "gravitino-host"); + Response response = ops.getIcebergRestServiceUri(""); + assertEquals("no-store", response.getHeaderString("Cache-Control")); + } +} 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 2c1f795afb..b86fa5c5dc 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 @@ -710,9 +710,12 @@ public class GravitinoConfig { /** * 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. + * metalake. Called only by the catalog connector manager's periodic metalake poll, which runs on + * the coordinator only — {@link #getDiscoveredIcebergRestUri} is therefore not by itself a valid + * routing signal on a worker node. The coordinator is responsible for embedding the discovered + * value into each catalog's own properties at registration time, so that it travels to every node + * through the {@code CREATE CATALOG} statement Trino replicates cluster-wide; see {@code + * CatalogRegister.generateCreateCatalogCommand}. * * @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 @@ -727,24 +730,29 @@ public class GravitinoConfig { } /** - * 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. + * Retrieves the Iceberg REST server endpoint discovered from the Gravitino server for the given + * metalake, with no fallback to the manually configured endpoint. Only valid on the node that + * runs the periodic discovery poll (the coordinator); see {@link #setDiscoveredIcebergRestUri}. * * @param metalake the metalake to resolve the endpoint for - * @return the Iceberg REST server endpoint, or an empty string when none is available + * @return the discovered endpoint, or an empty string when none is available */ - 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; - } + public String getDiscoveredIcebergRestUri(String metalake) { return discoveredIcebergRestUriByMetalake.getOrDefault(metalake, ""); } + /** + * Retrieves the manually configured {@code gravitino.iceberg.rest-uri}, if any. Unlike the + * discovered endpoint, this is plain local file configuration and is therefore identical and + * valid on every node — coordinator and workers alike. + * + * @return the manually configured Iceberg REST server endpoint, or an empty string when unset + */ + public String getManualIcebergRestUri() { + return config.getOrDefault( + GRAVITINO_ICEBERG_REST_URI.key, GRAVITINO_ICEBERG_REST_URI.defaultValue); + } + /** * Retrieves the properties passed through to the internal Trino Iceberg REST catalog, with the * {@code gravitino.iceberg.rest-catalog.} prefix rewritten to {@code iceberg.rest-catalog.}. diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegister.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegister.java index faca9c2499..35f722bb1e 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegister.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegister.java @@ -38,6 +38,7 @@ import java.util.Set; import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.trino.connector.GravitinoConfig; import org.apache.gravitino.trino.connector.GravitinoErrorCode; +import org.apache.gravitino.trino.connector.catalog.iceberg.IcebergConnectorAdapter; import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -304,12 +305,18 @@ public class CatalogRegister { private String generateCreateCatalogCommand(String name, GravitinoCatalog gravitinoCatalog) throws Exception { + // This statement is replicated by Trino to every node in the cluster, coordinator and workers + // alike, so it is the only place a value the coordinator alone knows (like the Iceberg REST + // server endpoint it discovered) can reach every node that will build this catalog's internal + // connector config. + GravitinoCatalog catalogToRegister = + IcebergConnectorAdapter.embedDiscoveredIcebergRestUri(gravitinoCatalog, config); return String.format( "CREATE CATALOG %s USING gravitino WITH ( \"%s\" = 'true', \"%s\" = '%s', %s)", name, GRAVITINO_DYNAMIC_CONNECTOR, GRAVITINO_DYNAMIC_CONNECTOR_CATALOG_CONFIG, - GravitinoCatalog.toJson(gravitinoCatalog), + GravitinoCatalog.toJson(catalogToRegister), config.toCatalogConfig()); } 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 5d81fd02ab..14aed17709 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 @@ -149,14 +149,14 @@ public class IcebergCatalogPropertyConverter extends CatalogPropertyConverter { * the backend path by {@link #applyCredentials}. * * @param catalog the Gravitino catalog to load - * @param gravitinoConfig the connector configuration holding the Iceberg REST server endpoint + * @param gravitinoConfig the connector configuration holding the Iceberg REST catalog's + * authentication and other pass-through settings + * @param restUri the Iceberg REST server endpoint to route through, resolved by the caller * @return the Trino Iceberg connector config - * @throws TrinoException if no Iceberg REST server endpoint is configured or discovered for this - * catalog's metalake + * @throws TrinoException if {@code restUri} is blank */ public Map<String, String> buildIcebergRestProperties( - GravitinoCatalog catalog, GravitinoConfig gravitinoConfig) { - String restUri = gravitinoConfig.getIcebergRestUri(catalog.getMetalake()); + GravitinoCatalog catalog, GravitinoConfig gravitinoConfig, String restUri) { 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. 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 4cdd9cd1c0..85df20d944 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 @@ -20,6 +20,7 @@ package org.apache.gravitino.trino.connector.catalog.iceberg; import static java.util.Collections.emptyList; +import com.google.common.collect.ImmutableMap; import io.trino.spi.session.PropertyMetadata; import java.util.HashMap; import java.util.List; @@ -44,6 +45,18 @@ public class IcebergConnectorAdapter implements CatalogConnectorAdapter { private static final String CONNECTOR_ICEBERG = "iceberg"; private static final String REST_CATALOG_BACKEND = "rest"; + private static final String ICEBERG_PROVIDER = "lakehouse-iceberg"; + + /** + * Synthetic catalog property carrying the Iceberg REST server endpoint the coordinator discovered + * for this catalog's metalake. {@link GravitinoConfig}'s own discovered-endpoint map is populated + * only on the coordinator (the periodic discovery poll never runs on a worker), so it cannot be + * read directly when building a catalog's internal connector config: every node needs the same + * routing decision for the same catalog. Embedding the resolved endpoint into the catalog itself, + * at registration time, means it travels to every node through the {@code CREATE CATALOG} + * statement Trino replicates cluster-wide, the same way any other catalog property does. + */ + static final String DISCOVERED_ICEBERG_REST_URI_PROPERTY = "__gravitino.iceberg.rest-uri"; private final IcebergPropertyMeta propertyMetadata; private final IcebergCatalogPropertyConverter catalogConverter; @@ -65,12 +78,21 @@ 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. 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())) + // Trino reaches the data. Whenever an Iceberg REST server endpoint is available 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 no endpoint is available, this falls back to translating catalog-backend as + // before — nothing to configure either way. + // + // The manual override is plain local config, so it is valid on every node as-is. The + // discovered endpoint is coordinator-only knowledge, so it is read from the catalog's own + // properties, where the coordinator embeds it at registration time (see + // embedDiscoveredIcebergRestUri), rather than from GravitinoConfig directly. + String restUri = config.getManualIcebergRestUri(); + if (StringUtils.isBlank(restUri)) { + restUri = catalog.getProperty(DISCOVERED_ICEBERG_REST_URI_PROPERTY, ""); + } + if (StringUtils.isNotBlank(restUri) && !REST_CATALOG_BACKEND.equalsIgnoreCase( catalog.getProperty(IcebergConstants.CATALOG_BACKEND, null))) { // `credentials` is intentionally unused here: with vended credentials enabled, Trino obtains @@ -81,7 +103,7 @@ public class IcebergConnectorAdapter implements CatalogConnectorAdapter { + " are not applied because the REST protocol vends one per table access.", catalog.getName(), credentials.length); - return catalogConverter.buildIcebergRestProperties(catalog, config); + return catalogConverter.buildIcebergRestProperties(catalog, config, restUri); } Map<String, String> connectorConfig = @@ -90,6 +112,41 @@ public class IcebergConnectorAdapter implements CatalogConnectorAdapter { return connectorConfig; } + /** + * Returns a copy of {@code catalog} with the Iceberg REST server endpoint the coordinator + * discovered for its metalake embedded as a synthetic property, if the catalog is a + * lakehouse-iceberg catalog and a discovered endpoint exists. Called only on the coordinator, + * before a catalog is registered with Trino, so that the routing decision reaches every node + * through the {@code CREATE CATALOG} statement — see {@link + * #DISCOVERED_ICEBERG_REST_URI_PROPERTY}. + * + * @param catalog the catalog about to be registered + * @param config the connector configuration holding the discovered endpoints + * @return {@code catalog} unchanged if it is not a lakehouse-iceberg catalog or no endpoint was + * discovered for its metalake; otherwise a copy with the endpoint embedded + */ + public static GravitinoCatalog embedDiscoveredIcebergRestUri( + GravitinoCatalog catalog, GravitinoConfig config) { + if (!ICEBERG_PROVIDER.equals(catalog.getProvider())) { + return catalog; + } + String discoveredUri = config.getDiscoveredIcebergRestUri(catalog.getMetalake()); + if (StringUtils.isBlank(discoveredUri)) { + return catalog; + } + Map<String, String> properties = + ImmutableMap.<String, String>builder() + .putAll(catalog.getProperties()) + .put(DISCOVERED_ICEBERG_REST_URI_PROPERTY, discoveredUri) + .buildKeepingLast(); + return new GravitinoCatalog( + catalog.getMetalake(), + catalog.getProvider(), + catalog.getName(), + properties, + catalog.getLastModifiedTime()); + } + @Override public String internalConnectorName() { return CONNECTOR_ICEBERG; 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 0528006823..5d03d2e507 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 @@ -305,7 +305,8 @@ public class TestGravitinoConfig { GravitinoConfig config = new GravitinoConfig(ImmutableMap.of("gravitino.metalake", "user_001")); // Nothing configured, and nothing discovered yet. - assertEquals("", config.getIcebergRestUri("user_001")); + assertEquals("", config.getManualIcebergRestUri()); + assertEquals("", config.getDiscoveredIcebergRestUri("user_001")); assertTrue(config.getIcebergRestCatalogConfig().isEmpty()); } @@ -323,9 +324,8 @@ public class TestGravitinoConfig { "client_id:client_secret"); GravitinoConfig config = new GravitinoConfig(configMap); - // 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")); + // A manually configured URI is plain local config, not scoped to any metalake. + assertEquals("http://127.0.0.1:9001/iceberg";, config.getManualIcebergRestUri()); Map<String, String> restCatalogConfig = config.getIcebergRestCatalogConfig(); assertEquals(2, restCatalogConfig.size()); @@ -341,29 +341,14 @@ public class TestGravitinoConfig { 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")); + assertEquals("http://irc-a:9001/iceberg";, config.getDiscoveredIcebergRestUri("metalake_a")); + assertEquals("http://irc-b:9001/iceberg";, config.getDiscoveredIcebergRestUri("metalake_b")); // Unknown/unmatched metalakes fall back to nothing. - assertEquals("", config.getIcebergRestUri("metalake_c")); + assertEquals("", config.getDiscoveredIcebergRestUri("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")); + assertEquals("", config.getDiscoveredIcebergRestUri("metalake_a")); } @Test 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 8ad738188f..ee3e448c29 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 @@ -353,7 +353,7 @@ public class TestIcebergCatalogPropertyConverter { Assertions.assertThrows( TrinoException.class, - () -> converter.buildIcebergRestProperties(catalog, icebergRestUnavailableConfig())); + () -> converter.buildIcebergRestProperties(catalog, icebergRestUnavailableConfig(), "")); } @Test @@ -366,12 +366,15 @@ public class TestIcebergCatalogPropertyConverter { .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. + // Gravitino server, embedded onto the catalog exactly as CatalogRegister does before + // registering it — this is the only way a worker (which never runs discovery itself) can see + // the same routing decision as the coordinator. Map<String, String> config = buildConnectorConfig( "catalog1", properties, - icebergRestDiscoveredConfig("test", "http://discovered:9001/iceberg";)); + icebergRestDiscoveredConfig("test", "http://discovered:9001/iceberg";), + /* embedDiscovery= */ true); Assertions.assertEquals("rest", config.get("iceberg.catalog.type")); Assertions.assertEquals( @@ -388,12 +391,36 @@ public class TestIcebergCatalogPropertyConverter { .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. + // to (the mock catalog's metalake is "test"); embedDiscoveredIcebergRestUri must not embed it. Map<String, String> config = buildConnectorConfig( "catalog1", properties, - icebergRestDiscoveredConfig("other_metalake", "http://other:9001/iceberg";)); + icebergRestDiscoveredConfig("other_metalake", "http://other:9001/iceberg";), + /* embedDiscovery= */ true); + + Assertions.assertEquals("jdbc", config.get("iceberg.catalog.type")); + Assertions.assertNull(config.get("iceberg.rest-catalog.uri")); + } + + @Test + public void testDiscoveredUriUnusedWithoutEmbedding() throws Exception { + // A worker never runs discovery, so its own GravitinoConfig never has a discovered value + // populated — this asserts that IcebergConnectorAdapter really does read the routing signal + // from the catalog, not from GravitinoConfig's discovered map directly. + 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(); + + Map<String, String> config = + buildConnectorConfig( + "catalog1", + properties, + icebergRestDiscoveredConfig("test", "http://discovered:9001/iceberg";), + /* embedDiscovery= */ false); Assertions.assertEquals("jdbc", config.get("iceberg.catalog.type")); Assertions.assertNull(config.get("iceberg.rest-catalog.uri")); @@ -692,6 +719,30 @@ public class TestIcebergCatalogPropertyConverter { .buildInternalConnectorConfig(new GravitinoCatalog("test", mockCatalog), credentials); } + /** + * Like {@link #buildConnectorConfig(String, Map, GravitinoConfig, Credential[])}, but for + * discovered-endpoint scenarios: {@code embedDiscovery} mirrors what {@code CatalogRegister} does + * before registering a catalog, so a discovered value only reaches the adapter the same way it + * would on a real node — through the catalog, never by reading {@code GravitinoConfig}'s + * discovered map directly. + */ + private static Map<String, String> buildConnectorConfig( + String catalogName, + Map<String, String> properties, + GravitinoConfig gravitinoConfig, + boolean embedDiscovery) + throws Exception { + Catalog mockCatalog = + TestGravitinoCatalog.mockCatalog( + catalogName, "lakehouse-iceberg", "test catalog", Catalog.Type.RELATIONAL, properties); + GravitinoCatalog catalog = new GravitinoCatalog("test", mockCatalog); + if (embedDiscovery) { + catalog = IcebergConnectorAdapter.embedDiscoveredIcebergRestUri(catalog, gravitinoConfig); + } + return new IcebergConnectorAdapter(gravitinoConfig) + .buildInternalConnectorConfig(catalog, new Credential[0]); + } + 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.
