[lake/paimon] Make PaimonSplit state survive Paimon class relocation (#3825)
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializer.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializer.java
index 147a79e..0181c94 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializer.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializer.java
@@ -18,39 +18,61 @@
package org.apache.fluss.lake.paimon.source;
+import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.DataInputViewStreamWrapper;
import org.apache.paimon.io.DataOutputViewStreamWrapper;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.utils.InstantiationUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectStreamClass;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import static org.apache.fluss.utils.Preconditions.checkState;
+
/** Serializer for paimon split. */
public class PaimonSplitSerializer implements SimpleVersionedSerializer<PaimonSplit> {
+ private static final Logger LOG = LoggerFactory.getLogger(PaimonSplitSerializer.class);
+
+ // VERSION_1 and VERSION_2 persisted DataSplit via Java serialization; kept read-only to
+ // restore state written by older versions and must never be changed.
private static final int VERSION_1 = 1;
// VERSION_2 additionally persists the partition values.
private static final int VERSION_2 = 2;
+ // VERSION_3 persists DataSplit via Paimon's own versioned binary protocol whose bytes contain
+ // no Java class names, so it survives class relocation (shading) in downstream distributions.
+ private static final int VERSION_3 = 3;
+
+ // Split type tags of VERSION_3, aligned with Paimon master's SplitSerializer type ids.
+ // Serialization must dispatch on the concrete split class: DataSplit subclasses append extra
+ // fields after the base payload (e.g. FallbackDataSplit), so deserializing them with the base
+ // DataSplit.deserialize would leave trailing bytes and shift all subsequent reads.
+ private static final int SPLIT_TYPE_DATA_SPLIT = 1;
+ private static final int SPLIT_TYPE_FALLBACK_DATA_SPLIT = 6;
@Override
public int getVersion() {
- return VERSION_2;
+ return VERSION_3;
}
@Override
public byte[] serialize(PaimonSplit paimonSplit) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
- DataSplit dataSplit = paimonSplit.dataSplit();
- InstantiationUtil.serializeObject(view, dataSplit);
+ serializeDataSplit(paimonSplit.dataSplit(), view);
view.writeBoolean(paimonSplit.isBucketUnAware());
List<String> partition = paimonSplit.partition();
view.writeInt(partition.size());
@@ -60,12 +82,66 @@
return out.toByteArray();
}
+ private void serializeDataSplit(DataSplit dataSplit, DataOutputViewStreamWrapper view)
+ throws IOException {
+ if (dataSplit.getClass() == DataSplit.class) {
+ view.writeByte(SPLIT_TYPE_DATA_SPLIT);
+ dataSplit.serialize(view);
+ } else if (dataSplit instanceof FallbackReadFileStoreTable.FallbackDataSplit) {
+ view.writeByte(SPLIT_TYPE_FALLBACK_DATA_SPLIT);
+ dataSplit.serialize(view);
+ } else {
+ // fail fast: an unknown subclass may append extra fields after the base payload,
+ // silently corrupting the stream on restore
+ throw new IOException("Unsupported DataSplit class: " + dataSplit.getClass().getName());
+ }
+ }
+
@Override
public PaimonSplit deserialize(int version, byte[] serialized) throws IOException {
+ switch (version) {
+ case VERSION_1:
+ case VERSION_2:
+ return deserializeLegacy(version, serialized);
+ case VERSION_3:
+ return deserializeV3(serialized);
+ default:
+ throw new IOException("Unsupported PaimonSplit serialization version: " + version);
+ }
+ }
+
+ private PaimonSplit deserializeV3(byte[] serialized) throws IOException {
+ ByteArrayInputStream in = new ByteArrayInputStream(serialized);
+ DataInputViewStreamWrapper view = new DataInputViewStreamWrapper(in);
+ int splitType = view.readByte();
+ DataSplit dataSplit;
+ switch (splitType) {
+ case SPLIT_TYPE_DATA_SPLIT:
+ dataSplit = DataSplit.deserialize(view);
+ break;
+ case SPLIT_TYPE_FALLBACK_DATA_SPLIT:
+ dataSplit = FallbackReadFileStoreTable.FallbackDataSplit.deserialize(view);
+ break;
+ default:
+ throw new IOException("Unsupported DataSplit type: " + splitType);
+ }
+ boolean isBucketUnAware = view.readBoolean();
+ int size = view.readInt();
+ List<String> partition = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ partition.add(view.readUTF());
+ }
+ return new PaimonSplit(dataSplit, isBucketUnAware, partition);
+ }
+
+ private PaimonSplit deserializeLegacy(int version, byte[] serialized) throws IOException {
+ LOG.debug("Restoring PaimonSplit from legacy state format (version {}).", version);
ByteArrayInputStream in = new ByteArrayInputStream(serialized);
DataSplit dataSplit;
try {
- dataSplit = InstantiationUtil.deserializeObject(in, getClass().getClassLoader());
+ RelocatingObjectInputStream ois =
+ new RelocatingObjectInputStream(in, getClass().getClassLoader());
+ dataSplit = (DataSplit) ois.readObject();
DataInputStream dis = new DataInputStream(in);
boolean isBucketUnAware = dis.readBoolean();
if (version == VERSION_1) {
@@ -73,15 +149,13 @@
// exposed through DataSplit.partition(). Preserve that old behavior.
return new PaimonSplit(
dataSplit, isBucketUnAware, readStringPartition(dataSplit.partition()));
- } else if (version == VERSION_2) {
+ } else {
int size = dis.readInt();
List<String> partition = new ArrayList<>(size);
for (int i = 0; i < size; i++) {
partition.add(dis.readUTF());
}
return new PaimonSplit(dataSplit, isBucketUnAware, partition);
- } else {
- throw new IOException("Unsupported PaimonSplit serialization version: " + version);
}
} catch (ClassNotFoundException e) {
throw new IOException("Failed to deserialize PaimonSplit", e);
@@ -99,4 +173,67 @@
}
return partitions;
}
+
+ /**
+ * An {@link java.io.ObjectInputStream} that restores legacy state written before Paimon classes
+ * were relocated (shaded): class names starting with the original {@code org.apache.paimon.}
+ * prefix in the serialization stream are remapped to the actual (possibly relocated) class
+ * names at deserialization time.
+ *
+ * <p>In non-relocated builds the actual prefix equals the original prefix, so the remapping
+ * degrades to a no-op and behavior is unchanged.
+ */
+ static class RelocatingObjectInputStream
+ extends InstantiationUtil.ClassLoaderObjectInputStream {
+
+ // The prefix must NOT appear as a plain string literal: shade plugins rewrite matching
+ // constant-pool strings, which would silently turn old and new prefixes into the same
+ // string. Build it at runtime instead.
+ private static final String ORIGINAL_PREFIX =
+ String.join(".", "org", "apache", "paimon") + ".";
+
+ // Derived from the actually loaded class, so any relocation prefix works; in
+ // non-relocated builds it equals ORIGINAL_PREFIX.
+ private static final String ACTUAL_PREFIX;
+
+ static {
+ String cls = DataSplit.class.getName();
+ // relocation only rewrites the package prefix, so the trailing part is stable; fail
+ // loudly instead of computing a wrong prefix silently
+ String suffix = "table.source.DataSplit";
+ checkState(cls.endsWith(suffix), "Unexpected DataSplit class name: %s", cls);
+ ACTUAL_PREFIX = cls.substring(0, cls.length() - suffix.length());
+ }
+
+ private final String originalPrefix;
+ private final String actualPrefix;
+
+ RelocatingObjectInputStream(InputStream in, ClassLoader cl) throws IOException {
+ this(in, cl, ORIGINAL_PREFIX, ACTUAL_PREFIX);
+ }
+
+ @VisibleForTesting
+ RelocatingObjectInputStream(
+ InputStream in, ClassLoader cl, String originalPrefix, String actualPrefix)
+ throws IOException {
+ super(in, cl);
+ this.originalPrefix = originalPrefix;
+ this.actualPrefix = actualPrefix;
+ }
+
+ @Override
+ protected Class<?> resolveClass(ObjectStreamClass desc)
+ throws IOException, ClassNotFoundException {
+ String name = desc.getName();
+ if (!actualPrefix.equals(originalPrefix) && name.startsWith(originalPrefix)) {
+ String relocated = actualPrefix + name.substring(originalPrefix.length());
+ try {
+ return Class.forName(relocated, false, classLoader);
+ } catch (ClassNotFoundException ignored) {
+ // fall back to the default resolution to keep the original exception path
+ }
+ }
+ return super.resolveClass(desc);
+ }
+ }
}
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializerTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializerTest.java
index e4c4c45..424b42b 100644
--- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializerTest.java
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializerTest.java
@@ -27,9 +27,12 @@
import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.io.DataInputViewStreamWrapper;
import org.apache.paimon.io.DataOutputViewStreamWrapper;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.InstantiationUtil;
import org.junit.jupiter.api.Test;
@@ -39,6 +42,7 @@
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
@@ -51,13 +55,13 @@
@Test
void testSerializeAndDeserialize() throws Exception {
- PaimonSplit originalPaimonSplit = createStringPartitionSplit();
- byte[] serialized = serializer.serialize(originalPaimonSplit);
- PaimonSplit deserialized = serializer.deserialize(serializer.getVersion(), serialized);
+ // partitioned pk-table split and bucket-unaware append-table split with empty partition
+ assertV3RoundTrip(createStringPartitionSplit());
- assertThat(deserialized.dataSplit()).isEqualTo(originalPaimonSplit.dataSplit());
- assertThat(deserialized.isBucketUnAware()).isEqualTo(originalPaimonSplit.isBucketUnAware());
- assertThat(deserialized.partition()).isEqualTo(originalPaimonSplit.partition());
+ PaimonSplit bucketUnAwareSplit = createBucketUnAwareSplit();
+ assertThat(bucketUnAwareSplit.isBucketUnAware()).isTrue();
+ assertThat(bucketUnAwareSplit.partition()).isEmpty();
+ assertV3RoundTrip(bucketUnAwareSplit);
}
@Test
@@ -83,22 +87,76 @@
}
@Test
- void testDeserializeVersion1PreservesStringPartition() throws Exception {
+ void testDeserializeLegacyBytes() throws Exception {
+ // VERSION_1/VERSION_2 bytes written by older versions must remain restorable after the
+ // VERSION_3 upgrade
PaimonSplit original = createStringPartitionSplit();
- byte[] serialized = serializeVersion1(original);
- PaimonSplit deserialized = serializer.deserialize(1, serialized);
+ // VERSION_1 did not store partition values, they were derived from DataSplit.partition()
+ PaimonSplit fromV1 = serializer.deserialize(1, serializeVersion1(original));
+ assertThat(fromV1.dataSplit()).isEqualTo(original.dataSplit());
+ assertThat(fromV1.isBucketUnAware()).isEqualTo(original.isBucketUnAware());
+ assertThat(fromV1.partition()).isEqualTo(Collections.singletonList("A"));
- assertThat(deserialized.dataSplit()).isEqualTo(original.dataSplit());
- assertThat(deserialized.isBucketUnAware()).isEqualTo(original.isBucketUnAware());
- assertThat(deserialized.partition()).isEqualTo(Collections.singletonList("A"));
+ PaimonSplit fromV2 = serializer.deserialize(2, serializeVersion2(original));
+ assertThat(fromV2.dataSplit()).isEqualTo(original.dataSplit());
+ assertThat(fromV2.isBucketUnAware()).isEqualTo(original.isBucketUnAware());
+ assertThat(fromV2.partition()).isEqualTo(original.partition());
}
@Test
- void testDeserializeWithInvalidData() {
+ void testDataSplitSubclassDispatch() throws Exception {
+ PaimonSplit original = createStringPartitionSplit();
+
+ // FallbackDataSplit appends the isFallback flag after the base payload; the subtype and
+ // its extra state must survive the round trip. Its constructors are private, so build one
+ // via its public deserialize API.
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
+ original.dataSplit().serialize(view);
+ view.writeBoolean(true);
+ DataSplit fallbackDataSplit =
+ FallbackReadFileStoreTable.FallbackDataSplit.deserialize(
+ new DataInputViewStreamWrapper(
+ new ByteArrayInputStream(out.toByteArray())));
+
+ PaimonSplit fallbackSplit = new PaimonSplit(fallbackDataSplit, false, original.partition());
+ PaimonSplit deserialized = assertV3RoundTrip(fallbackSplit);
+ assertThat(deserialized.dataSplit())
+ .isInstanceOf(FallbackReadFileStoreTable.FallbackDataSplit.class);
+
+ // unknown subclasses may append extra fields as well and must be rejected at write time
+ // instead of silently corrupting the stream on restore
+ PaimonSplit unknown = new PaimonSplit(new DataSplit() {}, false, Collections.emptyList());
+ assertThatThrownBy(() -> serializer.serialize(unknown))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unsupported DataSplit class");
+ }
+
+ @Test
+ void testDeserializeInvalidInput() {
byte[] invalidData = "invalid".getBytes();
assertThatThrownBy(() -> serializer.deserialize(1, invalidData))
.isInstanceOf(IOException.class);
+ assertThatThrownBy(() -> serializer.deserialize(3, invalidData))
+ .isInstanceOf(IOException.class);
+ assertThatThrownBy(() -> serializer.deserialize(99, new byte[0]))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unsupported PaimonSplit serialization version");
+ }
+
+ private PaimonSplit assertV3RoundTrip(PaimonSplit original) throws IOException {
+ byte[] serialized = serializer.serialize(original);
+ // VERSION_3 bytes must not embed the DataSplit class name, otherwise relocation (shading)
+ // in downstream distributions would break state restore again
+ assertThat(new String(serialized, StandardCharsets.ISO_8859_1))
+ .doesNotContain(DataSplit.class.getName());
+
+ PaimonSplit deserialized = serializer.deserialize(serializer.getVersion(), serialized);
+ assertThat(deserialized.dataSplit()).isEqualTo(original.dataSplit());
+ assertThat(deserialized.isBucketUnAware()).isEqualTo(original.isBucketUnAware());
+ assertThat(deserialized.partition()).isEqualTo(original.partition());
+ return deserialized;
}
private PaimonSplit createStringPartitionSplit() throws Exception {
@@ -127,6 +185,24 @@
return plan.get(0);
}
+ private PaimonSplit createBucketUnAwareSplit() throws Exception {
+ // an append table without primary key and bucket key is bucket-unaware
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "bucket_unaware_table");
+ Schema.Builder builder =
+ Schema.newBuilder().column("c1", DataTypes.INT()).column("c2", DataTypes.STRING());
+ createTable(tablePath, builder.build());
+ Table table = getTable(tablePath);
+
+ GenericRow record1 = GenericRow.of(12, BinaryString.fromString("a"));
+ writeRecord(tablePath, Collections.singletonList(record1));
+ Snapshot snapshot = table.latestSnapshot().get();
+
+ LakeSource<PaimonSplit> lakeSource = lakeStorage.createLakeSource(tablePath);
+ List<PaimonSplit> plan = lakeSource.createPlanner(snapshot::id).plan();
+
+ return plan.get(0);
+ }
+
private byte[] serializeVersion1(PaimonSplit paimonSplit) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
@@ -134,4 +210,17 @@
view.writeBoolean(paimonSplit.isBucketUnAware());
return out.toByteArray();
}
+
+ private byte[] serializeVersion2(PaimonSplit paimonSplit) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
+ InstantiationUtil.serializeObject(view, paimonSplit.dataSplit());
+ view.writeBoolean(paimonSplit.isBucketUnAware());
+ List<String> partition = paimonSplit.partition();
+ view.writeInt(partition.size());
+ for (String value : partition) {
+ view.writeUTF(value);
+ }
+ return out.toByteArray();
+ }
}
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/RelocatingObjectInputStreamTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/RelocatingObjectInputStreamTest.java
new file mode 100644
index 0000000..c125aa1
--- /dev/null
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/RelocatingObjectInputStreamTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.fluss.lake.paimon.source;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test case for {@link PaimonSplitSerializer.RelocatingObjectInputStream}. */
+class RelocatingObjectInputStreamTest {
+
+ private static final String ORIGINAL_PREFIX =
+ org.apache.fluss.lake.paimon.source.original.Probe.class.getPackage().getName() + ".";
+ private static final String RELOCATED_PREFIX =
+ org.apache.fluss.lake.paimon.source.relocated.Probe.class.getPackage().getName() + ".";
+
+ @Test
+ void testResolveClass() throws Exception {
+ // the stream always carries the original class name, as written by a non-relocated build
+ byte[] bytes =
+ javaSerialize(new org.apache.fluss.lake.paimon.source.original.Probe("hello"));
+
+ // relocated build: remapped to the relocated class, field values preserved
+ Object remapped = deserialize(bytes, ORIGINAL_PREFIX, RELOCATED_PREFIX);
+ assertThat(remapped)
+ .isInstanceOf(org.apache.fluss.lake.paimon.source.relocated.Probe.class);
+ assertThat(((org.apache.fluss.lake.paimon.source.relocated.Probe) remapped).value())
+ .isEqualTo("hello");
+
+ // non-relocated build (equal prefixes): remapping is a no-op
+ assertThat(deserialize(bytes, ORIGINAL_PREFIX, ORIGINAL_PREFIX))
+ .isInstanceOf(org.apache.fluss.lake.paimon.source.original.Probe.class);
+
+ // remapped class missing: falls back to default resolution by the original name
+ assertThat(deserialize(bytes, ORIGINAL_PREFIX, ORIGINAL_PREFIX + "nonexistent."))
+ .isInstanceOf(org.apache.fluss.lake.paimon.source.original.Probe.class);
+ }
+
+ private Object deserialize(byte[] bytes, String originalPrefix, String actualPrefix)
+ throws Exception {
+ try (PaimonSplitSerializer.RelocatingObjectInputStream in =
+ new PaimonSplitSerializer.RelocatingObjectInputStream(
+ new ByteArrayInputStream(bytes),
+ getClass().getClassLoader(),
+ originalPrefix,
+ actualPrefix)) {
+ return in.readObject();
+ }
+ }
+
+ private static byte[] javaSerialize(Object object) throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+ oos.writeObject(object);
+ }
+ return baos.toByteArray();
+ }
+}
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/original/Probe.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/original/Probe.java
new file mode 100644
index 0000000..3b85b70
--- /dev/null
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/original/Probe.java
@@ -0,0 +1,41 @@
+/*
+ * 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.fluss.lake.paimon.source.original;
+
+import java.io.Serializable;
+
+/**
+ * Test probe simulating a serializable class before relocation (shading). Its counterpart {@link
+ * org.apache.fluss.lake.paimon.source.relocated.Probe} has the same simple name and
+ * serialVersionUID but lives in a different package.
+ */
+public class Probe implements Serializable {
+
+ private static final long serialVersionUID = 42L;
+
+ private final String value;
+
+ public Probe(String value) {
+ this.value = value;
+ }
+
+ public String value() {
+ return value;
+ }
+}
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/relocated/Probe.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/relocated/Probe.java
new file mode 100644
index 0000000..e129fe8
--- /dev/null
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/relocated/Probe.java
@@ -0,0 +1,41 @@
+/*
+ * 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.fluss.lake.paimon.source.relocated;
+
+import java.io.Serializable;
+
+/**
+ * Test probe simulating a serializable class after relocation (shading). Its counterpart {@link
+ * org.apache.fluss.lake.paimon.source.original.Probe} has the same simple name and serialVersionUID
+ * but lives in a different package.
+ */
+public class Probe implements Serializable {
+
+ private static final long serialVersionUID = 42L;
+
+ private final String value;
+
+ public Probe(String value) {
+ this.value = value;
+ }
+
+ public String value() {
+ return value;
+ }
+}
