hudi-agent commented on code in PR #19376:
URL: https://github.com/apache/hudi/pull/19376#discussion_r3669554472
##########
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;
Review Comment:
🤖 nit: could you rename `rangeFailure` to `rangeFailureMessage`? The field
holds a pre-formatted error string, and without that hint a reader has to look
at the type (`Option<String>`) and the surrounding comment to figure out it's a
message rather than, say, an exception or enum value.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieEnumeratorStateSerializer.java:
##########
@@ -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<HoodieSourceSplitState> splitStates, Option<String>
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);
Review Comment:
🤖 nit: the bare `out.writeBoolean(false)` here is hard to follow without a
comment — could you add something like `// lastEnumeratedInstantOffset: absent`
so a reader doesn't have to cross-reference the main serializer to know what
field this writes?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
