This is an automated email from the ASF dual-hosted git repository. danny0405 pushed a commit to branch release-1.1.2 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit bc99b1d7208a5fef73ff923c2ee06059c67e606e Author: Joy <[email protected]> AuthorDate: Sat Jul 11 07:27:59 2026 +0800 fix(flink): prevent data loss on global failover for streaming writes (#19237) * fix(flink): prevent data loss on global failover for streaming writes --------- Co-authored-by: jiangyu84 <[email protected]> Co-authored-by: danny0405 <[email protected]> (cherry picked from commit be8a54c4ebdaf287993d24fefba60a58952a9c9e) --- .../apache/hudi/client/BaseHoodieWriteClient.java | 23 +++++++ .../metadata/HoodieBackedTableMetadataWriter.java | 9 +++ .../TestHoodieBackedTableMetadataWriter.java | 27 ++++++++ .../apache/hudi/client/HoodieFlinkWriteClient.java | 27 ++++++++ .../apache/hudi/client/TestFlinkWriteClient.java | 61 +++++++++++++++++- .../hudi/sink/StreamWriteOperatorCoordinator.java | 32 +++++++--- .../org/apache/hudi/sink/utils/EventBuffers.java | 2 +- .../sink/TestStreamWriteOperatorCoordinator.java | 72 +++++++++++++++++----- .../org/apache/hudi/sink/TestWriteMergeOnRead.java | 27 ++++++++ .../sink/utils/StreamWriteFunctionWrapper.java | 3 +- 10 files changed, 256 insertions(+), 27 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java index 7a033a7b4824..183b57eb5ee0 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java @@ -616,6 +616,29 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient } } + /** + * Performs post-commit cleanup when the instant is already completed and commit metadata is not + * available to invoke the regular post-commit hook. This can happen while recovering a streaming + * metadata-table write after failover. The table is recreated from the write configuration so its + * marker directory can still be removed, and the heartbeat is always stopped even if marker cleanup + * fails. + * + * @param instantTime the completed instant to clean up + */ + public void postCommit(String instantTime) { + try { + HoodieTable table = createTable(config); + context.setJobStatus(this.getClass().getSimpleName(), "Cleaning up marker directories for commit " + instantTime + " in table " + + config.getTableName()); + // Delete the marker directory for the instant. + WriteMarkersFactory.get(config.getMarkersType(), table, instantTime) + .quietDeleteMarkerDir(context, config.getMarkersDeleteParallelism()); + metrics.updateTableServiceInstantMetrics(table.getActiveTimeline()); + } finally { + this.heartbeatClient.stop(instantTime); + } + } + /** * Triggers cleaning and archival for the table of interest. This method is called outside of locks. So, internal callers should ensure they acquire lock whereever applicable. * @param table instance of {@link HoodieTable} of interest. diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java index d45cdaaedf9f..a5923d30951a 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java @@ -1317,6 +1317,12 @@ public abstract class HoodieBackedTableMetadataWriter<I, O> implements HoodieTab @Override public void completeStreamingCommit(String instantTime, HoodieEngineContext context, List<HoodieWriteStat> partialWriteStats, HoodieCommitMetadata metadata) { + if (metadataMetaClient.getActiveTimeline().filterCompletedInstants().containsInstant(instantTime)) { + LOG.info("Skipping streaming metadata commit completion for already completed instant {}", instantTime); + getWriteClient().postCommit(instantTime); + return; + } + List<HoodieWriteStat> allWriteStats = new ArrayList<>(partialWriteStats); // update metadata for left over partitions which does not have streaming writes support. allWriteStats.addAll(prepareAndWriteToNonStreamingPartitions(metadata, instantTime).map(WriteStatus::getStat).collectAsList()); @@ -1745,6 +1751,9 @@ public abstract class HoodieBackedTableMetadataWriter<I, O> implements HoodieTab public void close() throws Exception { if (metadata != null) { metadata.close(); + // Keep the closed reader reference: guarded update paths use its presence to proceed and + // mayBeReinitMetadataReader() detects the closed file-system view and reopens the reader. + // Nullifying it here would silently skip subsequent metadata updates, such as rollbacks. } if (writeClient != null) { writeClient.close(); diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriter.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriter.java index e583dead5ddc..329d6f2e0fca 100644 --- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriter.java +++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriter.java @@ -22,6 +22,7 @@ import org.apache.hudi.client.BaseHoodieWriteClient; import org.apache.hudi.common.config.HoodieMetadataConfig; import org.apache.hudi.common.data.HoodieData; import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.model.HoodieCommitMetadata; import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.table.HoodieTableMetaClient; @@ -41,6 +42,7 @@ import org.junit.jupiter.params.provider.CsvSource; import org.mockito.MockedStatic; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -53,11 +55,13 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.CALLS_REAL_METHODS; import static org.mockito.Mockito.RETURNS_DEEP_STUBS; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; class TestHoodieBackedTableMetadataWriter { @@ -76,6 +80,29 @@ class TestHoodieBackedTableMetadataWriter { when(metadataConfig.getMaxReaderBufferSize()).thenReturn(1024); } + @Test + void completeStreamingCommitSkipsAlreadyCompletedMetadataInstant() { + String instantTime = "20260709120000000"; + HoodieBackedTableMetadataWriter<List<HoodieRecord>, List<?>> metadataWriter = + mock(HoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS); + HoodieEngineContext engineContext = mock(HoodieEngineContext.class); + HoodieTableMetaClient metadataMetaClient = mock(HoodieTableMetaClient.class); + HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class); + HoodieTimeline completedTimeline = mock(HoodieTimeline.class); + BaseHoodieWriteClient writeClient = mock(BaseHoodieWriteClient.class); + + metadataWriter.metadataMetaClient = metadataMetaClient; + when(metadataMetaClient.getActiveTimeline()).thenReturn(activeTimeline); + when(activeTimeline.filterCompletedInstants()).thenReturn(completedTimeline); + when(completedTimeline.containsInstant(instantTime)).thenReturn(true); + when(metadataWriter.initializeWriteClient()).thenReturn(writeClient); + + metadataWriter.completeStreamingCommit(instantTime, engineContext, Collections.emptyList(), mock(HoodieCommitMetadata.class)); + + verify(writeClient).postCommit(instantTime); + verifyNoMoreInteractions(writeClient); + } + @ParameterizedTest @CsvSource(value = { "true,true,false,true", diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java index 14a108e21b8a..5c40e4def972 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java @@ -98,6 +98,22 @@ public class HoodieFlinkWriteClient<T> return FlinkHoodieIndexFactory.createIndex((HoodieFlinkEngineContext) context, config); } + /** + * Stop the heartbeat for the instant. + */ + public void cleanResources(String instantTime) { + getHeartbeatClient().stop(instantTime); + } + + /** + * Restart the heartbeat for a recommitted instant. + */ + public void restartHeartbeat(String instantTime) { + if (getConfig().getFailedWritesCleanPolicy().isLazy()) { + getHeartbeatClient().start(instantTime); + } + } + @Override public boolean commit(String instantTime, List<WriteStatus> writeStatuses, Option<Map<String, String>> extraMetadata, String commitActionType, Map<String, List<String>> partitionToReplacedFileIds, @@ -538,4 +554,15 @@ public class HoodieFlinkWriteClient<T> ((MiniBatchHandle) writeHandle).closeGracefully(); } } + + /** + * Flink keeps the heartbeat active when a commit attempt fails because the coordinator may need to + * recommit the instant after failover. Successful commits stop the heartbeat from {@code postCommit}, + * while {@link #cleanResources(String)} cleans up data-table resources for + * instants discarded or resolved by the coordinator. Therefore this generic hook is intentionally + * a no-op. + */ + @Override + public void releaseResources(String instantTime) { + } } diff --git a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestFlinkWriteClient.java b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestFlinkWriteClient.java index 33ae4238962a..0b532bbe5570 100644 --- a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestFlinkWriteClient.java +++ b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestFlinkWriteClient.java @@ -19,27 +19,43 @@ package org.apache.hudi.client; +import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient; +import org.apache.hudi.common.engine.EngineType; +import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy; +import org.apache.hudi.config.HoodieCleanConfig; import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.table.HoodieTable; import org.apache.hudi.testutils.HoodieFlinkClientTestHarness; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; import java.io.IOException; +import java.util.concurrent.atomic.AtomicBoolean; 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; public class TestFlinkWriteClient extends HoodieFlinkClientTestHarness { @BeforeEach - private void setup() throws IOException { + void setup() throws IOException { initPath(); initFileSystem(); initMetaClient(); } + @AfterEach + void teardown() throws IOException { + cleanupResources(); + } + @ParameterizedTest @ValueSource(booleans = {true, false}) public void testWriteClientAndTableServiceClientWithTimelineServer( @@ -61,4 +77,47 @@ public class TestFlinkWriteClient extends HoodieFlinkClientTestHarness { writeClient.close(); } + + @Test + public void testReleaseAndPostCommitResourcesForLazyFailedWrites() throws IOException { + HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder() + .withPath(metaClient.getBasePath()) + .withEngineType(EngineType.FLINK) + .withCleanConfig(HoodieCleanConfig.newBuilder() + .withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY) + .build()) + .build(); + + AtomicBoolean failTableCreation = new AtomicBoolean(false); + writeClient = new HoodieFlinkWriteClient(context, writeConfig) { + @Override + protected HoodieTable createTable(HoodieWriteConfig config) { + if (failTableCreation.get()) { + throw new HoodieException("Expected table creation failure"); + } + return super.createTable(config); + } + }; + String instantTime = "20260709120000000"; + writeClient.restartHeartbeat(instantTime); + + assertTrue(HoodieHeartbeatClient.heartbeatExists( + metaClient.getStorage(), metaClient.getBasePath().toString(), instantTime)); + + writeClient.releaseResources(instantTime); + assertTrue(HoodieHeartbeatClient.heartbeatExists( + metaClient.getStorage(), metaClient.getBasePath().toString(), instantTime)); + + writeClient.postCommit(instantTime); + assertFalse(HoodieHeartbeatClient.heartbeatExists( + metaClient.getStorage(), metaClient.getBasePath().toString(), instantTime)); + + String failedPostCommitInstantTime = "20260709120000001"; + writeClient.restartHeartbeat(failedPostCommitInstantTime); + failTableCreation.set(true); + assertThrows(HoodieException.class, () -> writeClient.postCommit(failedPostCommitInstantTime)); + assertFalse(HoodieHeartbeatClient.heartbeatExists( + metaClient.getStorage(), metaClient.getBasePath().toString(), failedPostCommitInstantTime)); + } + } diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java index a0d9a40f7257..4a47e18f0423 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java @@ -241,7 +241,7 @@ public class StreamWriteOperatorCoordinator if (OptionsResolver.isMultiWriter(conf)) { initClientIds(conf); } - restoreEvents(); + restoreEvents(Long.MAX_VALUE); } catch (Throwable throwable) { LOG.error("Failed to start operator coordinator.", throwable); context.failJob(throwable); @@ -311,8 +311,17 @@ public class StreamWriteOperatorCoordinator @Override public void resetToCheckpoint(long checkpointID, byte[] checkpointData) { if (checkpointData != null) { - initEventBufferIfNecessary(); - this.eventBuffers.addEventsToBuffer(SerializationUtils.deserialize(checkpointData)); + // resetToCheckpoint() is called in two cases: + // 1. The job is restarted from state, start() will be called later. + // 2. The job is recovered from global failover. The coordinator is already started, and start() will not be called again. + if (this.eventBuffers == null) { + // case1: the events restore is moved to start() since it requires the write/meta client instantiation. + initEventBuffer(); + this.eventBuffers.addEventsToBuffer(SerializationUtils.deserialize(checkpointData)); + } else { + // case2: restore the events directly. + restoreEvents(checkpointID); + } } } @@ -394,10 +403,11 @@ public class StreamWriteOperatorCoordinator // Utilities // ------------------------------------------------------------------------- - private void restoreEvents() { + private void restoreEvents(long checkpointId) { if (this.eventBuffers.nonEmpty()) { - final HoodieTimeline completedTimeline = this.metaClient.getActiveTimeline().filterCompletedInstants(); + final HoodieTimeline completedTimeline = this.metaClient.reloadActiveTimeline().filterCompletedInstants(); this.eventBuffers.getEventBufferStream() + .filter(entry -> entry.getKey() < checkpointId) .forEach(entry -> recommitInstant(completedTimeline, entry.getKey(), entry.getValue().getLeft(), entry.getValue().getRight())); this.metaClient.reloadActiveTimeline(); } @@ -460,6 +470,10 @@ public class StreamWriteOperatorCoordinator if (this.eventBuffers != null) { return; } + initEventBuffer(); + } + + private void initEventBuffer() { // initialize event buffer this.eventBuffers = EventBuffers.getInstance(conf); } @@ -497,10 +511,12 @@ public class StreamWriteOperatorCoordinator LOG.info("Recommit instant {}", instant); // Recommit should start heartbeat for lazy failed writes clean policy to avoid aborting for heartbeat expired; // The following up checkpoints would recommit the instant. - if (writeClient.getConfig().getFailedWritesCleanPolicy().isLazy()) { - writeClient.getHeartbeatClient().start(instant); - } + writeClient.restartHeartbeat(instant); commitInstant(checkpointId, instant, bootstrapBuffer); + } else { + // clean the corresponding event buffer if the instant is already committed. + eventBuffers.reset(checkpointId); + writeClient.cleanResources(instant); } } diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java index d4eb6a393883..c1bdac1d717d 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java @@ -154,7 +154,7 @@ public class EventBuffers implements Serializable { */ public Map<Long, Pair<String, WriteMetadataEvent[]>> getAllCompletedEvents() { return this.eventBuffers.entrySet().stream() - .filter(entry -> Arrays.stream(entry.getValue().getRight()).allMatch(event -> event == null || event.isLastBatch())) + .filter(entry -> Arrays.stream(entry.getValue().getRight()).allMatch(event -> event != null && event.isLastBatch())) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); } } diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java index 8a79c0ddbf11..5ae64be04f20 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java @@ -20,6 +20,7 @@ package org.apache.hudi.sink; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient; +import org.apache.hudi.common.fs.FSUtils; import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.model.HoodieWriteStat; @@ -43,6 +44,7 @@ import org.apache.hudi.sink.utils.MockCoordinatorExecutor; import org.apache.hudi.sink.utils.NonThrownExecutor; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration; +import org.apache.hudi.util.StreamerUtil; import org.apache.hudi.utils.TestConfigurations; import org.apache.hudi.utils.TestUtils; @@ -92,7 +94,7 @@ public class TestStreamWriteOperatorCoordinator { @BeforeEach public void before() throws Exception { - coordinator = createCoordinator(TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()), 2); + coordinator = startCoordinator(TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()), 2); } @AfterEach @@ -128,11 +130,44 @@ public class TestStreamWriteOperatorCoordinator { } } + /** + * Verifies both coordinator restore paths. In case 1, a newly constructed coordinator receives + * checkpoint data before {@link StreamWriteOperatorCoordinator#start()}, so it must deserialize + * the event buffers for restoration during start. In case 2, global failover resets an already + * started coordinator, so it recommits its live event buffers directly. + */ @Test public void testCheckpointAndRestore() throws Exception { + Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); + conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.MERGE_ON_READ.name()); + coordinator = startCoordinator(conf, 2); + + requestInstantTime(-1); + String instant = coordinator.getInstant(); + assertNotEquals("", instant); + + OperatorEvent event0 = createOperatorEvent(0, instant, "par1", true, 0.1); + OperatorEvent event1 = createOperatorEvent(1, instant, "par2", true, 0.2); + coordinator.handleEventFromOperator(0, event0); + coordinator.handleEventFromOperator(1, event1); + CompletableFuture<byte[]> future = new CompletableFuture<>(); coordinator.checkpointCoordinator(1, future); + + // Case 1: job restart restores checkpoint data before the coordinator starts. + try (StreamWriteOperatorCoordinator restoredCoordinator = createCoordinator(conf, 2)) { + restoredCoordinator.resetToCheckpoint(1, future.get()); + + WriteMetadataEvent[] eventBuffer = restoredCoordinator.getEventBuffer(-1); + assertEquals(2, eventBuffer.length); + } + + // Case 2: global failover recommits the live buffers of the already started coordinator. coordinator.resetToCheckpoint(1, future.get()); + + assertNull(coordinator.getEventBuffer()); + assertTrue(StreamerUtil.createMetaClient(conf).reloadActiveTimeline() + .filterCompletedInstants().containsInstant(instant)); } @Test @@ -154,7 +189,7 @@ public class TestStreamWriteOperatorCoordinator { public void testEventReset() throws Exception { Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.MERGE_ON_READ.name()); - coordinator = createCoordinator(conf, 2); + coordinator = startCoordinator(conf, 2); CompletableFuture<byte[]> future = new CompletableFuture<>(); coordinator.checkpointCoordinator(1, future); String instant = requestInstantTime(0); @@ -197,7 +232,7 @@ public class TestStreamWriteOperatorCoordinator { conf.setString(HoodieWriteConfig.ALLOW_EMPTY_COMMIT.key(), "false"); OperatorCoordinator.Context context = new MockOperatorCoordinatorContext(new OperatorID(), 2); - coordinator = createCoordinator(conf, 2); + coordinator = startCoordinator(conf, 2); final CompletableFuture<byte[]> future = new CompletableFuture<>(); coordinator.checkpointCoordinator(1, future); @@ -243,7 +278,7 @@ public class TestStreamWriteOperatorCoordinator { // override the default configuration Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); conf.setString(HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key(), HoodieFailedWritesCleaningPolicy.LAZY.name()); - coordinator = createCoordinator(conf, 1); + coordinator = startCoordinator(conf, 1); assertTrue(coordinator.getWriteClient().getConfig().getFailedWritesCleanPolicy().isLazy()); @@ -289,7 +324,7 @@ public class TestStreamWriteOperatorCoordinator { // override the default configuration Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); conf.set(FlinkOptions.HIVE_SYNC_ENABLED, true); - coordinator = createCoordinator(conf, 1); + coordinator = startCoordinator(conf, 1); String instant = mockWriteWithMetadata(0); assertNotEquals("", instant); @@ -307,7 +342,7 @@ public class TestStreamWriteOperatorCoordinator { int metadataCompactionDeltaCommits = 5; conf.set(FlinkOptions.METADATA_ENABLED, true); conf.set(FlinkOptions.METADATA_COMPACTION_DELTA_COMMITS, metadataCompactionDeltaCommits); - coordinator = createCoordinator(conf, 1); + coordinator = startCoordinator(conf, 1); String instant = coordinator.getInstant(); assertEquals("", instant); @@ -384,7 +419,7 @@ public class TestStreamWriteOperatorCoordinator { conf.set(FlinkOptions.METADATA_ENABLED, true); conf.set(FlinkOptions.METADATA_COMPACTION_DELTA_COMMITS, 20); conf.setString("hoodie.metadata.log.compaction.enable", "true"); - coordinator = createCoordinator(conf, 1); + coordinator = startCoordinator(conf, 1); String instant = coordinator.getInstant(); assertEquals("", instant); @@ -428,7 +463,7 @@ public class TestStreamWriteOperatorCoordinator { // override the default configuration Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); conf.set(FlinkOptions.METADATA_ENABLED, true); - coordinator = createCoordinator(conf, 1); + coordinator = startCoordinator(conf, 1); String instant = coordinator.getInstant(); assertEquals("", instant); @@ -452,7 +487,7 @@ public class TestStreamWriteOperatorCoordinator { metadataTableMetaClient.getActiveTimeline().transitionRequestedToInflight(HoodieActiveTimeline.DELTA_COMMIT_ACTION, instant); metadataTableMetaClient.reloadActiveTimeline(); // reset the coordinator to mimic the job failover. - coordinator = createCoordinator(conf, 1); + coordinator = startCoordinator(conf, 1); // write another commit with new instant on the metadata timeline instant = mockWriteWithMetadata(ckp); @@ -471,7 +506,7 @@ public class TestStreamWriteOperatorCoordinator { Logger logger = Mockito.mock(Logger.class); // avoid too many logs by executor NonThrownExecutor executor = NonThrownExecutor.builder(logger).waitForTasksFinish(true).build(); - try (StreamWriteOperatorCoordinator coordinator = createCoordinator(conf, 1)) { + try (StreamWriteOperatorCoordinator coordinator = startCoordinator(conf, 1)) { coordinator.start(); coordinator.setExecutor(executor); TimeUnit.SECONDS.sleep(5); // wait for handled bootstrap event @@ -509,7 +544,7 @@ public class TestStreamWriteOperatorCoordinator { conf.setString(HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key(), WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.name()); conf.setString("hoodie.write.lock.client.num_retries", "1"); - coordinator = createCoordinator(conf, 1); + coordinator = startCoordinator(conf, 1); String instant = coordinator.getInstant(); assertEquals("", instant); @@ -534,7 +569,7 @@ public class TestStreamWriteOperatorCoordinator { public void testCommitOnEmptyBatch() throws Exception { Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); conf.setString(HoodieWriteConfig.ALLOW_EMPTY_COMMIT.key(), "true"); - try (StreamWriteOperatorCoordinator coordinator = createCoordinator(conf, 2)) { + try (StreamWriteOperatorCoordinator coordinator = startCoordinator(conf, 2)) { // Coordinator start the instant String instant = requestInstantTime(coordinator, -1); @@ -578,15 +613,20 @@ public class TestStreamWriteOperatorCoordinator { } } - private static StreamWriteOperatorCoordinator createCoordinator(Configuration conf, int subTasks) throws Exception { - MockOperatorCoordinatorContext coordinatorContext = new MockOperatorCoordinatorContext(new OperatorID(), subTasks); - StreamWriteOperatorCoordinator coordinator = new StreamWriteOperatorCoordinator(conf, coordinatorContext); + private static StreamWriteOperatorCoordinator startCoordinator(Configuration conf, int subTasks) throws Exception { + StreamWriteOperatorCoordinator coordinator = createCoordinator(conf, subTasks); coordinator.start(); + MockOperatorCoordinatorContext coordinatorContext = (MockOperatorCoordinatorContext) coordinator.getContext(); coordinator.setExecutor(new MockCoordinatorExecutor(coordinatorContext)); coordinator.setInstantRequestExecutor(new MockCoordinatorExecutor(coordinatorContext)); return coordinator; } + private static StreamWriteOperatorCoordinator createCoordinator(Configuration conf, int subTasks) { + MockOperatorCoordinatorContext coordinatorContext = new MockOperatorCoordinatorContext(new OperatorID(), subTasks); + return new StreamWriteOperatorCoordinator(conf, coordinatorContext); + } + private String mockWriteWithMetadata(long checkpointId) { String instant = requestInstantTime(checkpointId); OperatorEvent event = createOperatorEvent(0, checkpointId, instant, "par1", false, true, 0.1); @@ -624,7 +664,7 @@ public class TestStreamWriteOperatorCoordinator { HoodieWriteStat writeStat = new HoodieWriteStat(); writeStat.setPartitionPath(partitionPath); writeStat.setFileId("fileId123"); - writeStat.setPath("path123"); + writeStat.setPath(partitionPath + "/" + FSUtils.makeBaseFileName(instant, "1-0-1", "fileId123", ".parquet")); writeStat.setFileSizeInBytes(123); writeStat.setTotalWriteBytes(123); writeStat.setNumWrites(1); diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java index 8331aa515251..5733803a2be9 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java @@ -267,6 +267,33 @@ public class TestWriteMergeOnRead extends TestWriteCopyOnWrite { .end(); } + @Test + public void testRecommitOnGlobalFailover() throws Exception { + conf.set(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT, 10_000L); + conf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, true); + + Map<String, String> expected = new HashMap<>(); + expected.put("par1", "[id1,par1,id1,Danny,23,1,par1]"); + expected.put("par2", "[id3,par2,id3,Julian,53,3,par2]"); + + preparePipeline(conf) + .consume(TestData.DATA_SET_PART1) + .emptyEventBuffer() + .checkpoint(1) + .assertNextEvent(1, "par1") + .consume(TestData.DATA_SET_PART3) + .checkpoint(2) + // both ckp-1 and ckp-2 are not committing + .assertNextEvent(1, "par2") + .checkCompletedInstantCount(0) + // Simulate the global failover path. The coordinator resets to ckp-2, recommits ckp-1 + // from the coordinator state, and recommits ckp-2 from the restored writer state. + .jobFailover() + .checkCompletedInstantCount(2) + .checkWrittenData(expected, 2) + .end(); + } + @Test public void testInsertDuplicateRecordsWithCDCMode() throws Exception { conf.set(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT, 10_000L); diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java index fc2d3f9fdf5e..3bd3b2fd3412 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java @@ -257,7 +257,8 @@ public class StreamWriteFunctionWrapper<I> implements TestFunctionWrapper<I> { } public void jobFailover() throws Exception { - coordinatorFails(); + // A same-graph global failover resets the running coordinator without restarting it. + resetCoordinatorToCheckpoint(); subTaskFails(0, 0); }
