diqiu50 commented on code in PR #14517:
URL: https://github.com/apache/gravitino/pull/14517#discussion_r4452164299
##########
catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogPropertiesMetadata.java:
##########
@@ -19,86 +19,293 @@
package org.apache.gravitino.catalog.glue;
import static
org.apache.gravitino.catalog.glue.GlueConstants.AWS_ACCESS_KEY_ID;
+import static
org.apache.gravitino.catalog.glue.GlueConstants.AWS_GLUE_API_CALL_ATTEMPT_TIMEOUT_MS;
+import static
org.apache.gravitino.catalog.glue.GlueConstants.AWS_GLUE_API_CALL_TIMEOUT_MS;
import static
org.apache.gravitino.catalog.glue.GlueConstants.AWS_GLUE_CATALOG_ID;
import static
org.apache.gravitino.catalog.glue.GlueConstants.AWS_GLUE_ENDPOINT;
+import static
org.apache.gravitino.catalog.glue.GlueConstants.AWS_GLUE_MAX_ATTEMPTS;
import static org.apache.gravitino.catalog.glue.GlueConstants.AWS_REGION;
import static
org.apache.gravitino.catalog.glue.GlueConstants.AWS_SECRET_ACCESS_KEY;
+import static
org.apache.gravitino.catalog.glue.GlueConstants.DEFAULT_AWS_GLUE_API_CALL_ATTEMPT_TIMEOUT_MS;
+import static
org.apache.gravitino.catalog.glue.GlueConstants.DEFAULT_AWS_GLUE_API_CALL_TIMEOUT_MS;
+import static
org.apache.gravitino.catalog.glue.GlueConstants.DEFAULT_AWS_GLUE_MAX_ATTEMPTS;
import static
org.apache.gravitino.catalog.glue.GlueConstants.DEFAULT_TABLE_FORMAT;
import static
org.apache.gravitino.catalog.glue.GlueConstants.DEFAULT_TABLE_FORMAT_FILTER;
import static
org.apache.gravitino.catalog.glue.GlueConstants.DEFAULT_TABLE_FORMAT_VALUE;
import static
org.apache.gravitino.catalog.glue.GlueConstants.TABLE_FORMAT_FILTER;
import static org.apache.gravitino.catalog.glue.GlueConstants.WAREHOUSE;
-import static
org.apache.gravitino.connector.PropertyEntry.stringOptionalPropertyEntry;
-import static
org.apache.gravitino.connector.PropertyEntry.stringRequiredPropertyEntry;
+import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Arrays;
+import java.util.Locale;
import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.regex.Pattern;
+import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.cloud.storage.AWSPropertiesMetadata;
import org.apache.gravitino.connector.BaseCatalogPropertiesMetadata;
import org.apache.gravitino.connector.PropertyEntry;
+import software.amazon.awssdk.regions.Region;
/** Properties metadata for the AWS Glue Data Catalog connector catalog-level
configuration. */
public class GlueCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetadata {
+ /** Accepted values of {@code default-table-format}, compared
case-insensitively. */
+ static final Set<String> DEFAULT_TABLE_FORMATS = ImmutableSet.of("iceberg",
"hive");
+
+ /** Accepted entries of {@code table-format-filter}, compared
case-insensitively. */
+ static final Set<String> TABLE_FORMAT_FILTER_VALUES =
+ ImmutableSet.of(DEFAULT_TABLE_FORMAT_FILTER, "hive", "iceberg", "delta",
"parquet");
+
+ private static final Pattern CATALOG_ID_PATTERN = Pattern.compile("\\d{12}");
+
+ private static final Pattern HOST_PORT_PATTERN =
+ Pattern.compile("[A-Za-z0-9]([A-Za-z0-9._-]*[A-Za-z0-9])?(:\\d{1,5})?");
+
private static final Map<String, PropertyEntry<?>> PROPERTIES_METADATA =
ImmutableMap.<String, PropertyEntry<?>>builder()
.put(
AWS_REGION,
- stringRequiredPropertyEntry(
+ stringEntry(
AWS_REGION,
"AWS region for the Glue Data Catalog (e.g. us-east-1)",
+ true /* required */,
true /* immutable */,
- false /* hidden */))
+ null /* defaultValue */,
+ GlueCatalogPropertiesMetadata::checkRegion))
.put(
AWS_GLUE_CATALOG_ID,
- stringOptionalPropertyEntry(
+ stringEntry(
AWS_GLUE_CATALOG_ID,
"The 12-digit AWS account ID that owns the Glue catalog."
+ " When omitted, defaults to the caller's AWS account
ID.",
+ false /* required */,
true /* immutable */,
null /* defaultValue */,
- false /* hidden */))
+ GlueCatalogPropertiesMetadata::checkCatalogId))
.put(AWS_ACCESS_KEY_ID, AWSPropertiesMetadata.AWS_ACCESS_KEY_ID)
.put(AWS_SECRET_ACCESS_KEY,
AWSPropertiesMetadata.AWS_SECRET_ACCESS_KEY)
.put(
AWS_GLUE_ENDPOINT,
- stringOptionalPropertyEntry(
+ stringEntry(
AWS_GLUE_ENDPOINT,
"Custom Glue endpoint URL for VPC endpoints or LocalStack
testing"
+ " (e.g. http://localhost:4566)",
+ false /* required */,
false /* immutable */,
null /* defaultValue */,
- false /* hidden */))
+ GlueCatalogPropertiesMetadata::checkEndpoint))
+ .put(
+ AWS_GLUE_API_CALL_TIMEOUT_MS,
+ positiveLongEntry(
+ AWS_GLUE_API_CALL_TIMEOUT_MS,
+ "Upper bound in milliseconds for one Glue API call made by
the catalog,"
+ + " including retries. Not applied to Iceberg table
I/O.",
+ DEFAULT_AWS_GLUE_API_CALL_TIMEOUT_MS))
+ .put(
+ AWS_GLUE_API_CALL_ATTEMPT_TIMEOUT_MS,
+ positiveLongEntry(
+ AWS_GLUE_API_CALL_ATTEMPT_TIMEOUT_MS,
+ "Upper bound in milliseconds for a single attempt of a Glue
API call.",
+ DEFAULT_AWS_GLUE_API_CALL_ATTEMPT_TIMEOUT_MS))
+ .put(
+ AWS_GLUE_MAX_ATTEMPTS,
+ new PropertyEntry.Builder<Integer>()
+ .withName(AWS_GLUE_MAX_ATTEMPTS)
+ .withDescription(
+ "Maximum number of attempts of a Glue API call made by
the catalog,"
+ + " including the first one. Not applied to Iceberg
table I/O.")
+ .withJavaType(Integer.class)
+ .withDefaultValue(DEFAULT_AWS_GLUE_MAX_ATTEMPTS)
+ .withDecoder(
+ value -> {
+ int attempts = Integer.parseInt(value.trim());
+ Preconditions.checkArgument(attempts > 0, "must be
positive");
+ return attempts;
+ })
+ .withEncoder(String::valueOf)
+ .build())
.put(
DEFAULT_TABLE_FORMAT,
- stringOptionalPropertyEntry(
+ stringEntry(
DEFAULT_TABLE_FORMAT,
"Default format for tables created via createTable().
Accepted: iceberg, hive.",
+ false /* required */,
false /* immutable */,
DEFAULT_TABLE_FORMAT_VALUE,
- false /* hidden */))
+ GlueCatalogPropertiesMetadata::checkDefaultTableFormat))
.put(
TABLE_FORMAT_FILTER,
- stringOptionalPropertyEntry(
+ stringEntry(
TABLE_FORMAT_FILTER,
- "Comma-separated table formats exposed by listTables() and
loadTable().",
+ "Comma-separated table formats exposed by the catalog.
Tables of other formats"
+ + " are absent from list, load, alter and drop, and
cannot be created."
+ + " Accepted: all, hive, iceberg, delta, parquet.",
+ false /* required */,
false /* immutable */,
DEFAULT_TABLE_FORMAT_FILTER,
- false /* hidden */))
+ GlueCatalogPropertiesMetadata::checkTableFormatFilter))
.put(
WAREHOUSE,
- stringRequiredPropertyEntry(
+ stringEntry(
WAREHOUSE,
"Base storage path used as warehouse when no explicit
location is set"
+ " at table creation time and the database declares no
LocationUri"
+ " (e.g. s3://my-bucket/warehouse)."
+ " Table location is then derived as
warehouse/database/table.",
+ true /* required */,
false /* immutable */,
- false /* hidden */))
+ null /* defaultValue */,
+ GlueCatalogPropertiesMetadata::checkWarehouse))
.build();
@Override
protected Map<String, PropertyEntry<?>> specificPropertyEntries() {
return PROPERTIES_METADATA;
}
+
+ /**
+ * Parses {@code aws-glue-endpoint} into a URI with an {@code http} or
{@code https} scheme and a
+ * host.
+ *
+ * @param endpoint the configured endpoint
+ * @return the parsed endpoint URI
+ * @throws IllegalArgumentException if the endpoint is not such a URI
+ */
+ static URI parseEndpoint(String endpoint) {
+ URI uri;
+ try {
+ uri = new URI(endpoint);
+ } catch (URISyntaxException e) {
+ throw new IllegalArgumentException(
+ String.format("Property '%s' is not a valid URI: %s",
AWS_GLUE_ENDPOINT, e.getMessage()),
+ e);
+ }
+ Preconditions.checkArgument(
+ ("http".equalsIgnoreCase(uri.getScheme()) ||
"https".equalsIgnoreCase(uri.getScheme()))
+ && hasHost(uri),
+ "Property '%s' must be an http or https URL with a host, got: %s",
+ AWS_GLUE_ENDPOINT,
+ endpoint);
+ return uri;
+ }
+
+ private static boolean hasHost(URI uri) {
+ // getHost() is null for host names with '_', such as Docker service
names, so such an
+ // authority is accepted when it is a plain host name with an optional
port.
+ return uri.getHost() != null
Review Comment:
We intend to leave port-range validation to the underlying AWS/HTTP client
rather than duplicate it in the connector. The local checks do not guarantee
that every unusable endpoint is rejected when the catalog is created. For this
PR, the focus is translating the resulting client failure into an appropriate
Gravitino exception and HTTP response. I'll verify the invalid-port failure
path and add regression coverage, fixing any conversion gap it exposes, and
clarify the scope of the create-time validation.
##########
catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueExceptionConverter.java:
##########
@@ -174,46 +320,150 @@ static RuntimeException toSchemaException(GlueException
e, String context) {
* @return a Gravitino or standard Java runtime exception
*/
static RuntimeException toTableException(GlueException e, String context) {
- if (isAuthenticationFailure(e)) {
- return toAuthenticationException(e, context);
- }
if (e instanceof EntityNotFoundException) {
return new NoSuchTableException(e, "%s does not exist", context);
}
if (e instanceof AlreadyExistsException) {
return new TableAlreadyExistsException(e, "%s already exists", context);
}
- if (e instanceof InvalidInputException) {
- return ExceptionMessages.illegalArgument(context, e);
+ return toGenericException(e, context, null);
+ }
+
+ /**
+ * Converts a failure raised by the Iceberg SDK into a Gravitino exception.
Iceberg reports Glue
+ * and S3 errors either as its own exception types or wrapped around the AWS
SDK exception, so the
+ * cause chain is walked from the outside in and the first recognized type
decides the result:
+ * Iceberg not-found, already-exists, validation and forbidden errors map to
the matching
+ * Gravitino types, a commit conflict maps to {@link
OptimisticLockException}, and AWS SDK causes
+ * go through {@link #toGenericException}. A commit whose state is unknown
is reported as such,
+ * because the change may have been applied. {@link
GravitinoRuntimeException}, {@link
+ * IllegalArgumentException} and {@link UnsupportedOperationException} are
returned as is.
+ *
+ * @param e the failure raised by an Iceberg catalog call
+ * @param context description of the operation context for error messages
+ * @return a Gravitino or standard Java runtime exception
+ */
+ static RuntimeException toIcebergException(RuntimeException e, String
context) {
+ if (e instanceof GravitinoRuntimeException
+ || e instanceof IllegalArgumentException
+ || e instanceof UnsupportedOperationException) {
+ return e;
}
- if (e instanceof AccessDeniedException) {
- return new ForbiddenException(e, "Glue error: %s: %s", context,
awsErrorDetail(e));
+ for (Throwable t = e; t != null; t = t.getCause() == t ? null :
t.getCause()) {
+ if (t instanceof CommitStateUnknownException) {
+ // A local validation failure underneath means nothing was sent, so
the commit did not
+ // happen. Any other cause, such as a transport failure, does not mean
the commit failed.
+ Throwable invalid = findCause(t,
org.apache.iceberg.exceptions.ValidationException.class);
+ if (invalid != null) {
+ return new IllegalArgumentException(
+ String.format("Invalid request for %s: %s", context,
describe(invalid)), e);
+ }
+ return new RuntimeException(
+ String.format(
+ "%s: the commit state is unknown and the change may have been
applied; check the"
+ + " table before retrying: %s",
+ context, describe(t.getCause() != null ? t.getCause() : t)),
+ e);
+ }
+ if (t instanceof CommitFailedException) {
+ Throwable awsCause = findCause(t, SdkException.class);
+ return awsCause == null || awsCause instanceof
ConcurrentModificationException
+ ? new OptimisticLockException(
+ e, "%s: the table was changed concurrently, retry: %s",
context, describe(t))
+ : toGenericException((SdkException) awsCause, context, null);
+ }
+ if (t instanceof AlreadyExistsException) {
+ return new TableAlreadyExistsException(
+ e, "%s: %s", context, awsErrorDetail((AlreadyExistsException) t));
+ }
+ if (t instanceof org.apache.iceberg.exceptions.AlreadyExistsException) {
+ return new TableAlreadyExistsException(e, "%s: %s", context,
describe(t));
+ }
+ if (t instanceof NoSuchNamespaceException) {
+ return new NoSuchSchemaException(e, "%s: %s", context, describe(t));
+ }
+ if (t instanceof org.apache.iceberg.exceptions.NoSuchTableException) {
+ return new NoSuchTableException(e, "%s: %s", context, describe(t));
+ }
+ if (t instanceof org.apache.iceberg.exceptions.NotFoundException) {
+ // Iceberg wraps a Glue EntityNotFoundException here; it may be the
table or its database.
+ return new NotFoundException(e, "%s: %s", context, describe(t));
+ }
+ if (t instanceof org.apache.iceberg.exceptions.ValidationException) {
+ return new IllegalArgumentException(
+ String.format("Invalid request for %s: %s", context, describe(t)),
e);
+ }
+ if (t instanceof org.apache.iceberg.exceptions.ForbiddenException) {
+ // A Glue denial deeper in the chain keeps its AWS error code and
message through the
+ // shared mapping; a denial by another service (such as S3) stays
forbidden.
+ Throwable awsCause = findCause(t, SdkException.class);
+ return awsCause instanceof GlueException
+ ? toGenericException((SdkException) awsCause, context, null)
+ : new ForbiddenException(
+ e, "%s: %s", context, describe(awsCause != null ? awsCause :
t));
+ }
+ if (t instanceof EntityNotFoundException) {
+ // Not wrapped by Iceberg; it may be the table or its database.
+ return new NotFoundException(e, "%s: %s", context, describe(t));
+ }
+ if (t instanceof SdkException) {
+ return toGenericException((SdkException) t, context, null);
+ }
}
- return new RuntimeException("Glue error: " + context + ": " +
awsErrorDetail(e), e);
+ return new RuntimeException(context + ": " + describe(e), e);
}
- private static RuntimeException toAuthenticationException(GlueException e,
String context) {
- return new RuntimeException(
- String.format(
- "Failed to authenticate with AWS Glue for %s. AWS rejected the
configured "
- + "credentials. Verify the '%s' and '%s' catalog properties,
or the configured "
- + "default AWS credential source. AWS error: %s",
- context,
- GlueConstants.AWS_ACCESS_KEY_ID,
- GlueConstants.AWS_SECRET_ACCESS_KEY,
- awsErrorDetail(e)),
- e);
+ private static String regionMismatchMessage(GlueException e, @Nullable
GlueClient client) {
+ GlueServiceClientConfiguration conf =
+ client == null ? null : client.serviceClientConfiguration();
+ String region = conf != null && conf.region() != null ? conf.region().id()
: "unknown";
+ String endpoint =
+ conf != null && conf.endpointOverride() != null &&
conf.endpointOverride().isPresent()
+ ? conf.endpointOverride().get().toString()
+ : "(not set)";
+ return String.format(
+ "The request was signed for a region that does not match the AWS Glue
endpoint. Check that"
+ + " '%s' (%s) and '%s' (%s) refer to the same region. AWS error:
%s",
+ GlueConstants.AWS_REGION,
+ region,
+ GlueConstants.AWS_GLUE_ENDPOINT,
+ endpoint,
+ awsErrorDetail(e));
+ }
+
+ @Nullable
+ private static Throwable findCause(Throwable t, Class<?> type) {
+ for (Throwable c = t.getCause(); c != null; c = c.getCause() == c ? null :
c.getCause()) {
+ if (type.isInstance(c)) {
+ return c;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Describes a throwable for an error message without the name of the
internal Iceberg catalog,
+ * which means nothing to the caller, falling back to the type name when
there is no message.
+ */
+ private static String describe(Throwable t) {
+ String message = StringUtils.defaultIfBlank(t.getMessage(),
t.getClass().getSimpleName());
+ return message.replace(GlueIcebergTableHelper.ICEBERG_CATALOG_NAME + ".",
"");
}
/**
* Renders the AWS-side detail of a Glue exception. AWS names the failing
action and the resource
* there, which is what the caller needs to act on; the error code is
prefixed so the failure can
- * be classified at a glance.
+ * be classified at a glance. Long messages, which quote the offending
request value, are
+ * shortened in the middle so the violated constraint at the end stays
visible.
*
* @param e the Glue exception to describe
* @return the AWS error code and message, or a best-effort description when
they are unavailable
*/
private static String awsErrorDetail(GlueException e) {
+ return StringUtils.abbreviateMiddle(rawAwsErrorDetail(e), " ... ",
MAX_AWS_MESSAGE_LENGTH);
Review Comment:
The intended scope is to shorten the top-level error message, while
retaining the original cause and full diagnostic stack. As you pointed out,
with includeErrorStackTrace enabled, the stack field can still contain the
original AWS message, and the Java client appends that stack to its exception
message. We will keep this existing diagnostic behavior and clarify the PR
description and relevant documentation accordingly: this change does not bound
the entire HTTP response or the final Java client exception message. We do not
plan to change the shared ErrorResponse or client stack handling in this PR.
--
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]
