fix(flink): fail fast when a bounded HoodieSource resumes a checkpoin…

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java

@@ -73,6 +73,9 @@
  */
 @Slf4j
 public class HoodieSource<T> extends FileIndexReader implements Source<T, HoodieSourceSplit, HoodieSplitEnumeratorState> {
+  /** 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<SplitReaderFunction<T>> readerFunctionSupplier;
   private final SerializableComparator<HoodieSourceSplit> splitComparator;
@@ -135,13 +138,30 @@ public SourceReader<T, HoodieSourceSplit> createReader(SourceReaderContext reade
   private SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> createEnumerator(
       SplitEnumeratorContext<HoodieSourceSplit> 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<String> readStartCommit = Option.of(conf.getOptional(FlinkOptions.READ_START_COMMIT).orElse(UNSET));
+    Option<String> 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<String> 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<HoodieSourceSplit, HoodieSplitEnumeratorState> createEnu
       splitProvider.onDiscoveredSplits(pendingSplits);
     }
 
-    if (scanContext.isStreaming()) {
+    if (streaming) {
       HoodieContinuousSplitDiscover discover = new DefaultHoodieSplitDiscover(
           scanContext);
 
@@ -163,10 +183,90 @@ private SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> createEnu
         List<HoodieSourceSplit> 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.
+   *
+   * <p>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.
+   *
+   * <p>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.
+   *
+   * <p>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<String> checkBoundedCommitRangeUnchanged(
+      String tableName,
+      HoodieSplitEnumeratorState state,
+      Option<String> configuredStart,
+      Option<String> configuredEnd) {
+    Option<String> checkpointedStart = state.getReadStartCommit();
+    Option<String> 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<String> checkpointed, Option<String> configured) {
+    return normalizeBound(checkpointed).equals(normalizeBound(configured));
+  }
+
+  private static Option<String> normalizeBound(Option<String> bound) {
+    return bound.map(value -> {
+      String trimmed = value.trim();
+      return trimmed.equalsIgnoreCase(FlinkOptions.START_COMMIT_EARLIEST)
+          ? FlinkOptions.START_COMMIT_EARLIEST
+          : trimmed;
+    });
+  }
+
   @VisibleForTesting
   List<HoodieSourceSplit> createBatchHoodieSplits() {
     final Configuration flinkConf = this.scanContext.getConf();

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<HoodieSplitEnumeratorState> {
-  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<HoodieSourceSplitState> splitStates = obj.getPendingSplitStates();
       out.writeInt(splitStates.size());
@@ -75,16 +87,37 @@ 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();
     }
   }
 
   @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<HoodieSourceSplitState> 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<String> readStartCommit = Option.empty();
+      Option<String> 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);
     }
   }
 }

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<HoodieSourceSplitState> pendingSplitStates;
   Option<String> lastEnumeratedInstant;
   Option<String> 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.
+   *
+   * <p>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<String> readStartCommit;
+  Option<String> 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<HoodieSourceSplitState> pendingSplitStates,
+      Option<String> lastEnumeratedInstant,
+      Option<String> lastEnumeratedInstantOffset) {
+    this(pendingSplitStates, lastEnumeratedInstant, lastEnumeratedInstantOffset,
+        Option.empty(), Option.empty());
+  }
 }

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<String> readStartCommit;
+  private final Option<String> 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<String> rangeFailure;
+
   public HoodieStaticSplitEnumerator(
-      String tableName, SplitEnumeratorContext<HoodieSourceSplit> enumeratorContext, HoodieSplitProvider provider) {
+      String tableName,
+      SplitEnumeratorContext<HoodieSourceSplit> enumeratorContext,
+      HoodieSplitProvider provider,
+      Option<String> readStartCommit,
+      Option<String> readEndCommit,
+      Option<String> 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