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 8d0780b701bd226e34b44acb524e3248cba6f028 Author: diqiu50 <[email protected]> AuthorDate: Fri Aug 21 15:02:53 2026 +0800 [#12546] improvement(trino-connector): Address review findings on registration status Do not report a healthy load loop when a metalake failed, keep the state of a catalog that could not be unregistered from Trino, surface the real reason the Trino connection failed, and survive an Error escaping the load loop. Fix the config key names in the troubleshooting docs. --- docs/trino-connector/supported-catalog.md | 22 +-- .../connector/catalog/CatalogConnectorManager.java | 156 ++++++++++++------ .../trino/connector/catalog/CatalogRegister.java | 18 ++- .../catalog/CatalogRegistrationState.java | 74 +++++++-- .../connector/system/GravitinoSystemConnector.java | 9 +- .../AlterCatalogStoredProcedure.java | 2 +- .../CreateCatalogStoredProcedure.java | 2 +- .../catalog/TestCatalogConnectorManager.java | 175 ++++++++++++++++++++- .../system/TestGravitinoSystemConnector.java | 46 ++++++ .../table/TestGravitinoSystemStatusTables.java | 2 +- 10 files changed, 422 insertions(+), 84 deletions(-) diff --git a/docs/trino-connector/supported-catalog.md b/docs/trino-connector/supported-catalog.md index 8762ce145b..29d70652e3 100644 --- a/docs/trino-connector/supported-catalog.md +++ b/docs/trino-connector/supported-catalog.md @@ -82,9 +82,11 @@ The result is like: Check catalog registration status: -`gravitino.system.catalog` lists the catalogs the Gravitino server knows about. A catalog listed -there is not necessarily usable in Trino: registering it is a separate step that can fail. When a -catalog does not show up in `SHOW CATALOGS`, `gravitino.system.catalog_status` says why. +`gravitino.system.catalog` lists the relational catalogs the Gravitino server knows about, minus any +that match `gravitino.trino.skip-catalog-patterns`. A catalog listed there is not necessarily usable +in Trino: registering it is a separate step that can fail. `gravitino.system.catalog_status` covers +every catalog the connector considered, including the ones `catalog` filters out, and says why each +one is or is not registered. ```sql select catalog_name, status, last_error from gravitino.system.catalog_status; @@ -109,7 +111,7 @@ The result is like: | `status` | One of `REGISTERED`, `FAILED`, `UNSUPPORTED` or `SKIPPED`. See the table below. | | `last_error` | The reason the catalog is not registered, `NULL` when it is. | | `last_attempt_time` | When the catalog was last processed, as an ISO-8601 UTC timestamp. | -| `last_success_time` | When the catalog was last registered successfully, `NULL` if it never was. | +| `last_success_time` | When the catalog was last registered successfully, `NULL` if it never was. Retained when a catalog later fails or becomes unsupported. | | `failure_count` | The number of consecutive failed attempts, `0` when the last attempt succeeded. | | Status | Meaning | @@ -129,15 +131,15 @@ select * from gravitino.system.load_status; | Column | Description | |------------------------|-------------------------------------------------------------------------------------------| -| `trino_started` | Whether the connector can reach the Trino server over JDBC. No catalog is registered until it can. | +| `trino_started` | Whether the Trino server has been observed reachable over JDBC. No catalog is registered until it is. Latched: once true it stays true, so it is not a liveness probe. | | `last_attempt_time` | When the loop last ran, as an ISO-8601 UTC timestamp. | | `last_success_time` | When the loop last completed successfully, `NULL` if it never did. | | `consecutive_failures` | The number of consecutive failed runs, `0` when the last run succeeded. | -| `last_error` | The reason the last run failed, `NULL` when it succeeded. | -| `metalake_errors` | A JSON map of metalake name to its last error, `NULL` when every metalake loaded. | +| `last_error` | The reason the last run did not complete, including waiting for Trino to start, `NULL` when it succeeded. | +| `metalake_errors` | A JSON map of metalake name to its last error, `NULL` when every metalake loaded. A metalake that fails here also fails the run as a whole. | Both tables are served by the coordinator and reflect the last refresh, which runs every -`gravitino.metadata.refresh-interval-second` seconds (10 by default). A catalog created moments ago +`gravitino.metadata.refresh-interval-seconds` seconds (10 by default). A catalog created moments ago may not have been processed yet. Example: @@ -250,12 +252,12 @@ Registration happens in the background, so a catalog that fails to register simp | Symptom | Likely cause | |--------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------| | A catalog is missing from `SHOW CATALOGS` | Query `gravitino.system.catalog_status` and read `status` and `last_error`, then follow the rows below | -| `status = FAILED`, `last_error` mentions `Access Denied` | The `gravitino.trino.user` lacks a Trino system role permitted to run `CREATE CATALOG` | +| `status = FAILED`, `last_error` mentions `Access Denied` | The `trino.jdbc.user` lacks a Trino system role permitted to run `CREATE CATALOG` | | `status = FAILED`, `last_error` mentions a configuration property | A `trino.bypass.` property is not accepted by the underlying Trino connector | | `status = UNSUPPORTED` | The catalog is not relational, or its provider is outside the supported list. `last_error` names the supported providers | | `status = SKIPPED` | The catalog matches `gravitino.trino.skip-catalog-patterns` | | The catalog has no row in `catalog_status` at all | The load loop never reached it. Check `gravitino.system.load_status` | -| `load_status.trino_started = false` | The connector cannot reach Trino over JDBC. Check `discovery.uri`, `gravitino.trino.user` and `gravitino.trino.password` | +| `load_status.trino_started = false` | The connector cannot reach Trino over JDBC. `last_error` carries the connection error. Check `discovery.uri`, `trino.jdbc.user` and `trino.jdbc.password` | | `load_status.last_error` mentions connection refused | The Gravitino server is unreachable. Check `gravitino.uri` | | `load_status.metalake_errors` names a metalake | That metalake could not be listed, the other metalakes are unaffected | | Querying `gravitino.system.catalog_status` itself fails | The entry catalog did not initialise. The error is reported when the entry catalog is created, check the Trino server log at startup | 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 f55752be30..d07ede3d01 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 @@ -28,6 +28,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.TreeMap; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; @@ -64,6 +65,7 @@ public class CatalogConnectorManager { private static final int NUMBER_EXECUTOR_THREAD = 1; private static final int LOAD_METALAKE_TIMEOUT = 60; + private static final int MAX_CAUSE_DEPTH = 32; private int metadataUpdateIntervalSecond = 10; @@ -75,7 +77,8 @@ public class CatalogConnectorManager { new ConcurrentHashMap<>(); // The registration state of every catalog seen by the load loop, keyed by the Trino catalog - // name. Written only by the load loop thread, read by query threads through the system tables. + // name. Written only by the load loop thread, read by query threads through the system tables + // and by stored procedure threads through describeRegistrationFailure(). private final ConcurrentHashMap<String, CatalogRegistrationState> catalogStates = new ConcurrentHashMap<>(); @@ -198,12 +201,16 @@ public class CatalogConnectorManager { lastLoadAttemptTimeMs = System.currentTimeMillis(); try { if (!catalogRegister.isTrinoStarted()) { - String message = "Waiting for the Trino server to start"; - if (!Objects.equals(lastLoadError, message)) { - LOG.info("{}.", message); - } - lastLoadError = message; + // Report why the connection failed. "Waiting for Trino" alone reads the same for a + // coordinator that is seconds from ready and for credentials that will never work. + String cause = catalogRegister.getLastConnectionError(); + String message = + cause == null + ? "Waiting for the Trino server to start" + : "Waiting for the Trino server to start, the last connection attempt failed: " + + cause; trinoStarted = false; + recordLoadFailure(message, null); return; } trinoStarted = true; @@ -234,10 +241,24 @@ public class CatalogConnectorManager { } } - lastSuccessfulLoadTimeMs = System.currentTimeMillis(); - recordLoadSuccess(); - } catch (Exception e) { - recordLoadFailure(toErrorMessage(e), e); + if (metalakeErrors.isEmpty()) { + lastSuccessfulLoadTimeMs = System.currentTimeMillis(); + recordLoadSuccess(); + } else { + // Some metalake failed. The loop reaching its last line is not a health signal, so do not + // advance the success time or clear the error, or load_status would report a healthy loop + // while no catalog is being registered at all. + recordLoadFailure( + String.format( + "%d of %d metalakes failed to load: %s", + metalakeErrors.size(), usedMetalakes.size(), new TreeMap<>(metalakeErrors)), + null); + } + } catch (Throwable t) { + // Catch Throwable, not Exception: scheduleWithFixedDelay silently cancels the task forever + // the first time the runnable throws, and loading a Trino connector plugin can raise + // NoClassDefFoundError. A dead loop must not look like a healthy one. + recordLoadFailure(toErrorMessage(t), t); } } @@ -249,18 +270,21 @@ public class CatalogConnectorManager { consecutiveLoadFailures.set(0); } - private void recordLoadFailure(String message, Exception cause) { + private void recordLoadFailure(String message, Throwable cause) { boolean changed = !Objects.equals(lastLoadError, message); lastLoadError = message; consecutiveLoadFailures.incrementAndGet(); - if (changed) { + if (!changed) { + LOG.debug("Failed to load catalogs from the Gravitino server: {}", message, cause); + } else if (trinoStarted) { LOG.error("Failed to load catalogs from the Gravitino server: {}", message, cause); } else { - LOG.debug("Failed to load catalogs from the Gravitino server: {}", message, cause); + // Trino not being up yet is the normal state during startup, not an error. + LOG.info("{}", message); } } - private void recordMetalakeError(String metalakeName, Exception cause) { + private void recordMetalakeError(String metalakeName, Throwable cause) { String message = toErrorMessage(cause); String previous = metalakeErrors.put(metalakeName, message); if (!Objects.equals(previous, message)) { @@ -270,20 +294,33 @@ public class CatalogConnectorManager { } } - private static String toErrorMessage(Exception e) { + private static String toErrorMessage(Throwable e) { // Report the root cause: the actual reason a registration failed, such as "Access Denied: // Cannot create catalog", is wrapped in several layers of TrinoException by the time it gets - // here, and the outer messages say nothing a user can act on. Do not use - // GravitinoErrorCode.toSimpleErrorMessage(), it throws a NullPointerException when the - // exception carries no message. + // here, and the outer messages say nothing a user can act on. The outermost message is kept + // as a prefix because it names the subsystem that failed. Do not use + // GravitinoErrorCode.toSimpleErrorMessage(), it throws on an exception with no message. Throwable rootCause = e; - while (rootCause.getCause() != null && rootCause.getCause() != rootCause) { - rootCause = rootCause.getCause(); + // Bound the walk: a cause chain can be cyclic, and this runs on the single load loop thread. + for (int depth = 0; depth < MAX_CAUSE_DEPTH; depth++) { + Throwable cause = rootCause.getCause(); + if (cause == null || cause == rootCause) { + break; + } + rootCause = cause; } - String message = rootCause.getMessage(); - if (StringUtils.isBlank(message)) { - message = e.getMessage(); + String rootMessage = describeThrowable(rootCause); + if (rootCause == e) { + return rootMessage; } + String outerMessage = e.getMessage(); + return StringUtils.isBlank(outerMessage) || outerMessage.contains(rootMessage) + ? rootMessage + : outerMessage + ": " + rootMessage; + } + + private static String describeThrowable(Throwable e) { + String message = e.getMessage(); return StringUtils.isBlank(message) ? e.getClass().getName() : message; } @@ -355,12 +392,10 @@ public class CatalogConnectorManager { presentTrinoNames.add(trinoCatalogName); if (skipCatalog(trinoCatalogName)) { recordCatalogState( - CatalogRegistrationState.notLoaded( + CatalogRegistrationState.skipped( metalakeName, catalogName, trinoCatalogName, - null, - CatalogRegistrationState.Status.SKIPPED, "Matched gravitino.trino.skip-catalog-patterns"), null); continue; @@ -384,19 +419,33 @@ public class CatalogConnectorManager { try { unloadCatalog(entry.getValue().getCatalog()); } catch (Exception e) { - LOG.error("Failed to remove catalog {}.", entry.getKey(), e); + // The catalog is gone from Gravitino but is still registered in Trino. Record it, or + // the pruning below would drop the row and the table would report nothing at all about + // a catalog that still shows up in SHOW CATALOGS. + GravitinoCatalog catalog = entry.getValue().getCatalog(); + recordCatalogState( + CatalogRegistrationState.failed( + metalakeName, + catalog.getName(), + entry.getKey(), + catalog.getProvider(), + "The catalog was deleted in Gravitino but could not be unregistered from Trino: " + + toErrorMessage(e)), + e); } } } // Drop the states of catalogs that no longer exist in the Gravitino server, including the - // states of catalogs that never had a connector. + // states of catalogs that never had a connector. A catalog whose connector could not be + // removed from Trino is kept, so that its failure stays visible for as long as it is real. catalogStates .values() .removeIf( state -> state.getMetalake().equals(metalakeName) - && !presentTrinoNames.contains(state.getTrinoCatalogName())); + && !presentTrinoNames.contains(state.getTrinoCatalogName()) + && !catalogConnectors.containsKey(state.getTrinoCatalogName())); // Load new catalogs belows to the metalake. for (String catalogName : catalogNames) { @@ -414,12 +463,11 @@ public class CatalogConnectorManager { CatalogRegistrationState.succeeded(gravitinoCatalog, trinoCatalogName), null); } else if (catalog.type() != Catalog.Type.RELATIONAL) { recordCatalogState( - CatalogRegistrationState.notLoaded( + CatalogRegistrationState.unsupported( metalakeName, catalogName, trinoCatalogName, gravitinoCatalog.getProvider(), - CatalogRegistrationState.Status.UNSUPPORTED, String.format( "Only relational catalogs are supported, the catalog type is %s", catalog.type())), @@ -428,12 +476,11 @@ public class CatalogConnectorManager { .getSupportedCatalogProviders() .contains(gravitinoCatalog.getProvider())) { recordCatalogState( - CatalogRegistrationState.notLoaded( + CatalogRegistrationState.unsupported( metalakeName, catalogName, trinoCatalogName, gravitinoCatalog.getProvider(), - CatalogRegistrationState.Status.UNSUPPORTED, String.format( "The catalog provider %s is not supported, the supported providers are %s", gravitinoCatalog.getProvider(), @@ -444,32 +491,26 @@ public class CatalogConnectorManager { recordCatalogState( CatalogRegistrationState.succeeded(gravitinoCatalog, trinoCatalogName), null); } - } catch (UnsupportedOperationException e) { - recordCatalogState( - CatalogRegistrationState.notLoaded( - metalakeName, - catalogName, - trinoCatalogName, - provider, - CatalogRegistrationState.Status.UNSUPPORTED, - toErrorMessage(e)), - null); } catch (Exception e) { recordCatalogState( CatalogRegistrationState.failed( - metalakeName, - catalogName, - trinoCatalogName, - provider, - toErrorMessage(e), - catalogStates.get(trinoCatalogName)), + metalakeName, catalogName, trinoCatalogName, provider, toErrorMessage(e)), e); } } } - private void recordCatalogState(CatalogRegistrationState state, Exception cause) { - CatalogRegistrationState previous = catalogStates.put(state.getTrinoCatalogName(), state); + private void recordCatalogState(CatalogRegistrationState newState, Throwable cause) { + // Merge under compute() so the history carried over cannot be lost to a concurrent record. + CatalogRegistrationState[] seen = new CatalogRegistrationState[1]; + CatalogRegistrationState state = + catalogStates.compute( + newState.getTrinoCatalogName(), + (name, previous) -> { + seen[0] = previous; + return newState.withHistoryOf(previous); + }); + CatalogRegistrationState previous = seen[0]; boolean changed = previous == null || previous.getStatus() != state.getStatus() @@ -678,17 +719,28 @@ public class CatalogConnectorManager { /** * Describes why a catalog is not registered in Trino, for use in error messages. * + * @param metalake the name of the metalake the catalog belongs to * @param trinoCatalogName the name the catalog would be registered under in Trino * @return a human readable explanation */ - public String describeRegistrationFailure(String trinoCatalogName) { + public String describeRegistrationFailure(String metalake, String trinoCatalogName) { CatalogRegistrationState state = catalogStates.get(trinoCatalogName); if (state != null && state.getLastError() != null) { return String.format("%s: %s", state.getStatus(), state.getLastError()); } + String metalakeError = metalakeErrors.get(metalake); + if (metalakeError != null) { + return String.format("Metalake %s could not be loaded: %s", metalake, metalakeError); + } if (lastLoadError != null) { return lastLoadError; } + if (state != null) { + // The catalog is registered, so the caller is looking at a change that did not take effect. + return String.format( + "The catalog is %s and the last load attempt did not pick up the change.", + state.getStatus()); + } return "The catalog has not been loaded yet, please retry later."; } 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 3df8045ffc..4c25ef36df 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 java.util.regex.Matcher; import java.util.regex.Pattern; import org.apache.commons.lang3.StringUtils; +import javax.annotation.Nullable; import org.apache.gravitino.trino.connector.GravitinoConfig; import org.apache.gravitino.trino.connector.GravitinoErrorCode; import org.apache.gravitino.trino.connector.catalog.iceberg.IcebergConnectorAdapter; @@ -82,6 +83,7 @@ public class CatalogRegister { private Connection connection; private boolean isStarted = false; + private volatile String lastConnectionError; private String catalogStoreDirectory; private GravitinoConfig config; @@ -93,13 +95,27 @@ public class CatalogRegister { String command = "SELECT 1"; try (Statement statement = connection.createStatement()) { isStarted = statement.execute(command); + lastConnectionError = null; return isStarted; } catch (Exception e) { - LOG.warn("Trino server is not started: {}", e.getMessage()); + // Keep the reason: wrong credentials, a wrong port and a coordinator that is still booting + // are indistinguishable to the caller otherwise, and only the first two are actionable. + lastConnectionError = e.getMessage() == null ? e.getClass().getName() : e.getMessage(); + LOG.warn("Trino server is not started: {}", lastConnectionError); return false; } } + /** + * Retrieves the error from the last failed attempt to reach the Trino server. + * + * @return the error message, null if the Trino server was reached + */ + @Nullable + String getLastConnectionError() { + return lastConnectionError; + } + /** * Initializes the catalog register with the specified Trino connector context and Gravitino * configuration. diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegistrationState.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegistrationState.java index 3ebb5dcb02..d553cc821b 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegistrationState.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegistrationState.java @@ -102,7 +102,6 @@ public final class CatalogRegistrationState { * @param trinoCatalogName the name the catalog would be registered under in Trino * @param provider the catalog provider, null if it could not be determined * @param error the error that prevented the registration - * @param previous the previous state of the catalog, null if the catalog was never seen before * @return the registration state */ public static CatalogRegistrationState failed( @@ -110,8 +109,7 @@ public final class CatalogRegistrationState { String catalogName, String trinoCatalogName, @Nullable String provider, - String error, - @Nullable CatalogRegistrationState previous) { + String error) { return new CatalogRegistrationState( metalake, catalogName, @@ -120,40 +118,94 @@ public final class CatalogRegistrationState { Status.FAILED, error, System.currentTimeMillis(), - previous == null ? 0 : previous.lastSuccessTimeMs, - previous == null ? 1 : previous.failureCount + 1); + 0, + 1); + } + + /** + * Creates a state for a catalog that is deliberately not registered because it matches {@code + * gravitino.trino.skip-catalog-patterns}. + * + * @param metalake the name of the metalake the catalog belongs to + * @param catalogName the name of the catalog in Gravitino + * @param trinoCatalogName the name the catalog would be registered under in Trino + * @param reason a human readable explanation of why the catalog is skipped + * @return the registration state + */ + public static CatalogRegistrationState skipped( + String metalake, String catalogName, String trinoCatalogName, String reason) { + return new CatalogRegistrationState( + metalake, + catalogName, + trinoCatalogName, + null, + Status.SKIPPED, + reason, + System.currentTimeMillis(), + 0, + 0); } /** - * Creates a state for a catalog that is intentionally not registered in Trino. + * Creates a state for a catalog the connector cannot register because of its type or provider. * * @param metalake the name of the metalake the catalog belongs to * @param catalogName the name of the catalog in Gravitino * @param trinoCatalogName the name the catalog would be registered under in Trino * @param provider the catalog provider, null if it could not be determined - * @param status the reason category, either {@link Status#SKIPPED} or {@link Status#UNSUPPORTED} - * @param reason a human readable explanation of why the catalog is not registered + * @param reason a human readable explanation of why the catalog is not supported * @return the registration state */ - public static CatalogRegistrationState notLoaded( + public static CatalogRegistrationState unsupported( String metalake, String catalogName, String trinoCatalogName, @Nullable String provider, - Status status, String reason) { return new CatalogRegistrationState( metalake, catalogName, trinoCatalogName, provider, - status, + Status.UNSUPPORTED, reason, System.currentTimeMillis(), 0, 0); } + /** + * Returns this state with the history carried over from the state it replaces. A catalog keeps + * the time it was last registered even after it starts failing, and consecutive failures are + * counted across attempts. + * + * @param previous the state being replaced, null if this catalog was never seen before + * @return this state if there is no history to carry over, a new state carrying it otherwise + */ + CatalogRegistrationState withHistoryOf(@Nullable CatalogRegistrationState previous) { + if (previous == null) { + return this; + } + // A registered catalog stamps its own success time; every other status keeps the last one. + long successTime = status == Status.REGISTERED ? lastSuccessTimeMs : previous.lastSuccessTimeMs; + // Consecutive failures only accumulate while the catalog keeps failing. Any other status + // interrupts the run, and its own count is already zero. + long failures = status == Status.FAILED ? previous.failureCount + 1 : failureCount; + if (successTime == lastSuccessTimeMs && failures == failureCount) { + return this; + } + return new CatalogRegistrationState( + metalake, + catalogName, + trinoCatalogName, + provider, + status, + lastError, + lastAttemptTimeMs, + successTime, + failures); + } + /** * Retrieves the name of the metalake the catalog belongs to. * diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/GravitinoSystemConnector.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/GravitinoSystemConnector.java index 8f5052ec93..82c336f390 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/GravitinoSystemConnector.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/GravitinoSystemConnector.java @@ -193,9 +193,12 @@ public class GravitinoSystemConnector implements Connector { } // The system table data lives on the coordinator only: the catalog load loop runs there, and - // the registration state it records is never replicated to workers. Splits are built in - // SplitManager.getSplits(), which also runs on the coordinator, so this is always set by the - // time it is read. It stays null on worker JVMs, where the behaviour is unchanged. + // the registration state it records is never replicated to workers. Set once by the + // coordinator's GravitinoConnectorFactory.create(), which necessarily runs before any query + // can reach the scheduler, and read by the scheduler through isRemotelyAccessible() and + // getAddresses() below. On a real worker JVM it is never set and the split keeps the previous + // remotely accessible behaviour; in a single JVM test runner the static is shared with the + // coordinator, which is harmless because only the coordinator's scheduler reads it. private static volatile HostAddress coordinatorAddress; /** diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java index caf469e06f..87a5bd83ce 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java @@ -145,7 +145,7 @@ public class AlterCatalogStoredProcedure extends GravitinoStoredProcedure { throw new TrinoException( GravitinoErrorCode.GRAVITINO_OPERATION_FAILED, "Update catalog failed due to the reloading process fails. " - + catalogConnectorManager.describeRegistrationFailure(trinoCatalogName)); + + catalogConnectorManager.describeRegistrationFailure(metalake, trinoCatalogName)); } LOG.info("Alter catalog {} in metalake {} successfully.", catalogName, metalake); diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java index 4fb8438a3a..d4d034d371 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java @@ -120,7 +120,7 @@ public class CreateCatalogStoredProcedure extends GravitinoStoredProcedure { throw new TrinoException( GravitinoErrorCode.GRAVITINO_OPERATION_FAILED, "Create catalog failed due to the loading process fails. " - + catalogConnectorManager.describeRegistrationFailure(trinoCatalogName)); + + catalogConnectorManager.describeRegistrationFailure(metalake, trinoCatalogName)); } LOG.info("Create catalog {} in metalake {} successfully.", catalogName, metalake); diff --git a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java index bc6f42f7ea..0b2a97d971 100644 --- a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java +++ b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java @@ -50,6 +50,7 @@ import org.apache.gravitino.trino.connector.GravitinoConfig; import org.apache.gravitino.trino.connector.GravitinoErrorCode; import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog; import org.junit.jupiter.api.Test; +import org.mockito.Mockito; public class TestCatalogConnectorManager { @@ -343,7 +344,7 @@ public class TestCatalogConnectorManager { CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); manager.loadMetalakeSync(); - String description = manager.describeRegistrationFailure("memory"); + String description = manager.describeRegistrationFailure("test", "memory"); assertTrue(description.contains("FAILED")); assertTrue(description.contains("Access Denied")); } @@ -417,7 +418,9 @@ public class TestCatalogConnectorManager { manager.loadMetalakeSync(); assertEquals(CatalogRegistrationState.Status.REGISTERED, singleState(manager).getStatus()); - when(fixture.metalake.listCatalogs()).thenThrow(new RuntimeException("Connection refused")); + Mockito.doThrow(new RuntimeException("Connection refused")) + .when(fixture.metalake) + .listCatalogs(); manager.loadMetalakeSync(); Map<String, String> metalakeErrors = manager.getMetalakeErrors(); @@ -457,6 +460,170 @@ public class TestCatalogConnectorManager { assertTrue(manager.getCatalogRegistrationStates().isEmpty()); } + @Test + public void testCatalogRecoversFromFailure() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("memory", "memory", Catalog.Type.RELATIONAL)); + doThrow(new TrinoException(GravitinoErrorCode.GRAVITINO_RUNTIME_ERROR, "Access Denied")) + .when(fixture.catalogRegister) + .registerCatalog(any(), any()); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + assertEquals(CatalogRegistrationState.Status.FAILED, singleState(manager).getStatus()); + + // The underlying problem is fixed; the row must not stay stuck on FAILED. + Mockito.reset(fixture.catalogRegister); + when(fixture.catalogRegister.isTrinoStarted()).thenReturn(true); + manager.loadMetalakeSync(); + + CatalogRegistrationState state = singleState(manager); + assertEquals(CatalogRegistrationState.Status.REGISTERED, state.getStatus()); + assertNull(state.getLastError()); + assertEquals(0, state.getFailureCount()); + assertTrue(state.getLastSuccessTimeMs() > 0); + } + + @Test + public void testFailureAfterSuccessKeepsLastSuccessTime() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("memory", "memory", Catalog.Type.RELATIONAL)); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + long successTime = singleState(manager).getLastSuccessTimeMs(); + assertTrue(successTime > 0); + + doThrow(new TrinoException(GravitinoErrorCode.GRAVITINO_RUNTIME_ERROR, "Access Denied")) + .when(fixture.catalogRegister) + .registerCatalog(any(), any()); + manager.loadMetalakeSync(); + + // A catalog that regressed must keep telling the user when it last worked. + CatalogRegistrationState state = singleState(manager); + assertEquals(CatalogRegistrationState.Status.FAILED, state.getStatus()); + assertEquals(1, state.getFailureCount()); + assertEquals(successTime, state.getLastSuccessTimeMs()); + } + + @Test + public void testUnsupportedCatalogKeepsLastSuccessTime() throws Exception { + LoadFixture fixture = new LoadFixture(); + Catalog catalog = mockCatalog("memory", "memory", Catalog.Type.RELATIONAL); + fixture.withCatalogs(catalog); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + long successTime = singleState(manager).getLastSuccessTimeMs(); + + // The provider changes to one the connector does not support. + when(catalog.provider()).thenReturn("unknown"); + manager.loadMetalakeSync(); + + CatalogRegistrationState state = singleState(manager); + assertEquals(CatalogRegistrationState.Status.UNSUPPORTED, state.getStatus()); + assertEquals(successTime, state.getLastSuccessTimeMs()); + } + + @Test + public void testSkippedCatalogSurvivesPruning() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("memory", "memory", Catalog.Type.RELATIONAL)); + + CatalogConnectorManager manager = + fixture.createManager(ImmutableMap.of("gravitino.trino.skip-catalog-patterns", "mem.*")); + manager.loadMetalakeSync(); + manager.loadMetalakeSync(); + + // The row must not flicker away on the second iteration. + assertEquals(CatalogRegistrationState.Status.SKIPPED, singleState(manager).getStatus()); + } + + @Test + public void testMetalakeErrorClearsOnRecovery() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("memory", "memory", Catalog.Type.RELATIONAL)); + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + + Mockito.doThrow(new RuntimeException("Connection refused")) + .when(fixture.metalake) + .listCatalogs(); + manager.loadMetalakeSync(); + assertEquals(1, manager.getMetalakeErrors().size()); + // A metalake that cannot be listed is not a healthy loop. + assertNotNull(manager.getLastLoadError()); + assertEquals(0, manager.getLastSuccessfulLoadTimeMs()); + + fixture.withCatalogs(mockCatalog("memory", "memory", Catalog.Type.RELATIONAL)); + manager.loadMetalakeSync(); + + assertTrue(manager.getMetalakeErrors().isEmpty()); + assertNull(manager.getLastLoadError()); + assertEquals(0, manager.getConsecutiveLoadFailures()); + assertTrue(manager.getLastSuccessfulLoadTimeMs() > 0); + } + + @Test + public void testUnloadFailureKeepsCatalogVisibleInState() throws Exception { + LoadFixture fixture = new LoadFixture(); + Catalog catalog = mockCatalog("memory", "memory", Catalog.Type.RELATIONAL); + fixture.withCatalogs(catalog); + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + + // Trino has the connector, so the manager will try to unload it when it disappears. + CatalogConnectorContext context = + manager.createCatalogConnectorContext( + "memory", createConnectorConfig(catalogConfigJson("test", "memory")), mockContext()); + when(context.getMetalake()).thenReturn(fixture.metalake); + when(context.getCatalog()) + .thenReturn(new GravitinoCatalog("test", "memory", "memory", ImmutableMap.of(), 0L)); + doThrow(new TrinoException(GravitinoErrorCode.GRAVITINO_RUNTIME_ERROR, "Access Denied")) + .when(fixture.catalogRegister) + .unregisterCatalog(any()); + Mockito.doReturn(new String[0]).when(fixture.metalake).listCatalogs(); + manager.loadMetalakeSync(); + + // The catalog is gone from Gravitino but still registered in Trino: the state must say so + // rather than disappear. + CatalogRegistrationState state = singleState(manager); + assertEquals(CatalogRegistrationState.Status.FAILED, state.getStatus()); + assertTrue(state.getLastError().contains("could not be unregistered")); + } + + @Test + public void testTrinoConnectionErrorIsReported() throws Exception { + LoadFixture fixture = new LoadFixture(); + when(fixture.catalogRegister.isTrinoStarted()).thenReturn(false); + when(fixture.catalogRegister.getLastConnectionError()) + .thenReturn("Authentication failed: Access Denied"); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + + // "Waiting for Trino" alone would read the same for a misconfiguration that never resolves. + assertFalse(manager.isTrinoStarted()); + assertTrue(manager.getLastLoadError().contains("Authentication failed")); + assertEquals(1, manager.getConsecutiveLoadFailures()); + } + + @Test + public void testErrorMessageFallsBackWhenNoMessageIsPresent() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("memory", "memory", Catalog.Type.RELATIONAL)); + doThrow(new IllegalStateException()) + .when(fixture.catalogRegister) + .registerCatalog(any(), any()); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + + // An exception with no message must not produce a null or empty last_error. + CatalogRegistrationState state = singleState(manager); + assertEquals(CatalogRegistrationState.Status.FAILED, state.getStatus()); + assertTrue(state.getLastError().contains("IllegalStateException")); + } + private CatalogConnectorManager createManager(ImmutableMap<String, String> configMap) throws Exception { return createManager(createCatalogConnectorFactory(), configMap); @@ -554,9 +721,9 @@ public class TestCatalogConnectorManager { String[] names = new String[catalogs.length]; for (int i = 0; i < catalogs.length; i++) { names[i] = catalogs[i].name(); - when(metalake.loadCatalog(names[i])).thenReturn(catalogs[i]); + Mockito.doReturn(catalogs[i]).when(metalake).loadCatalog(names[i]); } - when(metalake.listCatalogs()).thenReturn(names); + Mockito.doReturn(names).when(metalake).listCatalogs(); } CatalogConnectorManager createManager(Map<String, String> extraConfig) { diff --git a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/TestGravitinoSystemConnector.java b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/TestGravitinoSystemConnector.java index 11b5538b08..1622aff081 100644 --- a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/TestGravitinoSystemConnector.java +++ b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/TestGravitinoSystemConnector.java @@ -18,7 +18,10 @@ */ package org.apache.gravitino.trino.connector.system; +import io.trino.spi.HostAddress; import io.trino.spi.Page; +import io.trino.spi.connector.SchemaTableName; +import java.util.List; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.mockito.Mockito; @@ -64,4 +67,47 @@ public class TestGravitinoSystemConnector { Assertions.assertTrue(pageSource.isFinished()); } } + + @Test + public void testSplitIsRemotelyAccessibleUntilTheCoordinatorIsKnown() { + HostAddress previous = null; + try { + GravitinoSystemConnector.Split.setCoordinatorAddress(null); + GravitinoSystemConnector.Split split = mockSplit(); + + // Without a known coordinator the split keeps the original, unpinned behaviour. + Assertions.assertTrue(split.isRemotelyAccessible()); + Assertions.assertTrue(split.getAddresses().isEmpty()); + } finally { + GravitinoSystemConnector.Split.setCoordinatorAddress(previous); + } + } + + @Test + public void testSplitIsPinnedToTheCoordinator() { + try { + // The system tables are only populated on the coordinator, so a split must never be + // scheduled onto a worker. + HostAddress coordinator = HostAddress.fromParts("127.0.0.1", 8080); + GravitinoSystemConnector.Split.setCoordinatorAddress(coordinator); + GravitinoSystemConnector.Split split = mockSplit(); + + Assertions.assertFalse(split.isRemotelyAccessible()); + Assertions.assertEquals(List.of(coordinator), split.getAddresses()); + } finally { + GravitinoSystemConnector.Split.setCoordinatorAddress(null); + } + } + + private static final SchemaTableName TABLE_NAME = new SchemaTableName("system", "catalog"); + + private static GravitinoSystemConnector.Split mockSplit() { + // Mocked rather than subclassed: ConnectorSplit's abstract members differ across the + // supported Trino SPI versions, and this test compiles against all of them. + return Mockito.mock( + GravitinoSystemConnector.Split.class, + Mockito.withSettings() + .useConstructor(TABLE_NAME) + .defaultAnswer(Mockito.CALLS_REAL_METHODS)); + } } diff --git a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/table/TestGravitinoSystemStatusTables.java b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/table/TestGravitinoSystemStatusTables.java index 7139f1a163..43b60ecd7e 100644 --- a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/table/TestGravitinoSystemStatusTables.java +++ b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/table/TestGravitinoSystemStatusTables.java @@ -64,7 +64,7 @@ public class TestGravitinoSystemStatusTables { public void testCatalogStatusTableRendersFailedCatalog() { CatalogRegistrationState state = CatalogRegistrationState.failed( - "test", "memory", "memory", null, "Access Denied: Cannot create catalog memory", null); + "test", "memory", "memory", null, "Access Denied: Cannot create catalog memory"); Page page = loadCatalogStatusPage(List.of(state));
