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");