cshuo commented on code in PR #19537:
URL: https://github.com/apache/hudi/pull/19537#discussion_r3732973931
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexingCatchupTask.java:
##########
@@ -240,6 +249,121 @@ public void testNoHeartbeat() throws IOException {
assertTrue(task.awaitInstantCaughtUp(pendingInstantWithNoHeartbeat),
"Expected null as the instant's heartbeat has expired.");
}
+ @Test
+ public void testRunSkipsInstantAlreadyCommittedToMetadata() {
+ HoodieInstant instant =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, "002");
+ HoodieActiveTimeline metadataTimeline = mock(HoodieActiveTimeline.class);
+
when(metadataMetaClient.reloadActiveTimeline()).thenReturn(metadataTimeline);
+
when(metadataTimeline.filterCompletedInstants()).thenReturn(metadataTimeline);
+ when(metadataTimeline.filter(any())).thenReturn(metadataTimeline);
+ when(metadataTimeline.firstInstant()).thenReturn(Option.of(instant));
+
+ RunningIndexingCatchupTask task = runningTask(instant);
+ task.run();
+
+ assertEquals("002", task.currentCaughtupInstant);
+ assertEquals(0, task.writeActionsUpdated.get());
+ }
+
+ @Test
+ public void testRunUpdatesCompletedWriteActionInsideTransaction() {
+ HoodieInstant instant =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "002");
+ stubNoCompletedMetadataInstant();
+ RunningIndexingCatchupTask task = runningTask(instant);
+
+ task.run();
+
+ assertEquals(1, task.writeActionsUpdated.get());
+ assertEquals("002", task.currentCaughtupInstant);
+ verify(transactionManager).beginStateChange(Option.of(instant),
Option.empty());
+ verify(transactionManager).endStateChange(Option.of(instant));
+ }
+
+ @Test
+ public void testRunUpdatesCleanRestoreAndRollbackActions() throws
IOException {
+ HoodieInstant clean =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.CLEAN_ACTION, "002");
+ HoodieInstant restore =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.RESTORE_ACTION, "003");
+ HoodieInstant rollback =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.ROLLBACK_ACTION, "004");
+ stubNoCompletedMetadataInstant();
+ HoodieCleanMetadata cleanMetadata = mock(HoodieCleanMetadata.class);
+ HoodieRestoreMetadata restoreMetadata = mock(HoodieRestoreMetadata.class);
+ HoodieRollbackMetadata rollbackMetadata =
mock(HoodieRollbackMetadata.class);
+ HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class);
+ when(metaClient.getActiveTimeline()).thenReturn(activeTimeline);
+
when(activeTimeline.readRestoreMetadata(restore)).thenReturn(restoreMetadata);
+
when(activeTimeline.readRollbackMetadata(rollback)).thenReturn(rollbackMetadata);
+
+ try (MockedStatic<CleanerUtils> cleanerUtils =
mockStatic(CleanerUtils.class)) {
+ cleanerUtils.when(() -> CleanerUtils.getCleanerMetadata(metaClient,
clean)).thenReturn(cleanMetadata);
+ runningTask(clean, restore, rollback).run();
+ }
+
+ verify(metadataWriter).update(cleanMetadata, "002");
+ verify(metadataWriter).update(restoreMetadata, "003");
+ verify(metadataWriter).update(rollbackMetadata, "004");
+ verify(transactionManager).endStateChange(Option.of(clean));
+ verify(transactionManager).endStateChange(Option.of(restore));
+ verify(transactionManager).endStateChange(Option.of(rollback));
+ }
+
+ @Test
+ public void testRunRejectsUnexpectedCompletedActionAndReleasesTransaction() {
+ HoodieInstant instant =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.SAVEPOINT_ACTION, "002");
+ stubNoCompletedMetadataInstant();
+
+ assertThrows(IllegalStateException.class, () ->
runningTask(instant).run());
+ verify(transactionManager).endStateChange(Option.of(instant));
+ }
+
+ @Test
+ public void testAwaitInstantCaughtUpUsesKnownMetadataInstant() {
+ HoodieInstant instant =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.INFLIGHT,
HoodieTimeline.COMMIT_ACTION, "002");
+ RunningIndexingCatchupTask task = new RunningIndexingCatchupTask(
+ metadataWriter, Collections.singletonList(instant),
Collections.singleton("002"), metaClient, metadataMetaClient,
+ transactionManager, "001", engineContext, table, heartbeatClient);
+
+ assertTrue(task.awaitInstantCaughtUp(instant));
+ assertEquals("002", task.currentCaughtupInstant);
+ }
+
+ private RunningIndexingCatchupTask runningTask(HoodieInstant... instants) {
+ return new RunningIndexingCatchupTask(
+ metadataWriter, java.util.Arrays.asList(instants), new HashSet<>(),
metaClient, metadataMetaClient,
+ transactionManager, "001", engineContext, table, heartbeatClient);
+ }
+
+ private void stubNoCompletedMetadataInstant() {
+ HoodieActiveTimeline metadataTimeline = mock(HoodieActiveTimeline.class);
+
when(metadataMetaClient.reloadActiveTimeline()).thenReturn(metadataTimeline);
+
when(metadataTimeline.filterCompletedInstants()).thenReturn(metadataTimeline);
+ when(metadataTimeline.filter(any())).thenReturn(metadataTimeline);
+ when(metadataTimeline.firstInstant()).thenReturn(Option.empty());
+ }
+
+ static class RunningIndexingCatchupTask extends AbstractIndexingCatchupTask {
+ private final AtomicInteger writeActionsUpdated = new AtomicInteger();
+
+ RunningIndexingCatchupTask(HoodieTableMetadataWriter metadataWriter,
+ List<HoodieInstant> instantsToIndex,
+ Set<String> metadataCompletedInstants,
+ HoodieTableMetaClient metaClient,
+ HoodieTableMetaClient metadataMetaClient,
+ TransactionManager transactionManager,
+ String currentCaughtupInstant,
+ HoodieEngineContext engineContext,
+ HoodieTable table,
+ HoodieHeartbeatClient heartbeatClient) {
+ super(metadataWriter, instantsToIndex, metadataCompletedInstants,
metaClient, metadataMetaClient,
+ transactionManager, currentCaughtupInstant, engineContext, table,
heartbeatClient);
+ }
+
+ @Override
+ public void updateIndexForWriteAction(HoodieInstant instant) {
+ writeActionsUpdated.incrementAndGet();
+ currentCaughtupInstant = instant.requestedTime();
Review Comment:
**P1:** This test double advances `currentCaughtupInstant` itself, but
neither production catch-up implementation does so. The executor test also
supplies empty timelines and mocks `IndexingCatchupTask`, so it proves only
that `run()` was called. In production, `RunIndexActionExecutor` passes the
initial instant as a `String` and later builds `HoodieIndexCommitMetadata` from
its unchanged field; after processing completed instant `002`, it still reports
`001`. Please exercise a production catch-up task with a real completed
instant, assert the returned `indexUptoInstant`, and propagate the last
successfully processed instant back to the executor.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexActionExecutors.java:
##########
@@ -0,0 +1,341 @@
+/*
+ * 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.table.action.index;
+
+import org.apache.hudi.avro.model.HoodieIndexCommitMetadata;
+import org.apache.hudi.avro.model.HoodieIndexPartitionInfo;
+import org.apache.hudi.avro.model.HoodieIndexPlan;
+import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient;
+import org.apache.hudi.client.transaction.TransactionManager;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.WriteConcurrencyMode;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieArchivedTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieLockConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.core.transaction.lock.InProcessLockProvider;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieIndexException;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.metadata.HoodieTableMetadataWriter;
+import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.HoodieTable;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
+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.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestIndexActionExecutors {
+
+ private HoodieEngineContext context;
+ private HoodieTable table;
+ private HoodieTableMetaClient metaClient;
+ private HoodieTableConfig tableConfig;
+ private HoodieActiveTimeline activeTimeline;
+ private HoodieStorage storage;
+
+ @BeforeEach
+ void setUp() {
+ context = mock(HoodieEngineContext.class);
+ table = mock(HoodieTable.class);
+ metaClient = mock(HoodieTableMetaClient.class);
+ tableConfig = mock(HoodieTableConfig.class);
+ activeTimeline = mock(HoodieActiveTimeline.class);
+ storage = mock(HoodieStorage.class);
+ when(table.getMetaClient()).thenReturn(metaClient);
+ when(table.getActiveTimeline()).thenReturn(activeTimeline);
+ when(table.getInstantGenerator()).thenReturn(INSTANT_GENERATOR);
+ when(table.getStorage()).thenReturn(storage);
+ doReturn(getDefaultStorageConf()).when(storage).getConf();
+ doReturn(getDefaultStorageConf()).when(context).getStorageConf();
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+ when(metaClient.getInstantGenerator()).thenReturn(INSTANT_GENERATOR);
+ when(metaClient.getActiveTimeline()).thenReturn(activeTimeline);
+ when(metaClient.reloadActiveTimeline()).thenReturn(activeTimeline);
+
when(tableConfig.getMetadataPartitions()).thenReturn(Collections.emptySet());
+
when(tableConfig.getMetadataPartitionsInflight()).thenReturn(Collections.emptySet());
+ }
+
+ @Test
+ void testScheduleRejectsSingleWriterConfiguration() {
+ HoodieWriteConfig config =
HoodieWriteConfig.newBuilder().withPath("/table").build();
+ ScheduleIndexActionExecutor executor = new ScheduleIndexActionExecutor(
+ context, config, table, "002",
Collections.singletonList(MetadataPartitionType.COLUMN_STATS),
Collections.emptyList());
+
+ HoodieIndexException exception = assertThrows(HoodieIndexException.class,
executor::execute);
+
assertTrue(exception.getMessage().contains(HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key()));
+ }
+
+ @Test
+ void testScheduleCreatesPlanAndIsIdempotent() {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant completed = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "001");
+ HoodieTimeline completedTimeline = mock(HoodieTimeline.class);
+
when(activeTimeline.getContiguousCompletedWriteTimeline()).thenReturn(completedTimeline);
+ when(completedTimeline.lastInstant()).thenReturn(Option.of(completed));
+ ScheduleIndexActionExecutor executor = new ScheduleIndexActionExecutor(
+ context, config, table, "002",
Collections.singletonList(MetadataPartitionType.COLUMN_STATS),
Collections.emptyList());
+
+ Option<HoodieIndexPlan> plan = executor.execute();
+
+ assertTrue(plan.isPresent());
+ assertEquals(1, plan.get().getIndexPartitionInfos().size());
+ assertEquals(MetadataPartitionType.COLUMN_STATS.getPartitionPath(),
+ plan.get().getIndexPartitionInfos().get(0).getMetadataPartitionPath());
+ assertEquals("001",
plan.get().getIndexPartitionInfos().get(0).getIndexUptoInstant());
+ verify(activeTimeline).saveToPendingIndexAction(any(HoodieInstant.class),
any(HoodieIndexPlan.class));
+
+
when(tableConfig.getMetadataPartitions()).thenReturn(Set.of(MetadataPartitionType.COLUMN_STATS.getPartitionPath()));
+ assertFalse(executor.execute().isPresent());
+ }
+
+ @Test
+ void testScheduleAbortsWhenPendingPlanCannotBeSaved() {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant completed = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "001");
+ HoodieTimeline completedTimeline = mock(HoodieTimeline.class);
+
when(activeTimeline.getContiguousCompletedWriteTimeline()).thenReturn(completedTimeline);
+ when(completedTimeline.lastInstant()).thenReturn(Option.of(completed));
+ doThrow(new HoodieIOException("write failed"))
+
.when(activeTimeline).saveToPendingIndexAction(any(HoodieInstant.class),
any(HoodieIndexPlan.class));
+
+ try (MockedStatic<HoodieTableMetadataUtil> metadataUtil =
mockStatic(HoodieTableMetadataUtil.class)) {
+ metadataUtil.when(() ->
HoodieTableMetadataUtil.getInflightAndCompletedMetadataPartitions(tableConfig))
+ .thenReturn(Collections.emptySet());
+ metadataUtil.when(() -> HoodieTableMetadataUtil.metadataPartitionExists(
+ metaClient.getBasePath(), context,
MetadataPartitionType.COLUMN_STATS.getPartitionPath())).thenReturn(false);
+
+ Option<HoodieIndexPlan> result = new ScheduleIndexActionExecutor(
+ context, config, table, "002",
Collections.singletonList(MetadataPartitionType.COLUMN_STATS),
Collections.emptyList()).execute();
+
+ assertFalse(result.isPresent());
+
verify(activeTimeline).deleteInstantFileIfExists(INSTANT_GENERATOR.getIndexRequestedInstant("002"));
+ }
+ }
+
+ @Test
+ void testRunRejectsInvalidConfigurationAndMissingInstant() {
+ RunIndexActionExecutor invalidExecutor = new RunIndexActionExecutor(
+ context, HoodieWriteConfig.newBuilder().withPath("/table").build(),
table, "002");
+ assertThrows(HoodieIndexException.class, invalidExecutor::execute);
+
+ HoodieWriteConfig config = multiWriterConfig();
+
when(activeTimeline.filterPendingIndexTimeline()).thenReturn(activeTimeline);
+ when(activeTimeline.filter(any())).thenReturn(activeTimeline);
+ when(activeTimeline.lastInstant()).thenReturn(Option.empty());
+ assertThrows(HoodieIndexException.class,
+ () -> new RunIndexActionExecutor(context, config, table,
"002").execute());
+ }
+
+ @Test
+ void testRunRejectsUnreadableAndEmptyPlans() throws IOException {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant requested = requestedIndexInstant();
+ stubRequestedIndexInstant(requested);
+ doThrow(new IOException("read
failed")).when(activeTimeline).readIndexPlan(requested);
+ assertThrows(HoodieIndexException.class,
+ () -> new RunIndexActionExecutor(context, config, table,
"002").execute());
+
+ doReturn(new HoodieIndexPlan(1,
Collections.emptyList())).when(activeTimeline).readIndexPlan(requested);
+ assertThrows(HoodieIndexException.class,
+ () -> new RunIndexActionExecutor(context, config, table,
"002").execute());
+ }
+
+ @Test
+ void testRunRejectsPartitionThatAlreadyExists() throws IOException {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant requested = requestedIndexInstant();
+ stubRequestedIndexInstant(requested);
+ HoodieIndexPartitionInfo info = new HoodieIndexPartitionInfo(
+ 1, MetadataPartitionType.COLUMN_STATS.getPartitionPath(), "001",
Collections.emptyMap());
+ when(activeTimeline.readIndexPlan(requested)).thenReturn(new
HoodieIndexPlan(1, Collections.singletonList(info)));
+
when(tableConfig.getMetadataPartitions()).thenReturn(Set.of(MetadataPartitionType.COLUMN_STATS.getPartitionPath()));
+
+ assertThrows(HoodieIndexException.class,
+ () -> new RunIndexActionExecutor(context, config, table,
"002").execute());
+ }
+
+ @Test
+ void testRunInitializesFilesPartitionAndCompletesIndexInstant() throws
IOException {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant requested = requestedIndexInstant();
+ stubRequestedIndexInstant(requested);
+ HoodieIndexPartitionInfo info = new HoodieIndexPartitionInfo(
+ 1, MetadataPartitionType.FILES.getPartitionPath(), "001",
Collections.emptyMap());
+ when(activeTimeline.readIndexPlan(requested)).thenReturn(new
HoodieIndexPlan(1, Collections.singletonList(info)));
+ HoodieTableMetadataWriter writer = mock(HoodieTableMetadataWriter.class);
+ when(table.getIndexingMetadataWriter("002")).thenReturn(Option.of(writer));
Review Comment:
**P2:** This mock represents an `AutoCloseable` metadata writer, but the
FILES initialization branch obtains the writer and discards it without closing
it. Unlike the non-FILES branch, there is no try-with-resources; the real
writer's `close()` releases its metadata reader, write client, and view. A
focused `verify(writer).close()` assertion fails. Please close the writer in
the production FILES branch and assert closure on both successful
initialization and timeline-completion failure.
--
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]
