voonhous commented on code in PR #19726: URL: https://github.com/apache/hudi/pull/19726#discussion_r3888879114
########## 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: **nit:** 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); Review Comment: **nit:** #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. Feel free to ignore. Could we add a `@TempDir java.nio.file.Path tempDir;` field and pass it through? ```suggestion HiveTestUtil.setUp(Option.empty(), true, tempDir); ``` ########## 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: **minor:** `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. Not blocking. Would it be worth taking the flag as a parameter so the helper is safe to reuse? -- 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]
