From 175831376be31e03b2b0eb53e59efc1343c569ae Mon Sep 17 00:00:00 2001 From: Eric Yuan Date: Fri, 24 Jul 2026 12:55:37 -0700 Subject: [PATCH] fix(flink): fail fast when a bounded HoodieSource resumes a checkpoint with a changed commit range A bounded FLIP-27 HoodieSource read enumerates its splits once at job start. On a checkpoint or savepoint restore it reuses that persisted split set and never re-enumerates, so resuming after read.start-commit / read.end-commit were edited silently reads the checkpoint's old range with no error and no warning. Record the configured commit range in HoodieSplitEnumeratorState and compare it on restore of a bounded read, failing fast with an actionable message when it differs. An unconfigured bound is recorded as the empty string so that "recorded but unset" stays distinguishable from "not recorded at all"; a state carrying neither bound cannot be verified and is allowed through with a warning, keeping existing checkpoints restorable. Bounds are compared after normalization, so a change of case in the `earliest` sentinel or surrounding whitespace does not fail a job that is in fact reading the same data. The failure is raised from the enumerator's start() rather than at restore time. On the initial restore from a savepoint, OperatorCoordinatorHolder#resetToCheckpoint runs during ExecutionGraph construction, before lazyInitialize supplies the scheduler executor, so the failJob underneath DeferrableCoordinator#resetAndStart throws from checkInitialized() inside an unobserved whenComplete callback and the job comes up RUNNING with a coordinator that never started. By start() the context is initialized, so the failure reaches context.failJob and terminates the job. SuppressRestartsException keeps the restart strategy from looping on a mismatch that would recur on every restore. Streaming reads are deliberately left unguarded: their checkpointed offset supersedes read.start-commit by design (IncrementalInputSplits), so validating it would break legitimate resumes. Bump the enumerator state serializer 1 -> 2 for the new fields, and record the nested split serializer's own version in the payload so the two formats can evolve independently instead of the outer version being forwarded to the split deserializer. --- .../org/apache/hudi/source/HoodieSource.java | 106 ++++++++++++- .../HoodieEnumeratorStateSerializer.java | 53 ++++++- .../HoodieSplitEnumeratorState.java | 27 ++++ .../HoodieStaticSplitEnumerator.java | 49 +++++- .../apache/hudi/source/TestHoodieSource.java | 142 +++++++++++++++++ .../TestHoodieSourceCommitRangeGuard.java | 137 +++++++++++++++++ .../TestHoodieEnumeratorStateSerializer.java | 145 +++++++++++++++++- .../TestHoodieStaticSplitEnumerator.java | 42 ++++- 8 files changed, 691 insertions(+), 10 deletions(-) create mode 100644 hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieSourceCommitRangeGuard.java diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java index 7126dfd290ef7..b7805a935fe7c 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java @@ -73,6 +73,9 @@ */ @Slf4j public class HoodieSource extends FileIndexReader implements Source { + /** Recorded in place of a {@code read.*-commit} bound that is not configured. */ + private static final String UNSET = ""; + private final HoodieScanContext scanContext; private final SerializableSupplier> readerFunctionSupplier; private final SerializableComparator splitComparator; @@ -135,13 +138,30 @@ public SourceReader createReader(SourceReaderContext reade private SplitEnumerator createEnumerator( SplitEnumeratorContext enumContext, @Nullable HoodieSplitEnumeratorState enumeratorState) { + boolean streaming = scanContext.isStreaming(); + + // read.start-commit / read.end-commit as configured for this run. They are recorded in the + // checkpoint of a bounded read and re-checked on the next restore. UNSET stands in for an option + // that is not configured, so that "recorded but unset" stays distinguishable from "not recorded + // at all" (a streaming read, or a checkpoint written before this field existed). + Configuration conf = scanContext.getConf(); + Option readStartCommit = Option.of(conf.getOptional(FlinkOptions.READ_START_COMMIT).orElse(UNSET)); + Option readEndCommit = Option.of(conf.getOptional(FlinkOptions.READ_END_COMMIT).orElse(UNSET)); + // Deferred commit-range failure, raised from the enumerator's start() rather than here. See + // checkBoundedCommitRangeUnchanged for why the restore path cannot fail the job itself. + Option rangeFailure = Option.empty(); + HoodieSplitProvider splitProvider; HoodieSplitAssigner splitAssigner = HoodieSplitAssigners.createHoodieSplitAssigner( - scanContext.getConf(), enumContext.currentParallelism()); + conf, enumContext.currentParallelism()); if (enumeratorState == null) { splitProvider = new DefaultHoodieSplitProvider(splitAssigner); } else { + if (!streaming) { + rangeFailure = checkBoundedCommitRangeUnchanged( + tableName, enumeratorState, readStartCommit, readEndCommit); + } log.info( "Hoodie source restored {} splits from state for table {}", enumeratorState.getPendingSplitStates().size(), tableName); @@ -151,7 +171,7 @@ private SplitEnumerator createEnu splitProvider.onDiscoveredSplits(pendingSplits); } - if (scanContext.isStreaming()) { + if (streaming) { HoodieContinuousSplitDiscover discover = new DefaultHoodieSplitDiscover( scanContext); @@ -163,10 +183,90 @@ private SplitEnumerator createEnu List splits = createBatchHoodieSplits(); splitProvider.onDiscoveredSplits(splits); } - return new HoodieStaticSplitEnumerator(tableName, enumContext, splitProvider); + return new HoodieStaticSplitEnumerator( + tableName, enumContext, splitProvider, readStartCommit, readEndCommit, rangeFailure); } } + /** + * Returns the failure message for a bounded read being restored from a checkpoint taken under a + * different {@code read.start-commit} / {@code read.end-commit} range, or {@link Option#empty()} + * when the range is unchanged or cannot be verified. + * + *

