liugddx commented on code in PR #10402: URL: https://github.com/apache/seatunnel/pull/10402#discussion_r2778671297
########## seatunnel-api/src/main/java/org/apache/seatunnel/api/metalake/gravitino/GravitinoClient.java: ########## @@ -0,0 +1,229 @@ +/* + * 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.seatunnel.api.metalake.gravitino; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode; +import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting; + +import org.apache.seatunnel.api.metalake.MetalakeClient; +import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException; +import org.apache.seatunnel.common.utils.JsonUtils; + +import org.apache.http.HttpEntity; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.util.EntityUtils; + +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import static org.apache.seatunnel.api.table.schema.exception.SchemaEvolutionErrorCode.ERROR_INVALID_TABLE_URL; + +@Slf4j +public class GravitinoClient implements MetalakeClient { + + private static final String HEADER_ACCEPT = "Accept"; + private static final String MEDIA_TYPE_GRAVITINO_V1 = "application/vnd.gravitino.v1+json"; + private static final String JSON_FIELD_CATALOG = "catalog"; + private static final String JSON_FIELD_TABLE = "table"; + private static final String JSON_FIELD_PROPERTIES = "properties"; + private static final String ERROR_NO_RESPONSE_ENTITY = "No response entity"; + private static final String ERROR_MISSING_FIELD_TEMPLATE = "Response JSON has no '%s' field"; + private static final int MAX_RETRY_ATTEMPTS = 3; + private static final long RETRY_DELAY_MS = 2000; + private static final Pattern TABLE_URL_PATTERN = + Pattern.compile("/catalogs/([^/]+)/schemas/([^/]+)/tables/([^/]+)"); + + private final CloseableHttpClient httpClient; + + public GravitinoClient() { + this.httpClient = HttpClients.createDefault(); + } + + @VisibleForTesting + protected GravitinoClient(CloseableHttpClient httpClient) { + this.httpClient = httpClient; + } + + @Override + public JsonNode getMetaInfo(String sourceId, String metalakeUrl) throws IOException { + JsonNode rootNode = executeGetRequest(metalakeUrl + sourceId); + JsonNode catalogNode = getRequiredNode(rootNode, JSON_FIELD_CATALOG); + return getRequiredNode(catalogNode, JSON_FIELD_PROPERTIES); + } + + @Override + public JsonNode getTableSchema(String schemaHttpUrl) throws IOException { + JsonNode rootNode = executeGetRequest(schemaHttpUrl); + return getRequiredNode(rootNode, JSON_FIELD_TABLE); + } + + @Override + public TablePath getTableSchemaPath(String schemaHttpUrl) { + if (schemaHttpUrl == null || schemaHttpUrl.isEmpty()) { + throw new SeaTunnelRuntimeException( + ERROR_INVALID_TABLE_URL, "Table URL cannot be null or empty"); + } + final Matcher matcher = getMatcher(schemaHttpUrl); + String catalogName = matcher.group(1); + String schemaName = matcher.group(2); + String tableName = matcher.group(3); + return TablePath.of(catalogName, schemaName, tableName); + } + + private Matcher getMatcher(String schemaHttpUrl) { + Matcher matcher = TABLE_URL_PATTERN.matcher(schemaHttpUrl); + if (!matcher.find()) { + throw new SeaTunnelRuntimeException( + ERROR_INVALID_TABLE_URL, + String.format( + "Invalid table URL format: '%s'. " + + "Expected format: http://host/api/metalakes/{metalake}/catalogs/{catalog}/schemas/{schema}/tables/{table}";, + schemaHttpUrl)); + } + return matcher; + } + + /** + * Execute HTTP GET request and return parsed JSON response. Implements retry with exponential + * backoff for transient failures. + * + * @param url the request URL + * @return parsed JSON root node + * @throws IOException if network or parsing error occurs after all retries + */ + private JsonNode executeGetRequest(String url) throws IOException { + IOException lastException = null; + for (int attempt = 1; attempt <= MAX_RETRY_ATTEMPTS; attempt++) { + HttpGet request = new HttpGet(url); + request.addHeader(HEADER_ACCEPT, MEDIA_TYPE_GRAVITINO_V1); + try (CloseableHttpResponse response = httpClient.execute(request)) { + HttpEntity entity = response.getEntity(); + if (entity == null) { + throw new RuntimeException(ERROR_NO_RESPONSE_ENTITY); + } + try { + return JsonUtils.readTree(entity.getContent()); + } finally { + EntityUtils.consume(entity); + } Review Comment: The `response.getStatusLine().getStatusCode()` is not checked. If Gravitino returns a 4xx/5xx response, the code attempts to parse the error body as JSON, which may lead to misleading exceptions. Suggestion: Check the HTTP status code before reading the entity, and throw an exception containing the status code and body information for non-2xx responses. ########## seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/TableSourceFactory.java: ########## @@ -40,6 +43,18 @@ TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) { "The Factory has not been implemented and the deprecated Plugin will be used."); } + /** + * We can get the catalogTable list in the source configuration through this method + * + * @param context TableFactoryContext + */ + default List<CatalogTable> discoverTableSchemas(TableSourceFactoryContext context) { + try (TableSchemaDiscoverer metaLakeSchemaDiscoverer = + new TableSchemaDiscoverer(context, factoryIdentifier())) { + return metaLakeSchemaDiscoverer.discoverTableSchemas(); + } + } Review Comment: All TableSourceFactory implementations will inherit this method by default, and even if the MetaLake feature is not used, a MetalakeClient (HTTP connection) will be created during construction. Suggestion: Implement lazy initialization in the TableSchemaDiscoverer constructor, creating the client only when there is indeed a schema_url configuration. ########## seatunnel-api/src/main/java/org/apache/seatunnel/api/metalake/gravitino/GravitinoTableSchemaConvertor.java: ########## @@ -0,0 +1,436 @@ +/* + * 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.seatunnel.api.metalake.gravitino; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode; + +import org.apache.seatunnel.api.metalake.MetaLakeTableSchemaConvertor; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.Column; +import org.apache.seatunnel.api.table.catalog.ConstraintKey; +import org.apache.seatunnel.api.table.catalog.PhysicalColumn; +import org.apache.seatunnel.api.table.catalog.PrimaryKey; +import org.apache.seatunnel.api.table.catalog.TableIdentifier; +import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.api.table.catalog.TableSchema; +import org.apache.seatunnel.api.table.type.ArrayType; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.DecimalType; +import org.apache.seatunnel.api.table.type.LocalTimeType; +import org.apache.seatunnel.api.table.type.MapType; +import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.common.constants.MetaLakeType; +import org.apache.seatunnel.common.exception.CommonError; + +import scala.Tuple2; Review Comment: Introducing Scala dependencies in the pure Java seatunnel-api module may lead to unnecessary coupling. Recommendation: Use Java's built-in data structures (such as custom Pair classes or Map.Entry) as alternatives. ########## seatunnel-api/src/main/java/org/apache/seatunnel/api/metalake/gravitino/GravitinoClient.java: ########## @@ -0,0 +1,229 @@ +/* + * 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.seatunnel.api.metalake.gravitino; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode; +import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting; + +import org.apache.seatunnel.api.metalake.MetalakeClient; +import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException; +import org.apache.seatunnel.common.utils.JsonUtils; + +import org.apache.http.HttpEntity; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.util.EntityUtils; + +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import static org.apache.seatunnel.api.table.schema.exception.SchemaEvolutionErrorCode.ERROR_INVALID_TABLE_URL; + +@Slf4j +public class GravitinoClient implements MetalakeClient { + + private static final String HEADER_ACCEPT = "Accept"; + private static final String MEDIA_TYPE_GRAVITINO_V1 = "application/vnd.gravitino.v1+json"; + private static final String JSON_FIELD_CATALOG = "catalog"; + private static final String JSON_FIELD_TABLE = "table"; + private static final String JSON_FIELD_PROPERTIES = "properties"; + private static final String ERROR_NO_RESPONSE_ENTITY = "No response entity"; + private static final String ERROR_MISSING_FIELD_TEMPLATE = "Response JSON has no '%s' field"; + private static final int MAX_RETRY_ATTEMPTS = 3; + private static final long RETRY_DELAY_MS = 2000; + private static final Pattern TABLE_URL_PATTERN = + Pattern.compile("/catalogs/([^/]+)/schemas/([^/]+)/tables/([^/]+)"); + + private final CloseableHttpClient httpClient; + + public GravitinoClient() { + this.httpClient = HttpClients.createDefault(); + } + + @VisibleForTesting + protected GravitinoClient(CloseableHttpClient httpClient) { + this.httpClient = httpClient; + } + + @Override + public JsonNode getMetaInfo(String sourceId, String metalakeUrl) throws IOException { + JsonNode rootNode = executeGetRequest(metalakeUrl + sourceId); + JsonNode catalogNode = getRequiredNode(rootNode, JSON_FIELD_CATALOG); + return getRequiredNode(catalogNode, JSON_FIELD_PROPERTIES); + } + + @Override + public JsonNode getTableSchema(String schemaHttpUrl) throws IOException { + JsonNode rootNode = executeGetRequest(schemaHttpUrl); + return getRequiredNode(rootNode, JSON_FIELD_TABLE); + } + + @Override + public TablePath getTableSchemaPath(String schemaHttpUrl) { + if (schemaHttpUrl == null || schemaHttpUrl.isEmpty()) { + throw new SeaTunnelRuntimeException( + ERROR_INVALID_TABLE_URL, "Table URL cannot be null or empty"); + } + final Matcher matcher = getMatcher(schemaHttpUrl); + String catalogName = matcher.group(1); + String schemaName = matcher.group(2); + String tableName = matcher.group(3); + return TablePath.of(catalogName, schemaName, tableName); + } + + private Matcher getMatcher(String schemaHttpUrl) { + Matcher matcher = TABLE_URL_PATTERN.matcher(schemaHttpUrl); + if (!matcher.find()) { + throw new SeaTunnelRuntimeException( + ERROR_INVALID_TABLE_URL, + String.format( + "Invalid table URL format: '%s'. " + + "Expected format: http://host/api/metalakes/{metalake}/catalogs/{catalog}/schemas/{schema}/tables/{table}";, + schemaHttpUrl)); + } + return matcher; + } + + /** + * Execute HTTP GET request and return parsed JSON response. Implements retry with exponential + * backoff for transient failures. + * + * @param url the request URL + * @return parsed JSON root node + * @throws IOException if network or parsing error occurs after all retries + */ + private JsonNode executeGetRequest(String url) throws IOException { + IOException lastException = null; + for (int attempt = 1; attempt <= MAX_RETRY_ATTEMPTS; attempt++) { + HttpGet request = new HttpGet(url); + request.addHeader(HEADER_ACCEPT, MEDIA_TYPE_GRAVITINO_V1); + try (CloseableHttpResponse response = httpClient.execute(request)) { + HttpEntity entity = response.getEntity(); + if (entity == null) { + throw new RuntimeException(ERROR_NO_RESPONSE_ENTITY); + } + try { + return JsonUtils.readTree(entity.getContent()); + } finally { + EntityUtils.consume(entity); + } + } catch (IOException e) { + lastException = e; + // Check if exception is retryable + if (!isRetryableException(e) || attempt >= MAX_RETRY_ATTEMPTS) { + break; + } + // Exponential backoff delay before retry + long delayMs = RETRY_DELAY_MS; + log.debug( + "HTTP request to {} failed on attempt {}/{}, retrying in {}ms: {}", + url, + attempt, + MAX_RETRY_ATTEMPTS, + delayMs, + e.getMessage()); + sleepQuietly(delayMs); + } + } + throw new IOException( + String.format( + "Failed to execute HTTP request to %s after %d attempts", + url, MAX_RETRY_ATTEMPTS), + lastException); + } + + /** + * Determine if an exception is retryable. Certain exceptions like DNS resolution failures, SSL + * errors, or 4xx client errors should not be retried as they will likely fail again. + * + * @param e the exception to check + * @return true if the exception is retryable, false otherwise + */ + private boolean isRetryableException(IOException e) { + String message = e.getMessage(); + if (message == null) { + return true; + } + // Non-retryable error patterns + String lowerMessage = message.toLowerCase(); + if (lowerMessage.contains("unknownhost") + || lowerMessage.contains("dns") + || lowerMessage.contains("hostname") + || lowerMessage.contains("ssl") + || lowerMessage.contains("certificate") + || lowerMessage.contains("400") + || lowerMessage.contains("401") + || lowerMessage.contains("403") + || lowerMessage.contains("404")) { + log.debug("Non-retryable exception detected", e); + return false; + } Review Comment: It is unlikely that the message of an IOException contains HTTP status code strings (Apache HttpClient's IOException is typically a connection/IO-level error). These checks will not actually take effect. The correct approach is to first check the response status code in executeGetRequest, throwing a non-retryable exception for 4xx errors and retrying for 5xx errors. -- 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]
