voonhous commented on code in PR #19726:
URL: https://github.com/apache/hudi/pull/19726#discussion_r3888879103
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java:
##########
@@ -367,6 +367,30 @@ public static void createReplaceCommit(String instantTime,
String partitions, Wr
createReplaceCommitFile(replaceCommitMetadata, instantTime);
}
+ /**
+ * Writes the data files and the requested + inflight markers of an
INSERT_OVERWRITE replacecommit
+ * but not the completed file, i.e. a write that is still running. Complete
it with
+ * {@link #createReplaceCommitFile}.
+ */
+ public static HoodieReplaceCommitMetadata
startInsertOverwritePartition(String partitionPath, String instantTime)
+ throws IOException, URISyntaxException {
+ HoodieCommitMetadata commitMetadata = createPartition(partitionPath, true,
true, instantTime);
+ HoodieReplaceCommitMetadata replaceCommitMetadata = new
HoodieReplaceCommitMetadata();
+ commitMetadata.getPartitionToWriteStats().forEach((p, stats) ->
stats.forEach(s -> replaceCommitMetadata.addWriteStat(p, s)));
Review Comment:
This write-stat copy is load-bearing and nothing says so.
`TimelineUtils.isDeletePartition` counts `INSERT_OVERWRITE`
(`TimelineUtils.java:617-621`), so without it the new partition is resolved as
dropped rather than added, and it survives only via the delete-ts vs write-ts
filter at `TimelineUtils.java:167-170`.
This is also the one branch this suite protects that
`testHiveSyncWithMultiWriter` does not: flipping `GREATER_THAN` to
`GREATER_THAN_OR_EQUALS` on line 170 fails this test and passes all six legs of
that one.
Could a one-line comment record why the copy is required?
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncToolLongRunningWriteWatermark.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.hive;
+
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.hive.testutils.HiveTestUtil;
+import org.apache.hudi.sync.common.model.Partition;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
+import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_MODE;
+import static org.apache.hudi.hive.testutils.HiveTestUtil.TABLE_NAME;
+import static org.apache.hudi.hive.testutils.HiveTestUtil.hiveSyncProps;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A partition written by a long-running INSERT_OVERWRITE must still be
registered when the incremental
+ * sync's watermark moved past the write's instant time while it was in
flight. That only works when the
+ * sync client persists and reads the commit completion-time watermark ({@code
last_commit_completion_time_sync}),
+ * which drives the "hollow instant" lookup in {@code
TimelineUtils.getCommitsTimelineAfter}; a client that
+ * tracks {@code last_commit_time_sync} alone silently and permanently drops
the partition.
+ */
+public class TestHiveSyncToolLongRunningWriteWatermark {
Review Comment:
Optional, feel free to ignore. This is the module's second class to own
`HiveTestUtil`, and both carry `@AfterAll HiveTestUtil.shutdown()`, so a
surefire fork (`forkCount=1`, `reuseForks=true`) now pays a full HiveServer2
plus ZooKeeper stop and restart between the two.
Would folding this method and the two nested clients into `TestHiveSyncTool`
be worth it? It already has the `hiveSyncTool`/`hiveClient` fields and the same
lifecycle, and the nested classes work as `private static` there.
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncToolLongRunningWriteWatermark.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.hive;
+
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.hive.testutils.HiveTestUtil;
+import org.apache.hudi.sync.common.model.Partition;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
+import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_MODE;
+import static org.apache.hudi.hive.testutils.HiveTestUtil.TABLE_NAME;
+import static org.apache.hudi.hive.testutils.HiveTestUtil.hiveSyncProps;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A partition written by a long-running INSERT_OVERWRITE must still be
registered when the incremental
+ * sync's watermark moved past the write's instant time while it was in
flight. That only works when the
+ * sync client persists and reads the commit completion-time watermark ({@code
last_commit_completion_time_sync}),
+ * which drives the "hollow instant" lookup in {@code
TimelineUtils.getCommitsTimelineAfter}; a client that
+ * tracks {@code last_commit_time_sync} alone silently and permanently drops
the partition.
+ */
+public class TestHiveSyncToolLongRunningWriteWatermark {
+
+ private static final String EXISTING_PARTITION_INSTANT = "100";
+ private static final String LONG_RUNNING_INSERT_OVERWRITE_INSTANT = "101";
+ private static final String EMPTY_COMMIT_DURING_WRITE = "102";
+ private static final String EMPTY_COMMIT_AFTER_WRITE = "103";
+ private static final String NEW_PARTITION = "2026/08/04";
+ /** {@link org.apache.hudi.hive.SlashEncodedDayPartitionValueExtractor} maps
{@link #NEW_PARTITION} to this datestr value. */
+ private static final List<String> NEW_PARTITION_VALUES =
Collections.singletonList("2026-08-04");
+
+ private HiveSyncTool hiveSyncTool;
+ private HoodieHiveSyncClient hiveClient;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ HiveTestUtil.setUp(Option.empty(), true);
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(), "hms");
+ }
+
+ @AfterEach
+ void teardown() throws Exception {
+ closeHiveSyncTool();
+ HiveTestUtil.clear();
+ }
+
+ @AfterAll
+ static void cleanUpClass() throws IOException {
+ HiveTestUtil.shutdown();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testPartitionOfWriteCompletingAfterWatermarkAdvanced(boolean
clientPersistsCompletionTime) throws Exception {
+ // t=100: table with one partition, synced. Watermark = 100.
+ HiveTestUtil.createCOWTable(EXISTING_PARTITION_INSTANT, 1, true);
+ reSyncHiveTable(clientPersistsCompletionTime);
+ assertEquals(1, hiveClient.getAllPartitions(TABLE_NAME).size());
+ assertEquals(EXISTING_PARTITION_INSTANT,
hiveClient.getLastCommitTimeSynced(TABLE_NAME).get());
+
+ // t=101: INSERT_OVERWRITE into a brand-new partition starts (requested +
inflight, files on storage)
+ // and keeps running for a long time.
+ HoodieReplaceCommitMetadata longRunningWrite =
+ HiveTestUtil.startInsertOverwritePartition(NEW_PARTITION,
LONG_RUNNING_INSERT_OVERWRITE_INSTANT);
+
+ // t=102: a concurrent writer lands an empty commit (no partitions) while
101 is still in flight.
+ HiveTestUtil.addEmptyCommit(EMPTY_COMMIT_DURING_WRITE);
+
+ // The periodic sync runs now: 101 is pending so it is invisible, and the
watermark jumps to 102.
+ reSyncHiveTable(clientPersistsCompletionTime);
+ assertEquals(1, hiveClient.getAllPartitions(TABLE_NAME).size(), "pending
write must not be synced yet");
+ assertEquals(EMPTY_COMMIT_DURING_WRITE,
hiveClient.getLastCommitTimeSynced(TABLE_NAME).get());
+
+ // 101 completes: its instant time is now BELOW the watermark, while its
completion time (stamped into
+ // the completed file's name at creation) is later than 102's.
+ HiveTestUtil.createReplaceCommitFile(longRunningWrite,
LONG_RUNNING_INSERT_OVERWRITE_INSTANT);
+ assertTrue(HiveTestUtil.fileSystem.exists(new Path(HiveTestUtil.basePath,
NEW_PARTITION)),
+ "partition data is on storage");
+
+ // Next sync cycle.
+ reSyncHiveTable(clientPersistsCompletionTime);
+ List<Partition> partitionsAfterCompletion =
hiveClient.getAllPartitions(TABLE_NAME);
+
+ // t=103: more unrelated commits keep arriving; every later sync advances
the watermark further.
+ HiveTestUtil.addEmptyCommit(EMPTY_COMMIT_AFTER_WRITE);
+ reSyncHiveTable(clientPersistsCompletionTime);
+ List<Partition> partitionsAfterLaterSyncs =
hiveClient.getAllPartitions(TABLE_NAME);
+ assertEquals(EMPTY_COMMIT_AFTER_WRITE,
hiveClient.getLastCommitTimeSynced(TABLE_NAME).get());
+
+ if (clientPersistsCompletionTime) {
+ // Stock Hive client: the completion-time watermark rescues the hollow
instant.
+ assertEquals(2, partitionsAfterCompletion.size(), "hollow instant
rescued by completion-time watermark");
+ assertTrue(containsNewPartition(partitionsAfterCompletion), "the
registered partition is the INSERT_OVERWRITE's");
+ assertEquals(2, partitionsAfterLaterSyncs.size());
+ assertTrue(containsNewPartition(partitionsAfterLaterSyncs));
+ } else {
+ // Instant-time-only client: the partition is on storage with a
completed replacecommit, yet it is
+ // never added - and the watermark keeps moving, so no future
incremental sync will add it.
+ assertEquals(1, partitionsAfterCompletion.size(), "partition of the
late-completing write was skipped");
+ assertFalse(containsNewPartition(partitionsAfterCompletion));
+ assertEquals(1, partitionsAfterLaterSyncs.size(), "partition is
permanently lost to incremental sync");
+ assertFalse(containsNewPartition(partitionsAfterLaterSyncs));
+ // read through a stock client to show the completion key was never
written to the metastore
+ try (HiveSyncTool stockTool = new HiveSyncTool(hiveSyncProps,
HiveTestUtil.getHiveConf())) {
+ assertFalse(((HoodieHiveSyncClient)
stockTool.syncClient).getLastCommitCompletionTimeSynced(TABLE_NAME).isPresent());
Review Comment:
The negative leg pins the behaviour of a test-local subclass, but the same
loss happens with the **stock** client on any table last synced before
HUDI-6182: the metastore then has `last_commit_time_sync` and no completion
key, so `isAlreadySynced` takes `.orElse(true)` (`HiveSyncTool.java:327`) and
short-circuits every later cycle. `AdbSyncTool.java:190` still hard-codes
`Option.empty()`, so the shape is live in-repo.
Could the last cycle re-run with a stock `HiveSyncTool` and assert the
partition is still absent, so the arm pins the upgrade path rather than the
fake?
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java:
##########
@@ -229,6 +229,15 @@ public void teardown() throws Exception {
HiveTestUtil.clear();
}
+ @Test
+ public void testUpdateLastCommitTimeSyncedWithNoCompletedCommit() {
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(), "hms");
+ // created-but-never-written table: the active timeline has no completed
instant, the update must no-op
+ reInitHiveSyncClient();
+ assertDoesNotThrow(() ->
hiveClient.updateLastCommitTimeSynced(HiveTestUtil.TABLE_NAME));
Review Comment:
This pins the `!lastCommitSynced.isPresent()` early return, but its
JDBC-fallback sibling `updateLastCommitTimeSyncedViaJdbc`
(`HoodieHiveSyncClient.java:755-767`) writes both watermark keys and has no
coverage at all. The read half is already pinned at
`TestHoodieHiveSyncClientOperations.java:95`, and that file never calls
`updateLastCommitTimeSynced`.
Would it be worth one
`verify(fixture.jdbcMetadataOperator).setTableProperties(...)` on that existing
fallback fixture, so both halves are covered?
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java:
##########
@@ -229,6 +229,15 @@ public void teardown() throws Exception {
HiveTestUtil.clear();
}
+ @Test
+ public void testUpdateLastCommitTimeSyncedWithNoCompletedCommit() {
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(), "hms");
+ // created-but-never-written table: the active timeline has no completed
instant, the update must no-op
+ reInitHiveSyncClient();
+ assertDoesNotThrow(() ->
hiveClient.updateLastCommitTimeSynced(HiveTestUtil.TABLE_NAME));
+ assertFalse(hiveClient.tableExists(HiveTestUtil.TABLE_NAME), "the no-op
must not create or alter the metastore table");
Review Comment:
Optional nit. No branch of `updateLastCommitTimeSynced` can create a table,
so this holds before the call too. `assertDoesNotThrow` on the line above is
what discriminates: delete the early return at `HoodieHiveSyncClient.java:725`
and `client.getTable` throws.
Could this line go, or the comment note that line 237 is the real guard so a
later reader does not trim the wrong one?
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncToolLongRunningWriteWatermark.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.hive;
+
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.hive.testutils.HiveTestUtil;
+import org.apache.hudi.sync.common.model.Partition;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
+import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_MODE;
+import static org.apache.hudi.hive.testutils.HiveTestUtil.TABLE_NAME;
+import static org.apache.hudi.hive.testutils.HiveTestUtil.hiveSyncProps;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A partition written by a long-running INSERT_OVERWRITE must still be
registered when the incremental
+ * sync's watermark moved past the write's instant time while it was in
flight. That only works when the
+ * sync client persists and reads the commit completion-time watermark ({@code
last_commit_completion_time_sync}),
+ * which drives the "hollow instant" lookup in {@code
TimelineUtils.getCommitsTimelineAfter}; a client that
+ * tracks {@code last_commit_time_sync} alone silently and permanently drops
the partition.
+ */
+public class TestHiveSyncToolLongRunningWriteWatermark {
+
+ private static final String EXISTING_PARTITION_INSTANT = "100";
+ private static final String LONG_RUNNING_INSERT_OVERWRITE_INSTANT = "101";
+ private static final String EMPTY_COMMIT_DURING_WRITE = "102";
+ private static final String EMPTY_COMMIT_AFTER_WRITE = "103";
+ private static final String NEW_PARTITION = "2026/08/04";
+ /** {@link org.apache.hudi.hive.SlashEncodedDayPartitionValueExtractor} maps
{@link #NEW_PARTITION} to this datestr value. */
+ private static final List<String> NEW_PARTITION_VALUES =
Collections.singletonList("2026-08-04");
+
+ private HiveSyncTool hiveSyncTool;
+ private HoodieHiveSyncClient hiveClient;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ HiveTestUtil.setUp(Option.empty(), true);
Review Comment:
#18883 (`6e40cffcc92b`) added the 3-arg `setUp` so hive-sync tests root
their base path in a JUnit `@TempDir`, and moved `TestHiveSyncTool`,
`TestHiveIncrementalPuller` and `TestHudiHiveSyncJob` onto it. This is the only
2-arg caller left, so cleanup here depends on `clear()` running rather than on
JUnit.
Could we add a `@TempDir java.nio.file.Path tempDir;` field and pass it
through?
```suggestion
HiveTestUtil.setUp(Option.empty(), true, tempDir);
```
##########
hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java:
##########
@@ -901,7 +976,7 @@ private CompletableFuture<GetTableResponse>
getTableWithDefaultProps(String tabl
.parameters(new HashMap<>())
.storageDescriptor(storageDescriptor)
.partitionKeys(partitionColumns)
- .parameters(tableProperties)
+ .parameters(extraTableProperties)
Review Comment:
Optional nit. The builder already calls `.parameters(new HashMap<>())` a few
lines above and this overwrites it (SDK v2 builder setters are
last-write-wins), so the earlier call is dead. Pre-existing, but the new
parameter name makes it read as though the two merge.
Could the earlier `.parameters(new HashMap<>())` line be dropped, and this
parameter renamed `tableProperties`?
##########
hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java:
##########
@@ -941,11 +941,10 @@ public Option<String> getLastCommitTimeSynced(String
tableName) {
@Override
public Option<String> getLastCommitCompletionTimeSynced(String tableName) {
- // Get the last commit completion time from the TBLproperties
try {
return
Option.ofNullable(getInitialTable(tableName).parameters().getOrDefault(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC,
null));
} catch (Exception e) {
- throw new HoodieGlueSyncException("Failed to get the last commit
completion time synced from the table " + tableName, e);
+ throw new HoodieGlueSyncException("Fail to get last commit completion
time synced for " + tableId(databaseName, tableName), e);
Review Comment:
Optional nit. In a `test(sync)` PR, would it be worth leaving the message
rewording out? The new text diverges from the identical message still in
`HoodieHiveSyncClient.java:606`, so the two clients no longer read the same way
for the same failure.
The `HashMap` to `Map` widening below is worth keeping either way.
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java:
##########
@@ -367,6 +367,30 @@ public static void createReplaceCommit(String instantTime,
String partitions, Wr
createReplaceCommitFile(replaceCommitMetadata, instantTime);
}
+ /**
+ * Writes the data files and the requested + inflight markers of an
INSERT_OVERWRITE replacecommit
+ * but not the completed file, i.e. a write that is still running. Complete
it with
+ * {@link #createReplaceCommitFile}.
+ */
+ public static HoodieReplaceCommitMetadata
startInsertOverwritePartition(String partitionPath, String instantTime)
+ throws IOException, URISyntaxException {
+ HoodieCommitMetadata commitMetadata = createPartition(partitionPath, true,
true, instantTime);
+ HoodieReplaceCommitMetadata replaceCommitMetadata = new
HoodieReplaceCommitMetadata();
+ commitMetadata.getPartitionToWriteStats().forEach((p, stats) ->
stats.forEach(s -> replaceCommitMetadata.addWriteStat(p, s)));
+
replaceCommitMetadata.setOperationType(WriteOperationType.INSERT_OVERWRITE);
+
replaceCommitMetadata.setPartitionToReplaceFileIds(Collections.singletonMap(partitionPath,
new ArrayList<>()));
+
commitMetadata.getExtraMetadata().forEach(replaceCommitMetadata::addMetadata);
+ createMetaFile(basePath,
INSTANT_FILE_NAME_GENERATOR.makeRequestedReplaceFileName(instantTime),
Option.empty());
+ createMetaFile(basePath,
INSTANT_FILE_NAME_GENERATOR.makeInflightReplaceFileName(instantTime),
Option.empty());
+ createdTablesSet.add(DB_NAME + "." + TABLE_NAME);
+ return replaceCommitMetadata;
+ }
+
+ /** A completed commit that touched no partition, like a streaming writer's
empty micro-batch. */
+ public static void addEmptyCommit(String instantTime) throws IOException {
+ createCommitFileWithSchema(new HoodieCommitMetadata(), instantTime, true);
Review Comment:
`createCommitFileWithSchema(..., true)` stamps the simple schema into the
"empty" commit, and
`TableSchemaResolver.getTableSchemaFromLatestCommitMetadata`
(`TableSchemaResolver.java:202`) resolves from the latest commit carrying a
schema. Harmless for this simple-schema fixture, but a future caller on an
evolved-schema table would get a silent downgrade.
Would it be worth taking the flag as a parameter so the helper is safe to
reuse?
##########
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() {
Review Comment:
Optional nit. These two 8-line tests cover a single `getOrDefault`, and
`WhenAbsent`/`WhenPresent` differs from the `testX_Scenario` convention the
neighbours use (`testCreateOrReplaceTable_TableExists`,
`testGetAllPartitions_SinglePage`).
Could they fold into one
`testGetLastCommitCompletionTimeSynced_AbsentAndPresent`?
--
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]
