hudi-agent commented on code in PR #19726:
URL: https://github.com/apache/hudi/pull/19726#discussion_r3865957943
##########
hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java:
##########
@@ -881,13 +885,84 @@ void testUpdateHoodieWriterVersionThrowsGlueException()
throws ExecutionExceptio
assertTrue(ex.getMessage().contains(HoodieVersion.get()), "exception
message should mention the writer version");
}
+ @Test
+ void testUpdateLastCommitTimeSyncedPersistsCompletionTimeWatermark() throws
ExecutionException, InterruptedException, IOException {
+ String tableName = "test";
+ List<Column> columns =
Collections.singletonList(GlueTestUtil.getColumn("name", "string", "person's
name"));
+ List<Column> partitionKeys =
Collections.singletonList(GlueTestUtil.getColumn("city", "string", "person's
city"));
+ CompletableFuture<UpdateTableResponse> mockUpdateTableResponse =
mock(CompletableFuture.class);
+
Mockito.when(mockUpdateTableResponse.get()).thenReturn(UpdateTableResponse.builder().build());
+ Mockito.when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(getTableWithProps(tableName, columns, partitionKeys,
Collections.singletonMap("EXTERNAL", "TRUE")));
+
Mockito.when(mockAwsGlue.updateTable(any(UpdateTableRequest.class))).thenReturn(mockUpdateTableResponse);
+
+ // A second commit whose instant time sorts below the fixture's but whose
completion time is later:
+ // the instant-time watermark must come from the fixture commit and the
completion-time watermark from this one.
+ String earlierInstant = "100";
+ String earlierInstantCompletionTime = "20250101000001000";
+ GlueTestUtil.createCommitFile(earlierInstant,
earlierInstantCompletionTime);
+ // fresh client so its lazily loaded timeline sees both commits
+ String basePath =
GlueTestUtil.getHiveSyncConfig().getString(META_SYNC_BASE_PATH);
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts,
GlueTestUtil.getHiveSyncConfig(),
+
HoodieTableMetaClient.builder().setConf(HadoopFSUtils.getStorageConf(GlueTestUtil.getHadoopConf())).setBasePath(basePath).build());
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ ArgumentCaptor<UpdateTableRequest> captor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ verify(mockAwsGlue, times(1)).updateTable(captor.capture());
+ Map<String, String> params = captor.getValue().tableInput().parameters();
+ assertEquals(GlueTestUtil.INSTANT_TIME,
params.get(HOODIE_LAST_COMMIT_TIME_SYNC));
+ assertEquals(earlierInstantCompletionTime,
params.get(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC),
+ "incremental sync needs the completion-time watermark to pick up
commits that complete out of instant-time order");
+ assertEquals("TRUE", params.get("EXTERNAL"),
+ "updating the watermarks must merge into the existing table
parameters, not replace them");
+ }
+
+ @Test
+ void testUpdateLastCommitTimeSyncedWithNoCompletedCommit() throws
IOException {
+ String tableName = "test";
+ GlueTestUtil.deleteCommitFile(GlueTestUtil.INSTANT_TIME,
GlueTestUtil.INSTANT_COMPLETION_TIME);
+ String basePath =
GlueTestUtil.getHiveSyncConfig().getString(META_SYNC_BASE_PATH);
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts,
GlueTestUtil.getHiveSyncConfig(),
+
HoodieTableMetaClient.builder().setConf(HadoopFSUtils.getStorageConf(GlueTestUtil.getHadoopConf())).setBasePath(basePath).build());
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testGetLastCommitCompletionTimeSyncedWhenAbsent() {
+ String tableName = "test";
+ List<Column> columns =
Collections.singletonList(GlueTestUtil.getColumn("name", "string", "person's
name"));
+ List<Column> partitionKeys =
Collections.singletonList(GlueTestUtil.getColumn("city", "string", "person's
city"));
+
Mockito.when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenReturn(getTableWithDefaultProps(tableName,
columns, partitionKeys));
+
assertFalse(awsGlueSyncClient.getLastCommitCompletionTimeSynced(tableName).isPresent());
+ }
+
+ @Test
+ void testGetLastCommitCompletionTimeSyncedWhenPresent() {
+ String tableName = "test";
+ List<Column> columns =
Collections.singletonList(GlueTestUtil.getColumn("name", "string", "person's
name"));
+ List<Column> partitionKeys =
Collections.singletonList(GlueTestUtil.getColumn("city", "string", "person's
city"));
+ HashMap<String, String> props = new HashMap<>();
+ props.put(HOODIE_LAST_COMMIT_TIME_SYNC, "101");
Review Comment:
🤖 nit: this PR already switches `HashMap` to `Map` in the production code —
could you do the same here (`Map<String, String> props = new HashMap<>()`) for
consistency?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
