Skip to content

[hotfix] Fix union read can't restore issue #2321

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
luoyuxia merged 3 commits into from
Jan 8, 2026
Merged

[hotfix] Fix union read can't restore issue #2321

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ccacaaf
[hotfix] Fix union read can't restore issue
luoyuxia Jan 7, 2026
d0fc306
address comment v1
luoyuxia Jan 7, 2026
9c4cf8c
address comment again
luoyuxia Jan 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 52 additions & 11 deletions ...c/main/java/org/apache/fluss/flink/source/state/FlussSourceEnumeratorStateSerializer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

package org.apache.fluss.flink.source.state;

import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.flink.source.split.SourceSplitBase;
import org.apache.fluss.flink.source.split.SourceSplitSerializer;
import org.apache.fluss.lake.source.LakeSource;
Expand All @@ -37,17 +38,21 @@
import java.util.Map;
import java.util.Set;

import static org.apache.fluss.utils.Preconditions.checkNotNull;

@loserwang1024 loserwang1024 Jan 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would like to add what's the different with different version? such as:
image

/** A serializer for {@link SourceEnumeratorState}. */
public class FlussSourceEnumeratorStateSerializer
implements SimpleVersionedSerializer<SourceEnumeratorState> {

@Nullable private final LakeSource<LakeSplit> lakeSource;

private static final int VERSION_0 = 0;
private static final int VERSION_1 = 1;

private static final ThreadLocal<DataOutputSerializer> SERIALIZER_CACHE =
ThreadLocal.withInitial(() -> new DataOutputSerializer(64));

private static final int CURRENT_VERSION = VERSION_0;
private static final int CURRENT_VERSION = VERSION_1;

public FlussSourceEnumeratorStateSerializer(LakeSource<LakeSplit> lakeSource) {
this.lakeSource = lakeSource;
Expand All @@ -61,11 +66,28 @@ public int getVersion() {
@Override
public byte[] serialize(SourceEnumeratorState state) throws IOException {
final DataOutputSerializer out = SERIALIZER_CACHE.get();

// serialize assign bucket and partitions
serializeAssignBucketAndPartitions(
out, state.getAssignedBuckets(), state.getAssignedPartitions());

// serialize remain hybrid lake splits
serializeRemainingHybridLakeFlussSplits(out, state);

final byte[] result = out.getCopyOfBuffer();
out.clear();
return result;
}

private void serializeAssignBucketAndPartitions(
DataOutputSerializer out,
Set<TableBucket> assignedBuckets,
Map<Long, String> assignedPartitions)
throws IOException {
// write assigned buckets
out.writeInt(state.getAssignedBuckets().size());
for (TableBucket tableBucket : state.getAssignedBuckets()) {
out.writeInt(assignedBuckets.size());
for (TableBucket tableBucket : assignedBuckets) {
out.writeLong(tableBucket.getTableId());

// write partition
// if partition is not null
if (tableBucket.getPartitionId() != null) {
Expand All @@ -78,24 +100,29 @@ public byte[] serialize(SourceEnumeratorState state) throws IOException {
out.writeInt(tableBucket.getBucket());
}
// write assigned partitions
out.writeInt(state.getAssignedPartitions().size());
for (Map.Entry<Long, String> entry : state.getAssignedPartitions().entrySet()) {
out.writeInt(assignedPartitions.size());
for (Map.Entry<Long, String> entry : assignedPartitions.entrySet()) {
out.writeLong(entry.getKey());
out.writeUTF(entry.getValue());
}
}

@VisibleForTesting
protected byte[] serializeV0(SourceEnumeratorState state) throws IOException {
final DataOutputSerializer out = SERIALIZER_CACHE.get();
serializeAssignBucketAndPartitions(
out, state.getAssignedBuckets(), state.getAssignedPartitions());
if (lakeSource != null) {
serializeRemainingHybridLakeFlussSplits(out, state);
}

final byte[] result = out.getCopyOfBuffer();
out.clear();
return result;
}

@Override
public SourceEnumeratorState deserialize(int version, byte[] serialized) throws IOException {
if (version != VERSION_0) {
if (version != VERSION_0 && version != CURRENT_VERSION) {

@loserwang1024 loserwang1024 Jan 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

version != VERSION_0 && version != CURRENT_VERSION

When we support version 3 later, state with version 3 cannot downgrade. We have met this problem in many connectors. @leonardBang , CC, should we stop higher version's state?

And I also found a problem : we can never downgrade later to version 1. Because in later code will version 1 state: version != VERSION_0 is always true. Maybe we should add this to upgrade document later.

@luoyuxia luoyuxia Jan 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

1: so, what's the best pratice in here? Always make the version check pass and make sure version3 compitable with v2? From kafka connector code, seems the kafka connector will still have the problem?

2: what do you mean by saying we can never downgrade later to version 1? Seems it still the same problem to 1?

@loserwang1024 loserwang1024 Jan 8, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I understand it. But in your follower comment "for VERSION_1 and later" maybe inconsistency.

image

Maybe you can use switch.. case as kafka does. It's clearer
image

@luoyuxia luoyuxia Jan 8, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree, that's clearer

throw new IOException("Unknown version or corrupt state: " + version);
}
final DataInputDeserializer in = new DataInputDeserializer(serialized);
Expand Down Expand Up @@ -124,8 +151,18 @@ public SourceEnumeratorState deserialize(int version, byte[] serialized) throws
}

List<SourceSplitBase> remainingHybridLakeFlussSplits = null;
if (lakeSource != null) {
// todo: add a ut for serialize remaining hybrid lake fluss splits

if (version == VERSION_0) {
// For VERSION_0, deserialize remaining hybrid lake Fluss splits only when lakeSource is
// not null.
if (lakeSource != null) {
remainingHybridLakeFlussSplits = deserializeRemainingHybridLakeFlussSplits(in);
}
} else {
// For VERSION_1 and later, always attempt to deserialize remaining hybrid lake/Fluss
// splits. The serialized state encodes their presence via a boolean flag, so this
// logic no longer depends on the lakeSource flag. This unconditional deserialization
// is the intended behavior change compared to VERSION_0.
remainingHybridLakeFlussSplits = deserializeRemainingHybridLakeFlussSplits(in);
}

Expand Down Expand Up @@ -160,7 +197,11 @@ private List<SourceSplitBase> deserializeRemainingHybridLakeFlussSplits(
if (in.readBoolean()) {
int numSplits = in.readInt();
List<SourceSplitBase> splits = new ArrayList<>(numSplits);
SourceSplitSerializer sourceSplitSerializer = new SourceSplitSerializer(lakeSource);
SourceSplitSerializer sourceSplitSerializer =
new SourceSplitSerializer(
checkNotNull(
lakeSource,
"lake source must not be null when there are hybrid lake splits."));
int version = in.readInt();
for (int i = 0; i < numSplits; i++) {
int splitSizeInBytes = in.readInt();
Expand Down
6 changes: 4 additions & 2 deletions ...flink-common/src/main/java/org/apache/fluss/flink/source/state/SourceEnumeratorState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,12 +73,14 @@ public boolean equals(Object o) {
}
SourceEnumeratorState that = (SourceEnumeratorState) o;
return Objects.equals(assignedBuckets, that.assignedBuckets)
&& Objects.equals(assignedPartitions, that.assignedPartitions);
&& Objects.equals(assignedPartitions, that.assignedPartitions)
&& Objects.equals(
remainingHybridLakeFlussSplits, that.remainingHybridLakeFlussSplits);
}

@Override
public int hashCode() {
return Objects.hash(assignedBuckets, assignedPartitions);
return Objects.hash(assignedBuckets, assignedPartitions, remainingHybridLakeFlussSplits);
}

@Override
Expand Down
62 changes: 62 additions & 0 deletions ...rc/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,4 +92,66 @@ void testPendingSplitsCheckpointSerde() throws Exception {
/* check deserialized is equal to the original */
assertThat(deserializedSourceEnumeratorState).isEqualTo(sourceEnumeratorState);
}

@Test
void testV0Compatibility() throws Exception {
// serialize with v0,
int version = 0;
// test with lake source = null
FlussSourceEnumeratorStateSerializer serializer =
new FlussSourceEnumeratorStateSerializer(null);

Set<TableBucket> assignedBuckets =
new HashSet<>(Arrays.asList(new TableBucket(1, 0), new TableBucket(1, 4L, 1)));
Map<Long, String> assignedPartitions = new HashMap<>();
assignedPartitions.put(1L, "partition1");
assignedPartitions.put(2L, "partition2");
SourceEnumeratorState sourceEnumeratorState =
new SourceEnumeratorState(assignedBuckets, assignedPartitions, null);
byte[] serialized = serializer.serializeV0(sourceEnumeratorState);

// then deserialize
SourceEnumeratorState deserializedSourceEnumeratorState =
serializer.deserialize(version, serialized);
assertThat(deserializedSourceEnumeratorState).isEqualTo(sourceEnumeratorState);

// test with lake source is not null
serializer = new FlussSourceEnumeratorStateSerializer(new TestingLakeSource());
List<SourceSplitBase> remainingHybridLakeFlussSplits = new ArrayList<>();
// Add a LogSplit
TableBucket logSplitBucket = new TableBucket(1, 0);
LogSplit logSplit = new LogSplit(logSplitBucket, null, 100L);
remainingHybridLakeFlussSplits.add(logSplit);
sourceEnumeratorState =
new SourceEnumeratorState(
assignedBuckets, assignedPartitions, remainingHybridLakeFlussSplits);

serialized = serializer.serializeV0(sourceEnumeratorState);

// then deserialize
deserializedSourceEnumeratorState = serializer.deserialize(version, serialized);
assertThat(deserializedSourceEnumeratorState).isEqualTo(sourceEnumeratorState);
Comment thread
luoyuxia marked this conversation as resolved.
}

@Test
void testInconsistentLakeSourceSerde() throws Exception {
Comment thread
luoyuxia marked this conversation as resolved.
// test serialize with null lake source
FlussSourceEnumeratorStateSerializer serializer =
new FlussSourceEnumeratorStateSerializer(null);

Set<TableBucket> assignedBuckets =
new HashSet<>(Arrays.asList(new TableBucket(1, 0), new TableBucket(1, 4L, 1)));
Map<Long, String> assignedPartitions = new HashMap<>();
assignedPartitions.put(1L, "partition1");
assignedPartitions.put(2L, "partition2");
SourceEnumeratorState sourceEnumeratorState =
new SourceEnumeratorState(assignedBuckets, assignedPartitions, null);
byte[] serialized = serializer.serialize(sourceEnumeratorState);

// test deserialize with nonnull lake source
serializer = new FlussSourceEnumeratorStateSerializer(new TestingLakeSource());
SourceEnumeratorState deserializedSourceEnumeratorState =
serializer.deserialize(serializer.getVersion(), serialized);
assertThat(deserializedSourceEnumeratorState).isEqualTo(sourceEnumeratorState);
Comment thread
luoyuxia marked this conversation as resolved.
}
}