Copilot commented on code in PR #12547:
URL: https://github.com/apache/gravitino/pull/12547#discussion_r3921979156
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java:
##########
@@ -206,12 +238,149 @@ private void loadMetalake() {
icebergRestUriDiscovery.refresh(usedMetalake, config,
gravitinoClient);
loadCatalogs(metalake);
} catch (Exception e) {
- LOG.error(e, "Load Metalake %s failed.", usedMetalake);
+ recordMetalakeError(usedMetalake, e);
}
}
- } catch (Exception e) {
- LOG.error(e, "Error when loading metalake");
+
+ pruneMissingMetalakes(usedMetalakes);
+
+ if (metalakeErrors.isEmpty()) {
+ 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);
+ }
+ }
+
+ private void pruneMissingMetalakes(Set<String> usedMetalakes) {
+ // A metalake that was deleted, or that dropped out of the configuration,
leaves its catalog
+ // connectors, catalog rows and cached metalake handle behind. Without
this its catalogs keep
+ // reporting REGISTERED and stay live in Trino even though they no longer
exist, and
+ // getUsedMetalakes()/getMetalake() keep returning a metalake that isn't
loaded anymore.
+ for (Map.Entry<String, CatalogConnectorContext> entry :
catalogConnectors.entrySet()) {
+ GravitinoCatalog catalog = entry.getValue().getCatalog();
+ if (usedMetalakes.contains(catalog.getMetalake())) {
+ continue;
+ }
+ try {
+ unloadCatalog(catalog);
+ } catch (Exception e) {
+ // The metalake is gone but the catalog 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.
+ recordCatalogState(
+ CatalogRegistrationState.failed(
+ catalog.getMetalake(),
+ catalog.getName(),
+ entry.getKey(),
+ catalog.getProvider(),
+ "The metalake was removed but the catalog could not be
unregistered from Trino: "
+ + toErrorMessage(e)),
+ e);
+ }
+ }
+
+ // A catalog whose connector could not be removed from Trino keeps its
state, the same way the
+ // per-catalog pruning in loadCatalogs() does, so the failure recorded
just above stays visible
+ // for as long as the connector still shows up in SHOW CATALOGS.
+ catalogStates
+ .values()
+ .removeIf(
+ state ->
+ !usedMetalakes.contains(state.getMetalake())
+ &&
!catalogConnectors.containsKey(state.getTrinoCatalogName()));
+ metalakeErrors.keySet().removeIf(metalakeName ->
!usedMetalakes.contains(metalakeName));
+ metalakes.keySet().removeIf(metalakeName ->
!usedMetalakes.contains(metalakeName));
+ }
+
+ private void recordLoadSuccess() {
+ if (loadOutcome.lastError != null) {
+ LOG.info("The Gravitino catalog load loop recovered.");
}
+ loadOutcome = new LoadOutcome(System.currentTimeMillis(), null, 0);
+ }
+
+ private void recordLoadFailure(String message, Throwable cause) {
+ LoadOutcome previous = loadOutcome;
+ boolean changed = !Objects.equals(previous.lastError, message);
+ loadOutcome =
+ new LoadOutcome(previous.lastSuccessTimeMs, message,
previous.consecutiveFailures + 1);
+ if (!changed) {
+ LOG.warn(
+ "Failed to load catalogs from the Gravitino server: %s%n%s",
+ message, CatalogRegister.describe(cause));
+ } else if (trinoStarted) {
+ LOG.error(
+ "Failed to load catalogs from the Gravitino server: %s%n%s",
+ message, CatalogRegister.describe(cause));
Review Comment:
`recordLoadFailure` is called with a null cause both while waiting for Trino
and when aggregating metalake errors. On the first metalake failure this branch
calls `CatalogRegister.describe(null)`, throws a `NullPointerException`, and
the outer catch replaces the real `load_status.last_error` with that secondary
failure; repeated startup failures hit the other logging branch in the same
way. Guard the optional cause before rendering it.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java:
##########
@@ -146,16 +144,27 @@ public Connector create(
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);
+ // Only the coordinator runs the load loop, so it is the only node
holding the
+ // registration state the system tables report. Pin their splits to
it.
+ GravitinoSystemConnector.Split.setCoordinatorAddress(
+ getCurrentNodeAddress(trinoConnectorContext));
catalogConnectorManager.start();
}
} catch (Exception e) {
+ if (catalogConnectorManagerStartTriggered) {
+ // Discard the half-started manager instead of keeping it: leaving
it published would
+ // let the next create() skip the startup and hand out a connector
whose load loop
+ // never started. start() fails inside CatalogRegister.init(),
before any catalog is
+ // registered in Trino, so closing it here leaves nothing behind,
and the next create()
+ // can start over once the configuration is fixed.
+ catalogConnectorManager.shutdown();
+ catalogConnectorManager = null;
+ catalogConnectorManagerStartTriggered = false;
Review Comment:
A manager may already own live dynamic connectors when the static catalog
arrives—the dynamic-first path is explicitly supported above. If `start()` then
fails, shutting down and discarding that shared manager closes its Gravitino
client and executor and loses its connector map while those dynamic connectors
remain installed in Trino. Recovery should dispose only the failed startup
resources or otherwise preserve/rebind existing dynamic connector contexts
instead of unconditionally shutting down the manager.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java:
##########
@@ -232,67 +401,197 @@ public GravitinoMetalake retrieveMetalake(String
metalakeName) {
}
private void loadCatalogs(GravitinoMetalake metalake) {
- List<String> catalogNames;
+ String metalakeName = metalake.name();
+ String[] allCatalogNames;
try {
- catalogNames =
- Arrays.stream(metalake.listCatalogs())
- .filter(id -> !skipCatalog(getTrinoCatalogName(metalake.name(),
id)))
- .collect(Collectors.toList());
+ allCatalogNames = metalake.listCatalogs();
} catch (Exception e) {
- LOG.error(e, "Failed to list catalogs in metalake %s.", metalake.name());
+ // Keep the existing catalog states untouched, a transient listing
failure must not turn
+ // healthy catalogs into failed ones. The load status system table
reports the cause.
+ recordMetalakeError(metalakeName, e);
return;
}
+ metalakeErrors.remove(metalakeName);
+
+ // The Trino names of every catalog the Gravitino server currently
reports, including the
+ // catalogs that are intentionally not registered.
+ Set<String> presentTrinoNames = new HashSet<>();
+ List<String> catalogNames = new ArrayList<>();
+ for (String catalogName : allCatalogNames) {
+ String trinoCatalogName = getTrinoCatalogName(metalakeName, catalogName);
+ presentTrinoNames.add(trinoCatalogName);
+ if (skipCatalog(trinoCatalogName)) {
+ recordCatalogState(
+ CatalogRegistrationState.skipped(
+ metalakeName,
+ catalogName,
+ trinoCatalogName,
+ "Matched gravitino.trino.skip-catalog-patterns"),
+ null);
+ continue;
+ }
+ catalogNames.add(catalogName);
+ }
- LOG.debug("Load metalake %s's catalogs. catalogs: %s.", metalake.name(),
catalogNames);
+ LOG.debug("Load metalake %s's catalogs. catalogs: %s.", metalakeName,
catalogNames);
// Delete those catalogs that have been deleted in Gravitino server
- Set<String> catalogNameStrings =
- catalogNames.stream()
- .map(id -> getTrinoCatalogName(metalake.name(), id))
- .collect(Collectors.toSet());
+ Set<String> catalogNameStrings = new HashSet<>();
+ for (String catalogName : catalogNames) {
+ catalogNameStrings.add(getTrinoCatalogName(metalakeName, catalogName));
+ }
for (Map.Entry<String, CatalogConnectorContext> entry :
catalogConnectors.entrySet()) {
if (!catalogNameStrings.contains(entry.getKey())
&&
// Skip the catalog doesn't belong to this metalake.
- entry.getValue().getMetalake().name().equals(metalake.name())) {
+ entry.getValue().getMetalake().name().equals(metalakeName)) {
try {
unloadCatalog(entry.getValue().getCatalog());
} catch (Exception e) {
- LOG.error(e, "Failed to remove catalog %s.", entry.getKey());
+ // 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. 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())
+ &&
!catalogConnectors.containsKey(state.getTrinoCatalogName()));
+
// Load new catalogs belows to the metalake.
- catalogNames.stream()
- .forEach(
- (String catalogName) -> {
- try {
- Catalog catalog = metalake.loadCatalog(catalogName);
- Map<String, String> properties = propsWithSecrets(catalog);
- GravitinoCatalog gravitinoCatalog =
- new GravitinoCatalog(metalake.name(), catalog, properties);
- if
(catalogConnectors.containsKey(getTrinoCatalogName(gravitinoCatalog))) {
- // Reload catalogs that have been updated in Gravitino
server.
- reloadCatalog(gravitinoCatalog);
- } else {
- if (catalog.type() == Catalog.Type.RELATIONAL
- && catalogConnectorFactory
- .getSupportedCatalogProviders()
- .contains(gravitinoCatalog.getProvider())) {
- loadCatalog(gravitinoCatalog);
- }
- }
- } catch (UnsupportedOperationException e) {
- LOG.warn(
- "Unsupported catalog type for catalog %s in metalake %s:
%s",
- catalogName, metalake.name(), e.getMessage());
- } catch (Exception e) {
- LOG.error(
- e, "Failed to load metalake %s's catalog %s.",
metalake.name(), catalogName);
- }
+ for (String catalogName : catalogNames) {
+ String trinoCatalogName = getTrinoCatalogName(metalakeName, catalogName);
+ // Known before the catalog is even loaded, since it only depends on the
name.
+ boolean alreadyRegistered =
catalogConnectors.containsKey(trinoCatalogName);
+ // Tracked outside the try so that a failure can still report the
provider it knows about.
+ String provider = null;
+ try {
+ Catalog catalog = metalake.loadCatalog(catalogName);
+ // Registration deliberately carries only the visible properties. The
resolved secrets are
+ // added by each node in createCatalogConnectorContext(), so that they
never reach the
+ // CREATE CATALOG statement, the catalog properties file Trino
persists from it, or
+ // anything that quotes either of them back.
+ GravitinoCatalog gravitinoCatalog =
+ new GravitinoCatalog(metalakeName, catalog, visibleProps(catalog));
+ provider = gravitinoCatalog.getProvider();
+ if (alreadyRegistered) {
+ // Reload catalogs that have been updated in Gravitino server.
+ reloadCatalog(gravitinoCatalog);
+ recordCatalogState(
+ CatalogRegistrationState.succeeded(gravitinoCatalog,
trinoCatalogName), null);
Review Comment:
This records a fresh successful-registration timestamp even when
`reloadCatalog()` returns without issuing any registration because the catalog
is unchanged. Consequently `last_success_time` advances on every refresh and
duplicates `last_attempt_time`, contrary to the documented meaning “last
registered successfully.” Preserve the previous success time on the no-op path,
for example by having `reloadCatalog` report whether it actually re-registered.
This issue also appears on line 496 of the same file.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableLoadStatus.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.system.table;
+
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.BooleanType.BOOLEAN;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.airlift.log.Logger;
+import io.trino.spi.Page;
+import io.trino.spi.block.BlockBuilder;
+import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorTableMetadata;
+import io.trino.spi.connector.SchemaTableName;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
+import
org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager.LoadOutcome;
+
+/**
+ * An implementation of the load status system table.
+ *
+ * <p>It reports the health of the loop that registers Apache Gravitino
catalogs into Trino. A
+ * failure that prevents the loop from listing catalogs at all, such as an
unreachable Gravitino
+ * server, has no catalog to attach itself to and is only visible here.
+ */
+public class GravitinoSystemTableLoadStatus extends GravitinoSystemTable {
+
+ private static final Logger LOG =
Logger.get(GravitinoSystemTableLoadStatus.class);
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ /** The name of the load status system table. */
+ public static final SchemaTableName TABLE_NAME =
+ new SchemaTableName(SYSTEM_TABLE_SCHEMA_NAME, "load_status");
+
+ private static final ConnectorTableMetadata TABLE_METADATA =
+ new ConnectorTableMetadata(
+ TABLE_NAME,
+ List.of(
+
ColumnMetadata.builder().setName("trino_started").setType(BOOLEAN).build(),
+
ColumnMetadata.builder().setName("last_attempt_time").setType(VARCHAR).build(),
+
ColumnMetadata.builder().setName("last_success_time").setType(VARCHAR).build(),
+
ColumnMetadata.builder().setName("consecutive_failures").setType(BIGINT).build(),
+
ColumnMetadata.builder().setName("last_error").setType(VARCHAR).build(),
+
ColumnMetadata.builder().setName("metalake_errors").setType(VARCHAR).build()));
+
+ private final CatalogConnectorManager catalogConnectorManager;
+ private final String metalake;
+
+ /**
+ * Constructs a new GravitinoSystemTableLoadStatus.
+ *
+ * @param catalogConnectorManager the manager for catalog connectors
+ * @param metalake the metalake to report errors for
+ */
+ public GravitinoSystemTableLoadStatus(
+ CatalogConnectorManager catalogConnectorManager, String metalake) {
+ this.catalogConnectorManager = catalogConnectorManager;
+ this.metalake = metalake;
+ }
+
+ @Override
+ public Page loadPageData() {
+ BlockBuilder trinoStartedColumnBuilder = BOOLEAN.createBlockBuilder(null,
1);
+ BlockBuilder lastAttemptTimeColumnBuilder =
VARCHAR.createBlockBuilder(null, 1);
+ BlockBuilder lastSuccessTimeColumnBuilder =
VARCHAR.createBlockBuilder(null, 1);
+ BlockBuilder consecutiveFailuresColumnBuilder =
BIGINT.createBlockBuilder(null, 1);
+ BlockBuilder lastErrorColumnBuilder = VARCHAR.createBlockBuilder(null, 1);
+ BlockBuilder metalakeErrorsColumnBuilder =
VARCHAR.createBlockBuilder(null, 1);
+
+ // Read once so the three fields below come from the same attempt: reading
them through three
+ // separate calls could otherwise mix a fresh success time with a stale
error from a load that
+ // completed in between the calls.
+ LoadOutcome loadOutcome = catalogConnectorManager.getLoadOutcome();
+
+ BOOLEAN.writeBoolean(trinoStartedColumnBuilder,
catalogConnectorManager.isTrinoStarted());
+ writeTime(lastAttemptTimeColumnBuilder,
catalogConnectorManager.getLastLoadAttemptTimeMs());
Review Comment:
Only three fields are captured in `LoadOutcome`; `trinoStarted`,
`last_attempt_time`, and `metalake_errors` are read from separate volatile/map
state that the load loop mutates before publishing the outcome. A query racing
a refresh can therefore return contradictory diagnostics (for example
`last_error = NULL` with a non-null `metalake_errors`, or `trino_started =
false` with the previous successful outcome). Publish all row fields together
in one immutable completed-attempt snapshot.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
