This is an automated email from the ASF dual-hosted git repository. diqiu50 pushed a commit to branch trino-irc-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 378829b44be2038c8830fe5e30ec89a80568a9c9 Author: Yuhui <[email protected]> AuthorDate: Mon Aug 24 20:27:35 2026 +0800 [Cherry-pick to branch-1.3] [#12542] improvement(trino-connector): Support TLS and session role for the internal JDBC connection (#12543) --- docs/trino-connector/configuration.md | 125 ++++++- .../gravitino/trino/connector/GravitinoConfig.java | 242 ++++++++++++- .../trino/connector/GravitinoConnectorFactory.java | 49 ++- .../connector/catalog/CatalogConnectorManager.java | 17 +- .../trino/connector/catalog/CatalogRegister.java | 214 +++++++++++- .../trino/connector/TestGravitinoConfig.java | 132 +++++++ .../TestGravitinoConnectorFactoryStart.java | 152 +++++++++ .../connector/catalog/TestCatalogRegister.java | 380 +++++++++++++++++++++ 8 files changed, 1277 insertions(+), 34 deletions(-) diff --git a/docs/trino-connector/configuration.md b/docs/trino-connector/configuration.md index 3053ef646d..5bcb98636e 100644 --- a/docs/trino-connector/configuration.md +++ b/docs/trino-connector/configuration.md @@ -7,18 +7,28 @@ license: "This software is licensed under the Apache License version 2." ## Introduction -| Property | Type | Default Value | Description | Required | -|---------------------------------------------|---------|-----------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------| -| connector.name | string | (none) | The `connector.name` defines the type of Trino connector, this value is always 'gravitino'. | Yes | -| gravitino.metalake | string | (none) | The `gravitino.metalake` defines which metalake in Gravitino server the Trino connector uses. Trino connector should set it at start, the value of `gravitino.metalake` needs to be a valid name, Trino connector can detect and load the metalake with catalogs, schemas and tables once created and keep in sync. | Yes | -| gravitino.uri | string | http://localhost:8090 | The `gravitino.uri` defines the connection URL of the Gravitino server, the default value is `http://localhost:8090`. Trino connector can detect and connect to Gravitino server once it is ready, no need to start Gravitino server beforehand. | No | -| trino.jdbc.user | string | admin | The jdbc user name of current Trino. | NO | -| trino.jdbc.password | string | (none) | The jdbc password of current Trino. | NO | -| gravitino.metadata.refresh-interval-seconds | integer | 10 | The `gravitino.metadata.refresh-interval-seconds` defines the interval in seconds to refresh metadata from Gravitino server, the default value is 10 seconds. | No | -| gravitino.trino.skip-version-validation | boolean | false | The `gravitino.trino.skip-version-validation` defines whether to skip Trino version validation. Gravitino supports Trino versions between 435 and 478. If this option is `true`, unsupported Trino versions can still be used, but compatibility is not guaranteed. | No | -| 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 | +| Property | Type | Default Value | Description | Required | +|---------------------------------------------|---------|-----------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------| +| connector.name | string | (none) | The `connector.name` defines the type of Trino connector, this value is always 'gravitino'. | Yes | +| gravitino.metalake | string | (none) | The `gravitino.metalake` defines which metalake in Gravitino server the Trino connector uses. Trino connector should set it at start, the value of `gravitino.metalake` needs to be a valid name, Trino connector can detect and load the metalake with catalogs, schemas and tables once created and keep in sync. | Yes | +| gravitino.uri | string | http://localhost:8090 | The `gravitino.uri` defines the connection URL of the Gravitino server, the default value is `http://localhost:8090`. Trino connector can detect and connect to Gravitino server once it is ready, no need to start Gravitino server beforehand. | No | +| trino.jdbc.user | string | admin | The jdbc user name of current Trino. | NO | +| trino.jdbc.password | string | (none) | The jdbc password of current Trino. | NO | +| trino.jdbc.ssl.enabled | boolean | (derived) | Whether the internal JDBC connection to the Trino coordinator uses TLS. If not set, it is derived from the scheme of the Trino `discovery.uri`, so a coordinator whose `discovery.uri` is `https://...` needs no explicit setting. | No | +| trino.jdbc.ssl.truststore.path | string | (none) | Path of the truststore holding the Trino coordinator certificate. If omitted, the default JVM truststore is used. Requires TLS, which is enabled automatically for an HTTPS `discovery.uri` or explicitly with `trino.jdbc.ssl.enabled=true`, and a `trino.jdbc.ssl.verification` other than `NONE`. | No | +| trino.jdbc.ssl.truststore.password | string | (none) | Password of the truststore configured by `trino.jdbc.ssl.truststore.path`. Requires TLS and `trino.jdbc.ssl.truststore.path`, otherwise the connector fails to start. | No | +| trino.jdbc.ssl.truststore.type | string | (none) | Type of the truststore, for example `JKS` or `PKCS12`. If omitted, the default JVM truststore type is used. Requires TLS and `trino.jdbc.ssl.truststore.path`, otherwise the connector fails to start. | No | +| trino.jdbc.ssl.keystore.path | string | (none) | Path of the keystore holding the client certificate presented to the coordinator, for coordinators that require mutual TLS. Requires TLS, which is enabled automatically for an HTTPS `discovery.uri` or explicitly with `trino.jdbc.ssl.enabled=true`, and a `trino.jdbc.ssl.verification` other than `NONE`. See the note on mutual TLS below. | No | +| trino.jdbc.ssl.keystore.password | string | (none) | Password of the keystore configured by `trino.jdbc.ssl.keystore.path`. Requires TLS and `trino.jdbc.ssl.keystore.path`, otherwise the connector fails to start. | No | +| trino.jdbc.ssl.keystore.type | string | (none) | Type of the keystore, for example `JKS` or `PKCS12`. If omitted, the default JVM keystore type is used. Requires TLS and `trino.jdbc.ssl.keystore.path`, otherwise the connector fails to start. | No | +| trino.jdbc.ssl.verification | string | FULL | Certificate verification mode of the internal JDBC connection: `FULL`, `CA` or `NONE`. Any value other than `FULL` requires TLS, which may be derived from an HTTPS `discovery.uri`. `NONE` disables certificate verification entirely and should only be used for troubleshooting. | No | +| trino.jdbc.roles | string | (none) | Session roles applied to the internal JDBC connection, for example `system:sysadmin`. Required by deployments that only allow `CREATE CATALOG` with a privileged role. | No | +| trino.jdbc.properties. | string | (none) | The configuration key prefix for raw Trino JDBC driver properties, see [Connecting to a TLS-enabled coordinator](#connecting-to-a-tls-enabled-coordinator). | No | +| gravitino.metadata.refresh-interval-seconds | integer | 10 | The `gravitino.metadata.refresh-interval-seconds` defines the interval in seconds to refresh metadata from Gravitino server, the default value is 10 seconds. | No | +| gravitino.trino.skip-version-validation | boolean | false | The `gravitino.trino.skip-version-validation` defines whether to skip Trino version validation. Gravitino supports Trino versions between 435 and 478. If this option is `true`, unsupported Trino versions can still be used, but compatibility is not guaranteed. | No | +| 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 | To configure the Gravitino client, use properties prefixed with `gravitino.client.`. These properties will directly passed to the Gravitino client. @@ -26,6 +36,97 @@ To configure the Gravitino client, use properties prefixed with `gravitino.clien Multi-metalake mode (`gravitino.use-single-metalake=false`) is supported on Trino connector versions 435-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. +## Connecting to a TLS-enabled coordinator + +The Gravitino Trino connector registers catalogs by connecting back to the Trino coordinator over +JDBC and running `CREATE CATALOG` / `DROP CATALOG`. This connection is established when the +connector starts and is reused by the metadata refresh loop, so it must be configured for the +coordinator's own TLS and authorization settings. + +```properties +connector.name=gravitino +gravitino.metalake=metalake +gravitino.uri=http://localhost:8090 + +# The internal JDBC connection to the coordinator. +trino.jdbc.user=admin +trino.jdbc.password=YourSecureTrinoPassword +trino.jdbc.ssl.truststore.path=/etc/trino/truststore.jks +trino.jdbc.ssl.truststore.password=YourSecureTruststorePassword +# Required when the deployment only allows CREATE CATALOG with a privileged role. +trino.jdbc.roles=system:sysadmin +``` + +`trino.jdbc.ssl.enabled` may be omitted when the Trino `discovery.uri` uses the `https` scheme, as +it is derived from that scheme by default. When `discovery.uri` omits the port, the default port of +its scheme is used, that is `443` for `https` and `80` for `http`. + +The `trino.jdbc.ssl.*` configurations are only meaningful on a TLS-enabled connection. Setting any +of them while TLS is disabled fails the connector at startup rather than being silently ignored, so +a misconfigured truststore never degrades into a plaintext connection. Likewise, the truststore +password and type require a truststore path, and the keystore password and type require a keystore +path: without one the driver falls back to its default, which they would not apply to. + +If the coordinator certificate is signed by a CA the JVM does not trust, import it into a +truststore and point `trino.jdbc.ssl.truststore.path` at it: + +```shell +# Export the coordinator certificate, then import it into a dedicated truststore. +openssl s_client -showcerts -connect coordinator.example.com:8443 </dev/null \ + | openssl x509 -outform PEM > coordinator.pem +keytool -importcert -noprompt -alias trino-coordinator -file coordinator.pem \ + -keystore /etc/trino/truststore.jks -storepass YourSecureTruststorePassword +``` + +:::caution +`trino.jdbc.ssl.verification=NONE` disables certificate verification completely and exposes the +connection to man-in-the-middle attacks. Use it only for troubleshooting; import the coordinator +certificate into a truststore instead. +::: + +The `trino.jdbc.*` properties are used by the coordinator only. They are never copied into the +catalogs the connector creates, so the credentials they hold do not reach the generated +`CREATE CATALOG` statements or the Trino catalog properties files. + +### Mutual TLS and certificate authentication + +A coordinator that requires mutual TLS also expects a client certificate. Point +`trino.jdbc.ssl.keystore.path` at the keystore holding it: + +```properties +trino.jdbc.ssl.keystore.path=/etc/trino/client.p12 +trino.jdbc.ssl.keystore.password=YourSecureKeystorePassword +``` + +:::note +The mutual TLS configuration is provided for completeness and has not been verified against a +coordinator that requires client certificates. +::: + +Trino can also be configured with `http-server.authentication.type=CERTIFICATE`, where the client +certificate itself is the login and the coordinator derives the username from the certificate +subject. The connector does not support that authentication type today: it always authenticates +with `trino.jdbc.user`, so a coordinator configured this way rejects the internal JDBC connection. + +### Passing arbitrary JDBC driver properties + +Any Trino JDBC driver property that has no dedicated configuration can be passed through with the +`trino.jdbc.properties.` prefix. The prefix is stripped and the remainder is handed to the driver +verbatim, overriding the value derived from the dedicated `trino.jdbc.*` configurations: + +```properties +trino.jdbc.properties.KerberosRemoteServiceName=trino +trino.jdbc.properties.SSLKeyStorePath=/etc/trino/client.p12 +trino.jdbc.properties.SSLKeyStorePassword=YourSecureKeystorePassword +``` + +Properties passed through this prefix are handed to the driver without validation, unlike the +dedicated `trino.jdbc.*` configurations. An unknown name or an invalid value therefore surfaces as a +driver error when the connection is established, not as a configuration error. + +See the [Trino JDBC driver documentation](https://trino.io/docs/current/client/jdbc.html) for the +full list of supported property names. + ## Authentication The Gravitino Trino connector supports authenticating to the Gravitino server using Simple, Basic, OAuth, and Kerberos authentication. For detailed authentication configuration, refer to [Trino Connector Authentication](./authentication.md). 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 1186fcd1c2..817d9f6306 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 @@ -28,6 +28,7 @@ import java.net.URI; import java.util.ArrayList; import java.util.HashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Properties; import java.util.regex.Pattern; @@ -49,6 +50,8 @@ public class GravitinoConfig { public static final String TRINO_CATALOG_STORE = "catalog.store"; /** The Trino catalog management. */ public static final String TRINO_CATALOG_MANAGEMENT = "catalog.management"; + /** The common prefix of all internal Trino JDBC connection configurations. */ + private static final String TRINO_JDBC_CONFIG_PREFIX = "trino.jdbc."; // Trino config default value /** The Trino catalog config directory default value. */ @@ -57,6 +60,10 @@ public class GravitinoConfig { public static final String TRINO_CATALOG_STORE_DEFAULT_VALUE = "file"; /** The Trino catalog management default value. */ public static final String TRINO_CATALOG_MANAGEMENT_DEFAULT_VALUE = "static"; + /** The default port used when the Trino `discovery.uri` omits it and the scheme is http. */ + private static final int HTTP_DEFAULT_PORT = 80; + /** The default port used when the Trino `discovery.uri` omits it and the scheme is https. */ + private static final int HTTPS_DEFAULT_PORT = 443; // The Trino configuration of etc/config.properties /** The Trino configuration. */ @@ -134,6 +141,86 @@ public class GravitinoConfig { new ConfigEntry( "trino.jdbc.password", "The JDBC password for connecting to Trino", "", false); + private static final ConfigEntry TRINO_JDBC_SSL_ENABLED = + new ConfigEntry( + "trino.jdbc.ssl.enabled", + "Whether the internal JDBC connection to the Trino coordinator uses TLS. " + + "If not set, it is derived from the scheme of the Trino `discovery.uri`.", + "", + false); + + private static final ConfigEntry TRINO_JDBC_SSL_TRUSTSTORE_PATH = + new ConfigEntry( + "trino.jdbc.ssl.truststore.path", + "Path of the truststore holding the Trino coordinator certificate. " + + "If omitted, the default JVM truststore is used.", + "", + false); + + private static final ConfigEntry TRINO_JDBC_SSL_TRUSTSTORE_PASSWORD = + new ConfigEntry( + "trino.jdbc.ssl.truststore.password", + "Password of the truststore configured by `trino.jdbc.ssl.truststore.path`", + "", + false); + + private static final ConfigEntry TRINO_JDBC_SSL_TRUSTSTORE_TYPE = + new ConfigEntry( + "trino.jdbc.ssl.truststore.type", + "Type of the truststore, for example JKS or PKCS12. " + + "If omitted, the default JVM truststore type is used.", + "", + false); + + private static final ConfigEntry TRINO_JDBC_SSL_KEYSTORE_PATH = + new ConfigEntry( + "trino.jdbc.ssl.keystore.path", + "Path of the keystore holding the client certificate presented to the Trino coordinator, " + + "used by coordinators that require mutual TLS.", + "", + false); + + private static final ConfigEntry TRINO_JDBC_SSL_KEYSTORE_PASSWORD = + new ConfigEntry( + "trino.jdbc.ssl.keystore.password", + "Password of the keystore configured by `trino.jdbc.ssl.keystore.path`", + "", + false); + + private static final ConfigEntry TRINO_JDBC_SSL_KEYSTORE_TYPE = + new ConfigEntry( + "trino.jdbc.ssl.keystore.type", + "Type of the keystore, for example JKS or PKCS12. " + + "If omitted, the default JVM keystore type is used.", + "", + false); + + private static final ConfigEntry TRINO_JDBC_SSL_VERIFICATION = + new ConfigEntry( + "trino.jdbc.ssl.verification", + "Certificate verification mode of the internal JDBC connection: FULL, CA or NONE. " + + "NONE disables certificate verification and should only be used for troubleshooting.", + "FULL", + false); + + private static final ConfigEntry TRINO_JDBC_ROLES = + new ConfigEntry( + "trino.jdbc.roles", + "Session roles applied to the internal JDBC connection, for example `system:sysadmin`. " + + "Required by deployments that only allow CREATE CATALOG with a privileged role.", + "", + false); + + private static final ConfigEntry TRINO_JDBC_EXTRA_PROPERTIES_PREFIX = + new ConfigEntry( + "trino.jdbc.properties.", + "Prefix for Trino JDBC driver properties. Any property beginning with this prefix is " + + "passed to the driver verbatim with the prefix removed " + + "(e.g., trino.jdbc.properties.KerberosRemoteServiceName=trino), " + + "overriding the properties derived from the dedicated `trino.jdbc.*` configurations.", + "", + false); + private static final ConfigEntry GRAVITINO_METADATA_REFRESH_INTERVAL_SECOND = new ConfigEntry( "gravitino.metadata.refresh-interval-seconds", @@ -282,15 +369,25 @@ public class GravitinoConfig { * @return the Trino JDBC URI */ public String getTrinoJdbcURI() { - String uriString = ""; + URI trinoURI = parseDiscoveryUri(); + int port = trinoURI.getPort(); + if (port < 0) { + // `discovery.uri` may omit the port, for example a TLS coordinator behind a load balancer + // on the standard HTTPS port. Fall back to the default port of the scheme. + port = isHttpsScheme(trinoURI) ? HTTPS_DEFAULT_PORT : HTTP_DEFAULT_PORT; + } + return String.format("jdbc:trino://%s:%s", trinoURI.getHost(), port); + } + + private URI parseDiscoveryUri() { + String uriString; if (config.containsKey(TRINO_DISCOVERY_URI)) { uriString = config.get(TRINO_DISCOVERY_URI); } else { uriString = trinoConfig.getProperty(TRINO_DISCOVERY_URI); } try { - URI trinoURI = new URI(uriString); - return String.format("jdbc:trino://%s:%s", trinoURI.getHost(), trinoURI.getPort()); + return new URI(uriString); } catch (Exception e) { throw new TrinoException( GravitinoErrorCode.GRAVITINO_MISSING_CONFIG, @@ -298,6 +395,25 @@ public class GravitinoConfig { } } + private static boolean parseBooleanConfig(String key, String value) { + // `Boolean.parseBoolean` maps every unrecognized value to false, which would silently disable + // TLS for a typo such as `yes`, even when the `discovery.uri` scheme implies it should be on. + if ("true".equalsIgnoreCase(value)) { + return true; + } + if ("false".equalsIgnoreCase(value)) { + return false; + } + throw new TrinoException( + GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT, + String.format( + "Invalid value for config '%s': expected true or false, got: %s", key, value)); + } + + private static boolean isHttpsScheme(URI uri) { + return "https".equalsIgnoreCase(uri.getScheme()); + } + /** * Retrieves the region. * @@ -340,6 +456,119 @@ public class GravitinoConfig { return config.getOrDefault(TRINO_JDBC_PASSWORD.key, TRINO_JDBC_PASSWORD.defaultValue); } + /** + * Returns whether the internal JDBC connection to the Trino coordinator uses TLS. + * + * <p>If `trino.jdbc.ssl.enabled` is not set, the value is derived from the scheme of the Trino + * `discovery.uri`, which is `https` on a TLS enabled coordinator. + * + * @return true if the internal JDBC connection uses TLS + */ + public boolean isTrinoJdbcSslEnabled() { + String value = config.get(TRINO_JDBC_SSL_ENABLED.key); + if (StringUtils.isNotBlank(value)) { + return parseBooleanConfig(TRINO_JDBC_SSL_ENABLED.key, value.trim()); + } + return isHttpsScheme(parseDiscoveryUri()); + } + + /** + * Retrieves the truststore path of the internal JDBC connection. + * + * @return the truststore path, or an empty string if not configured + */ + public String getTrinoJdbcSslTruststorePath() { + return config.getOrDefault( + TRINO_JDBC_SSL_TRUSTSTORE_PATH.key, TRINO_JDBC_SSL_TRUSTSTORE_PATH.defaultValue); + } + + /** + * Retrieves the truststore password of the internal JDBC connection. + * + * @return the truststore password, or an empty string if not configured + */ + public String getTrinoJdbcSslTruststorePassword() { + return config.getOrDefault( + TRINO_JDBC_SSL_TRUSTSTORE_PASSWORD.key, TRINO_JDBC_SSL_TRUSTSTORE_PASSWORD.defaultValue); + } + + /** + * Retrieves the truststore type of the internal JDBC connection. + * + * @return the truststore type, or an empty string if not configured + */ + public String getTrinoJdbcSslTruststoreType() { + return config.getOrDefault( + TRINO_JDBC_SSL_TRUSTSTORE_TYPE.key, TRINO_JDBC_SSL_TRUSTSTORE_TYPE.defaultValue); + } + + /** + * Retrieves the keystore path of the internal JDBC connection. + * + * @return the keystore path, or an empty string if not configured + */ + public String getTrinoJdbcSslKeystorePath() { + return config.getOrDefault( + TRINO_JDBC_SSL_KEYSTORE_PATH.key, TRINO_JDBC_SSL_KEYSTORE_PATH.defaultValue); + } + + /** + * Retrieves the keystore password of the internal JDBC connection. + * + * @return the keystore password, or an empty string if not configured + */ + public String getTrinoJdbcSslKeystorePassword() { + return config.getOrDefault( + TRINO_JDBC_SSL_KEYSTORE_PASSWORD.key, TRINO_JDBC_SSL_KEYSTORE_PASSWORD.defaultValue); + } + + /** + * Retrieves the keystore type of the internal JDBC connection. + * + * @return the keystore type, or an empty string if not configured + */ + public String getTrinoJdbcSslKeystoreType() { + return config.getOrDefault( + TRINO_JDBC_SSL_KEYSTORE_TYPE.key, TRINO_JDBC_SSL_KEYSTORE_TYPE.defaultValue); + } + + /** + * Retrieves the certificate verification mode of the internal JDBC connection. + * + * @return the verification mode, one of FULL, CA or NONE + */ + public String getTrinoJdbcSslVerification() { + String value = config.get(TRINO_JDBC_SSL_VERIFICATION.key); + if (StringUtils.isBlank(value)) { + return TRINO_JDBC_SSL_VERIFICATION.defaultValue; + } + return value.trim().toUpperCase(Locale.ROOT); + } + + /** + * Retrieves the session roles applied to the internal JDBC connection. + * + * @return the session roles, or an empty string if not configured + */ + public String getTrinoJdbcRoles() { + return config.getOrDefault(TRINO_JDBC_ROLES.key, TRINO_JDBC_ROLES.defaultValue); + } + + /** + * Retrieves the Trino JDBC driver properties configured with the `trino.jdbc.properties.` prefix. + * + * @return a map of driver property names, with the prefix removed, to their values + */ + public Map<String, String> getTrinoJdbcExtraProperties() { + return config.entrySet().stream() + .filter(entry -> entry.getKey().startsWith(TRINO_JDBC_EXTRA_PROPERTIES_PREFIX.key)) + .filter(entry -> entry.getKey().length() > TRINO_JDBC_EXTRA_PROPERTIES_PREFIX.key.length()) + .collect( + Collectors.toMap( + entry -> entry.getKey().substring(TRINO_JDBC_EXTRA_PROPERTIES_PREFIX.key.length()), + Map.Entry::getValue)); + } + /** * Retrieves the catalog connector factory class name. * @@ -359,6 +588,13 @@ public class GravitinoConfig { public String toCatalogConfig() { List<String> stringList = new ArrayList<>(); for (Map.Entry<String, ConfigEntry> entry : CONFIG_DEFINITIONS.entrySet()) { + // The `trino.jdbc.*` configurations are only used by the coordinator to connect back to + // Trino. They must not be propagated to the dynamic catalogs, otherwise credentials such as + // the JDBC password and the truststore password would end up in the generated CREATE CATALOG + // statement, which is logged and persisted into the Trino catalog properties files. + if (entry.getKey().startsWith(TRINO_JDBC_CONFIG_PREFIX)) { + continue; + } String value = config.get(entry.getKey()); if (value != null) { stringList.add(String.format("\"%s\"='%s'", entry.getKey(), value)); diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java index 5315342713..6308b7d1bc 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java @@ -64,6 +64,7 @@ public class GravitinoConnectorFactory implements ConnectorFactory { private GravitinoSystemTableFactory gravitinoSystemTableFactory; private CatalogConnectorManager catalogConnectorManager; + private boolean catalogConnectorManagerStartTriggered = false; private GravitinoAdminClient client; private int trinoVersion; @@ -87,6 +88,16 @@ public class GravitinoConnectorFactory implements ConnectorFactory { return catalogConnectorManager; } + /** + * Returns whether starting the catalog connector manager has been triggered. + * + * @return true if starting the catalog connector manager has been triggered + */ + @VisibleForTesting + public boolean isCatalogConnectorManagerStartTriggered() { + return catalogConnectorManagerStartTriggered; + } + /** * This function call by Trino creates a connector. It creates GravitinoSystemConnector at first. * Another time's it get GravitinoConnector by CatalogConnectorManager @@ -104,9 +115,14 @@ public class GravitinoConnectorFactory implements ConnectorFactory { GravitinoConfig config = new GravitinoConfig(requiredConfig); synchronized (this) { + // Keep the version check out of the try below so that it keeps reporting its own error code + // instead of being wrapped as a generic initialization failure. if (catalogConnectorManager == null) { checkTrinoSpiVersion(trinoConnectorContext, config); - try { + } + + try { + if (catalogConnectorManager == null) { CatalogRegister catalogRegister = new CatalogRegister(); CatalogConnectorFactory catalogConnectorFactory = createCatalogConnectorFactory(config); @@ -115,16 +131,31 @@ public class GravitinoConnectorFactory implements ConnectorFactory { catalogRegister, catalogConnectorFactory, this::getTrinoCatalogName); catalogConnectorManager.config(config, client); - if (isCoordinator(trinoConnectorContext)) { - catalogConnectorManager.start(); - } - gravitinoSystemTableFactory = new GravitinoSystemTableFactory(catalogConnectorManager); - } catch (Exception e) { - String message = "Initialization of the GravitinoConnector failed " + e.getMessage(); - LOG.error(message); - throw new TrinoException(GRAVITINO_RUNTIME_ERROR, message, e); } + + // The `trino.jdbc.*` settings that CatalogRegister needs to connect back to the + // coordinator are deliberately not propagated to the dynamic catalogs, so they are only + // present in the configuration of the static connector. Trino does not guarantee that the + // static catalog is loaded before the catalogs Gravitino created, therefore the manager is + // started from the static connector only, re-applying its configuration in case a dynamic + // connector was created first. + if (!catalogConnectorManagerStartTriggered + && !config.isDynamicConnector() + && isCoordinator(trinoConnectorContext)) { + // Triggered before start() on purpose: everything that makes it fail is a + // configuration error, and retrying on the next create() would only open another + // connection. + catalogConnectorManagerStartTriggered = true; + // Only the configuration is re-applied here: rebuilding the Gravitino client would leak + // the one a dynamic connector may have already built. + catalogConnectorManager.updateConfig(config); + catalogConnectorManager.start(); + } + } catch (Exception e) { + String message = "Initialization of the GravitinoConnector failed " + e.getMessage(); + LOG.error(message); + throw new TrinoException(GRAVITINO_RUNTIME_ERROR, message, e); } } 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 7539a0b439..34872d2e2f 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 @@ -114,8 +114,7 @@ public class CatalogConnectorManager { * @param client the Gravitino admin client */ public void config(GravitinoConfig config, GravitinoAdminClient client) { - Preconditions.checkArgument(config != null, "config must not be null"); - this.config = config; + updateConfig(config); if (client == null) { String authType = config.getClientConfig().getOrDefault(GravitinoAuthProvider.AUTH_TYPE_KEY, "none"); @@ -142,6 +141,20 @@ public class CatalogConnectorManager { } else { this.gravitinoClient = client; } + } + + /** + * Updates the Gravitino configuration, leaving the Gravitino client untouched. + * + * <p>Used to re-apply the configuration of the static connector when a dynamic connector created + * the manager first, so that the catalog register is started with the `trino.jdbc.*` settings + * that are only present in the static configuration. + * + * @param config the Gravitino configuration + */ + public void updateConfig(GravitinoConfig config) { + Preconditions.checkArgument(config != null, "config must not be null"); + this.config = config; this.metadataUpdateIntervalSecond = Integer.parseInt(config.getMetadataRefreshIntervalSecond()); this.targetMetalake = config.getMetalake(); } 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 09bf1cd658..faca9c2499 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 @@ -21,17 +21,21 @@ package org.apache.gravitino.trino.connector.catalog; import static org.apache.gravitino.trino.connector.GravitinoConfig.GRAVITINO_DYNAMIC_CONNECTOR; import static org.apache.gravitino.trino.connector.GravitinoConfig.GRAVITINO_DYNAMIC_CONNECTOR_CATALOG_CONFIG; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableSet; import io.trino.jdbc.TrinoDriver; import io.trino.spi.TrinoException; import java.io.File; import java.nio.file.Files; import java.nio.file.Path; import java.sql.Connection; -import java.sql.DriverManager; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; +import java.util.Map; import java.util.Properties; +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.metadata.GravitinoCatalog; @@ -49,6 +53,12 @@ public class CatalogRegister { private static final int EXECUTE_QUERY_MAX_RETRIES = 6; private static final int EXECUTE_QUERY_BACKOFF_TIME_SECOND = 5; + private static final String SSL_VERIFICATION_FULL = "FULL"; + private static final String SSL_VERIFICATION_CA = "CA"; + private static final String SSL_VERIFICATION_NONE = "NONE"; + private static final Set<String> SSL_VERIFICATION_MODES = + ImmutableSet.of(SSL_VERIFICATION_FULL, SSL_VERIFICATION_CA, SSL_VERIFICATION_NONE); + private Connection connection; private boolean isStarted = false; private String catalogStoreDirectory; @@ -80,17 +90,16 @@ public class CatalogRegister { this.config = config; TrinoDriver driver = new TrinoDriver(); - DriverManager.registerDriver(driver); - - Properties properties = new Properties(); - properties.put("user", config.getTrinoUser()); - properties.put("password", config.getTrinoPassword()); + Properties properties = buildJdbcProperties(config); + String jdbcUri = config.getTrinoJdbcURI(); try { - connection = driver.connect(config.getTrinoJdbcURI(), properties); + connection = driver.connect(jdbcUri, properties); } catch (SQLException e) { throw new TrinoException( GravitinoErrorCode.GRAVITINO_RUNTIME_ERROR, - "Failed to initialize the Trino connection.", + String.format( + "Failed to initialize the Trino connection to %s (TLS %s).", + jdbcUri, config.isTrinoJdbcSslEnabled() ? "enabled" : "disabled"), e); } @@ -104,6 +113,195 @@ public class CatalogRegister { } } + /** + * Builds the JDBC properties used by the internal connection to the Trino coordinator. + * + * <p>The properties derived from the dedicated {@code trino.jdbc.*} configurations are applied + * first, then the raw driver properties configured with the {@code trino.jdbc.properties.} prefix + * are applied on top of them, so that any driver property can be overridden. + * + * @param config the Gravitino configuration + * @return the JDBC properties + */ + @VisibleForTesting + static Properties buildJdbcProperties(GravitinoConfig config) { + boolean sslEnabled = config.isTrinoJdbcSslEnabled(); + String verification = config.getTrinoJdbcSslVerification(); + String truststorePath = config.getTrinoJdbcSslTruststorePath(); + String truststorePassword = config.getTrinoJdbcSslTruststorePassword(); + String truststoreType = config.getTrinoJdbcSslTruststoreType(); + String keystorePath = config.getTrinoJdbcSslKeystorePath(); + String keystorePassword = config.getTrinoJdbcSslKeystorePassword(); + String keystoreType = config.getTrinoJdbcSslKeystoreType(); + String roles = config.getTrinoJdbcRoles(); + + validateSslConfig( + sslEnabled, + verification, + truststorePath, + truststorePassword, + truststoreType, + keystorePath, + keystorePassword, + keystoreType); + + Properties properties = new Properties(); + properties.put("user", config.getTrinoUser()); + String password = config.getTrinoPassword(); + if (StringUtils.isNotEmpty(password)) { + properties.put("password", password); + } + + if (sslEnabled) { + properties.put("SSL", "true"); + properties.put("SSLVerification", verification); + if (StringUtils.isNotBlank(truststorePath)) { + properties.put("SSLTrustStorePath", truststorePath); + } + if (StringUtils.isNotEmpty(truststorePassword)) { + properties.put("SSLTrustStorePassword", truststorePassword); + } + if (StringUtils.isNotBlank(truststoreType)) { + properties.put("SSLTrustStoreType", truststoreType); + } + if (StringUtils.isNotBlank(keystorePath)) { + properties.put("SSLKeyStorePath", keystorePath); + } + if (StringUtils.isNotEmpty(keystorePassword)) { + properties.put("SSLKeyStorePassword", keystorePassword); + } + if (StringUtils.isNotBlank(keystoreType)) { + properties.put("SSLKeyStoreType", keystoreType); + } + } + + if (StringUtils.isNotBlank(roles)) { + properties.put("roles", roles); + } + + Map<String, String> extraProperties = config.getTrinoJdbcExtraProperties(); + if (!extraProperties.isEmpty()) { + // Log the names only, the values may contain credentials. + LOG.debug("Applying extra Trino JDBC properties: {}", extraProperties.keySet()); + extraProperties.keySet().stream() + .filter(key -> key.startsWith("SSL") && properties.containsKey(key)) + .forEach( + key -> + LOG.warn( + "Extra Trino JDBC property '{}' overrides the TLS setting derived from the " + + "dedicated configuration and is applied without validation", + key)); + properties.putAll(extraProperties); + } + return properties; + } + + private static void validateSslConfig( + boolean sslEnabled, + String verification, + String truststorePath, + String truststorePassword, + String truststoreType, + String keystorePath, + String keystorePassword, + String keystoreType) { + if (!SSL_VERIFICATION_MODES.contains(verification)) { + throw new TrinoException( + GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT, + String.format( + "Invalid value for config 'trino.jdbc.ssl.verification': expected one of %s, got: %s", + SSL_VERIFICATION_MODES, verification)); + } + + if (!sslEnabled) { + if (!SSL_VERIFICATION_FULL.equals(verification)) { + throw new TrinoException( + GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT, + "Config 'trino.jdbc.ssl.verification' requires TLS to be enabled either by an HTTPS " + + "'discovery.uri' or by 'trino.jdbc.ssl.enabled=true'"); + } + checkRequiresSslEnabled("trino.jdbc.ssl.truststore.path", truststorePath); + checkRequiresSslEnabled("trino.jdbc.ssl.truststore.password", truststorePassword); + checkRequiresSslEnabled("trino.jdbc.ssl.truststore.type", truststoreType); + checkRequiresSslEnabled("trino.jdbc.ssl.keystore.path", keystorePath); + checkRequiresSslEnabled("trino.jdbc.ssl.keystore.password", keystorePassword); + checkRequiresSslEnabled("trino.jdbc.ssl.keystore.type", keystoreType); + return; + } + + validateKeystoreConfig(verification, keystorePath, keystorePassword, keystoreType); + + if (StringUtils.isBlank(truststorePath)) { + // The driver falls back to the default JVM truststore, which the password and the type of a + // truststore that was never configured have nothing to apply to. + checkRequires( + "trino.jdbc.ssl.truststore.password", + truststorePassword, + "trino.jdbc.ssl.truststore.path"); + checkRequires( + "trino.jdbc.ssl.truststore.type", truststoreType, "trino.jdbc.ssl.truststore.path"); + return; + } + + if (SSL_VERIFICATION_NONE.equals(verification)) { + throw new TrinoException( + GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT, + "Config 'trino.jdbc.ssl.truststore.path' cannot be used with " + + "'trino.jdbc.ssl.verification' = NONE"); + } + if (!Files.exists(Path.of(truststorePath))) { + throw new TrinoException( + GravitinoErrorCode.GRAVITINO_MISSING_CONFIG, + String.format( + "The truststore file configured by 'trino.jdbc.ssl.truststore.path' does not exist: %s", + truststorePath)); + } + } + + private static void validateKeystoreConfig( + String verification, String keystorePath, String keystorePassword, String keystoreType) { + if (StringUtils.isBlank(keystorePath)) { + checkRequires( + "trino.jdbc.ssl.keystore.password", keystorePassword, "trino.jdbc.ssl.keystore.path"); + checkRequires("trino.jdbc.ssl.keystore.type", keystoreType, "trino.jdbc.ssl.keystore.path"); + return; + } + if (SSL_VERIFICATION_NONE.equals(verification)) { + // The driver rejects the keystore properties in this combination, so fail with a config + // error here rather than letting it surface as a connection failure. + throw new TrinoException( + GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT, + "Config 'trino.jdbc.ssl.keystore.path' cannot be used with " + + "'trino.jdbc.ssl.verification' = NONE"); + } + if (!Files.exists(Path.of(keystorePath))) { + throw new TrinoException( + GravitinoErrorCode.GRAVITINO_MISSING_CONFIG, + String.format( + "The keystore file configured by 'trino.jdbc.ssl.keystore.path' does not exist: %s", + keystorePath)); + } + } + + private static void checkRequires(String key, String value, String requiredKey) { + if (StringUtils.isNotEmpty(value)) { + throw new TrinoException( + GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT, + String.format("Config '%s' requires '%s' to be set", key, requiredKey)); + } + } + + private static void checkRequiresSslEnabled(String key, String value) { + if (StringUtils.isNotEmpty(value)) { + throw new TrinoException( + GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT, + String.format( + "Config '%s' requires TLS to be enabled either by an HTTPS 'discovery.uri' or by " + + "'trino.jdbc.ssl.enabled=true'", + key)); + } + } + private String generateCreateCatalogCommand(String name, GravitinoCatalog gravitinoCatalog) throws Exception { return String.format( 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 8aa65b4b69..2d486b52c0 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 @@ -21,6 +21,7 @@ package org.apache.gravitino.trino.connector; import static org.apache.gravitino.trino.connector.GravitinoErrorCode.GRAVITINO_MISSING_CONFIG; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertThrowsExactly; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -168,6 +169,137 @@ public class TestGravitinoConfig { assertTrue(catalogConfig.contains("\"gravitino.user\"='admin'")); } + @Test + public void testTrinoJdbcConfigDefaults() { + GravitinoConfig config = + new GravitinoConfig( + ImmutableMap.of("gravitino.metalake", "user_001", "discovery.uri", "http://host:8080";)); + + assertEquals("admin", config.getTrinoUser()); + assertEquals("", config.getTrinoPassword()); + assertFalse(config.isTrinoJdbcSslEnabled()); + assertEquals("FULL", config.getTrinoJdbcSslVerification()); + assertEquals("", config.getTrinoJdbcSslTruststorePath()); + assertEquals("", config.getTrinoJdbcSslTruststorePassword()); + assertEquals("", config.getTrinoJdbcSslTruststoreType()); + assertEquals("", config.getTrinoJdbcRoles()); + assertTrue(config.getTrinoJdbcExtraProperties().isEmpty()); + } + + @Test + public void testTrinoJdbcSslEnabledDerivedFromDiscoveryUri() { + GravitinoConfig config = + new GravitinoConfig( + ImmutableMap.of( + "gravitino.metalake", "user_001", "discovery.uri", "https://host:8443";)); + + assertTrue(config.isTrinoJdbcSslEnabled()); + assertEquals("jdbc:trino://host:8443", config.getTrinoJdbcURI()); + } + + @Test + public void testTrinoJdbcExtraProperties() { + GravitinoConfig config = + new GravitinoConfig( + ImmutableMap.of( + "gravitino.metalake", + "user_001", + "discovery.uri", + "http://host:8080";, + "trino.jdbc.properties.KerberosRemoteServiceName", + "trino", + "trino.jdbc.properties.SSLKeyStorePath", + "/etc/trino/client.p12")); + + Map<String, String> extraProperties = config.getTrinoJdbcExtraProperties(); + assertEquals(2, extraProperties.size()); + assertEquals("trino", extraProperties.get("KerberosRemoteServiceName")); + assertEquals("/etc/trino/client.p12", extraProperties.get("SSLKeyStorePath")); + } + + @Test + public void testToCatalogConfigExcludesTrinoJdbcProperties() { + GravitinoConfig config = + new GravitinoConfig( + ImmutableMap.of( + "gravitino.metalake", + "user_001", + "trino.jdbc.user", + "admin", + "trino.jdbc.password", + "jdbc-secret", + "trino.jdbc.ssl.truststore.password", + "truststore-secret", + "trino.jdbc.properties.SSLKeyStorePassword", + "keystore-secret")); + + // The internal JDBC connection settings are coordinator only. They must never reach the + // generated CREATE CATALOG statement, which is logged and persisted to the catalog files. + String catalogConfig = config.toCatalogConfig(); + assertFalse(catalogConfig.contains("trino.jdbc.")); + assertFalse(catalogConfig.contains("secret")); + assertTrue(catalogConfig.contains("\"gravitino.metalake\"='user_001'")); + } + + @Test + public void testTrinoJdbcUriUsesSchemeDefaultPort() { + GravitinoConfig httpsConfig = + new GravitinoConfig( + ImmutableMap.of("gravitino.metalake", "user_001", "discovery.uri", "https://host";)); + assertEquals("jdbc:trino://host:443", httpsConfig.getTrinoJdbcURI()); + + GravitinoConfig httpConfig = + new GravitinoConfig( + ImmutableMap.of("gravitino.metalake", "user_001", "discovery.uri", "http://host";)); + assertEquals("jdbc:trino://host:80", httpConfig.getTrinoJdbcURI()); + } + + @Test + public void testBlankSslVerificationFallsBackToDefault() { + GravitinoConfig config = + new GravitinoConfig( + ImmutableMap.of( + "gravitino.metalake", + "user_001", + "discovery.uri", + "http://host:8080";, + "trino.jdbc.ssl.verification", + " ")); + + assertEquals("FULL", config.getTrinoJdbcSslVerification()); + } + + @Test + public void testInvalidSslEnabledIsRejected() { + GravitinoConfig config = + new GravitinoConfig( + ImmutableMap.of( + "gravitino.metalake", + "user_001", + // An HTTPS discovery.uri would have derived true, so a typo must not silently + // fall back to false. + "discovery.uri", + "https://host:8443";, + "trino.jdbc.ssl.enabled", + "yes")); + + TrinoException e = assertThrows(TrinoException.class, config::isTrinoJdbcSslEnabled); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.enabled")); + assertTrue(e.getMessage().contains("expected true or false")); + } + + @Test + public void testSslEnabledAcceptsMixedCase() { + GravitinoConfig config = + new GravitinoConfig( + ImmutableMap.of( + "gravitino.metalake", "user_001", + "discovery.uri", "http://host:8080";, + "trino.jdbc.ssl.enabled", " TRUE ")); + + assertTrue(config.isTrinoJdbcSslEnabled()); + } + 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/TestGravitinoConnectorFactoryStart.java b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorFactoryStart.java new file mode 100644 index 0000000000..b104b4d509 --- /dev/null +++ b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorFactoryStart.java @@ -0,0 +1,152 @@ +/* + * 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.trino.connector; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.google.common.collect.ImmutableMap; +import io.trino.spi.connector.ConnectorContext; +import java.nio.file.Path; +import java.util.HashMap; +import java.util.Map; +import org.apache.gravitino.client.GravitinoAdminClient; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Verifies that the catalog connector manager is only started from the static connector. + * + * <p>The `trino.jdbc.*` settings are not propagated to the dynamic catalogs, so starting the + * manager from a dynamic connector would make the internal JDBC connection lose its credentials and + * TLS settings. + */ +public class TestGravitinoConnectorFactoryStart { + + // Nothing listens on this port, but the Trino driver connects lazily, so the register still + // initializes successfully. + private static final String COORDINATOR_URI = "http://127.0.0.1:1";; + + @TempDir private static Path catalogStoreDir; + + private static Map<String, String> staticConfig() { + Map<String, String> config = new HashMap<>(); + config.put("gravitino.metalake", "test"); + config.put("discovery.uri", COORDINATOR_URI); + config.put("catalog.config-dir", catalogStoreDir.toString()); + config.put("trino.jdbc.user", "gravitino"); + return config; + } + + private static Map<String, String> dynamicConfig() { + Map<String, String> config = new HashMap<>(); + config.put("gravitino.metalake", "test"); + // A dynamic catalog config carries neither `discovery.uri` nor any `trino.jdbc.*` setting. + config.put(GravitinoConfig.GRAVITINO_DYNAMIC_CONNECTOR, "true"); + config.put(GravitinoConfig.GRAVITINO_DYNAMIC_CONNECTOR_CATALOG_CONFIG, "{}"); + return config; + } + + private static ConnectorContext mockContext() { + ConnectorContext context = mock(ConnectorContext.class); + when(context.getSpiVersion()).thenReturn("478"); + return context; + } + + /** A factory that always reports itself as running on the coordinator. */ + private static class CoordinatorFactory extends GravitinoConnectorFactory { + CoordinatorFactory(GravitinoAdminClient client) { + super(client); + } + + @Override + protected boolean isCoordinator(ConnectorContext connectorContext) { + return true; + } + } + + private static CoordinatorFactory newFactory() { + return new CoordinatorFactory(mock(GravitinoAdminClient.class)); + } + + @Test + public void testStaticConnectorStartsTheManager() { + CoordinatorFactory factory = newFactory(); + + assertNotNull(factory.create("gravitino", staticConfig(), mockContext())); + assertTrue(factory.isCatalogConnectorManagerStartTriggered()); + } + + @Test + public void testDynamicConnectorDoesNotStartTheManager() { + CoordinatorFactory factory = newFactory(); + + Exception e = + assertThrows( + Exception.class, () -> factory.create("catalog", dynamicConfig(), mockContext())); + + // It fails later while building the dynamic catalog, never by connecting back to Trino. + assertFalse( + e.getMessage().contains("jdbc:trino://"), + "The manager must not be started from a dynamic connector, got: " + e.getMessage()); + assertNotNull(factory.getCatalogConnectorManager()); + assertFalse(factory.isCatalogConnectorManagerStartTriggered()); + } + + @Test + public void testStaticConnectorStartsTheManagerAfterADynamicOne() { + CoordinatorFactory factory = newFactory(); + + // The coordinator restarted and Trino loaded a Gravitino-created catalog first. + assertThrows(Exception.class, () -> factory.create("catalog", dynamicConfig(), mockContext())); + + // The static catalog is loaded afterwards and must start the manager with its own config. + // Had the dynamic config been used instead, `discovery.uri` would have been missing and the + // register would have failed to build the JDBC URI. + assertNotNull(factory.create("gravitino", staticConfig(), mockContext())); + assertTrue(factory.isCatalogConnectorManagerStartTriggered()); + } + + @Test + public void testStartIsAttemptedOnlyOnce() { + CoordinatorFactory factory = newFactory(); + + Map<String, String> brokenConfig = staticConfig(); + brokenConfig.put("catalog.config-dir", "/not/exists/catalog"); + assertThrows(Exception.class, () -> factory.create("gravitino", brokenConfig, mockContext())); + + // Everything that makes start() fail is a configuration error, so the next create() must not + // try again: a second init() would open another connection and abandon the first one. + assertNotNull(factory.create("gravitino", brokenConfig, mockContext())); + } + + @Test + public void testDynamicConfigCarriesNoJdbcSettings() { + GravitinoConfig config = new GravitinoConfig(ImmutableMap.copyOf(staticConfig())); + + // What a dynamic catalog would receive: no `trino.jdbc.*`, no `discovery.uri`. + String catalogConfig = config.toCatalogConfig(); + assertFalse(catalogConfig.contains("trino.jdbc.")); + assertFalse(catalogConfig.contains("discovery.uri")); + } +} diff --git a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogRegister.java b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogRegister.java new file mode 100644 index 0000000000..2effd98843 --- /dev/null +++ b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogRegister.java @@ -0,0 +1,380 @@ +/* + * 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.trino.connector.catalog; + +import static org.apache.gravitino.trino.connector.GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT; +import static org.apache.gravitino.trino.connector.GravitinoErrorCode.GRAVITINO_MISSING_CONFIG; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.trino.spi.TrinoException; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import org.apache.gravitino.trino.connector.GravitinoConfig; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +public class TestCatalogRegister { + + @TempDir private static Path tempDir; + + private static GravitinoConfig config(Map<String, String> extraConfig) { + Map<String, String> configMap = new HashMap<>(); + configMap.put("gravitino.metalake", "test"); + configMap.put("discovery.uri", "http://localhost:8080";); + configMap.putAll(extraConfig); + return new GravitinoConfig(configMap); + } + + private static Path createStoreFile() throws IOException { + return Files.createTempFile(tempDir, "store", ".jks"); + } + + @Test + public void testDefaultProperties() { + Properties properties = CatalogRegister.buildJdbcProperties(config(Map.of())); + + assertEquals("admin", properties.get("user")); + // An empty password must not be sent to the driver. + assertNull(properties.get("password")); + assertNull(properties.get("SSL")); + assertNull(properties.get("roles")); + } + + @Test + public void testPasswordIsSetWhenNotEmpty() { + Properties properties = + CatalogRegister.buildJdbcProperties( + config(Map.of("trino.jdbc.user", "gravitino", "trino.jdbc.password", "secret"))); + + assertEquals("gravitino", properties.get("user")); + assertEquals("secret", properties.get("password")); + } + + @Test + public void testHttpsDiscoveryUriWithTruststore() throws IOException { + String truststore = createStoreFile().toString(); + Properties properties = + CatalogRegister.buildJdbcProperties( + config( + Map.of( + "discovery.uri", "https://localhost:8443";, + "trino.jdbc.ssl.truststore.path", truststore, + "trino.jdbc.ssl.truststore.password", "truststore-secret", + "trino.jdbc.ssl.truststore.type", "PKCS12"))); + + assertEquals("true", properties.get("SSL")); + assertEquals("FULL", properties.get("SSLVerification")); + assertEquals(truststore, properties.get("SSLTrustStorePath")); + assertEquals("truststore-secret", properties.get("SSLTrustStorePassword")); + assertEquals("PKCS12", properties.get("SSLTrustStoreType")); + } + + @Test + public void testSslEnabledDerivedFromDiscoveryUri() { + GravitinoConfig config = config(Map.of("discovery.uri", "https://localhost:8443";)); + assertTrue(config.isTrinoJdbcSslEnabled()); + + Properties properties = CatalogRegister.buildJdbcProperties(config); + assertEquals("true", properties.get("SSL")); + assertEquals("FULL", properties.get("SSLVerification")); + // Without a configured truststore the driver falls back to the default JVM truststore. + assertNull(properties.get("SSLTrustStorePath")); + } + + @Test + public void testExplicitSslEnabledOverridesDiscoveryUri() { + GravitinoConfig config = + config( + Map.of("discovery.uri", "https://localhost:8443";, "trino.jdbc.ssl.enabled", "false")); + assertFalse(config.isTrinoJdbcSslEnabled()); + assertNull(CatalogRegister.buildJdbcProperties(config).get("SSL")); + } + + @Test + public void testSslEnabledIgnoresSurroundingWhitespace() { + GravitinoConfig config = config(Map.of("trino.jdbc.ssl.enabled", " true ")); + assertTrue(config.isTrinoJdbcSslEnabled()); + assertEquals("true", CatalogRegister.buildJdbcProperties(config).get("SSL")); + } + + @Test + public void testSslVerificationIsNormalized() { + Properties properties = + CatalogRegister.buildJdbcProperties( + config( + Map.of("trino.jdbc.ssl.enabled", "true", "trino.jdbc.ssl.verification", " none "))); + + assertEquals("NONE", properties.get("SSLVerification")); + } + + @Test + public void testRoles() { + Properties properties = + CatalogRegister.buildJdbcProperties(config(Map.of("trino.jdbc.roles", "system:sysadmin"))); + + assertEquals("system:sysadmin", properties.get("roles")); + } + + @Test + public void testExtraPropertiesArePassedThroughAndOverride() { + Properties properties = + CatalogRegister.buildJdbcProperties( + config( + Map.of( + "trino.jdbc.ssl.enabled", "true", + "trino.jdbc.properties.KerberosRemoteServiceName", "trino", + "trino.jdbc.properties.SSLVerification", "CA"))); + + assertEquals("trino", properties.get("KerberosRemoteServiceName")); + // The escape hatch wins over the value derived from the dedicated configuration. + assertEquals("CA", properties.get("SSLVerification")); + } + + @Test + public void testHttpsDiscoveryUriWithKeystore() throws IOException { + String keystore = createStoreFile().toString(); + Properties properties = + CatalogRegister.buildJdbcProperties( + config( + Map.of( + "discovery.uri", "https://localhost:8443";, + "trino.jdbc.ssl.keystore.path", keystore, + "trino.jdbc.ssl.keystore.password", "keystore-secret", + "trino.jdbc.ssl.keystore.type", "PKCS12"))); + + assertEquals(keystore, properties.get("SSLKeyStorePath")); + assertEquals("keystore-secret", properties.get("SSLKeyStorePassword")); + assertEquals("PKCS12", properties.get("SSLKeyStoreType")); + } + + @Test + public void testKeystoreWithoutSslEnabled() throws IOException { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config(Map.of("trino.jdbc.ssl.keystore.path", createStoreFile().toString())))); + + assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.keystore.path")); + } + + @Test + public void testKeystorePasswordWithoutKeystorePath() { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config( + Map.of( + "trino.jdbc.ssl.enabled", "true", + "trino.jdbc.ssl.keystore.password", "keystore-secret")))); + + assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.keystore.path")); + } + + @Test + public void testKeystoreFileNotFound() { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config( + Map.of( + "trino.jdbc.ssl.enabled", "true", + "trino.jdbc.ssl.keystore.path", "/not/exists/keystore.p12")))); + + assertEquals(GRAVITINO_MISSING_CONFIG.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("does not exist")); + } + + @Test + public void testKeystoreWithVerificationNone() throws IOException { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config( + Map.of( + "trino.jdbc.ssl.enabled", "true", + "trino.jdbc.ssl.verification", "NONE", + "trino.jdbc.ssl.keystore.path", createStoreFile().toString())))); + + assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("NONE")); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.keystore.path")); + } + + @Test + public void testInvalidSslVerification() { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config( + Map.of( + "trino.jdbc.ssl.enabled", "true", + "trino.jdbc.ssl.verification", "PARTIAL")))); + + assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.verification")); + } + + @Test + public void testSslVerificationWithoutSslEnabled() { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config(Map.of("trino.jdbc.ssl.verification", "NONE")))); + + assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.enabled")); + } + + @Test + public void testTruststoreWithoutSslEnabled() throws IOException { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config( + Map.of("trino.jdbc.ssl.truststore.path", createStoreFile().toString())))); + + assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.truststore.path")); + } + + @Test + public void testTruststorePasswordWithoutSslEnabled() { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config(Map.of("trino.jdbc.ssl.truststore.password", "truststore-secret")))); + + assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.truststore.password")); + } + + @Test + public void testTruststoreTypeWithoutSslEnabled() { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config(Map.of("trino.jdbc.ssl.truststore.type", "PKCS12")))); + + assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.truststore.type")); + } + + @Test + public void testBlankSslVerificationFallsBackToDefault() { + Properties properties = + CatalogRegister.buildJdbcProperties( + config(Map.of("trino.jdbc.ssl.enabled", "true", "trino.jdbc.ssl.verification", ""))); + + assertEquals("FULL", properties.get("SSLVerification")); + } + + @Test + public void testTruststorePasswordWithoutTruststorePath() { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config( + Map.of( + "trino.jdbc.ssl.enabled", "true", + "trino.jdbc.ssl.truststore.password", "truststore-secret")))); + + assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.truststore.path")); + } + + @Test + public void testTruststoreTypeWithoutTruststorePath() { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config( + Map.of( + "trino.jdbc.ssl.enabled", "true", + "trino.jdbc.ssl.truststore.type", "PKCS12")))); + + assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.truststore.path")); + } + + @Test + public void testTruststoreWithVerificationNone() throws IOException { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config( + Map.of( + "trino.jdbc.ssl.enabled", "true", + "trino.jdbc.ssl.verification", "NONE", + "trino.jdbc.ssl.truststore.path", createStoreFile().toString())))); + + assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("NONE")); + } + + @Test + public void testTruststoreFileNotFound() { + TrinoException e = + assertThrows( + TrinoException.class, + () -> + CatalogRegister.buildJdbcProperties( + config( + Map.of( + "trino.jdbc.ssl.enabled", "true", + "trino.jdbc.ssl.truststore.path", "/not/exists/truststore.jks")))); + + assertEquals(GRAVITINO_MISSING_CONFIG.toErrorCode(), e.getErrorCode()); + assertTrue(e.getMessage().contains("does not exist")); + assertTrue(e.getMessage().contains("trino.jdbc.ssl.truststore.path")); + } +}
