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 c6d09661c6c3fcc4a587d4ff340956f480a45f30 Author: Shuo Cheng <[email protected]> AuthorDate: Mon Feb 16 14:47:55 2026 +0800 fix(flink): Use blocking instant generation when CDC is enabled (#18206) * fix(flink): Use blocking instant generation when CDC is enabled --------- Co-authored-by: danny0405 <[email protected]> (cherry picked from commit bce8d598c4a5ba3ea8e370ee7bfa645c1119396a) --- .../hudi/common/model/HoodieCommitMetadata.java | 34 ++++++++-- .../hudi/common/table/cdc/HoodieCDCExtractor.java | 5 +- .../hudi/common/table/cdc/HoodieCDCFileSplit.java | 13 +++- .../common/model/TestHoodieCommitMetadata.java | 36 +++++++--- .../apache/hudi/configuration/OptionsResolver.java | 8 ++- .../org/apache/hudi/sink/TestWriteMergeOnRead.java | 76 ++++++++++++++++++++++ .../hudi/source/TestIncrementalInputSplits.java | 44 +++++++++++++ .../test/java/org/apache/hudi/utils/TestData.java | 21 +++++- 8 files changed, 217 insertions(+), 20 deletions(-) diff --git a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieCommitMetadata.java b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieCommitMetadata.java index 5d8f7e271a7d..bc8e539d8da6 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieCommitMetadata.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieCommitMetadata.java @@ -248,23 +248,47 @@ public class HoodieCommitMetadata implements Serializable { } /** - * parse the bytes of deltacommit, and get the base file and the log files belonging to this - * provided file group. + * For a log file, get the dependent file slice from the commit metadata. + * The file slice is utilized to infer the row-level changes from the given log file payloads. + * + * @param inputStream The commit metadata input stream + * @param fileGroupId The file group id + * @param currentLogFile The log file */ - public static Option<Pair<String, List<String>>> getFileSliceForFileGroupFromDeltaCommit(InputStream inputStream, HoodieFileGroupId fileGroupId) { + public static Option<Pair<String, List<String>>> getDependentFileSliceForFileGroupFromDeltaCommit( + InputStream inputStream, HoodieFileGroupId fileGroupId, String currentLogFile) { try { org.apache.hudi.avro.model.HoodieCommitMetadata commitMetadata = deserializeAvroMetadata(inputStream, org.apache.hudi.avro.model.HoodieCommitMetadata.class); Map<String,List<org.apache.hudi.avro.model.HoodieWriteStat>> partitionToWriteStatsMap = commitMetadata.getPartitionToWriteStats(); + List<org.apache.hudi.avro.model.HoodieWriteStat> targetWriteStats = new ArrayList<>(); for (Map.Entry<String, List<org.apache.hudi.avro.model.HoodieWriteStat>> partitionToWriteStat: partitionToWriteStatsMap.entrySet()) { for (org.apache.hudi.avro.model.HoodieWriteStat writeStat: partitionToWriteStat.getValue()) { HoodieFileGroupId fgId = new HoodieFileGroupId(partitionToWriteStat.getKey(), writeStat.getFileId()); if (fgId.equals(fileGroupId)) { - return Option.of(Pair.of(writeStat.getBaseFile() == null ? "" : writeStat.getBaseFile(), writeStat.getLogFiles())); + targetWriteStats.add(writeStat); } } } - return Option.empty(); + if (targetWriteStats.isEmpty()) { + return Option.empty(); + } else if (targetWriteStats.size() == 1) { + org.apache.hudi.avro.model.HoodieWriteStat writeStat = targetWriteStats.get(0); + return Option.of(Pair.of(writeStat.getBaseFile() == null ? "" : writeStat.getBaseFile(), + writeStat.getLogFiles().stream().filter(logFile -> !logFile.equals(currentLogFile)).collect(Collectors.toList()))); + } else { + // There are two cases that multiple write-stats are generated for the same file group within one commit: + // 1). log file rolls over(the file size exceeds the upper threshold); + // 2). eager flush from flink memory buffer(when memory buffer reaches the limit). + String baseFile = ""; + List<String> logFiles = new ArrayList<>(); + for (org.apache.hudi.avro.model.HoodieWriteStat writeStat: targetWriteStats) { + baseFile = writeStat.getBaseFile() == null ? "" : writeStat.getBaseFile(); + logFiles.addAll(writeStat.getLogFiles()); + } + // filter out the log files written after the given log file. + return Option.of(Pair.of(baseFile, logFiles.stream().filter(f -> f.compareTo(currentLogFile) < 0).distinct().sorted().collect(Collectors.toList()))); + } } catch (Exception e) { throw new HoodieException("Fail to parse the base file and log files from DeltaCommit", e); } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java index 4e60027a0538..e47b8720f289 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java @@ -329,8 +329,8 @@ public class HoodieCDCExtractor { if (instant.getAction().equals(DELTA_COMMIT_ACTION)) { String currentLogFileName = new StoragePath(currentLogFile).getName(); Option<Pair<String, List<String>>> fileSliceOpt = - HoodieCommitMetadata.getFileSliceForFileGroupFromDeltaCommit( - metaClient.getActiveTimeline().getInstantContentStream(instant), fgId); + HoodieCommitMetadata.getDependentFileSliceForFileGroupFromDeltaCommit( + metaClient.getActiveTimeline().getInstantContentStream(instant), fgId, currentLogFileName); if (fileSliceOpt.isPresent()) { Pair<String, List<String>> fileSlice = fileSliceOpt.get(); try { @@ -338,7 +338,6 @@ public class HoodieCDCExtractor { ? null : new HoodieBaseFile(storage.getPathInfo(new StoragePath(partitionPath, fileSlice.getLeft()))); List<StoragePath> logFilePaths = fileSlice.getRight().stream() - .filter(logFile -> !logFile.equals(currentLogFileName)) .map(logFile -> new StoragePath(partitionPath, logFile)) .collect(Collectors.toList()); List<HoodieLogFile> logFiles = storage.listDirectEntries(logFilePaths).stream() diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCFileSplit.java b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCFileSplit.java index d5be88bfeaec..c33df8414bd2 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCFileSplit.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCFileSplit.java @@ -125,6 +125,17 @@ public class HoodieCDCFileSplit implements Serializable, Comparable<HoodieCDCFil @Override public int compareTo(HoodieCDCFileSplit o) { - return this.instant.compareTo(o.instant); + int cmpResult = this.instant.compareTo(o.instant); + if (cmpResult == 0 && this.cdcInferCase == HoodieCDCInferenceCase.LOG_FILE + && this.beforeFileSlice.isPresent() && o.getBeforeFileSlice().isPresent()) { + // For mor table, multiple writes may occur for the same file group for a single instant, + // producing multiple log files that have monotonically increasing file version. + // + // A deterministic ordering between splits from the same file group is required to infer row changes for readers. + // beforeFileSlice` includes the log files already existed before the write of the log file in this split, which can be used as the + // tie-breaker to preserve the relative order of multiple log files. + return Math.toIntExact(this.beforeFileSlice.get().getLogFiles().count() - o.getBeforeFileSlice().get().getLogFiles().count()); + } + return cmpResult; } } diff --git a/hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieCommitMetadata.java b/hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieCommitMetadata.java index 477737e120f9..35c521530c62 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieCommitMetadata.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieCommitMetadata.java @@ -126,7 +126,7 @@ public class TestHoodieCommitMetadata { } @Test - public void testGetFileSliceForFileGroupFromDeltaCommit() throws IOException { + public void testGetDependentFileSliceForFileGroupFromDeltaCommit() throws IOException { org.apache.hudi.avro.model.HoodieCommitMetadata commitMetadata = new org.apache.hudi.avro.model.HoodieCommitMetadata(); org.apache.hudi.avro.model.HoodieWriteStat writeStat1 = createWriteStat("111", "111base", Arrays.asList("1.log", "2.log")); org.apache.hudi.avro.model.HoodieWriteStat writeStat2 = createWriteStat("111", "111base", Arrays.asList("3.log", "4.log")); @@ -138,21 +138,41 @@ public class TestHoodieCommitMetadata { byte[] serializedCommitMetadata = TimelineMetadataUtils.serializeAvroMetadata( commitMetadata, org.apache.hudi.avro.model.HoodieCommitMetadata.class).get(); - Option<Pair<String, List<String>>> result = HoodieCommitMetadata.getFileSliceForFileGroupFromDeltaCommit( - new ByteArrayInputStream(serializedCommitMetadata), new HoodieFileGroupId("partition1", "111")); + Option<Pair<String, List<String>>> result = HoodieCommitMetadata.getDependentFileSliceForFileGroupFromDeltaCommit( + new ByteArrayInputStream(serializedCommitMetadata), new HoodieFileGroupId("partition1", "111"), "4.log"); assertTrue(result.isPresent()); assertEquals("111base", result.get().getKey()); - assertEquals(2, result.get().getValue().size()); + assertEquals(1, result.get().getValue().size()); assertEquals("3.log", result.get().getValue().get(0)); - assertEquals("4.log", result.get().getValue().get(1)); - result = HoodieCommitMetadata.getFileSliceForFileGroupFromDeltaCommit( - new ByteArrayInputStream(serializedCommitMetadata), new HoodieFileGroupId("partition1", "222")); + result = HoodieCommitMetadata.getDependentFileSliceForFileGroupFromDeltaCommit( + new ByteArrayInputStream(serializedCommitMetadata), new HoodieFileGroupId("partition1", "222"), "5.log"); assertTrue(result.isPresent()); assertTrue(result.get().getKey().isEmpty()); + assertEquals(0, result.get().getValue().size()); + + org.apache.hudi.avro.model.HoodieWriteStat writeStat4 = createWriteStat("111", null, Arrays.asList("1.log")); + org.apache.hudi.avro.model.HoodieWriteStat writeStat5 = createWriteStat("111", null, Arrays.asList("2.log")); + partitionToWriteStatsMap = new HashMap<>(); + // simulate the scenario like eager flushing of MOR ingestion in flink writer. + partitionToWriteStatsMap.put("partition1", Arrays.asList(writeStat4, writeStat5)); + commitMetadata.setPartitionToWriteStats(partitionToWriteStatsMap); + serializedCommitMetadata = TimelineMetadataUtils.serializeAvroMetadata( + commitMetadata, org.apache.hudi.avro.model.HoodieCommitMetadata.class).get(); + + result = HoodieCommitMetadata.getDependentFileSliceForFileGroupFromDeltaCommit( + new ByteArrayInputStream(serializedCommitMetadata), new HoodieFileGroupId("partition1", "111"), "1.log"); + assertTrue(result.isPresent()); + assertEquals("", result.get().getKey()); + assertEquals(0, result.get().getValue().size()); + + result = HoodieCommitMetadata.getDependentFileSliceForFileGroupFromDeltaCommit( + new ByteArrayInputStream(serializedCommitMetadata), new HoodieFileGroupId("partition1", "111"), "2.log"); + assertTrue(result.isPresent()); + assertEquals("", result.get().getKey()); assertEquals(1, result.get().getValue().size()); - assertEquals("5.log", result.get().getValue().get(0)); + assertEquals("1.log", result.get().getValue().get(0)); } @Test diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java index c55c41651ada..8f17b38e1062 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java @@ -493,9 +493,15 @@ public class OptionsResolver { /** * Returns whether the writers should use blocking instant time generation. + * + * <p>Blocking instant generation is enabled only for upsert workloads that require strict + * instant ordering, i.e. upsert on COW tables, or upsert with CDC enabled. + * + * <p>When this returns {@code true}, writer tasks wait for commit acknowledgement with timeout + * ({@link FlinkOptions#WRITE_COMMIT_ACK_TIMEOUT}). When this returns {@code false}. */ public static boolean isBlockingInstantGeneration(Configuration conf) { - return isCowTable(conf) && isUpsertOperation(conf); + return (isCowTable(conf) || conf.get(FlinkOptions.CDC_ENABLED)) && isUpsertOperation(conf); } /** 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 6deef9590170..8331aa515251 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 @@ -20,20 +20,34 @@ package org.apache.hudi.sink; import org.apache.hudi.common.config.HoodieStorageConfig; import org.apache.hudi.common.model.EventTimeAvroPayload; +import org.apache.hudi.common.model.HoodieCommitMetadata; import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.config.HoodieClusteringConfig; import org.apache.hudi.config.HoodieIndexConfig; import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.util.StreamerUtil; import org.apache.hudi.utils.TestData; import org.apache.flink.configuration.Configuration; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; +import static org.apache.hudi.utils.TestData.insertRow; +import static org.junit.jupiter.api.Assertions.assertEquals; + /** * Test cases for delta stream write. */ @@ -253,6 +267,68 @@ public class TestWriteMergeOnRead extends TestWriteCopyOnWrite { .end(); } + @Test + public void testInsertDuplicateRecordsWithCDCMode() throws Exception { + conf.set(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT, 10_000L); + conf.set(FlinkOptions.CDC_ENABLED, true); + + Map<String, String> expected = new HashMap<>(); + expected.put("par1", "[id1,par1,id1,Danny,23,1,par1]"); + + List<RowData> insertData = Collections.singletonList( + insertRow(StringData.fromString("id1"), StringData.fromString("Danny"), 23, + TimestampData.fromEpochMillis(1), StringData.fromString("par1"))); + + TestHarness testHarness = preparePipeline() + .consume(insertData) + .checkpoint(1) + .allDataFlushed() + .handleEvents(1); + + Thread t1 = new Thread(() -> { + try { + Thread.sleep(3000); + testHarness.checkpointComplete(1); + testHarness.checkWrittenData(expected, 1); + } catch (Exception e) { + throw new HoodieException(e); + } + }); + t1.start(); + + testHarness + .consume(insertData) + .checkpoint(2) + .allDataFlushed(); + t1.join(); + + testHarness.handleEvents(1) + .checkpointComplete(2) + .checkWrittenData(expected, 1) + .end(); + + // validate the metadata of the above two inserts, specifically the `prevCommit` in the write stat, the expected + // result should be: + // 1) the prev commit time for the first commit is itself. + // 2) the prev commit time for the second commit is the first instant time, which ensures the deterministic update sequence. + HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(conf); + List<HoodieInstant> completedDeltaInstants = + metaClient.reloadActiveTimeline().getDeltaCommitTimeline().filterCompletedInstants().getInstants(); + assertEquals(2, completedDeltaInstants.size()); + HoodieInstant firstInstant = completedDeltaInstants.get(0); + HoodieInstant secondInstant = completedDeltaInstants.get(1); + HoodieCommitMetadata firstCommitMetadata = metaClient.getActiveTimeline().readCommitMetadata(firstInstant); + HoodieCommitMetadata secondCommitMetadata = metaClient.getActiveTimeline().readCommitMetadata(secondInstant); + + List<HoodieWriteStat> firstWriteStats = firstCommitMetadata.getWriteStats(); + assertEquals(1, firstWriteStats.size()); + assertEquals(firstInstant.requestedTime(), firstWriteStats.get(0).getPrevCommit()); + + List<HoodieWriteStat> secondWriteStats = secondCommitMetadata.getWriteStats(); + assertEquals(1, secondWriteStats.size()); + assertEquals(firstInstant.requestedTime(), secondWriteStats.get(0).getPrevCommit()); + } + @Override protected Map<String, String> getExpectedBeforeCheckpointComplete() { return EXPECTED1; diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java index fb9275c2fc35..15777e919ca6 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java @@ -24,6 +24,8 @@ import org.apache.hudi.common.model.HoodieCommitMetadata; import org.apache.hudi.common.model.HoodieReplaceCommitMetadata; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.common.table.cdc.HoodieCDCFileSplit; +import org.apache.hudi.common.table.cdc.HoodieCDCInferenceCase; import org.apache.hudi.common.table.log.InstantRange; import org.apache.hudi.common.table.read.IncrementalQueryAnalyzer; import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; @@ -40,8 +42,11 @@ import org.apache.hudi.sink.partitioner.profile.WriteProfiles; import org.apache.hudi.source.prune.ColumnStatsProbe; import org.apache.hudi.source.prune.PartitionPruners; import org.apache.hudi.storage.StoragePathInfo; +import org.apache.hudi.table.format.cdc.CdcInputSplit; +import org.apache.hudi.table.format.mor.MergeOnReadInputSplit; import org.apache.hudi.utils.TestConfigurations; import org.apache.hudi.utils.TestData; +import org.apache.hudi.utils.TestUtils; import org.apache.flink.configuration.Configuration; import org.apache.flink.core.fs.Path; @@ -76,6 +81,7 @@ import static org.apache.hudi.common.table.timeline.InstantComparison.compareTim import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertIterableEquals; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -83,6 +89,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue; * Test cases for {@link IncrementalInputSplits}. */ public class TestIncrementalInputSplits extends HoodieCommonTestHarness { + // for RowData write function: to trigger buffer flush with batch size exceeded by 3 rows, each record is 48 bytes + private static final double BATCH_SIZE_MB = 0.00013; @BeforeEach void init() { @@ -559,4 +567,40 @@ public class TestIncrementalInputSplits extends HoodieCommonTestHarness { .findFirst() .orElse(-1); } + + @Test + void testOrderingOfCDCInputSplitsWithEagerFlushing() throws Exception { + metaClient = HoodieTestUtils.init(basePath, HoodieTableType.MERGE_ON_READ); + Configuration conf = TestConfigurations.getDefaultConf(basePath); + conf.set(FlinkOptions.WRITE_BATCH_SIZE, BATCH_SIZE_MB); + conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.MERGE_ON_READ.name()); + // Insert test data + TestData.writeData(TestData.DATA_SET_INSERT_DUPLICATES, 1, conf); + + String firstCommit = TestUtils.getFirstCompleteInstant(basePath); + conf.set(FlinkOptions.READ_START_COMMIT, firstCommit); + + IncrementalInputSplits iis = IncrementalInputSplits.builder() + .conf(conf) + .path(new Path(basePath)) + .rowType(TestConfigurations.ROW_TYPE) + .build(); + + List<MergeOnReadInputSplit> splits = iis.inputSplits(metaClient, true).getInputSplits(); + assertEquals(1, splits.size()); + assertInstanceOf(CdcInputSplit.class, splits.get(0)); + CdcInputSplit cdcInputSplit = (CdcInputSplit) splits.get(0); + + // there are two cdc file splits since eager flushing happens. + HoodieCDCFileSplit[] cdcFileSplits = cdcInputSplit.getChanges(); + assertEquals(2, cdcFileSplits.length); + // instant fields for these two splits are the same + assertEquals(cdcFileSplits[0].getInstant(), cdcFileSplits[1].getInstant()); + assertEquals(HoodieCDCInferenceCase.LOG_FILE, cdcFileSplits[0].getCdcInferCase()); + assertEquals(HoodieCDCInferenceCase.LOG_FILE, cdcFileSplits[1].getCdcInferCase()); + // validate the order of splits with same instant + assertTrue(cdcFileSplits[0].getBeforeFileSlice().isPresent()); + assertTrue(cdcFileSplits[1].getBeforeFileSlice().isPresent()); + assertTrue(cdcFileSplits[0].getCdcFiles().get(0).compareTo(cdcFileSplits[1].getCdcFiles().get(0)) < 0); + } } diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestData.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestData.java index 6b5f79c6872b..02617ef9829a 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestData.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestData.java @@ -545,6 +545,21 @@ public class TestData { public static void writeData( List<RowData> dataBuffer, Configuration conf) throws Exception { + writeData(dataBuffer, 0, conf); + } + + /** + * Write a list of row data with Hoodie format base on the given configuration. + * + * @param dataBuffer The data buffer to write + * @param numEagerFlush The number of eager flushing + * @param conf The flink configuration + * @throws Exception if error occurs + */ + public static void writeData( + List<RowData> dataBuffer, + int numEagerFlush, + Configuration conf) throws Exception { final TestFunctionWrapper<RowData> funcWrapper = getWritePipeline(conf.get(FlinkOptions.PATH), conf); funcWrapper.openFunction(); @@ -555,8 +570,10 @@ public class TestData { // this triggers the data write and event send funcWrapper.checkpointFunction(1); - final OperatorEvent nextEvent = funcWrapper.getNextEvent(); - funcWrapper.getCoordinator().handleEventFromOperator(0, nextEvent); + for (int i = 0; i < numEagerFlush + 1; i++) { + final OperatorEvent nextEvent = funcWrapper.getNextEvent(); + funcWrapper.getCoordinator().handleEventFromOperator(0, nextEvent); + } funcWrapper.checkpointComplete(1); funcWrapper.close();
