yihua commented on code in PR #9347:
URL: https://github.com/apache/hudi/pull/9347#discussion_r1283849039
##########
pom.xml:
##########
@@ -130,6 +130,8 @@
<orc.flink.version>1.5.6</orc.flink.version>
<airlift.version>0.16</airlift.version>
<prometheus.version>0.8.0</prometheus.version>
+ <aws.sdk.httpclient.version>4.5.13</aws.sdk.httpclient.version>
+ <aws.sdk.httpcore.version>4.4.13</aws.sdk.httpcore.version>
Review Comment:
Any reason we pick different versions here?
##########
hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java:
##########
@@ -143,31 +142,32 @@ public void addPartitionsToTable(String tableName,
List<String> partitionsToAdd)
LOG.info("Adding " + partitionsToAdd.size() + " partition(s) in table " +
tableId(databaseName, tableName));
try {
Table table = getTable(awsGlue, databaseName, tableName);
- StorageDescriptor sd = table.getStorageDescriptor();
+ StorageDescriptor sd = table.storageDescriptor();
List<PartitionInput> partitionInputs =
partitionsToAdd.stream().map(partition -> {
- StorageDescriptor partitionSd = sd.clone();
String fullPartitionPath = FSUtils.getPartitionPath(getBasePath(),
partition).toString();
List<String> partitionValues =
partitionValueExtractor.extractPartitionValuesInPath(partition);
- partitionSd.setLocation(fullPartitionPath);
- return new
PartitionInput().withValues(partitionValues).withStorageDescriptor(partitionSd);
+ StorageDescriptor partitionSD = sd.copy(copySd ->
copySd.location(fullPartitionPath));
+ return
PartitionInput.builder().values(partitionValues).storageDescriptor(partitionSD).build();
}).collect(Collectors.toList());
- List<Future<BatchCreatePartitionResult>> futures = new ArrayList<>();
+ List<CompletableFuture<BatchCreatePartitionResponse>> futures = new
ArrayList<>();
for (List<PartitionInput> batch :
CollectionUtils.batches(partitionInputs, MAX_PARTITIONS_PER_REQUEST)) {
- BatchCreatePartitionRequest request = new
BatchCreatePartitionRequest();
-
request.withDatabaseName(databaseName).withTableName(tableName).withPartitionInputList(batch);
- futures.add(awsGlue.batchCreatePartitionAsync(request));
+ BatchCreatePartitionRequest request =
BatchCreatePartitionRequest.builder()
+
.databaseName(databaseName).tableName(tableName).partitionInputList(batch).build();
+ futures.add(awsGlue.batchCreatePartition(request));
}
- for (Future<BatchCreatePartitionResult> future : futures) {
- BatchCreatePartitionResult result = future.get();
- if (CollectionUtils.nonEmpty(result.getErrors())) {
- if (result.getErrors().stream().allMatch((error) ->
"AlreadyExistsException".equals(error.getErrorDetail().getErrorCode()))) {
- LOG.warn("Partitions already exist in glue: " +
result.getErrors());
+ for (CompletableFuture<BatchCreatePartitionResponse> future : futures) {
+ BatchCreatePartitionResponse response = future.get();
+ if (CollectionUtils.nonEmpty(response.errors())) {
+ if (response.errors().stream()
+ .allMatch(
+ (error) ->
"AlreadyExistsException".equals(error.errorDetail().errorCode()))) {
Review Comment:
I assume the error name is the same here and this should still work. Is the
error case tested?
##########
hudi-utilities/pom.xml:
##########
@@ -450,6 +450,14 @@
<scope>test</scope>
</dependency>
+ <!-- AWS Services -->
+ <!--
https://mvnrepository.com/artifact/software.amazon.awssdk/aws-java-sdk-sqs -->
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>sqs</artifactId>
+ <version>${aws.sdk.version}</version>
+ </dependency>
Review Comment:
Same question as before:
https://github.com/apache/hudi/pull/8441/files#r1164644285
Should this be moved to `hudi-aws` module (`hudi-utilities` module has
already relied on `hudi-aws` module)?
##########
hudi-aws/src/main/java/org/apache/hudi/aws/transaction/lock/DynamoDBBasedLockProvider.java:
##########
@@ -153,45 +153,53 @@ public LockItem getLock() {
return lock;
}
- private AmazonDynamoDB getDynamoDBClient() {
+ private DynamoDbClient getDynamoDBClient() {
String region =
this.dynamoDBLockConfiguration.getString(DynamoDbBasedLockConfig.DYNAMODB_LOCK_REGION);
String endpointURL =
this.dynamoDBLockConfiguration.contains(DynamoDbBasedLockConfig.DYNAMODB_ENDPOINT_URL.key())
- ?
this.dynamoDBLockConfiguration.getString(DynamoDbBasedLockConfig.DYNAMODB_ENDPOINT_URL)
- :
RegionUtils.getRegion(region).getServiceEndpoint(AmazonDynamoDB.ENDPOINT_PREFIX);
- AwsClientBuilder.EndpointConfiguration dynamodbEndpoint =
- new AwsClientBuilder.EndpointConfiguration(endpointURL, region);
- return AmazonDynamoDBClientBuilder.standard()
- .withEndpointConfiguration(dynamodbEndpoint)
-
.withCredentials(HoodieAWSCredentialsProviderFactory.getAwsCredentialsProvider(dynamoDBLockConfiguration.getProps()))
+ ?
this.dynamoDBLockConfiguration.getString(DynamoDbBasedLockConfig.DYNAMODB_ENDPOINT_URL)
+ :
DynamoDbClient.serviceMetadata().endpointFor(Region.of(region)).toString();
+
+ if (!endpointURL.startsWith("https://";) ||
!endpointURL.startsWith("http://";)) {
Review Comment:
could the endpoint URL start without the HTTP prefix?
--
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]
