wombatu-kun commented on code in PR #19876:
URL: https://github.com/apache/hudi/pull/19876#discussion_r3974925346
##########
hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java:
##########
@@ -1010,4 +1034,631 @@ void testUpdateTableSchema_issuesOneUpdateTable() {
verify(mockAwsGlue, times(1)).updateTable(any(UpdateTableRequest.class));
verify(mockAwsGlue,
never()).batchUpdatePartition(any(BatchUpdatePartitionRequest.class));
}
+
+ @Test
+ void testGetPartitionsFromList_returnsPartitionsKnownToGlue() {
+ String tableName = "tbl";
+ software.amazon.awssdk.services.glue.model.Partition gluePartition =
+ software.amazon.awssdk.services.glue.model.Partition.builder()
+ .values("2024-01-15")
+
.storageDescriptor(StorageDescriptor.builder().location("s3://base/2024/01/15").build())
+ .build();
+ ArgumentCaptor<BatchGetPartitionRequest> captor =
ArgumentCaptor.forClass(BatchGetPartitionRequest.class);
+ when(mockAwsGlue.batchGetPartition(captor.capture()))
+ .thenReturn(CompletableFuture.completedFuture(
+
BatchGetPartitionResponse.builder().partitions(gluePartition).build()));
+
+ List<Partition> result =
awsGlueSyncClient.getPartitionsFromList(tableName, Arrays.asList("2024/01/15",
"2024/01/16"));
+
+ assertEquals(1, result.size(), "only the partition Glue knows about is
returned");
+ assertEquals(Collections.singletonList("2024-01-15"),
result.get(0).getValues());
+ assertEquals("s3://base/2024/01/15", result.get(0).getStorageLocation());
+
+ BatchGetPartitionRequest sent = captor.getValue();
+ assertEquals(GlueTestUtil.DB_NAME, sent.databaseName());
+ assertEquals(tableName, sent.tableName());
+ assertEquals(Arrays.asList(Collections.singletonList("2024-01-15"),
Collections.singletonList("2024-01-16")),
+
sent.partitionsToGet().stream().map(PartitionValueList::values).collect(Collectors.toList()),
+ "the requested partitions are the extracted partition values, not the
storage paths");
+ }
+
+ @Test
+ void testGetPartitionsFromList_emptyListDoesNotCallGlue() {
+ assertTrue(awsGlueSyncClient.getPartitionsFromList("tbl",
Collections.emptyList()).isEmpty());
+ verify(mockAwsGlue,
never()).batchGetPartition(any(BatchGetPartitionRequest.class));
+ }
+
+ @Test
+ void testGetMetastoreSchema_mergesColumnsAndPartitionKeys() {
+ String tableName = "tbl";
+ List<Column> columns = Arrays.asList(GlueTestUtil.getColumn("name",
"string", null),
+ GlueTestUtil.getColumn("age", "int", null));
+ List<Column> partitionKeys =
Collections.singletonList(GlueTestUtil.getColumn("datestr", "string", null));
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(getTableWithDefaultProps(tableName, columns,
partitionKeys));
+
+ Map<String, String> schema =
awsGlueSyncClient.getMetastoreSchema(tableName);
+
+ assertEquals(3, schema.size());
+ assertEquals("STRING", schema.get("name"), "column types are upper cased");
+ assertEquals("INT", schema.get("age"));
+ assertEquals("STRING", schema.get("datestr"), "partition keys are merged
into the schema");
+ }
+
+ @Test
+ void testGetMetastoreSchema_wrapsGlueFailure() {
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.getMetastoreSchema("tbl"));
+ assertTrue(ex.getMessage().contains("Fail to get schema for table"));
+ }
+
+ @Test
+ void testGetLastCommitTimeSynced_readsTableParameters() {
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(HOODIE_LAST_COMMIT_TIME_SYNC, "100");
+ parameters.put(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC, "110");
+ Table withSyncTimes = tableWithParameters("synced", parameters);
+ Table withoutSyncTimes = tableWithParameters("unsynced", new HashMap<>());
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withSyncTimes).build()))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withoutSyncTimes).build()));
+
+ assertEquals("100",
awsGlueSyncClient.getLastCommitTimeSynced("synced").get());
+ assertEquals("110",
awsGlueSyncClient.getLastCommitCompletionTimeSynced("synced").get());
+ // the table is cached per name, so the second table name triggers the
second stubbed response
+
assertFalse(awsGlueSyncClient.getLastCommitTimeSynced("unsynced").isPresent());
+
assertFalse(awsGlueSyncClient.getLastCommitCompletionTimeSynced("unsynced").isPresent());
+ verify(mockAwsGlue, times(2)).getTable(any(GetTableRequest.class));
+ }
+
+ @Test
+ void testGetStorageFieldSchemas_readsFieldsAndDocsFromStorage() {
+ Map<String, FieldSchema> byName =
awsGlueSyncClient.getStorageFieldSchemas().stream()
+ .collect(Collectors.toMap(FieldSchema::getName, f -> f));
+
+ assertEquals("int", byName.get("id").getType());
+ assertEquals(GlueTestUtil.ID_FIELD_DOC,
byName.get("id").getComment().get());
+ assertEquals("string", byName.get("name").getType());
+ assertEquals(GlueTestUtil.NAME_FIELD_DOC,
byName.get("name").getComment().get());
+ assertTrue(byName.containsKey("_hoodie_commit_time"), "metadata fields are
part of the storage schema");
+ }
+
+ @Test
+ void testManagePartitionIndexes_disabledDeactivatesFlagAndDropsIndexes()
throws Exception {
+ String tableName = "tbl";
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(GLUE_PARTITION_INDEX_ENABLE, "true");
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
parameters)).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(partitionIndexDescriptor("idx_one",
"datestr"))
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("false",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE),
+ "partition index usage is deactivated when the feature is off");
+ assertEquals(Collections.singletonList("idx_one"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()));
+ verify(mockAwsGlue,
never()).createPartitionIndex(any(CreatePartitionIndexRequest.class));
+ }
+
+ @Test
+ void
testManagePartitionIndexes_enabledDropsStaleIndexesAndCreatesMissingOnes()
throws Exception {
+ String tableName = "tbl";
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS_ENABLE.key(),
"true");
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS.key(),
"datestr;hour,region");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ // the table has no partition_filtering.enabled parameter, so indexing has
to be activated first
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
+ PartitionIndexDescriptor keptIndex = partitionIndexDescriptor("kept_idx",
"datestr", "hour");
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex,
partitionIndexDescriptor("stale_idx", "old_col"))
+ .build()))
+ // after a drop the index list is re-read
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex)
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+ ArgumentCaptor<CreatePartitionIndexRequest> createCaptor =
ArgumentCaptor.forClass(CreatePartitionIndexRequest.class);
+ when(mockAwsGlue.createPartitionIndex(createCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(CreatePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("true",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE));
+ assertEquals(Collections.singletonList("stale_idx"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()),
+ "only the index that is no longer configured is dropped");
+ assertEquals(1, createCaptor.getAllValues().size(), "the already existing
index is not recreated");
+ PartitionIndex created = createCaptor.getValue().partitionIndex();
+ assertEquals(Collections.singletonList("region"), created.keys());
+ assertEquals("hudi_managed_[region]", created.indexName());
+ verify(mockAwsGlue,
times(2)).getPartitionIndexes(any(GetPartitionIndexesRequest.class));
Review Comment:
The enabled-path test starts from a table with no
`partition_filtering.enabled` and one stale index, so the steady state every
later sync hits - flag already true, index list already matching the config -
never runs, and the `indexesChanges` re-read guard stays untested. Worth a
third case asserting a single `getPartitionIndexes` and no `updateTable` or
`deletePartitionIndex`?
##########
hudi-aws/src/test/java/org/apache/hudi/aws/utils/TestDynamoTableUtils.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.hudi.aws.utils;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
+import software.amazon.awssdk.services.dynamodb.model.CreateTableRequest;
+import software.amazon.awssdk.services.dynamodb.model.CreateTableResponse;
+import software.amazon.awssdk.services.dynamodb.model.DeleteTableRequest;
+import software.amazon.awssdk.services.dynamodb.model.DeleteTableResponse;
+import software.amazon.awssdk.services.dynamodb.model.DescribeTableRequest;
+import software.amazon.awssdk.services.dynamodb.model.DescribeTableResponse;
+import software.amazon.awssdk.services.dynamodb.model.ResourceInUseException;
+import
software.amazon.awssdk.services.dynamodb.model.ResourceNotFoundException;
+import software.amazon.awssdk.services.dynamodb.model.TableDescription;
+import software.amazon.awssdk.services.dynamodb.model.TableStatus;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests {@link DynamoTableUtils} against a mocked DynamoDB client. The
polling helpers are always
+ * given an explicit, short timeout so the tests stay fast and deterministic.
+ */
+@ExtendWith(MockitoExtension.class)
+class TestDynamoTableUtils {
+
+ private static final String TABLE_NAME = "lock_table";
+ private static final int TIMEOUT_MS = 300;
+ private static final int INTERVAL_MS = 50;
+
+ @Mock
+ private DynamoDbClient dynamoDb;
+
+ @Test
+ void testWaitUntilExists_returnsOnTheFirstDescription() throws Exception {
+
when(dynamoDb.describeTable(any(DescribeTableRequest.class))).thenReturn(describeResponse(TableStatus.CREATING));
+
+ DynamoTableUtils.waitUntilExists(dynamoDb, TABLE_NAME, TIMEOUT_MS,
INTERVAL_MS);
+
+ ArgumentCaptor<DescribeTableRequest> captor =
ArgumentCaptor.forClass(DescribeTableRequest.class);
+ verify(dynamoDb, times(1)).describeTable(captor.capture());
+ assertEquals(TABLE_NAME, captor.getValue().tableName(),
+ "any table status is enough to prove the table exists");
+ }
+
+ @Test
+ void testWaitUntilExists_pollsUntilTheTableShowsUp() throws Exception {
+ when(dynamoDb.describeTable(any(DescribeTableRequest.class)))
+ .thenThrow(ResourceNotFoundException.builder().message("not there
yet").build())
+ .thenReturn(describeResponse(TableStatus.ACTIVE));
+
+ DynamoTableUtils.waitUntilExists(dynamoDb, TABLE_NAME, TIMEOUT_MS,
INTERVAL_MS);
+
+ verify(dynamoDb, times(2)).describeTable(any(DescribeTableRequest.class));
+ }
+
+ @Test
+ void testWaitUntilExists_throwsWhenTheTableNeverShowsUp() {
+ when(dynamoDb.describeTable(any(DescribeTableRequest.class)))
+ .thenThrow(ResourceNotFoundException.builder().message("not there
yet").build());
+
+ SdkClientException ex = assertThrows(SdkClientException.class,
+ () -> DynamoTableUtils.waitUntilExists(dynamoDb, TABLE_NAME,
TIMEOUT_MS, INTERVAL_MS));
+
+ assertTrue(ex.getMessage().contains(TABLE_NAME + " never returned a
result"));
+ verify(dynamoDb,
atLeast(2)).describeTable(any(DescribeTableRequest.class));
+ }
+
+ @Test
+ void testWaitUntilActive_returnsWhenTheTableIsActive() throws Exception {
+ when(dynamoDb.describeTable(any(DescribeTableRequest.class)))
+ .thenReturn(describeResponse(TableStatus.CREATING))
+ .thenReturn(describeResponse(TableStatus.ACTIVE));
+
+ DynamoTableUtils.waitUntilActive(dynamoDb, TABLE_NAME, TIMEOUT_MS,
INTERVAL_MS);
+
+ verify(dynamoDb, times(2)).describeTable(any(DescribeTableRequest.class));
+ }
+
+ @Test
+ void testWaitUntilActive_throwsWhenTheTableStaysInAnotherState() {
+
when(dynamoDb.describeTable(any(DescribeTableRequest.class))).thenReturn(describeResponse(TableStatus.CREATING));
+
+ DynamoTableUtils.TableNeverTransitionedToStateException ex =
+
assertThrows(DynamoTableUtils.TableNeverTransitionedToStateException.class,
+ () -> DynamoTableUtils.waitUntilActive(dynamoDb, TABLE_NAME,
TIMEOUT_MS, INTERVAL_MS));
+
+ assertTrue(ex.getMessage().contains(TABLE_NAME + " never transitioned to
desired state of ACTIVE"));
+ }
+
+ @Test
+ void testWaitUntilActive_throwsWhenTheTableNeverAppears() {
+ when(dynamoDb.describeTable(any(DescribeTableRequest.class)))
+ .thenThrow(ResourceNotFoundException.builder().message("not there
yet").build());
+
+ assertThrows(DynamoTableUtils.TableNeverTransitionedToStateException.class,
+ () -> DynamoTableUtils.waitUntilActive(dynamoDb, TABLE_NAME,
TIMEOUT_MS, INTERVAL_MS),
+ "a table that never gets described is reported the same way as one
stuck in another state");
+ }
+
+ @Test
+ void testDefaultTimeoutOverloadsReturnAsSoonAsTheTableIsReady() throws
Exception {
+
when(dynamoDb.describeTable(any(DescribeTableRequest.class))).thenReturn(describeResponse(TableStatus.ACTIVE));
+
+ // both overloads poll before sleeping, so an already-ready table returns
without waiting
+ DynamoTableUtils.waitUntilExists(dynamoDb, TABLE_NAME);
Review Comment:
This calls the two-arg overloads, whose defaults are a 20 minute timeout and
a 10 second interval, so the class javadoc's "always given an explicit, short
timeout" does not hold here and nothing bounds the runtime if the poll and the
sleep ever swapped order. Add `@Timeout` to this test and reword the javadoc?
##########
hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java:
##########
@@ -1010,4 +1034,631 @@ void testUpdateTableSchema_issuesOneUpdateTable() {
verify(mockAwsGlue, times(1)).updateTable(any(UpdateTableRequest.class));
verify(mockAwsGlue,
never()).batchUpdatePartition(any(BatchUpdatePartitionRequest.class));
}
+
+ @Test
+ void testGetPartitionsFromList_returnsPartitionsKnownToGlue() {
+ String tableName = "tbl";
+ software.amazon.awssdk.services.glue.model.Partition gluePartition =
+ software.amazon.awssdk.services.glue.model.Partition.builder()
+ .values("2024-01-15")
+
.storageDescriptor(StorageDescriptor.builder().location("s3://base/2024/01/15").build())
+ .build();
+ ArgumentCaptor<BatchGetPartitionRequest> captor =
ArgumentCaptor.forClass(BatchGetPartitionRequest.class);
+ when(mockAwsGlue.batchGetPartition(captor.capture()))
+ .thenReturn(CompletableFuture.completedFuture(
+
BatchGetPartitionResponse.builder().partitions(gluePartition).build()));
+
+ List<Partition> result =
awsGlueSyncClient.getPartitionsFromList(tableName, Arrays.asList("2024/01/15",
"2024/01/16"));
+
+ assertEquals(1, result.size(), "only the partition Glue knows about is
returned");
+ assertEquals(Collections.singletonList("2024-01-15"),
result.get(0).getValues());
+ assertEquals("s3://base/2024/01/15", result.get(0).getStorageLocation());
+
+ BatchGetPartitionRequest sent = captor.getValue();
+ assertEquals(GlueTestUtil.DB_NAME, sent.databaseName());
+ assertEquals(tableName, sent.tableName());
+ assertEquals(Arrays.asList(Collections.singletonList("2024-01-15"),
Collections.singletonList("2024-01-16")),
+
sent.partitionsToGet().stream().map(PartitionValueList::values).collect(Collectors.toList()),
+ "the requested partitions are the extracted partition values, not the
storage paths");
+ }
+
+ @Test
+ void testGetPartitionsFromList_emptyListDoesNotCallGlue() {
+ assertTrue(awsGlueSyncClient.getPartitionsFromList("tbl",
Collections.emptyList()).isEmpty());
+ verify(mockAwsGlue,
never()).batchGetPartition(any(BatchGetPartitionRequest.class));
+ }
+
+ @Test
+ void testGetMetastoreSchema_mergesColumnsAndPartitionKeys() {
+ String tableName = "tbl";
+ List<Column> columns = Arrays.asList(GlueTestUtil.getColumn("name",
"string", null),
+ GlueTestUtil.getColumn("age", "int", null));
+ List<Column> partitionKeys =
Collections.singletonList(GlueTestUtil.getColumn("datestr", "string", null));
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(getTableWithDefaultProps(tableName, columns,
partitionKeys));
+
+ Map<String, String> schema =
awsGlueSyncClient.getMetastoreSchema(tableName);
+
+ assertEquals(3, schema.size());
+ assertEquals("STRING", schema.get("name"), "column types are upper cased");
+ assertEquals("INT", schema.get("age"));
+ assertEquals("STRING", schema.get("datestr"), "partition keys are merged
into the schema");
+ }
+
+ @Test
+ void testGetMetastoreSchema_wrapsGlueFailure() {
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.getMetastoreSchema("tbl"));
+ assertTrue(ex.getMessage().contains("Fail to get schema for table"));
+ }
+
+ @Test
+ void testGetLastCommitTimeSynced_readsTableParameters() {
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(HOODIE_LAST_COMMIT_TIME_SYNC, "100");
+ parameters.put(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC, "110");
+ Table withSyncTimes = tableWithParameters("synced", parameters);
+ Table withoutSyncTimes = tableWithParameters("unsynced", new HashMap<>());
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withSyncTimes).build()))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withoutSyncTimes).build()));
+
+ assertEquals("100",
awsGlueSyncClient.getLastCommitTimeSynced("synced").get());
+ assertEquals("110",
awsGlueSyncClient.getLastCommitCompletionTimeSynced("synced").get());
+ // the table is cached per name, so the second table name triggers the
second stubbed response
+
assertFalse(awsGlueSyncClient.getLastCommitTimeSynced("unsynced").isPresent());
+
assertFalse(awsGlueSyncClient.getLastCommitCompletionTimeSynced("unsynced").isPresent());
+ verify(mockAwsGlue, times(2)).getTable(any(GetTableRequest.class));
+ }
+
+ @Test
+ void testGetStorageFieldSchemas_readsFieldsAndDocsFromStorage() {
+ Map<String, FieldSchema> byName =
awsGlueSyncClient.getStorageFieldSchemas().stream()
+ .collect(Collectors.toMap(FieldSchema::getName, f -> f));
+
+ assertEquals("int", byName.get("id").getType());
+ assertEquals(GlueTestUtil.ID_FIELD_DOC,
byName.get("id").getComment().get());
+ assertEquals("string", byName.get("name").getType());
+ assertEquals(GlueTestUtil.NAME_FIELD_DOC,
byName.get("name").getComment().get());
+ assertTrue(byName.containsKey("_hoodie_commit_time"), "metadata fields are
part of the storage schema");
+ }
+
+ @Test
+ void testManagePartitionIndexes_disabledDeactivatesFlagAndDropsIndexes()
throws Exception {
+ String tableName = "tbl";
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(GLUE_PARTITION_INDEX_ENABLE, "true");
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
parameters)).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(partitionIndexDescriptor("idx_one",
"datestr"))
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("false",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE),
+ "partition index usage is deactivated when the feature is off");
+ assertEquals(Collections.singletonList("idx_one"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()));
+ verify(mockAwsGlue,
never()).createPartitionIndex(any(CreatePartitionIndexRequest.class));
+ }
+
+ @Test
+ void
testManagePartitionIndexes_enabledDropsStaleIndexesAndCreatesMissingOnes()
throws Exception {
+ String tableName = "tbl";
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS_ENABLE.key(),
"true");
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS.key(),
"datestr;hour,region");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ // the table has no partition_filtering.enabled parameter, so indexing has
to be activated first
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
+ PartitionIndexDescriptor keptIndex = partitionIndexDescriptor("kept_idx",
"datestr", "hour");
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex,
partitionIndexDescriptor("stale_idx", "old_col"))
+ .build()))
+ // after a drop the index list is re-read
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex)
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+ ArgumentCaptor<CreatePartitionIndexRequest> createCaptor =
ArgumentCaptor.forClass(CreatePartitionIndexRequest.class);
+ when(mockAwsGlue.createPartitionIndex(createCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(CreatePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("true",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE));
+ assertEquals(Collections.singletonList("stale_idx"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()),
+ "only the index that is no longer configured is dropped");
+ assertEquals(1, createCaptor.getAllValues().size(), "the already existing
index is not recreated");
+ PartitionIndex created = createCaptor.getValue().partitionIndex();
+ assertEquals(Collections.singletonList("region"), created.keys());
+ assertEquals("hudi_managed_[region]", created.indexName());
+ verify(mockAwsGlue,
times(2)).getPartitionIndexes(any(GetPartitionIndexesRequest.class));
+ }
+
+ @Test
+ void testParsePartitionsIndexConfig_keepsOnlyTheFirstThreeIndexes() {
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS.key(),
"a;b,c,d,e");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ assertEquals(Arrays.asList(Arrays.asList("a", "b"),
Collections.singletonList("c"), Collections.singletonList("d")),
+ awsGlueSyncClient.parsePartitionsIndexConfig(), "glue supports at most
three partition indexes");
+ }
+
+ @Test
+ void testUpdateLastCommitTimeSynced_writesTimelineInstantToTableParameters()
{
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ ArgumentCaptor<UpdateTableRequest> captor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(captor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder().build()));
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ Map<String, String> parameters =
captor.getValue().tableInput().parameters();
+ assertEquals(GlueTestUtil.INSTANT_TIME,
parameters.get(HOODIE_LAST_COMMIT_TIME_SYNC),
+ "the last instant of the active timeline is synced");
+ assertEquals(GlueTestUtil.COMPLETION_TIME,
parameters.get(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC),
+ "the completion time of that instant is synced alongside it");
+ assertTrue(captor.getValue().skipArchive(), "table archiving is skipped by
default");
+ }
+
+ /**
+ * An indexation already in flight surfaces as an {@link
ExecutionException}, anything else lands in the
+ * catch-all. Neither may fail the commit-time sync.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void
testUpdateLastCommitTimeSynced_partitionIndexFailureDoesNotFailTheSync(boolean
asExecutionFailure) throws Exception {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ when(mockAwsGlue.updateTable(any(UpdateTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+ if (asExecutionFailure) {
+ CompletableFuture<GetPartitionIndexesResponse> failed =
mock(CompletableFuture.class);
+ when(failed.get()).thenThrow(new ExecutionException(new
RuntimeException("indexing in progress")));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class))).thenReturn(failed);
+ } else {
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class))).thenThrow(new
RuntimeException("boom"));
+ }
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ verify(mockAwsGlue, times(1)).updateTable(any(UpdateTableRequest.class));
Review Comment:
Deleting the `ExecutionException` arm this parameterization targets just
drops the failure into the catch-all below it, which swallows identically, so
both values of `asExecutionFailure` pass on this one assertion. Is the `true`
case meant to pin that arm, or is it here as scenario documentation?
##########
hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java:
##########
@@ -1010,4 +1034,631 @@ void testUpdateTableSchema_issuesOneUpdateTable() {
verify(mockAwsGlue, times(1)).updateTable(any(UpdateTableRequest.class));
verify(mockAwsGlue,
never()).batchUpdatePartition(any(BatchUpdatePartitionRequest.class));
}
+
+ @Test
+ void testGetPartitionsFromList_returnsPartitionsKnownToGlue() {
+ String tableName = "tbl";
+ software.amazon.awssdk.services.glue.model.Partition gluePartition =
+ software.amazon.awssdk.services.glue.model.Partition.builder()
+ .values("2024-01-15")
+
.storageDescriptor(StorageDescriptor.builder().location("s3://base/2024/01/15").build())
+ .build();
+ ArgumentCaptor<BatchGetPartitionRequest> captor =
ArgumentCaptor.forClass(BatchGetPartitionRequest.class);
+ when(mockAwsGlue.batchGetPartition(captor.capture()))
+ .thenReturn(CompletableFuture.completedFuture(
+
BatchGetPartitionResponse.builder().partitions(gluePartition).build()));
+
+ List<Partition> result =
awsGlueSyncClient.getPartitionsFromList(tableName, Arrays.asList("2024/01/15",
"2024/01/16"));
+
+ assertEquals(1, result.size(), "only the partition Glue knows about is
returned");
+ assertEquals(Collections.singletonList("2024-01-15"),
result.get(0).getValues());
+ assertEquals("s3://base/2024/01/15", result.get(0).getStorageLocation());
+
+ BatchGetPartitionRequest sent = captor.getValue();
+ assertEquals(GlueTestUtil.DB_NAME, sent.databaseName());
+ assertEquals(tableName, sent.tableName());
+ assertEquals(Arrays.asList(Collections.singletonList("2024-01-15"),
Collections.singletonList("2024-01-16")),
+
sent.partitionsToGet().stream().map(PartitionValueList::values).collect(Collectors.toList()),
+ "the requested partitions are the extracted partition values, not the
storage paths");
+ }
+
+ @Test
+ void testGetPartitionsFromList_emptyListDoesNotCallGlue() {
+ assertTrue(awsGlueSyncClient.getPartitionsFromList("tbl",
Collections.emptyList()).isEmpty());
+ verify(mockAwsGlue,
never()).batchGetPartition(any(BatchGetPartitionRequest.class));
+ }
+
+ @Test
+ void testGetMetastoreSchema_mergesColumnsAndPartitionKeys() {
+ String tableName = "tbl";
+ List<Column> columns = Arrays.asList(GlueTestUtil.getColumn("name",
"string", null),
+ GlueTestUtil.getColumn("age", "int", null));
+ List<Column> partitionKeys =
Collections.singletonList(GlueTestUtil.getColumn("datestr", "string", null));
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(getTableWithDefaultProps(tableName, columns,
partitionKeys));
+
+ Map<String, String> schema =
awsGlueSyncClient.getMetastoreSchema(tableName);
+
+ assertEquals(3, schema.size());
+ assertEquals("STRING", schema.get("name"), "column types are upper cased");
+ assertEquals("INT", schema.get("age"));
+ assertEquals("STRING", schema.get("datestr"), "partition keys are merged
into the schema");
+ }
+
+ @Test
+ void testGetMetastoreSchema_wrapsGlueFailure() {
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.getMetastoreSchema("tbl"));
+ assertTrue(ex.getMessage().contains("Fail to get schema for table"));
+ }
+
+ @Test
+ void testGetLastCommitTimeSynced_readsTableParameters() {
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(HOODIE_LAST_COMMIT_TIME_SYNC, "100");
+ parameters.put(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC, "110");
+ Table withSyncTimes = tableWithParameters("synced", parameters);
+ Table withoutSyncTimes = tableWithParameters("unsynced", new HashMap<>());
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withSyncTimes).build()))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withoutSyncTimes).build()));
+
+ assertEquals("100",
awsGlueSyncClient.getLastCommitTimeSynced("synced").get());
+ assertEquals("110",
awsGlueSyncClient.getLastCommitCompletionTimeSynced("synced").get());
+ // the table is cached per name, so the second table name triggers the
second stubbed response
+
assertFalse(awsGlueSyncClient.getLastCommitTimeSynced("unsynced").isPresent());
+
assertFalse(awsGlueSyncClient.getLastCommitCompletionTimeSynced("unsynced").isPresent());
+ verify(mockAwsGlue, times(2)).getTable(any(GetTableRequest.class));
+ }
+
+ @Test
+ void testGetStorageFieldSchemas_readsFieldsAndDocsFromStorage() {
+ Map<String, FieldSchema> byName =
awsGlueSyncClient.getStorageFieldSchemas().stream()
+ .collect(Collectors.toMap(FieldSchema::getName, f -> f));
+
+ assertEquals("int", byName.get("id").getType());
+ assertEquals(GlueTestUtil.ID_FIELD_DOC,
byName.get("id").getComment().get());
+ assertEquals("string", byName.get("name").getType());
+ assertEquals(GlueTestUtil.NAME_FIELD_DOC,
byName.get("name").getComment().get());
+ assertTrue(byName.containsKey("_hoodie_commit_time"), "metadata fields are
part of the storage schema");
+ }
+
+ @Test
+ void testManagePartitionIndexes_disabledDeactivatesFlagAndDropsIndexes()
throws Exception {
+ String tableName = "tbl";
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(GLUE_PARTITION_INDEX_ENABLE, "true");
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
parameters)).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(partitionIndexDescriptor("idx_one",
"datestr"))
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("false",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE),
+ "partition index usage is deactivated when the feature is off");
+ assertEquals(Collections.singletonList("idx_one"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()));
+ verify(mockAwsGlue,
never()).createPartitionIndex(any(CreatePartitionIndexRequest.class));
+ }
+
+ @Test
+ void
testManagePartitionIndexes_enabledDropsStaleIndexesAndCreatesMissingOnes()
throws Exception {
+ String tableName = "tbl";
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS_ENABLE.key(),
"true");
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS.key(),
"datestr;hour,region");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ // the table has no partition_filtering.enabled parameter, so indexing has
to be activated first
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
+ PartitionIndexDescriptor keptIndex = partitionIndexDescriptor("kept_idx",
"datestr", "hour");
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex,
partitionIndexDescriptor("stale_idx", "old_col"))
+ .build()))
+ // after a drop the index list is re-read
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex)
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+ ArgumentCaptor<CreatePartitionIndexRequest> createCaptor =
ArgumentCaptor.forClass(CreatePartitionIndexRequest.class);
+ when(mockAwsGlue.createPartitionIndex(createCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(CreatePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("true",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE));
+ assertEquals(Collections.singletonList("stale_idx"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()),
+ "only the index that is no longer configured is dropped");
+ assertEquals(1, createCaptor.getAllValues().size(), "the already existing
index is not recreated");
+ PartitionIndex created = createCaptor.getValue().partitionIndex();
+ assertEquals(Collections.singletonList("region"), created.keys());
+ assertEquals("hudi_managed_[region]", created.indexName());
+ verify(mockAwsGlue,
times(2)).getPartitionIndexes(any(GetPartitionIndexesRequest.class));
+ }
+
+ @Test
+ void testParsePartitionsIndexConfig_keepsOnlyTheFirstThreeIndexes() {
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS.key(),
"a;b,c,d,e");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ assertEquals(Arrays.asList(Arrays.asList("a", "b"),
Collections.singletonList("c"), Collections.singletonList("d")),
+ awsGlueSyncClient.parsePartitionsIndexConfig(), "glue supports at most
three partition indexes");
+ }
+
+ @Test
+ void testUpdateLastCommitTimeSynced_writesTimelineInstantToTableParameters()
{
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ ArgumentCaptor<UpdateTableRequest> captor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(captor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder().build()));
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ Map<String, String> parameters =
captor.getValue().tableInput().parameters();
+ assertEquals(GlueTestUtil.INSTANT_TIME,
parameters.get(HOODIE_LAST_COMMIT_TIME_SYNC),
+ "the last instant of the active timeline is synced");
+ assertEquals(GlueTestUtil.COMPLETION_TIME,
parameters.get(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC),
+ "the completion time of that instant is synced alongside it");
+ assertTrue(captor.getValue().skipArchive(), "table archiving is skipped by
default");
+ }
+
+ /**
+ * An indexation already in flight surfaces as an {@link
ExecutionException}, anything else lands in the
+ * catch-all. Neither may fail the commit-time sync.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void
testUpdateLastCommitTimeSynced_partitionIndexFailureDoesNotFailTheSync(boolean
asExecutionFailure) throws Exception {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ when(mockAwsGlue.updateTable(any(UpdateTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+ if (asExecutionFailure) {
+ CompletableFuture<GetPartitionIndexesResponse> failed =
mock(CompletableFuture.class);
+ when(failed.get()).thenThrow(new ExecutionException(new
RuntimeException("indexing in progress")));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class))).thenReturn(failed);
+ } else {
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class))).thenThrow(new
RuntimeException("boom"));
+ }
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ verify(mockAwsGlue, times(1)).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateLastCommitTimeSynced_wrapsGlueFailure() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ when(mockAwsGlue.updateTable(any(UpdateTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.updateLastCommitTimeSynced(tableName));
+ assertTrue(ex.getMessage().contains("Fail to update last sync commit
time"));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_emptyPropertiesSkipUpdate() {
+ assertFalse(awsGlueSyncClient.updateSerdeProperties("tbl",
Collections.emptyMap(), false));
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_unchangedPropertiesSkipUpdate() {
+ String tableName = "tbl";
+ Map<String, String> serdeProperties = new HashMap<>();
+ serdeProperties.put("serialization.format", "1");
+ serdeProperties.put("path", "s3://base");
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+
GetTableResponse.builder().table(tableWithSerdeProperties(tableName,
serdeProperties)).build()));
+
+ assertFalse(awsGlueSyncClient.updateSerdeProperties(tableName, new
HashMap<>(serdeProperties), false));
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_changedPropertiesRewriteSerdeInfo() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+
GetTableResponse.builder().table(tableWithSerdeProperties(tableName,
+ serdePropertiesOf("serialization.format", "1", "location",
"s3://old"))).build()));
+ ArgumentCaptor<UpdateTableRequest> captor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(captor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
+ Map<String, String> serdeProperties = new HashMap<>();
+ serdeProperties.put("path", "s3://new");
+ assertTrue(awsGlueSyncClient.updateSerdeProperties(tableName,
serdeProperties, false));
Review Comment:
All four new `updateSerdeProperties` tests pass `false` for
`useRealtimeFormat`, and `AWSGlueCatalogSyncClient` never reads that parameter,
while `HiveSyncTool` passes `true` when it syncs the MOR `_rt` table. Is the
flag meant to be a no-op for Glue, or is a `true` case worth adding here?
##########
hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java:
##########
@@ -1010,4 +1034,631 @@ void testUpdateTableSchema_issuesOneUpdateTable() {
verify(mockAwsGlue, times(1)).updateTable(any(UpdateTableRequest.class));
verify(mockAwsGlue,
never()).batchUpdatePartition(any(BatchUpdatePartitionRequest.class));
}
+
+ @Test
+ void testGetPartitionsFromList_returnsPartitionsKnownToGlue() {
+ String tableName = "tbl";
+ software.amazon.awssdk.services.glue.model.Partition gluePartition =
+ software.amazon.awssdk.services.glue.model.Partition.builder()
+ .values("2024-01-15")
+
.storageDescriptor(StorageDescriptor.builder().location("s3://base/2024/01/15").build())
+ .build();
+ ArgumentCaptor<BatchGetPartitionRequest> captor =
ArgumentCaptor.forClass(BatchGetPartitionRequest.class);
+ when(mockAwsGlue.batchGetPartition(captor.capture()))
+ .thenReturn(CompletableFuture.completedFuture(
+
BatchGetPartitionResponse.builder().partitions(gluePartition).build()));
+
+ List<Partition> result =
awsGlueSyncClient.getPartitionsFromList(tableName, Arrays.asList("2024/01/15",
"2024/01/16"));
+
+ assertEquals(1, result.size(), "only the partition Glue knows about is
returned");
+ assertEquals(Collections.singletonList("2024-01-15"),
result.get(0).getValues());
+ assertEquals("s3://base/2024/01/15", result.get(0).getStorageLocation());
+
+ BatchGetPartitionRequest sent = captor.getValue();
+ assertEquals(GlueTestUtil.DB_NAME, sent.databaseName());
+ assertEquals(tableName, sent.tableName());
+ assertEquals(Arrays.asList(Collections.singletonList("2024-01-15"),
Collections.singletonList("2024-01-16")),
+
sent.partitionsToGet().stream().map(PartitionValueList::values).collect(Collectors.toList()),
+ "the requested partitions are the extracted partition values, not the
storage paths");
+ }
+
+ @Test
+ void testGetPartitionsFromList_emptyListDoesNotCallGlue() {
+ assertTrue(awsGlueSyncClient.getPartitionsFromList("tbl",
Collections.emptyList()).isEmpty());
+ verify(mockAwsGlue,
never()).batchGetPartition(any(BatchGetPartitionRequest.class));
+ }
+
+ @Test
+ void testGetMetastoreSchema_mergesColumnsAndPartitionKeys() {
+ String tableName = "tbl";
+ List<Column> columns = Arrays.asList(GlueTestUtil.getColumn("name",
"string", null),
+ GlueTestUtil.getColumn("age", "int", null));
+ List<Column> partitionKeys =
Collections.singletonList(GlueTestUtil.getColumn("datestr", "string", null));
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(getTableWithDefaultProps(tableName, columns,
partitionKeys));
+
+ Map<String, String> schema =
awsGlueSyncClient.getMetastoreSchema(tableName);
+
+ assertEquals(3, schema.size());
+ assertEquals("STRING", schema.get("name"), "column types are upper cased");
+ assertEquals("INT", schema.get("age"));
+ assertEquals("STRING", schema.get("datestr"), "partition keys are merged
into the schema");
+ }
+
+ @Test
+ void testGetMetastoreSchema_wrapsGlueFailure() {
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.getMetastoreSchema("tbl"));
+ assertTrue(ex.getMessage().contains("Fail to get schema for table"));
+ }
+
+ @Test
+ void testGetLastCommitTimeSynced_readsTableParameters() {
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(HOODIE_LAST_COMMIT_TIME_SYNC, "100");
+ parameters.put(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC, "110");
+ Table withSyncTimes = tableWithParameters("synced", parameters);
+ Table withoutSyncTimes = tableWithParameters("unsynced", new HashMap<>());
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withSyncTimes).build()))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withoutSyncTimes).build()));
+
+ assertEquals("100",
awsGlueSyncClient.getLastCommitTimeSynced("synced").get());
+ assertEquals("110",
awsGlueSyncClient.getLastCommitCompletionTimeSynced("synced").get());
+ // the table is cached per name, so the second table name triggers the
second stubbed response
+
assertFalse(awsGlueSyncClient.getLastCommitTimeSynced("unsynced").isPresent());
+
assertFalse(awsGlueSyncClient.getLastCommitCompletionTimeSynced("unsynced").isPresent());
+ verify(mockAwsGlue, times(2)).getTable(any(GetTableRequest.class));
+ }
+
+ @Test
+ void testGetStorageFieldSchemas_readsFieldsAndDocsFromStorage() {
+ Map<String, FieldSchema> byName =
awsGlueSyncClient.getStorageFieldSchemas().stream()
+ .collect(Collectors.toMap(FieldSchema::getName, f -> f));
+
+ assertEquals("int", byName.get("id").getType());
+ assertEquals(GlueTestUtil.ID_FIELD_DOC,
byName.get("id").getComment().get());
+ assertEquals("string", byName.get("name").getType());
+ assertEquals(GlueTestUtil.NAME_FIELD_DOC,
byName.get("name").getComment().get());
+ assertTrue(byName.containsKey("_hoodie_commit_time"), "metadata fields are
part of the storage schema");
+ }
+
+ @Test
+ void testManagePartitionIndexes_disabledDeactivatesFlagAndDropsIndexes()
throws Exception {
+ String tableName = "tbl";
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(GLUE_PARTITION_INDEX_ENABLE, "true");
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
parameters)).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(partitionIndexDescriptor("idx_one",
"datestr"))
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("false",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE),
+ "partition index usage is deactivated when the feature is off");
+ assertEquals(Collections.singletonList("idx_one"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()));
+ verify(mockAwsGlue,
never()).createPartitionIndex(any(CreatePartitionIndexRequest.class));
+ }
+
+ @Test
+ void
testManagePartitionIndexes_enabledDropsStaleIndexesAndCreatesMissingOnes()
throws Exception {
+ String tableName = "tbl";
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS_ENABLE.key(),
"true");
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS.key(),
"datestr;hour,region");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ // the table has no partition_filtering.enabled parameter, so indexing has
to be activated first
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
+ PartitionIndexDescriptor keptIndex = partitionIndexDescriptor("kept_idx",
"datestr", "hour");
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex,
partitionIndexDescriptor("stale_idx", "old_col"))
+ .build()))
+ // after a drop the index list is re-read
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex)
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+ ArgumentCaptor<CreatePartitionIndexRequest> createCaptor =
ArgumentCaptor.forClass(CreatePartitionIndexRequest.class);
+ when(mockAwsGlue.createPartitionIndex(createCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(CreatePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("true",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE));
+ assertEquals(Collections.singletonList("stale_idx"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()),
+ "only the index that is no longer configured is dropped");
+ assertEquals(1, createCaptor.getAllValues().size(), "the already existing
index is not recreated");
+ PartitionIndex created = createCaptor.getValue().partitionIndex();
+ assertEquals(Collections.singletonList("region"), created.keys());
+ assertEquals("hudi_managed_[region]", created.indexName());
+ verify(mockAwsGlue,
times(2)).getPartitionIndexes(any(GetPartitionIndexesRequest.class));
+ }
+
+ @Test
+ void testParsePartitionsIndexConfig_keepsOnlyTheFirstThreeIndexes() {
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS.key(),
"a;b,c,d,e");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ assertEquals(Arrays.asList(Arrays.asList("a", "b"),
Collections.singletonList("c"), Collections.singletonList("d")),
+ awsGlueSyncClient.parsePartitionsIndexConfig(), "glue supports at most
three partition indexes");
+ }
+
+ @Test
+ void testUpdateLastCommitTimeSynced_writesTimelineInstantToTableParameters()
{
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ ArgumentCaptor<UpdateTableRequest> captor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(captor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder().build()));
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ Map<String, String> parameters =
captor.getValue().tableInput().parameters();
+ assertEquals(GlueTestUtil.INSTANT_TIME,
parameters.get(HOODIE_LAST_COMMIT_TIME_SYNC),
+ "the last instant of the active timeline is synced");
+ assertEquals(GlueTestUtil.COMPLETION_TIME,
parameters.get(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC),
+ "the completion time of that instant is synced alongside it");
+ assertTrue(captor.getValue().skipArchive(), "table archiving is skipped by
default");
+ }
+
+ /**
+ * An indexation already in flight surfaces as an {@link
ExecutionException}, anything else lands in the
+ * catch-all. Neither may fail the commit-time sync.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void
testUpdateLastCommitTimeSynced_partitionIndexFailureDoesNotFailTheSync(boolean
asExecutionFailure) throws Exception {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ when(mockAwsGlue.updateTable(any(UpdateTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+ if (asExecutionFailure) {
+ CompletableFuture<GetPartitionIndexesResponse> failed =
mock(CompletableFuture.class);
+ when(failed.get()).thenThrow(new ExecutionException(new
RuntimeException("indexing in progress")));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class))).thenReturn(failed);
+ } else {
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class))).thenThrow(new
RuntimeException("boom"));
+ }
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ verify(mockAwsGlue, times(1)).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateLastCommitTimeSynced_wrapsGlueFailure() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ when(mockAwsGlue.updateTable(any(UpdateTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.updateLastCommitTimeSynced(tableName));
+ assertTrue(ex.getMessage().contains("Fail to update last sync commit
time"));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_emptyPropertiesSkipUpdate() {
+ assertFalse(awsGlueSyncClient.updateSerdeProperties("tbl",
Collections.emptyMap(), false));
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_unchangedPropertiesSkipUpdate() {
+ String tableName = "tbl";
+ Map<String, String> serdeProperties = new HashMap<>();
+ serdeProperties.put("serialization.format", "1");
+ serdeProperties.put("path", "s3://base");
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+
GetTableResponse.builder().table(tableWithSerdeProperties(tableName,
serdeProperties)).build()));
+
+ assertFalse(awsGlueSyncClient.updateSerdeProperties(tableName, new
HashMap<>(serdeProperties), false));
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_changedPropertiesRewriteSerdeInfo() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+
GetTableResponse.builder().table(tableWithSerdeProperties(tableName,
+ serdePropertiesOf("serialization.format", "1", "location",
"s3://old"))).build()));
+ ArgumentCaptor<UpdateTableRequest> captor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(captor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
+ Map<String, String> serdeProperties = new HashMap<>();
+ serdeProperties.put("path", "s3://new");
+ assertTrue(awsGlueSyncClient.updateSerdeProperties(tableName,
serdeProperties, false));
+
+ SerDeInfo sent =
captor.getValue().tableInput().storageDescriptor().serdeInfo();
+
assertEquals("org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe",
sent.serializationLibrary(),
+ "the serde class is derived from the base file format");
+ assertEquals("s3://new", sent.parameters().get("path"));
+ assertEquals("1", sent.parameters().get("serialization.format"), "the
serialization format is defaulted in");
+ }
+
+ @Test
+ void testUpdateSerdeProperties_wrapsGlueFailure() {
+
when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(EntityNotFoundException.class);
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.updateSerdeProperties("tbl", new
HashMap<>(Collections.singletonMap("path", "s3://new")), false));
+ assertTrue(ex.getMessage().contains("Failed to update table serde info for
table"));
+ }
+
+ @Test
+ void testCreateTable_existingTableIsNotRecreated() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(getTableWithDefaultProps(tableName,
Collections.emptyList(), Collections.emptyList()));
+
+ awsGlueSyncClient.createTable(tableName, GlueTestUtil.getSimpleSchema(),
"inputFormat", "outputFormat",
+ "serde", new HashMap<>(), new HashMap<>());
+
+ verify(mockAwsGlue, never()).createTable(any(CreateTableRequest.class));
+ }
+
+ @Test
+ void testTableExists_wrapsNonEntityNotFoundExecutionFailure() throws
Exception {
+ CompletableFuture<GetTableResponse> failed = mock(CompletableFuture.class);
+ when(failed.get()).thenThrow(new ExecutionException(new
RuntimeException("boom")));
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenReturn(failed);
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
() -> awsGlueSyncClient.tableExists("tbl"));
+ assertTrue(ex.getMessage().contains("Fail to get table"));
+ }
+
+ @Test
+ void testTableExists_wrapsClientFailure() {
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
() -> awsGlueSyncClient.tableExists("tbl"));
+ assertTrue(ex.getMessage().contains("Fail to get table"));
+ }
+
+ @Test
+ void testDatabaseExists_wrapsClientFailure() {
+ when(mockAwsGlue.getDatabase(any(GetDatabaseRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
() -> awsGlueSyncClient.databaseExists("db"));
+ assertTrue(ex.getMessage().contains("Fail to check if database exists"));
+ }
+
+ @Test
+ void testDropTable_interruptionRestoresTheInterruptFlag() throws Exception {
+ CompletableFuture<DeleteTableResponse> failed =
mock(CompletableFuture.class);
+ when(failed.get()).thenThrow(new InterruptedException("interrupted"));
+
when(mockAwsGlue.deleteTable(any(DeleteTableRequest.class))).thenReturn(failed);
+
+ assertThrows(HoodieGlueSyncException.class, () ->
awsGlueSyncClient.dropTable("tbl"));
+ assertTrue(Thread.interrupted(), "the interrupt flag is restored for
handlers up the stack");
+ }
+
+ @Test
+ void testBuildAsyncClient_appliesTheConfiguredEndpointAndRegion() {
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+ props.setProperty(HoodieAWSConfig.AWS_GLUE_ENDPOINT.key(),
"https://glue.eu-west-1.amazonaws.com";);
+ props.setProperty(HoodieAWSConfig.AWS_GLUE_REGION.key(), "eu-west-1");
+
+ try (MockedStatic<GlueAsyncClient> glueStatic =
mockStatic(GlueAsyncClient.class);
+ MockedStatic<StsClient> stsStatic = mockStatic(StsClient.class)) {
+ GlueAsyncClientBuilder builder = mock(GlueAsyncClientBuilder.class);
+ glueStatic.when(GlueAsyncClient::builder).thenReturn(builder);
+ when(builder.credentialsProvider(any())).thenReturn(builder);
+ when(builder.endpointOverride(any(URI.class))).thenReturn(builder);
+ when(builder.region(any(Region.class))).thenReturn(builder);
+ when(builder.build()).thenReturn(mockAwsGlue);
+ stsStatic.when(StsClient::create).thenReturn(mockSts);
+
+ new AWSGlueCatalogSyncClient(new HiveSyncConfig(props),
GlueTestUtil.getMetaClient());
+
+
verify(builder).endpointOverride(URI.create("https://glue.eu-west-1.amazonaws.com";));
Review Comment:
Both `buildAsyncClient` tests set `hoodie.aws.glue.endpoint`, so the default
deployment with neither endpoint nor region configured - the arm that leaves
the builder untouched - is never exercised. The same static-mock harness with
both unset, asserting `endpointOverride` and `region` are never called, would
cover it.
--
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]
