voonhous commented on code in PR #19876:
URL: https://github.com/apache/hudi/pull/19876#discussion_r3975336237
##########
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:
Added
`testManagePartitionIndexes_enabledWithTheConfiguredIndexesInPlaceChangesNothing`:
flag already `true`, one existing index equal to the configured fields. It
asserts a single `getPartitionIndexes` and no `updateTable`,
`deletePartitionIndex` or `createPartitionIndex`, so the `indexesChanges`
re-read guard staying false is pinned. Done in bb7fccea3715.
##########
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:
The two arms differ only in the log line (`An indexation process is
currently running.` vs `Something went wrong with partition index`); neither
rethrows, so there is nothing observable to tell them apart. Kept both values
so each catch line runs, renamed the parameter to `throughExecutionException`,
and the javadoc now says the effect is the same by design. Done in bb7fccea3715.
--
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]