A bounded read enumerates its splits once, at job start, and a restore reuses that persisted + * split set without re-enumerating. Resuming after the range was edited would therefore read the + * checkpoint's old range and silently ignore the configured one. + * + *

The message is returned rather than thrown, and is raised by {@link + * HoodieStaticSplitEnumerator#start()} instead. Throwing from here does not fail the job on the + * case that matters — the initial restore from a savepoint or retained checkpoint. As + * {@code OperatorCoordinatorHolder#resetToCheckpoint} documents, that first call happens during + * ExecutionGraph construction, before {@code lazyInitialize} has supplied the scheduler executor. + * {@code RecreateOnResetOperatorCoordinator$DeferrableCoordinator#resetAndStart} does catch the + * throw and call {@code cleanAndFailJob}, but the {@code failJob} underneath it begins with + * {@code checkInitialized()}, which at that point throws {@code IllegalStateException} from inside + * the unobserved {@code closingFuture.whenComplete(...)} callback. The job then comes up RUNNING + * with a coordinator that never started: no splits, no throughput, and checkpoints that never + * complete. By {@code start()} the context is initialized — + * {@code OperatorCoordinatorHolder#start()} asserts it — so the failure reaches + * {@code context.failJob} and terminates the job. + * + *

A state carrying no range at all cannot be verified and is allowed through with a warning, so + * that checkpoints written by serializer VERSION 1 stay restorable. Note that a VERSION 2 state + * written by the streaming enumerator also carries no range, so a streaming-to-bounded + * reconfiguration takes the same unverified path. + */ + @VisibleForTesting + static Option checkBoundedCommitRangeUnchanged( + String tableName, + HoodieSplitEnumeratorState state, + Option configuredStart, + Option configuredEnd) { + Option checkpointedStart = state.getReadStartCommit(); + Option checkpointedEnd = state.getReadEndCommit(); + if (!checkpointedStart.isPresent() && !checkpointedEnd.isPresent()) { + log.warn( + "Restoring bounded read for table {} from a checkpoint that records no commit range " + + "(written by serializer VERSION 1, or by a streaming read). Cannot verify it was " + + "taken with the configured range [{}, {}]; resuming anyway.", + tableName, configuredStart.orElse(UNSET), configuredEnd.orElse(UNSET)); + return Option.empty(); + } + if (boundsMatch(checkpointedStart, configuredStart) && boundsMatch(checkpointedEnd, configuredEnd)) { + return Option.empty(); + } + return Option.of(String.format( + "Refusing to resume bounded read for table %s: read.start-commit/read.end-commit changed " + + "since the checkpoint was taken.%n checkpoint range: [%s, %s]%n" + + " configured range: [%s, %s]%n" + + "A bounded read enumerates its splits once at job start and reuses them on restore, so " + + "resuming would read the checkpoint's range and ignore the configured one. To read the " + + "new range, start the job fresh instead of resuming: submit without a savepoint or " + + "retained checkpoint, or use a new checkpoint directory.", + tableName, + checkpointedStart.orElse(UNSET), checkpointedEnd.orElse(UNSET), + configuredStart.orElse(UNSET), configuredEnd.orElse(UNSET))); + } + + /** + * Whether two recorded bounds select the same data. Compared after normalization so that a purely + * cosmetic edit does not fail the job: the {@code earliest} sentinel is matched case-insensitively, + * the way {@link org.apache.hudi.configuration.OptionsResolver} matches it everywhere else, and + * surrounding whitespace is ignored. + */ + private static boolean boundsMatch(Option checkpointed, Option configured) { + return normalizeBound(checkpointed).equals(normalizeBound(configured)); + } + + private static Option normalizeBound(Option bound) { + return bound.map(value -> { + String trimmed = value.trim(); + return trimmed.equalsIgnoreCase(FlinkOptions.START_COMMIT_EARLIEST) + ? FlinkOptions.START_COMMIT_EARLIEST + : trimmed; + }); + } + @VisibleForTesting List createBatchHoodieSplits() { final Configuration flinkConf = this.scanContext.getConf(); diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieEnumeratorStateSerializer.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieEnumeratorStateSerializer.java index 74b868d965d6c..4fa61d2f1fd0c 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieEnumeratorStateSerializer.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieEnumeratorStateSerializer.java @@ -40,7 +40,14 @@ */ @Internal public class HoodieEnumeratorStateSerializer implements SimpleVersionedSerializer { - private static final int VERSION = 1; + private static final int VERSION = 2; + /** + * Version of {@link HoodieSourceSplitSerializer} that produced the nested split payloads of a + * VERSION 1 state, which did not record it. VERSION 2 onwards writes the actual version, so the + * two formats can evolve independently. + */ + private static final int LEGACY_SPLIT_SERIALIZER_VERSION = 1; + private final HoodieSourceSplitSerializer splitSerializer = new HoodieSourceSplitSerializer(); @Override @@ -53,6 +60,11 @@ public byte[] serialize(HoodieSplitEnumeratorState obj) throws IOException { try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); DataOutputStream out = new DataOutputStream(baos)) { + // Serialize the version of the nested split serializer (added in VERSION 2) so the split + // payloads below are decoded with the version that wrote them rather than with this + // serializer's own version. + out.writeInt(splitSerializer.getVersion()); + // Serialize pending split states Collection splitStates = obj.getPendingSplitStates(); out.writeInt(splitStates.size()); @@ -75,6 +87,18 @@ public byte[] serialize(HoodieSplitEnumeratorState obj) throws IOException { out.writeUTF(obj.getLastEnumeratedInstantOffset().get()); } + // Serialize readStartCommit (added in VERSION 2) + out.writeBoolean(obj.getReadStartCommit().isPresent()); + if (obj.getReadStartCommit().isPresent()) { + out.writeUTF(obj.getReadStartCommit().get()); + } + + // Serialize readEndCommit (added in VERSION 2) + out.writeBoolean(obj.getReadEndCommit().isPresent()); + if (obj.getReadEndCommit().isPresent()) { + out.writeUTF(obj.getReadEndCommit().get()); + } + out.flush(); return baos.toByteArray(); } @@ -82,9 +106,18 @@ public byte[] serialize(HoodieSplitEnumeratorState obj) throws IOException { @Override public HoodieSplitEnumeratorState deserialize(int version, byte[] serialized) throws IOException { + if (version > VERSION) { + throw new IOException( + "Cannot deserialize Hoodie enumerator state written by a newer serializer version " + + version + "; this serializer supports up to version " + VERSION); + } try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized); DataInputStream in = new DataInputStream(bais)) { + // The nested split serializer version is recorded from VERSION 2 onwards; VERSION 1 payloads + // were always written by split serializer version 1. + int splitVersion = version >= 2 ? in.readInt() : LEGACY_SPLIT_SERIALIZER_VERSION; + // Deserialize pending split states int splitCount = in.readInt(); List splitStates = new ArrayList<>(splitCount); @@ -94,7 +127,7 @@ public HoodieSplitEnumeratorState deserialize(int version, byte[] serialized) th in.readFully(splitBytes); String statusName = in.readUTF(); splitStates.add(new HoodieSourceSplitState( - splitSerializer.deserialize(version, splitBytes), + splitSerializer.deserialize(splitVersion, splitBytes), HoodieSourceSplitStatus.valueOf(statusName))); } @@ -114,7 +147,21 @@ public HoodieSplitEnumeratorState deserialize(int version, byte[] serialized) th lastEnumeratedInstantOffset = Option.empty(); } - return new HoodieSplitEnumeratorState(splitStates, lastEnumeratedInstant, lastEnumeratedInstantOffset); + // Deserialize the commit range (added in VERSION 2). VERSION 1 checkpoints do not carry it, so + // both bounds stay empty and the restore-time range check is skipped for them. + Option readStartCommit = Option.empty(); + Option readEndCommit = Option.empty(); + if (version >= 2) { + if (in.readBoolean()) { + readStartCommit = Option.of(in.readUTF()); + } + if (in.readBoolean()) { + readEndCommit = Option.of(in.readUTF()); + } + } + + return new HoodieSplitEnumeratorState( + splitStates, lastEnumeratedInstant, lastEnumeratedInstantOffset, readStartCommit, readEndCommit); } } } diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieSplitEnumeratorState.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieSplitEnumeratorState.java index b7750e1672614..0068122061f6f 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieSplitEnumeratorState.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieSplitEnumeratorState.java @@ -21,6 +21,7 @@ import org.apache.hudi.common.util.Option; import org.apache.hudi.source.split.HoodieSourceSplitState; +import lombok.AllArgsConstructor; import lombok.Value; import java.io.Serializable; @@ -30,9 +31,35 @@ * State of Hoodie split enumerator. Mainly include the states of pending splits of split provider. */ @Value +@AllArgsConstructor public class HoodieSplitEnumeratorState implements Serializable { Collection pendingSplitStates; Option lastEnumeratedInstant; Option lastEnumeratedInstantOffset; + /** + * The {@code read.start-commit} / {@code read.end-commit} bounds configured when this checkpoint + * was taken, recorded only for bounded reads. Both hold {@code Option.of("")} when the option was + * not configured, so that "recorded but unset" stays distinguishable from "not recorded at all"; + * both are {@link Option#empty()} for streaming reads and for checkpoints written by serializer + * VERSION 1, which predates this field. + * + *

A bounded read's split set is frozen at enumeration time and is NOT re-derived on restore, so + * {@code HoodieSource} compares these against the configured bounds and fails fast when they + * differ. See {@code HoodieSource#checkBoundedCommitRangeUnchanged}. + */ + Option readStartCommit; + Option readEndCommit; + + /** + * Backward-compatible constructor for callers that do not record a commit range: the streaming + * enumerator and pre-existing tests. Leaves both bounds {@link Option#empty()}. + */ + public HoodieSplitEnumeratorState( + Collection pendingSplitStates, + Option lastEnumeratedInstant, + Option lastEnumeratedInstantOffset) { + this(pendingSplitStates, lastEnumeratedInstant, lastEnumeratedInstantOffset, + Option.empty(), Option.empty()); + } } diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieStaticSplitEnumerator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieStaticSplitEnumerator.java index ec6aa955bce5d..f794042f03f17 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieStaticSplitEnumerator.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/HoodieStaticSplitEnumerator.java @@ -18,19 +18,66 @@ package org.apache.hudi.source.enumerator; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; import org.apache.hudi.source.split.HoodieSourceSplit; import org.apache.hudi.source.split.HoodieSplitProvider; +import lombok.extern.slf4j.Slf4j; import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.runtime.execution.SuppressRestartsException; /** * Static Hoodie split enumerator that only handles with a bounded number of hudi commits. */ +@Slf4j public class HoodieStaticSplitEnumerator extends AbstractHoodieSplitEnumerator { + // The read.start-commit / read.end-commit bounds this bounded read was enumerated with, persisted + // in the enumerator checkpoint so a later restore can detect that they changed. See + // HoodieSource#checkBoundedCommitRangeUnchanged. + private final Option readStartCommit; + private final Option readEndCommit; + // Deferred commit-range failure, present only when this enumerator was restored from a checkpoint + // whose range differs from the configured one. Raised from start() rather than at restore time, + // because on the initial restore from a savepoint the coordinator context is not yet initialized + // and its failJob cannot terminate the job. See HoodieSource#checkBoundedCommitRangeUnchanged. + private final Option rangeFailure; + public HoodieStaticSplitEnumerator( - String tableName, SplitEnumeratorContext enumeratorContext, HoodieSplitProvider provider) { + String tableName, + SplitEnumeratorContext enumeratorContext, + HoodieSplitProvider provider, + Option readStartCommit, + Option readEndCommit, + Option rangeFailure) { super(tableName, enumeratorContext, provider); + this.readStartCommit = readStartCommit; + this.readEndCommit = readEndCommit; + this.rangeFailure = rangeFailure; + } + + /** + * Raises the deferred commit-range failure before starting split discovery. Flink's + * single-threaded coordinator executor guarantees this runs before any other enumerator callback, + * and by this point {@code OperatorCoordinatorHolder#start()} has asserted that the coordinator + * context is initialized, so the throw reaches {@code context.failJob} and terminates the job. + * {@link SuppressRestartsException} keeps the restart strategy from looping on it: the same + * mismatch would recur on every restore from the same checkpoint. + */ + @Override + public void start() { + if (rangeFailure.isPresent()) { + log.error(rangeFailure.get()); + throw new SuppressRestartsException(new HoodieException(rangeFailure.get())); + } + super.start(); + } + + @Override + public HoodieSplitEnumeratorState snapshotState(long checkpointId) { + return new HoodieSplitEnumeratorState( + splitProvider.state(), Option.empty(), Option.empty(), readStartCommit, readEndCommit); } @Override diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieSource.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieSource.java index 1f975d078f217..eb2ff744bbc9f 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieSource.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieSource.java @@ -23,11 +23,14 @@ import org.apache.hudi.common.schema.HoodieSchema; import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.testutils.HoodieTestUtils; +import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.PartitionPathEncodeUtils; import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.configuration.HadoopConfigurations; import org.apache.hudi.index.HoodieIndex; import org.apache.hudi.index.bucket.BucketIdentifier; +import org.apache.hudi.source.enumerator.HoodieContinuousSplitEnumerator; +import org.apache.hudi.source.enumerator.HoodieSplitEnumeratorState; import org.apache.hudi.source.prune.ColumnStatsProbe; import org.apache.hudi.source.prune.PartitionPruners; import org.apache.hudi.source.reader.HoodieRecordEmitter; @@ -43,7 +46,15 @@ import org.apache.hudi.utils.TestData; import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.api.connector.source.ReaderInfo; +import org.apache.flink.api.connector.source.SourceEvent; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.api.connector.source.SplitsAssignment; import org.apache.flink.configuration.Configuration; +import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup; +import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; +import org.apache.flink.runtime.execution.SuppressRestartsException; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.data.RowData; import org.apache.flink.table.expressions.CallExpression; @@ -62,11 +73,15 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.function.BiConsumer; import java.util.function.Function; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; /** @@ -481,4 +496,131 @@ private HoodieSource createHoodieSourceWithPruner( metaClient, new HoodieRecordEmitter<>()); } + + // --------------------------------------------------------------------------------------------- + // Bounded commit-range guard: end-to-end wiring through restoreEnumerator. The detection logic + // itself is unit-tested in TestHoodieSourceCommitRangeGuard; these verify it is actually hooked up. + // --------------------------------------------------------------------------------------------- + + @Test + public void testRestoreBoundedEnumeratorFailsWhenCommitRangeChanged() throws Exception { + metaClient = HoodieTestUtils.init(tempDir.getAbsolutePath(), HoodieTableType.COPY_ON_WRITE); + conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.COPY_ON_WRITE.name()); + conf.set(FlinkOptions.READ_AS_STREAMING, false); + conf.set(FlinkOptions.READ_START_COMMIT, "20260226000000"); + + HoodieSource source = createHoodieSource(conf, metaClient); + // Checkpoint taken under a different start commit. + SplitEnumerator enumerator = + source.restoreEnumerator(new StubEnumeratorContext(), stateWithRange("20260225000000", "")); + + SuppressRestartsException ex = assertThrows(SuppressRestartsException.class, enumerator::start); + assertTrue(ex.getCause().getMessage().contains("20260225000000"), + "The failure raised from start() should name the checkpoint range"); + } + + @Test + public void testRestoreBoundedEnumeratorResumesWhenCommitRangeUnchanged() throws Exception { + metaClient = HoodieTestUtils.init(tempDir.getAbsolutePath(), HoodieTableType.COPY_ON_WRITE); + conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.COPY_ON_WRITE.name()); + conf.set(FlinkOptions.READ_AS_STREAMING, false); + conf.set(FlinkOptions.READ_START_COMMIT, "20260226000000"); + + HoodieSource source = createHoodieSource(conf, metaClient); + SplitEnumerator enumerator = + source.restoreEnumerator(new StubEnumeratorContext(), stateWithRange("20260226000000", "")); + + enumerator.start(); + // And the range is carried forward into the next checkpoint, so the guard keeps working. + assertEquals(Option.of("20260226000000"), enumerator.snapshotState(1L).getReadStartCommit()); + } + + @Test + public void testFreshBoundedEnumeratorRecordsConfiguredCommitRange() throws Exception { + metaClient = HoodieTestUtils.init(tempDir.getAbsolutePath(), HoodieTableType.COPY_ON_WRITE); + conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.COPY_ON_WRITE.name()); + conf.set(FlinkOptions.READ_AS_STREAMING, false); + TestData.writeData(TestData.DATA_SET_INSERT, conf); + metaClient.reloadActiveTimeline(); + + HoodieSource source = createHoodieSource(conf, metaClient); + SplitEnumerator enumerator = + source.createEnumerator(new StubEnumeratorContext()); + + // Neither bound is configured, so both are recorded as the empty string rather than left absent + // -- that is what lets a later restore tell "recorded but unset" from "not recorded at all". + HoodieSplitEnumeratorState state = enumerator.snapshotState(1L); + assertEquals(Option.of(""), state.getReadStartCommit()); + assertEquals(Option.of(""), state.getReadEndCommit()); + } + + @Test + public void testRestoreStreamingEnumeratorIgnoresCommitRangeChange() throws Exception { + metaClient = HoodieTestUtils.init(tempDir.getAbsolutePath(), HoodieTableType.COPY_ON_WRITE); + conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.COPY_ON_WRITE.name()); + conf.set(FlinkOptions.READ_AS_STREAMING, true); + conf.set(FlinkOptions.READ_START_COMMIT, "20260226000000"); + + HoodieSource source = createHoodieSource(conf, metaClient); + // A streaming read resumes from its checkpointed offset, which deliberately supersedes + // read.start-commit, so a changed start commit must NOT fail the job. + SplitEnumerator enumerator = + source.restoreEnumerator(new StubEnumeratorContext(), stateWithRange("20260225000000", "")); + + assertNotNull(enumerator, "Streaming restore should not be range-checked"); + assertEquals(HoodieContinuousSplitEnumerator.class, enumerator.getClass()); + } + + private static HoodieSplitEnumeratorState stateWithRange(String start, String end) { + return new HoodieSplitEnumeratorState( + Collections.emptyList(), Option.empty(), Option.empty(), Option.of(start), Option.of(end)); + } + + /** + * Minimal {@link SplitEnumeratorContext} for constructing an enumerator. Only the members the + * enumerator touches at construction and start time are implemented. + */ + private static class StubEnumeratorContext implements SplitEnumeratorContext { + + @Override + public SplitEnumeratorMetricGroup metricGroup() { + return UnregisteredMetricsGroup.createSplitEnumeratorMetricGroup(); + } + + @Override + public void sendEventToSourceReader(int subtaskId, SourceEvent event) { + } + + @Override + public int currentParallelism() { + return 1; + } + + @Override + public Map registeredReaders() { + return Collections.emptyMap(); + } + + @Override + public void assignSplits(SplitsAssignment newSplitAssignments) { + } + + @Override + public void signalNoMoreSplits(int subtask) { + } + + @Override + public void callAsync(Callable callable, BiConsumer handler) { + } + + @Override + public void callAsync( + Callable callable, BiConsumer handler, long initialDelay, long period) { + } + + @Override + public void runInCoordinatorThread(Runnable runnable) { + runnable.run(); + } + } } diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieSourceCommitRangeGuard.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieSourceCommitRangeGuard.java new file mode 100644 index 0000000000000..88f3e63f4f597 --- /dev/null +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieSourceCommitRangeGuard.java @@ -0,0 +1,137 @@ +/* + * 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.source; + +import org.apache.hudi.common.util.Option; +import org.apache.hudi.source.enumerator.HoodieSplitEnumeratorState; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests for {@link HoodieSource#checkBoundedCommitRangeUnchanged}: a bounded read enumerates its + * splits once at job start and reuses them on restore, so resuming a checkpoint taken under a + * different {@code read.start-commit} / {@code read.end-commit} range would read the old range and + * silently ignore the configured one. + * + *

Detection returns the failure message rather than throwing; it is raised from the enumerator's + * {@code start()} — see {@code TestHoodieStaticSplitEnumerator#testStartFailsJobWhenCommitRangeChanged}. + */ +public class TestHoodieSourceCommitRangeGuard { + + private static final String START = "20260226000000"; + private static final String END = "20260227000000"; + + @Test + public void testUnchangedRangeResumes() { + assertFalse(HoodieSource.checkBoundedCommitRangeUnchanged( + "table", stateWithRange(Option.of(START), Option.of(END)), Option.of(START), Option.of(END)) + .isPresent()); + } + + @Test + public void testChangedStartCommitReportsActionableFailure() { + // The motivating case: a bounded read resubmitted from a retained checkpoint with a new range. + Option failure = HoodieSource.checkBoundedCommitRangeUnchanged( + "table", + stateWithRange(Option.of(START), Option.of(END)), + Option.of("20260225000000"), + Option.of(END)); + + assertTrue(failure.isPresent(), "A changed start commit must be reported"); + assertTrue(failure.get().contains(START), "Message should show the checkpoint range"); + assertTrue(failure.get().contains("20260225000000"), "Message should show the configured range"); + assertTrue(failure.get().contains("checkpoint directory"), + "Message should explain how to run the new range"); + } + + @Test + public void testChangedEndCommitReportsFailure() { + assertTrue(HoodieSource.checkBoundedCommitRangeUnchanged( + "table", + stateWithRange(Option.of(START), Option.of(END)), + Option.of(START), + Option.of("20260228000000")) + .isPresent()); + } + + @Test + public void testNewlyConfiguredBoundReportsFailure() { + // The checkpoint recorded the range as unset (empty string, not an absent Option). Adding a + // start commit afterwards is still a range change and must be caught. + assertTrue(HoodieSource.checkBoundedCommitRangeUnchanged( + "table", + stateWithRange(Option.of(""), Option.of("")), + Option.of(START), + Option.of("")) + .isPresent()); + } + + @Test + public void testUnsetRangeOnBothSidesResumes() { + // A bounded snapshot read configures neither bound; both sides record the empty string and match. + assertEquals(Option.empty(), HoodieSource.checkBoundedCommitRangeUnchanged( + "table", stateWithRange(Option.of(""), Option.of("")), Option.of(""), Option.of(""))); + } + + @Test + public void testLegacyCheckpointWithoutRangeResumes() { + // Checkpoints written by serializer VERSION 1 carry no range at all, so nothing can be verified. + // They must still restore, otherwise the guard would break every existing bounded job. + assertFalse(HoodieSource.checkBoundedCommitRangeUnchanged( + "table", stateWithRange(Option.empty(), Option.empty()), Option.of(START), Option.of(END)) + .isPresent()); + } + + @Test + public void testEarliestSentinelIsMatchedCaseInsensitively() { + // OptionsResolver matches the `earliest` sentinel with equalsIgnoreCase everywhere else, so + // `earliest` and `EARLIEST` select the same data. Failing the job over the difference in case + // would be a false positive that no restart can clear. + assertFalse(HoodieSource.checkBoundedCommitRangeUnchanged( + "table", stateWithRange(Option.of("earliest"), Option.of("")), Option.of("EARLIEST"), Option.of("")) + .isPresent()); + } + + @Test + public void testSurroundingWhitespaceDoesNotCountAsAChange() { + assertFalse(HoodieSource.checkBoundedCommitRangeUnchanged( + "table", stateWithRange(Option.of(START), Option.of("")), Option.of(" " + START + " "), Option.of("")) + .isPresent()); + } + + @Test + public void testEarliestToExplicitCommitIsStillAChange() { + // Normalization must not blunt the guard: `earliest` and a real instant are different scopes. + assertTrue(HoodieSource.checkBoundedCommitRangeUnchanged( + "table", stateWithRange(Option.of("earliest"), Option.of("")), Option.of(START), Option.of("")) + .isPresent()); + } + + private static HoodieSplitEnumeratorState stateWithRange( + Option readStartCommit, Option readEndCommit) { + return new HoodieSplitEnumeratorState( + Collections.emptyList(), Option.empty(), Option.empty(), readStartCommit, readEndCommit); + } +} diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieEnumeratorStateSerializer.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieEnumeratorStateSerializer.java index 7ff61f9f8ef08..7da59efce0e65 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieEnumeratorStateSerializer.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieEnumeratorStateSerializer.java @@ -20,11 +20,16 @@ import org.apache.hudi.common.util.Option; import org.apache.hudi.source.split.HoodieSourceSplit; +import org.apache.hudi.source.split.HoodieSourceSplitSerializer; import org.apache.hudi.source.split.HoodieSourceSplitState; import org.apache.hudi.source.split.HoodieSourceSplitStatus; import org.junit.jupiter.api.Test; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; @@ -35,6 +40,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; /** @@ -248,7 +254,7 @@ public void testSerializeAndDeserializeWithDifferentSplitStatuses() throws IOExc @Test public void testGetVersion() { - assertEquals(1, serializer.getVersion()); + assertEquals(2, serializer.getVersion()); } @Test @@ -479,6 +485,143 @@ public void testSerializeStateWithMixedSplitStates() throws IOException { assertEquals(Integer.MAX_VALUE, deserializedStates.get(2).getSplit().getFileOffset()); } + @Test + public void testSerializeAndDeserializeCommitRange() throws IOException { + HoodieSplitEnumeratorState original = new HoodieSplitEnumeratorState( + Collections.emptyList(), + Option.empty(), + Option.empty(), + Option.of("20260226000000"), + Option.of("20260227000000") + ); + + byte[] serialized = serializer.serialize(original); + HoodieSplitEnumeratorState deserialized = serializer.deserialize(serializer.getVersion(), serialized); + + assertEquals(Option.of("20260226000000"), deserialized.getReadStartCommit()); + assertEquals(Option.of("20260227000000"), deserialized.getReadEndCommit()); + } + + @Test + public void testRecordedButUnsetCommitRangeRoundTrips() throws IOException { + // A bounded read with neither option configured records the empty string rather than an absent + // Option, so a restore can tell "recorded but unset" apart from "not recorded at all". + HoodieSplitEnumeratorState original = new HoodieSplitEnumeratorState( + Collections.emptyList(), + Option.empty(), + Option.empty(), + Option.of(""), + Option.of("") + ); + + HoodieSplitEnumeratorState deserialized = + serializer.deserialize(serializer.getVersion(), serializer.serialize(original)); + + assertEquals(Option.of(""), deserialized.getReadStartCommit()); + assertEquals(Option.of(""), deserialized.getReadEndCommit()); + } + + @Test + public void testCommitRangeDefaultsEmptyViaLegacyConstructor() throws IOException { + // The 3-arg constructor (streaming enumerator + pre-existing call sites) records no range; it + // must round-trip as absent so the restore-time range check is skipped. + HoodieSplitEnumeratorState original = new HoodieSplitEnumeratorState( + Collections.emptyList(), + Option.of("20240122120000"), + Option.empty() + ); + + HoodieSplitEnumeratorState deserialized = + serializer.deserialize(serializer.getVersion(), serializer.serialize(original)); + + assertFalse(deserialized.getReadStartCommit().isPresent()); + assertFalse(deserialized.getReadEndCommit().isPresent()); + } + + @Test + public void testDeserializeVersion1Payload() throws IOException { + // A VERSION 1 checkpoint: no nested split-serializer version at the head, no commit range at the + // tail. It must still restore, with its splits intact and the range absent. + HoodieSourceSplit split = createTestSplit(7, "file7", "/partition7"); + byte[] v1Bytes = serializeAsVersion1( + Collections.singletonList(new HoodieSourceSplitState(split, HoodieSourceSplitStatus.ASSIGNED)), + Option.of("20240122120000")); + + HoodieSplitEnumeratorState deserialized = serializer.deserialize(1, v1Bytes); + + assertEquals(1, deserialized.getPendingSplitStates().size()); + HoodieSourceSplitState restored = deserialized.getPendingSplitStates().iterator().next(); + assertEquals(7, restored.getSplit().getSplitNum()); + assertEquals("file7", restored.getSplit().getFileId()); + assertEquals(HoodieSourceSplitStatus.ASSIGNED, restored.getStatus()); + assertEquals(Option.of("20240122120000"), deserialized.getLastEnumeratedInstant()); + assertFalse(deserialized.getReadStartCommit().isPresent()); + assertFalse(deserialized.getReadEndCommit().isPresent()); + } + + @Test + public void testNestedSplitVersionIsRecordedNotInheritedFromOuterVersion() throws IOException { + // The outer state format and the nested split format version independently. VERSION 2 records + // the split serializer's own version in the payload so the splits are decoded with the version + // that wrote them, rather than with whatever the outer version happens to be. + HoodieSourceSplit split = createTestSplit(3, "file3", "/partition3"); + HoodieSplitEnumeratorState original = new HoodieSplitEnumeratorState( + Collections.singletonList(new HoodieSourceSplitState(split, HoodieSourceSplitStatus.UNASSIGNED)), + Option.empty(), + Option.empty(), + Option.of("20260226000000"), + Option.of("") + ); + + byte[] serialized = serializer.serialize(original); + + try (DataInputStream in = new DataInputStream(new ByteArrayInputStream(serialized))) { + assertEquals(new HoodieSourceSplitSerializer().getVersion(), in.readInt(), + "VERSION 2 payloads must lead with the nested split serializer version"); + } + // And the payload still round-trips end to end with that leading int in place. + HoodieSplitEnumeratorState deserialized = serializer.deserialize(2, serialized); + assertEquals(1, deserialized.getPendingSplitStates().size()); + assertEquals("file3", deserialized.getPendingSplitStates().iterator().next().getSplit().getFileId()); + assertEquals(Option.of("20260226000000"), deserialized.getReadStartCommit()); + } + + @Test + public void testDeserializeRejectsNewerVersion() throws IOException { + byte[] serialized = serializer.serialize(new HoodieSplitEnumeratorState( + Collections.emptyList(), Option.empty(), Option.empty())); + + IOException ex = assertThrows(IOException.class, + () -> serializer.deserialize(serializer.getVersion() + 1, serialized)); + assertTrue(ex.getMessage().contains("newer serializer version")); + } + + /** + * Writes the VERSION 1 payload layout: split states, then lastEnumeratedInstant and + * lastEnumeratedInstantOffset. No nested split-serializer version, no commit range. + */ + private byte[] serializeAsVersion1( + List splitStates, Option lastEnumeratedInstant) throws IOException { + HoodieSourceSplitSerializer splitSerializer = new HoodieSourceSplitSerializer(); + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(baos)) { + out.writeInt(splitStates.size()); + for (HoodieSourceSplitState splitState : splitStates) { + byte[] splitBytes = splitSerializer.serialize(splitState.getSplit()); + out.writeInt(splitBytes.length); + out.write(splitBytes); + out.writeUTF(splitState.getStatus().name()); + } + out.writeBoolean(lastEnumeratedInstant.isPresent()); + if (lastEnumeratedInstant.isPresent()) { + out.writeUTF(lastEnumeratedInstant.get()); + } + out.writeBoolean(false); + out.flush(); + return baos.toByteArray(); + } + } + /** * Helper method to create a test HoodieSourceSplit. */ diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieStaticSplitEnumerator.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieStaticSplitEnumerator.java index 70a564bd6fb4e..dc5b5e2df1b6f 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieStaticSplitEnumerator.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieStaticSplitEnumerator.java @@ -19,6 +19,7 @@ package org.apache.hudi.source.enumerator; import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; import org.apache.hudi.source.split.DefaultHoodieSplitProvider; import org.apache.hudi.source.split.HoodieSourceSplit; import org.apache.hudi.source.split.SplitRequestEvent; @@ -31,6 +32,7 @@ import org.apache.flink.api.connector.source.SplitsAssignment; import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup; import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; +import org.apache.flink.runtime.execution.SuppressRestartsException; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -45,6 +47,7 @@ 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.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -64,7 +67,7 @@ public class TestHoodieStaticSplitEnumerator { public void setUp() { context = new MockSplitEnumeratorContext(); splitProvider = new DefaultHoodieSplitProvider(new HoodieSplitNumberAssigner(2)); - enumerator = new HoodieStaticSplitEnumerator("test-table", context, splitProvider); + enumerator = new HoodieStaticSplitEnumerator("test-table", context, splitProvider, Option.of(""), Option.of(""), Option.empty()); split1 = createTestSplit(1, "file1"); split2 = createTestSplit(2, "file2"); @@ -77,6 +80,41 @@ public void testStartEnumerator() { // Verify start doesn't throw exception } + @Test + public void testSnapshotStateCarriesCommitRange() throws Exception { + // A bounded (static) enumerator must persist the commit range it was enumerated with, so a later + // restore can detect that the range changed. See HoodieSource#checkBoundedCommitRangeUnchanged. + HoodieStaticSplitEnumerator scoped = new HoodieStaticSplitEnumerator( + "test-table", context, splitProvider, Option.of("20260226000000"), Option.of(""), Option.empty()); + + HoodieSplitEnumeratorState state = scoped.snapshotState(1L); + + assertEquals(Option.of("20260226000000"), state.getReadStartCommit()); + assertEquals(Option.of(""), state.getReadEndCommit()); + } + + @Test + public void testStartFailsJobWhenCommitRangeChanged() { + // A bounded read restored from a checkpoint whose commit range changed must fail the job from + // start(), not from the restore path. On the initial restore from a savepoint, the coordinator + // context is not yet initialized (OperatorCoordinatorHolder#resetToCheckpoint runs during + // ExecutionGraph construction, before lazyInitialize), so failJob's checkInitialized() throws + // inside an unobserved whenComplete callback and the job comes up RUNNING with no enumerator. + // By start(), OperatorCoordinatorHolder#start() has asserted the context is initialized. + // SuppressRestartsException stops the restart strategy looping: the mismatch would recur on + // every restore from the same checkpoint. + String message = "Refusing to resume bounded read for table test-table: " + + "read.start-commit/read.end-commit changed since the checkpoint was taken. " + + "Start the job fresh instead of resuming (use a new checkpoint directory)."; + HoodieStaticSplitEnumerator poisoned = new HoodieStaticSplitEnumerator( + "test-table", context, splitProvider, Option.of(""), Option.of(""), Option.of(message)); + + SuppressRestartsException ex = assertThrows(SuppressRestartsException.class, poisoned::start); + assertInstanceOf(HoodieException.class, ex.getCause(), "Cause should carry the actionable details"); + assertTrue(ex.getCause().getMessage().contains("checkpoint directory"), + "The operator-facing message must survive on the cause"); + } + @Test public void testHandleSplitRequest() { splitProvider.onDiscoveredSplits(Arrays.asList(split1, split2)); @@ -226,7 +264,7 @@ public void testHandleSourceEventWithAttemptNumber() { public void testConstructorWithNullMetricGroup() { MockSplitEnumeratorContext nullMetricContext = new MockSplitEnumeratorContext(true); HoodieStaticSplitEnumerator enumeratorWithNullMetrics = - new HoodieStaticSplitEnumerator("test-table", nullMetricContext, splitProvider); + new HoodieStaticSplitEnumerator("test-table", nullMetricContext, splitProvider, Option.of(""), Option.of(""), Option.empty()); assertNotNull(enumeratorWithNullMetrics, "Enumerator should initialize without NPE when metricGroup is null");