This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 91f32af2403e test(aws): cover the Glue sync client, DynamoTableUtils
and HoodieAWSConfig (#19876)
91f32af2403e is described below
commit 91f32af2403e1f315e665e19ea4f1e319f1ed969
Author: voonhous <[email protected]>
AuthorDate: Thu Sep 10 13:55:15 2026 +0800
test(aws): cover the Glue sync client, DynamoTableUtils and HoodieAWSConfig
(#19876)
test(aws): cover the Glue sync client, DynamoTableUtils and HoodieAWSConfig
(#19876)
Test-only change, all with Mockito and no AWS access.
- TestAWSGlueSyncClient: 42 new tests on the mocked GlueAsyncClient
fixture covering partition-index management (enable flag, stale
index drop, missing index create, the three-index cap, and the
steady state that re-reads nothing), updateLastCommitTimeSynced
including the swallowed partition-index failures and the empty
timeline, updateSerdeProperties on all four exits and both values
of useRealtimeFormat, getMetastoreSchema, getStorageFieldSchemas,
getPartitionsFromList, the endpoint, region and default arms of
buildAsyncClient, and the failure-wrapping branches of create,
drop, exists, update and addPartitions. Each asserts the request
sent to Glue, not only the return value.
- TestDynamoTableUtils (new, 13): waitUntilExists and waitUntilActive
with short explicit timeouts, the default-overload path bounded by
@Timeout, createTableIfNotExists and deleteTableIfExists on both
exits.
- TestHoodieAWSConfig (new, 4): every builder method, build()
defaults, fromProperties, fromFile.
- TestS3StorageLockClient: two more branches of the upsert error
handling.
- GlueTestUtil wrote its commit as raw JSON to .hoodie/101.commit,
which the 1.x timeline layout never reads, so the fixture table had
an empty timeline and no schema. It now writes the commit through
DefaultCommitMetadataSerDe under .hoodie/timeline/ and records the
table schema.
---
.../hudi/aws/sync/TestAWSGlueSyncClient.java | 717 +++++++++++++++++++++
.../apache/hudi/aws/testutils/GlueTestUtil.java | 59 +-
.../transaction/lock/TestS3StorageLockClient.java | 29 +
.../hudi/aws/utils/TestDynamoTableUtils.java | 184 ++++++
.../apache/hudi/config/TestHoodieAWSConfig.java | 114 ++++
5 files changed, 1095 insertions(+), 8 deletions(-)
diff --git
a/hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java
b/hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java
index 5ff5ca1c738b..7e4c667f586f 100644
--- a/hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java
+++ b/hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java
@@ -24,6 +24,7 @@ import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.config.GlueCatalogSyncClientConfig;
+import org.apache.hudi.config.HoodieAWSConfig;
import org.apache.hudi.hive.HiveSyncConfig;
import org.apache.hudi.hive.SchemaDifference;
import org.apache.hudi.storage.StoragePath;
@@ -41,35 +42,50 @@ import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.Mock;
+import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.glue.GlueAsyncClient;
+import software.amazon.awssdk.services.glue.GlueAsyncClientBuilder;
import software.amazon.awssdk.services.glue.GlueServiceClientConfiguration;
import software.amazon.awssdk.services.glue.model.BatchCreatePartitionRequest;
import software.amazon.awssdk.services.glue.model.BatchCreatePartitionResponse;
import software.amazon.awssdk.services.glue.model.BatchDeletePartitionRequest;
import software.amazon.awssdk.services.glue.model.BatchDeletePartitionResponse;
+import software.amazon.awssdk.services.glue.model.BatchGetPartitionRequest;
+import software.amazon.awssdk.services.glue.model.BatchGetPartitionResponse;
+import
software.amazon.awssdk.services.glue.model.BatchUpdatePartitionFailureEntry;
import software.amazon.awssdk.services.glue.model.BatchUpdatePartitionRequest;
import
software.amazon.awssdk.services.glue.model.BatchUpdatePartitionRequestEntry;
import software.amazon.awssdk.services.glue.model.BatchUpdatePartitionResponse;
import software.amazon.awssdk.services.glue.model.Column;
import software.amazon.awssdk.services.glue.model.CreateDatabaseRequest;
import software.amazon.awssdk.services.glue.model.CreateDatabaseResponse;
+import software.amazon.awssdk.services.glue.model.CreatePartitionIndexRequest;
+import software.amazon.awssdk.services.glue.model.CreatePartitionIndexResponse;
import software.amazon.awssdk.services.glue.model.CreateTableRequest;
import software.amazon.awssdk.services.glue.model.CreateTableResponse;
import software.amazon.awssdk.services.glue.model.Database;
+import software.amazon.awssdk.services.glue.model.DeletePartitionIndexRequest;
+import software.amazon.awssdk.services.glue.model.DeletePartitionIndexResponse;
import software.amazon.awssdk.services.glue.model.DeleteTableRequest;
import software.amazon.awssdk.services.glue.model.DeleteTableResponse;
import software.amazon.awssdk.services.glue.model.EntityNotFoundException;
import software.amazon.awssdk.services.glue.model.ErrorDetail;
import software.amazon.awssdk.services.glue.model.GetDatabaseRequest;
import software.amazon.awssdk.services.glue.model.GetDatabaseResponse;
+import software.amazon.awssdk.services.glue.model.GetPartitionIndexesRequest;
+import software.amazon.awssdk.services.glue.model.GetPartitionIndexesResponse;
import software.amazon.awssdk.services.glue.model.GetPartitionsRequest;
import software.amazon.awssdk.services.glue.model.GetPartitionsResponse;
import software.amazon.awssdk.services.glue.model.GetTableRequest;
import software.amazon.awssdk.services.glue.model.GetTableResponse;
+import software.amazon.awssdk.services.glue.model.KeySchemaElement;
import software.amazon.awssdk.services.glue.model.PartitionError;
+import software.amazon.awssdk.services.glue.model.PartitionIndex;
+import software.amazon.awssdk.services.glue.model.PartitionIndexDescriptor;
+import software.amazon.awssdk.services.glue.model.PartitionValueList;
import software.amazon.awssdk.services.glue.model.SerDeInfo;
import software.amazon.awssdk.services.glue.model.StorageDescriptor;
import software.amazon.awssdk.services.glue.model.Table;
@@ -82,18 +98,24 @@ import
software.amazon.awssdk.services.sts.model.GetCallerIdentityRequest;
import software.amazon.awssdk.services.sts.model.GetCallerIdentityResponse;
import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
import static org.apache.hudi.aws.testutils.GlueTestUtil.glueSyncProps;
import static org.apache.hudi.common.table.HoodieTableConfig.DATABASE_NAME;
import static
org.apache.hudi.common.table.HoodieTableConfig.HOODIE_TABLE_NAME_KEY;
+import static
org.apache.hudi.sync.common.HoodieMetaSyncOperations.HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC;
+import static
org.apache.hudi.sync.common.HoodieMetaSyncOperations.HOODIE_LAST_COMMIT_TIME_SYNC;
import static org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_BASE_PATH;
import static
org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_DATABASE_NAME;
import static
org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_TABLE_NAME;
@@ -105,6 +127,7 @@ import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -113,6 +136,7 @@ import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
class TestAWSGlueSyncClient {
private static final String CATALOG_ID = "DEFAULT_AWS_ACCOUNT_ID";
+ private static final String GLUE_PARTITION_INDEX_ENABLE =
"partition_filtering.enabled";
@Mock
private GlueAsyncClient mockAwsGlue;
@@ -1010,4 +1034,697 @@ class TestAWSGlueSyncClient {
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
testManagePartitionIndexes_enabledWithTheConfiguredIndexesInPlaceChangesNothing()
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");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ 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()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(partitionIndexDescriptor("kept_idx",
"datestr", "hour"))
+ .build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ // nothing was dropped, so the index list is not re-read
+ verify(mockAwsGlue,
times(1)).getPartitionIndexes(any(GetPartitionIndexesRequest.class));
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ verify(mockAwsGlue,
never()).deletePartitionIndex(any(DeletePartitionIndexRequest.class));
+ verify(mockAwsGlue,
never()).createPartitionIndex(any(CreatePartitionIndexRequest.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. The two parameter
values exist to run each catch arm;
+ * both arms only log, so the observable effect is the same by design.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void
testUpdateLastCommitTimeSynced_partitionIndexFailureDoesNotFailTheSync(boolean
throughExecutionException) 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 (throughExecutionException) {
+ 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));
+ }
+
+ // useRealtimeFormat is not read by the Glue client: unlike the Hive client
it never picks an input format
+ // from the flag, it only rewrites the serde parameters, so both values must
produce the same request.
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testUpdateSerdeProperties_changedPropertiesRewriteSerdeInfo(boolean
useRealtimeFormat) {
+ String tableName = "tbl";
+ Table table = tableWithSerdeProperties(tableName,
+ serdePropertiesOf("serialization.format", "1", "location",
"s3://old"));
+ Table tableWithFormats = table.toBuilder()
+ .storageDescriptor(table.storageDescriptor().toBuilder()
+ .inputFormat("org.apache.hudi.hadoop.HoodieParquetInputFormat")
+
.outputFormat("org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat")
+ .build())
+ .build();
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithFormats).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, useRealtimeFormat));
+
+ StorageDescriptor sentStorageDescriptor =
captor.getValue().tableInput().storageDescriptor();
+ SerDeInfo sent = sentStorageDescriptor.serdeInfo();
+
assertEquals("org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe",
sent.serializationLibrary(),
+ "the serde class is derived from the base file format, not from
useRealtimeFormat");
+ assertEquals("s3://new", sent.parameters().get("path"));
+ assertEquals("1", sent.parameters().get("serialization.format"), "the
serialization format is defaulted in");
+ assertEquals(tableWithFormats.storageDescriptor().inputFormat(),
sentStorageDescriptor.inputFormat(),
+ "the input format is carried over from the catalog for both values of
useRealtimeFormat");
+ assertEquals(tableWithFormats.storageDescriptor().outputFormat(),
sentStorageDescriptor.outputFormat(),
+ "the output format is carried over from the catalog for both values of
useRealtimeFormat");
+ }
+
+ @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 = mockGlueClientBuilder(glueStatic);
+ 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";));
+ verify(builder).region(Region.of("eu-west-1"));
+ }
+ }
+
+ @Test
+ void testBuildAsyncClient_rejectsAMalformedEndpoint() {
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+ props.setProperty(HoodieAWSConfig.AWS_GLUE_ENDPOINT.key(), "https://glue
eu-west-1.amazonaws.com");
+ HiveSyncConfig config = new HiveSyncConfig(props);
+
+ try (MockedStatic<GlueAsyncClient> glueStatic =
mockStatic(GlueAsyncClient.class)) {
+ mockGlueClientBuilder(glueStatic);
+
+ RuntimeException ex = assertThrows(RuntimeException.class,
+ () -> new AWSGlueCatalogSyncClient(config,
GlueTestUtil.getMetaClient()));
+ assertTrue(ex.getCause() instanceof URISyntaxException, "the malformed
endpoint is reported as its parse failure");
+ }
+ }
+
+ @Test
+ void testBuildAsyncClient_withoutAnEndpointOrRegionKeepsTheSdkDefaults() {
+ HiveSyncConfig config = GlueTestUtil.getHiveSyncConfig();
+
+ try (MockedStatic<GlueAsyncClient> glueStatic =
mockStatic(GlueAsyncClient.class);
+ MockedStatic<StsClient> stsStatic = mockStatic(StsClient.class)) {
+ GlueAsyncClientBuilder builder = mockGlueClientBuilder(glueStatic);
+ when(builder.build()).thenReturn(mockAwsGlue);
+ stsStatic.when(StsClient::create).thenReturn(mockSts);
+
+ new AWSGlueCatalogSyncClient(config, GlueTestUtil.getMetaClient());
+
+ verify(builder).credentialsProvider(any());
+ verify(builder).build();
+ verify(builder, never()).endpointOverride(any(URI.class));
+ verify(builder, never()).region(any(Region.class));
+ }
+ }
+
+ @Test
+ void testReplicationOperationsAreUnsupported() {
+ assertThrows(UnsupportedOperationException.class, () ->
awsGlueSyncClient.getLastReplicatedTime("tbl"));
+ assertThrows(UnsupportedOperationException.class, () ->
awsGlueSyncClient.updateLastReplicatedTimeStamp("tbl", "101"));
+ assertThrows(UnsupportedOperationException.class, () ->
awsGlueSyncClient.deleteLastReplicatedTimeStamp("tbl"));
+ }
+
+ @Test
+ void testGeneratePushDownFilter_delegatesToTheGlueFilterGenerator() {
+ assertEquals("datestr = '2024-01-15'",
awsGlueSyncClient.generatePushDownFilter(
+ Collections.singletonList("2024/01/15"), Collections.singletonList(new
FieldSchema("datestr", "string"))));
+ }
+
+ @Test
+ void testGetPartitionsFromList_wrapsGlueFailure() throws Exception {
+ CompletableFuture<BatchGetPartitionResponse> failed =
mock(CompletableFuture.class);
+ when(failed.get()).thenThrow(new ExecutionException(new
RuntimeException("boom")));
+
when(mockAwsGlue.batchGetPartition(any(BatchGetPartitionRequest.class))).thenReturn(failed);
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.getPartitionsFromList("tbl",
Collections.singletonList("2024/01/15")));
+ assertTrue(ex.getMessage().contains("Failed to get all partitions for
table"));
+ }
+
+ @Test
+ void testAddPartitionsToTable_nonAlreadyExistsErrorsFailTheSync() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder()
+ .table(Table.builder().name(tableName)
+
.storageDescriptor(StorageDescriptor.builder().location("s3://base").build()).build())
+ .build()));
+ PartitionError error = PartitionError.builder()
+
.errorDetail(ErrorDetail.builder().errorCode("AccessDeniedException").build()).build();
+
when(mockAwsGlue.batchCreatePartition(any(BatchCreatePartitionRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+
BatchCreatePartitionResponse.builder().errors(Collections.singletonList(error)).build()));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.addPartitionsToTable(tableName,
Collections.singletonList("2024/01/15")));
+ assertTrue(ex.getCause().getCause().getMessage().contains("Fail to add
partitions to"),
+ "an error that is not AlreadyExists fails the sync");
+ }
+
+ @Test
+ void testUpdatePartitionsToTable_errorResponsesFailTheSync() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder()
+ .table(Table.builder().name(tableName)
+
.storageDescriptor(StorageDescriptor.builder().location("s3://base").build()).build())
+ .build()));
+ BatchUpdatePartitionFailureEntry error =
BatchUpdatePartitionFailureEntry.builder()
+ .partitionValueList("2024-01-15")
+
.errorDetail(ErrorDetail.builder().errorCode("AccessDeniedException").build()).build();
+
when(mockAwsGlue.batchUpdatePartition(any(BatchUpdatePartitionRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+
BatchUpdatePartitionResponse.builder().errors(Collections.singletonList(error)).build()));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.updatePartitionsToTable(tableName,
Collections.singletonList("2024/01/15")));
+ assertTrue(ex.getCause().getCause().getMessage().contains("Fail to update
partitions to"));
+ }
+
+ @Test
+ void testPartitionIndexEnableAccessors_wrapGlueFailures() {
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ assertTrue(assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.getPartitionIndexEnable("tbl"))
+ .getMessage().contains("Fail to get parameter
partition_filtering.enabled"));
+ assertTrue(assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.updatePartitionIndexEnable("tbl", true))
+ .getMessage().contains("Fail to update parameter
partition_filtering.enabled"));
+ }
+
+ @Test
+ void testLastCommitTimeAccessors_wrapGlueFailures() {
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ assertTrue(assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.getLastCommitTimeSynced("tbl"))
+ .getMessage().contains("Fail to get last sync commit time"));
+ assertTrue(assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.getLastCommitCompletionTimeSynced("other"))
+ .getMessage().contains("Failed to get the last commit completion time
synced"));
+ }
+
+ @Test
+ void testUpdateTableProperties_propertiesAlreadyInTheCatalogSkipUpdate() {
+ String tableName = "tbl";
+ Map<String, String> existing = new HashMap<>();
+ existing.put("hudi.metadata-listing-enabled", "FALSE");
+ existing.put(HOODIE_LAST_COMMIT_TIME_SYNC, "100");
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
existing)).build()));
+
+ Map<String, String> update = new HashMap<>();
+ update.put(HOODIE_LAST_COMMIT_TIME_SYNC, "100");
+ assertFalse(awsGlueSyncClient.updateTableProperties(tableName, update));
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testStorageSchemaReads_failWhenTheTableHasNoCommits() throws
IOException {
+ AWSGlueCatalogSyncClient clientWithoutCommits =
clientForTableWithoutCommits();
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(getTableWithDefaultProps("tbl", Collections.emptyList(),
Collections.emptyList()));
+
+ assertTrue(assertThrows(HoodieGlueSyncException.class,
clientWithoutCommits::getStorageFieldSchemas)
+ .getMessage().contains("Failed to get field schemas from storage"));
+ assertTrue(assertThrows(HoodieGlueSyncException.class,
+ () -> clientWithoutCommits.updateTableComments("tbl",
Collections.emptyList(), Collections.emptyList()))
+ .getMessage().contains("Failed to get schema's doc from storage"));
+ }
+
+ @Test
+ void testUpdateLastCommitTimeSynced_withoutACommitNothingIsSynced() throws
IOException {
+ AWSGlueCatalogSyncClient clientWithoutCommits =
clientForTableWithoutCommits();
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters("tbl", new
HashMap<>())).build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder().build()));
+
+ clientWithoutCommits.updateLastCommitTimeSynced("tbl");
+
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateTableSchema_wrapsGlueFailure() {
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+ HoodieSchema schema = GlueTestUtil.getSimpleSchema();
+ SchemaDifference schemaDiff = SchemaDifference.newBuilder(schema, new
HashMap<>()).build();
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.updateTableSchema("tbl", schema, schemaDiff));
+ assertTrue(ex.getMessage().contains("Fail to update definition for
table"));
+ }
+
+ @Test
+ void testUpdateTableSchema_cascadeWithoutPartitionsIssuesNoPartitionUpdate()
{
+ String tableName = GlueTestUtil.TABLE_NAME;
+ Table table = tableWithColumns(tableName,
+
Collections.singletonList(Column.builder().name("name").type("string").build()),
+
Collections.singletonList(Column.builder().name("datestr").type("string").build()));
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(table).build()));
+ when(mockAwsGlue.updateTable(any(UpdateTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+ when(mockAwsGlue.getPartitions(any(GetPartitionsRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionsResponse.builder().nextToken(null).build()));
+
+ HoodieSchema schema = GlueTestUtil.getSimpleSchema();
+ awsGlueSyncClient.updateTableSchema(tableName, schema,
+ SchemaDifference.newBuilder(schema, new
HashMap<>()).updateTableColumn("name", "string").build());
+
+ verify(mockAwsGlue,
never()).batchUpdatePartition(any(BatchUpdatePartitionRequest.class));
+ }
+
+ @Test
+ void testCreateOrReplaceTable_wrapsFailureOfTheReplace() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(getTableWithDefaultProps(tableName,
Collections.emptyList(), Collections.emptyList()));
+ when(mockAwsGlue.deleteTable(any(DeleteTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.createOrReplaceTable(tableName,
GlueTestUtil.getSimpleSchema(), "inputFormat",
+ "outputFormat", "serde", new HashMap<>(), new HashMap<>()));
+ assertTrue(ex.getMessage().contains("Fail to recreate the table"));
+ }
+
+ @Test
+ void testCreateTable_wrapsGlueFailure() throws Exception {
+ String tableName = "tbl";
+ CompletableFuture<GetTableResponse> notFound =
mock(CompletableFuture.class);
+ when(notFound.get()).thenThrow(new
ExecutionException(EntityNotFoundException.builder().build()));
+
when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenReturn(notFound);
+ when(mockAwsGlue.createTable(any(CreateTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.createTable(tableName,
GlueTestUtil.getSimpleSchema(), "inputFormat",
+ "outputFormat", "serde", new HashMap<>(), new HashMap<>()));
+ assertTrue(ex.getMessage().contains("Fail to create"));
+ }
+
+ @Test
+ void testCreateDatabase_wrapsGlueFailure() throws Exception {
+ String dbName = "db";
+ CompletableFuture<GetDatabaseResponse> notFound =
mock(CompletableFuture.class);
+ when(notFound.get()).thenThrow(new
ExecutionException(EntityNotFoundException.builder().build()));
+
when(mockAwsGlue.getDatabase(any(GetDatabaseRequest.class))).thenReturn(notFound);
+ CompletableFuture<CreateDatabaseResponse> failed =
mock(CompletableFuture.class);
+ when(failed.get()).thenThrow(new ExecutionException(new
RuntimeException("boom")));
+
when(mockAwsGlue.createDatabase(any(CreateDatabaseRequest.class))).thenReturn(failed);
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.createDatabase(dbName));
+ assertTrue(ex.getMessage().contains("Fail to create database"));
+ }
+
+ /** Stubs {@code GlueAsyncClient.builder()} onto a mocked builder; callers
add whatever else they exercise. */
+ private static GlueAsyncClientBuilder
mockGlueClientBuilder(MockedStatic<GlueAsyncClient> glueStatic) {
+ GlueAsyncClientBuilder builder = mock(GlueAsyncClientBuilder.class);
+ glueStatic.when(GlueAsyncClient::builder).thenReturn(builder);
+ when(builder.credentialsProvider(any())).thenReturn(builder);
+ return builder;
+ }
+
+ private AWSGlueCatalogSyncClient clientForTableWithoutCommits() throws
IOException {
+ HoodieTableMetaClient withoutCommits =
GlueTestUtil.createTableWithoutCommits();
+ TypedProperties props =
TypedProperties.copy(GlueTestUtil.getHiveSyncConfig().getProps());
+ props.setProperty(META_SYNC_BASE_PATH.key(),
withoutCommits.getBasePath().toString());
+ return new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), withoutCommits);
+ }
+
+ private static Map<String, String> serdePropertiesOf(String...
keysAndValues) {
+ Map<String, String> properties = new HashMap<>();
+ for (int i = 0; i < keysAndValues.length; i += 2) {
+ properties.put(keysAndValues[i], keysAndValues[i + 1]);
+ }
+ return properties;
+ }
+
+ private static PartitionIndexDescriptor partitionIndexDescriptor(String
indexName, String... keys) {
+ return PartitionIndexDescriptor.builder()
+ .indexName(indexName)
+ .keys(Arrays.stream(keys).map(key ->
KeySchemaElement.builder().name(key).build()).collect(Collectors.toList()))
+ .build();
+ }
+
+ private static Table tableWithParameters(String tableName, Map<String,
String> parameters) {
+ return tableWithColumns(tableName,
Collections.singletonList(Column.builder().name("name").type("string").build()),
+
Collections.singletonList(Column.builder().name("datestr").type("string").build()))
+ .toBuilder()
+ .parameters(parameters)
+ .build();
+ }
+
+ private static Table tableWithColumns(String tableName, List<Column>
columns, List<Column> partitionKeys) {
+ return Table.builder()
+ .name(tableName)
+ .databaseName(GlueTestUtil.DB_NAME)
+ .tableType("COPY_ON_WRITE")
+ .parameters(new HashMap<>())
+
.storageDescriptor(StorageDescriptor.builder().location("s3://base").columns(columns).build())
+ .partitionKeys(partitionKeys)
+ .build();
+ }
+
+ private static Table tableWithSerdeProperties(String tableName, Map<String,
String> serdeProperties) {
+ Table table = tableWithColumns(tableName,
+
Collections.singletonList(Column.builder().name("name").type("string").build()),
+
Collections.singletonList(Column.builder().name("datestr").type("string").build()));
+ return table.toBuilder()
+ .storageDescriptor(table.storageDescriptor().toBuilder()
+
.serdeInfo(SerDeInfo.builder().serializationLibrary("serde").parameters(serdeProperties).build())
+ .build())
+ .build();
+ }
}
diff --git
a/hudi-aws/src/test/java/org/apache/hudi/aws/testutils/GlueTestUtil.java
b/hudi-aws/src/test/java/org/apache/hudi/aws/testutils/GlueTestUtil.java
index 0c352730494e..a84809d1a0aa 100644
--- a/hudi-aws/src/test/java/org/apache/hudi/aws/testutils/GlueTestUtil.java
+++ b/hudi-aws/src/test/java/org/apache/hudi/aws/testutils/GlueTestUtil.java
@@ -26,10 +26,13 @@ import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import
org.apache.hudi.common.table.timeline.versioning.DefaultCommitMetadataSerDe;
import
org.apache.hudi.common.table.timeline.versioning.DefaultInstantFileNameGenerator;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hive.HiveSyncConfig;
import org.apache.hudi.hive.SlashEncodedDayPartitionValueExtractor;
+import org.apache.hudi.storage.HoodieInstantWriter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -38,12 +41,12 @@ import org.apache.hadoop.fs.Path;
import software.amazon.awssdk.services.glue.model.Column;
import java.io.IOException;
-import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.time.Instant;
import java.util.Arrays;
import static
org.apache.hudi.common.table.HoodieTableMetaClient.METAFOLDER_NAME;
+import static
org.apache.hudi.common.table.HoodieTableMetaClient.TIMELINEFOLDER_NAME;
import static
org.apache.hudi.config.GlueCatalogSyncClientConfig.GLUE_SYNC_DATABASE_NAME;
import static
org.apache.hudi.config.GlueCatalogSyncClientConfig.GLUE_SYNC_TABLE_NAME;
import static
org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_BATCH_SYNC_PARTITION_NUM;
@@ -58,7 +61,13 @@ public class GlueTestUtil {
public static TypedProperties glueSyncProps;
public static final String DB_NAME = "testdb";
public static final String TABLE_NAME = "test1";
+ public static final String TABLE_DOC = "example table doc";
+ public static final String ID_FIELD_DOC = "the record id";
+ public static final String NAME_FIELD_DOC = "the record name";
+ public static final String INSTANT_TIME = "101";
+ public static final String COMPLETION_TIME = "102";
private static String basePath;
+ private static String basePathWithoutCommits;
public static FileSystem fileSystem;
private static HiveSyncConfig hiveSyncConfig;
private static Configuration hadoopConf;
@@ -85,6 +94,10 @@ public class GlueTestUtil {
public static void clear() throws IOException {
fileSystem.delete(new Path(basePath), true);
+ if (basePathWithoutCommits != null) {
+ fileSystem.delete(new Path(basePathWithoutCommits), true);
+ basePathWithoutCommits = null;
+ }
}
public static void teardown() throws IOException {
@@ -110,9 +123,25 @@ public class GlueTestUtil {
.setPayloadClass(HoodieAvroPayload.class)
.initTable(HadoopFSUtils.getStorageConf(new Configuration()),
basePath);
- String instantTime = "101";
HoodieCommitMetadata commitMetadata = new HoodieCommitMetadata(false);
- createMetaFile(basePath, new
DefaultInstantFileNameGenerator().makeCommitFileName(instantTime),
commitMetadata);
+ // the commit carries the table schema so the storage-schema read paths
have something to resolve
+ commitMetadata.addMetadata(HoodieCommitMetadata.SCHEMA_KEY,
getDocumentedSchema().toString());
+ // completed instants carry their completion time in the file name,
without it the timeline skips the file
+ createMetaFile(basePath,
+ new DefaultInstantFileNameGenerator().makeCommitFileName(INSTANT_TIME
+ "_" + COMPLETION_TIME), commitMetadata);
+ }
+
+ /**
+ * A table in its own base path with no commits at all, for the paths that
face an empty timeline.
+ * Its directory is removed by {@link #clear()} along with the main table.
+ */
+ public static HoodieTableMetaClient createTableWithoutCommits() throws
IOException {
+ basePathWithoutCommits =
Files.createTempDirectory("glueClientNoCommitTest" +
Instant.now().toEpochMilli()).toUri().toString();
+ return HoodieTableMetaClient.newTableBuilder()
+ .setTableType(HoodieTableType.COPY_ON_WRITE)
+ .setTableName(TABLE_NAME)
+ .setPayloadClass(HoodieAvroPayload.class)
+ .initTable(HadoopFSUtils.getStorageConf(new Configuration()),
basePathWithoutCommits);
}
public static HoodieSchema getSimpleSchema() {
@@ -123,13 +152,27 @@ public class GlueTestUtil {
));
}
+ /**
+ * Same shape as {@link #getSimpleSchema()} but with a record doc and
per-field docs, so the
+ * comment/doc read paths have something to return.
+ */
+ public static HoodieSchema getDocumentedSchema() {
+ return HoodieSchema.createRecord("example_schema", null, TABLE_DOC,
+ Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.INT), ID_FIELD_DOC, null),
+ HoodieSchemaField.of("name",
HoodieSchema.create(HoodieSchemaType.STRING), NAME_FIELD_DOC, null)
+ ));
+ }
+
private static void createMetaFile(String basePath, String fileName,
HoodieCommitMetadata metadata)
throws IOException {
- byte[] bytes = metadata.toJsonString().getBytes(StandardCharsets.UTF_8);
- Path fullPath = new Path(basePath + "/" + METAFOLDER_NAME + "/" +
fileName);
- FSDataOutputStream fsout = fileSystem.create(fullPath, true);
- fsout.write(bytes);
- fsout.close();
+ Path fullPath = new Path(basePath + "/" + METAFOLDER_NAME + "/" +
TIMELINEFOLDER_NAME + "/" + fileName);
+ Option<HoodieInstantWriter> writer = new
DefaultCommitMetadataSerDe().getInstantWriter(metadata);
+ try (FSDataOutputStream fsout = fileSystem.create(fullPath, true)) {
+ if (writer.isPresent()) {
+ writer.get().writeToStream(fsout);
+ }
+ }
}
public static Column getColumn(String name, String type, String comment) {
diff --git
a/hudi-aws/src/test/java/org/apache/hudi/aws/transaction/lock/TestS3StorageLockClient.java
b/hudi-aws/src/test/java/org/apache/hudi/aws/transaction/lock/TestS3StorageLockClient.java
index f8514ca116e9..ea97156bc125 100644
---
a/hudi-aws/src/test/java/org/apache/hudi/aws/transaction/lock/TestS3StorageLockClient.java
+++
b/hudi-aws/src/test/java/org/apache/hudi/aws/transaction/lock/TestS3StorageLockClient.java
@@ -243,6 +243,35 @@ class TestS3StorageLockClient {
verify(mockLogger).warn(contains("internal server error"), eq(OWNER_ID),
eq(LOCK_FILE_PATH), eq(ex503));
}
+ @Test
+ void testTryUpsertLockFile_unexpectedS3StatusIsReportedAsUnknownError() {
+ StorageLockData lockData = new StorageLockData(false, 6000L, "myTxOwner");
+ AwsServiceException ex403 = S3Exception.builder().statusCode(403).build();
+ when(mockS3Client.putObject(any(PutObjectRequest.class),
any(RequestBody.class))).thenThrow(ex403);
+
+ Pair<LockUpsertResult, Option<StorageLockFile>> result =
+ lockService.tryUpsertLockFile(lockData, Option.empty());
+
+ assertEquals(UNKNOWN_ERROR, result.getLeft());
+ assertTrue(result.getRight().isEmpty());
+ verify(mockLogger).warn(contains("Error writing lock file"), eq(OWNER_ID),
eq(LOCK_FILE_PATH), eq(ex403));
+ }
+
+ @Test
+ void testTryRenewLockFile_unexpectedErrorIsReportedRatherThanThrown() {
+ StorageLockData lockData = new StorageLockData(false, 7000L, "myTxOwner");
+ StorageLockFile prevLockFile = new StorageLockFile(lockData, "old-etag");
+ AwsServiceException ex400 =
AwsServiceException.builder().statusCode(400).build();
+ when(mockS3Client.putObject(any(PutObjectRequest.class),
any(RequestBody.class))).thenThrow(ex400);
+
+ Pair<LockUpsertResult, Option<StorageLockFile>> result =
+ lockService.tryUpsertLockFile(lockData, Option.of(prevLockFile));
+
+ assertEquals(UNKNOWN_ERROR, result.getLeft(), "a renewal keeps the failure
to itself, only creation throws early");
+ assertTrue(result.getRight().isEmpty());
+ verify(mockLogger).error(contains("Unexpected SDK error"), eq(OWNER_ID),
eq(LOCK_FILE_PATH), eq(ex400));
+ }
+
@Test
void testTryCreateLockFile_unexpectedError() {
StorageLockData lockData = new StorageLockData(false, 8000L, "myTxOwner");
diff --git
a/hudi-aws/src/test/java/org/apache/hudi/aws/utils/TestDynamoTableUtils.java
b/hudi-aws/src/test/java/org/apache/hudi/aws/utils/TestDynamoTableUtils.java
new file mode 100644
index 000000000000..3df884561664
--- /dev/null
+++ b/hudi-aws/src/test/java/org/apache/hudi/aws/utils/TestDynamoTableUtils.java
@@ -0,0 +1,184 @@
+/*
+ * 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.Timeout;
+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 java.util.concurrent.TimeUnit;
+
+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. Most tests
hand the polling helpers a
+ * short explicit timeout so they bound their own runtime; the one that
exercises the default-timeout
+ * overloads (20 minutes, polled every 10 seconds) is bounded by {@code
@Timeout} instead.
+ */
+@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
+ @Timeout(value = 10, unit = TimeUnit.SECONDS)
+ 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);
+ DynamoTableUtils.waitUntilActive(dynamoDb, TABLE_NAME);
+
+ verify(dynamoDb, times(2)).describeTable(any(DescribeTableRequest.class));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"-1, 10", "100, 0", "100, 100", "100, 200"})
+ void testWaitUntilExists_rejectsInvalidTimeoutAndInterval(int timeout, int
interval) {
+ assertThrows(IllegalArgumentException.class,
+ () -> DynamoTableUtils.waitUntilExists(dynamoDb, TABLE_NAME, timeout,
interval));
+ }
+
+ @Test
+ void testCreateTableIfNotExists_reportsWhetherTheTableWasCreated() {
+ CreateTableRequest request =
CreateTableRequest.builder().tableName(TABLE_NAME).build();
+ when(dynamoDb.createTable(request))
+ .thenReturn(CreateTableResponse.builder().build())
+ .thenThrow(ResourceInUseException.builder().message("already
there").build());
+
+ assertTrue(DynamoTableUtils.createTableIfNotExists(dynamoDb, request));
+ assertFalse(DynamoTableUtils.createTableIfNotExists(dynamoDb, request),
+ "an already existing table is not an error");
+ }
+
+ @Test
+ void testDeleteTableIfExists_reportsWhetherTheTableWasDeleted() {
+ DeleteTableRequest request =
DeleteTableRequest.builder().tableName(TABLE_NAME).build();
+ when(dynamoDb.deleteTable(request))
+ .thenReturn(DeleteTableResponse.builder().build())
+
.thenThrow(ResourceNotFoundException.builder().message("gone").build());
+
+ assertTrue(DynamoTableUtils.deleteTableIfExists(dynamoDb, request));
+ assertFalse(DynamoTableUtils.deleteTableIfExists(dynamoDb, request),
+ "a missing table is not an error");
+ }
+
+ private static DescribeTableResponse describeResponse(TableStatus status) {
+ return DescribeTableResponse.builder()
+
.table(TableDescription.builder().tableName(TABLE_NAME).tableStatus(status).build())
+ .build();
+ }
+}
diff --git
a/hudi-aws/src/test/java/org/apache/hudi/config/TestHoodieAWSConfig.java
b/hudi-aws/src/test/java/org/apache/hudi/config/TestHoodieAWSConfig.java
new file mode 100644
index 000000000000..8c7fb4a8c43a
--- /dev/null
+++ b/hudi-aws/src/test/java/org/apache/hudi/config/TestHoodieAWSConfig.java
@@ -0,0 +1,114 @@
+/*
+ * 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.config;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Properties;
+
+import static
org.apache.hudi.config.DynamoDbBasedLockConfig.DYNAMODB_LOCK_BILLING_MODE;
+import static
org.apache.hudi.config.DynamoDbBasedLockConfig.DYNAMODB_LOCK_PARTITION_KEY;
+import static
org.apache.hudi.config.DynamoDbBasedLockConfig.DYNAMODB_LOCK_READ_CAPACITY;
+import static
org.apache.hudi.config.DynamoDbBasedLockConfig.DYNAMODB_LOCK_REGION;
+import static
org.apache.hudi.config.DynamoDbBasedLockConfig.DYNAMODB_LOCK_TABLE_NAME;
+import static
org.apache.hudi.config.DynamoDbBasedLockConfig.DYNAMODB_LOCK_WRITE_CAPACITY;
+import static
org.apache.hudi.config.GlueCatalogSyncClientConfig.GLUE_SKIP_TABLE_ARCHIVE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+class TestHoodieAWSConfig {
+
+ @Test
+ void testBuilderWritesEveryPropertyItIsGiven() {
+ HoodieAWSConfig config = HoodieAWSConfig.newBuilder()
+ .withAccessKey("access-key")
+ .withSecretKey("secret-key")
+ .withSessionToken("session-token")
+ .withAssumeRoleARN("arn:aws:iam::123456789012:role/hudi")
+ .withAssumeRoleExternalID("external-id")
+ .withAssumeRoleSessionName("session-name")
+ .withDynamoDBTable("lock_table")
+ .withDynamoDBPartitionKey("partition-key")
+ .withDynamoDBRegion("eu-west-1")
+ .withDynamoDBBillingMode("PAY_PER_REQUEST")
+ .withDynamoDBReadCapacity("12")
+ .withDynamoDBWriteCapacity("13")
+ .withGlueSkipTableArchive("false")
+ .build();
+
+ assertEquals("access-key", config.getAWSAccessKey());
+ assertEquals("secret-key", config.getAWSSecretKey());
+ assertEquals("session-token", config.getAWSSessionToken());
+ assertEquals("arn:aws:iam::123456789012:role/hudi",
config.getAWSAssumeRoleARN());
+ assertEquals("external-id", config.getAWSAssumeRoleExternalID());
+ assertEquals("session-name", config.getAWSAssumeRoleSessionName());
+
+ Properties props = config.getProps();
+ assertEquals("access-key",
props.getProperty(HoodieAWSConfig.AWS_ACCESS_KEY.key()));
+ assertEquals("secret-key",
props.getProperty(HoodieAWSConfig.AWS_SECRET_KEY.key()));
+ assertEquals("session-token",
props.getProperty(HoodieAWSConfig.AWS_SESSION_TOKEN.key()));
+ assertEquals("arn:aws:iam::123456789012:role/hudi",
props.getProperty(HoodieAWSConfig.AWS_ASSUME_ROLE_ARN.key()));
+ assertEquals("external-id",
props.getProperty(HoodieAWSConfig.AWS_ASSUME_ROLE_EXTERNAL_ID.key()));
+ assertEquals("session-name",
props.getProperty(HoodieAWSConfig.AWS_ASSUME_ROLE_SESSION_NAME.key()));
+ assertEquals("lock_table",
props.getProperty(DYNAMODB_LOCK_TABLE_NAME.key()));
+ assertEquals("partition-key",
props.getProperty(DYNAMODB_LOCK_PARTITION_KEY.key()));
+ assertEquals("eu-west-1", props.getProperty(DYNAMODB_LOCK_REGION.key()));
+ assertEquals("PAY_PER_REQUEST",
props.getProperty(DYNAMODB_LOCK_BILLING_MODE.key()));
+ assertEquals("12", props.getProperty(DYNAMODB_LOCK_READ_CAPACITY.key()));
+ assertEquals("13", props.getProperty(DYNAMODB_LOCK_WRITE_CAPACITY.key()));
+ assertEquals("false", props.getProperty(GLUE_SKIP_TABLE_ARCHIVE.key()));
+ }
+
+ @Test
+ void testBuildFillsInTheSessionNameDefaultOnly() {
+ HoodieAWSConfig config = HoodieAWSConfig.newBuilder().build();
+
+ assertEquals(HoodieAWSConfig.AWS_ASSUME_ROLE_SESSION_NAME.defaultValue(),
config.getAWSAssumeRoleSessionName());
+
assertFalse(config.getProps().containsKey(HoodieAWSConfig.AWS_ACCESS_KEY.key()),
+ "properties without a default stay unset");
+ }
+
+ @Test
+ void testFromPropertiesCopiesTheGivenProperties() {
+ Properties given = new Properties();
+ given.setProperty(HoodieAWSConfig.AWS_ACCESS_KEY.key(), "from-properties");
+
+ HoodieAWSConfig config =
HoodieAWSConfig.newBuilder().fromProperties(given).build();
+
+ assertEquals("from-properties", config.getAWSAccessKey());
+ }
+
+ @Test
+ void testFromFileReadsTheGivenPropertiesFile(@TempDir Path tempDir) throws
IOException {
+ File propertiesFile = tempDir.resolve("aws.properties").toFile();
+ Files.write(propertiesFile.toPath(),
+ (HoodieAWSConfig.AWS_ACCESS_KEY.key() + "=from-file\n"
+ + HoodieAWSConfig.AWS_SECRET_KEY.key() +
"=secret-from-file\n").getBytes("UTF-8"));
+
+ HoodieAWSConfig config =
HoodieAWSConfig.newBuilder().fromFile(propertiesFile).build();
+
+ assertEquals("from-file", config.getAWSAccessKey());
+ assertEquals("secret-from-file", config.getAWSSecretKey());
+ }
+}
