minor fix

luoyuxia · luoyuxia · commit c29d0f680cf6 · 2025-12-26T09:59:40.000+08:00
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/CommittedLakeSnapshot.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/CommittedLakeSnapshot.java
@@ -27,6 +27,7 @@
 public class CommittedLakeSnapshot {
 
     private final long lakeSnapshotId;
+
     private final Map<String, String> snapshotProperties;
 
     public CommittedLakeSnapshot(long lakeSnapshotId, Map<String, String> snapshotProperties) {
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeCommitter.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeCommitter.java
@@ -36,6 +36,12 @@
 @PublicEvolving
 public interface LakeCommitter<WriteResult, CommittableT> extends AutoCloseable {
 
+    /**
+     * The property key used to store the file path of lake table bucket offsets in snapshot
+     * properties.
+     */
+    String FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY = "fluss-offsets";
+
     /**
      * Converts a list of write results to a committable object.
      *
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java
@@ -56,7 +56,7 @@
  * information in Fluss:
  *
  * <ul>
- *   <li><b>Prepare phase</b> ({@link #prepareCommit}): Sends log end offsets to the FLuss cluster,
+ *   <li><b>Prepare phase</b> ({@link #prepareCommit}): Sends log end offsets to the Fluss cluster,
  *       which merges them with the previous log end offsets and stores the merged snapshot data in
  *       a file. Returns the file path where the snapshot metadata is stored.
  *   <li><b>Commit phase</b> ({@link #commit}): Sends the lake snapshot metadata (including snapshot
@@ -112,7 +112,7 @@ String prepareCommit(long tableId, TablePath tablePath, Map<TableBucket, Long> l
             if (prepareCommitResp.hasErrorCode()) {
                 throw ApiError.fromErrorMessage(prepareCommitResp).exception();
             } else {
-                return checkNotNull(prepareCommitResp).getLakeTableSnapshotFilePath();
+                return checkNotNull(prepareCommitResp).getLakeTableBucketOffsetsPath();
             }
         } catch (Exception e) {
             throw new IOException(
@@ -126,7 +126,7 @@ String prepareCommit(long tableId, TablePath tablePath, Map<TableBucket, Long> l
     void commit(
             long tableId,
             long lakeSnapshotId,
-            String lakeSnapshotPath,
+            String lakeBucketOffsetsPath,
             Map<TableBucket, Long> logEndOffsets,
             Map<TableBucket, Long> logMaxTieredTimestamps)
             throws IOException {
@@ -135,7 +135,7 @@ void commit(
                     toCommitLakeTableSnapshotRequest(
                             tableId,
                             lakeSnapshotId,
-                            lakeSnapshotPath,
+                            lakeBucketOffsetsPath,
                             logEndOffsets,
                             logMaxTieredTimestamps);
             List<PbCommitLakeTableSnapshotRespForTable> commitLakeTableSnapshotRespForTables =
@@ -200,15 +200,15 @@ private PrepareCommitLakeTableSnapshotRequest toPrepareCommitLakeTableSnapshotRe
      *
      * @param tableId the table ID
      * @param snapshotId the lake snapshot ID
-     * @param lakeSnapshotPath the file path where the snapshot metadata is stored
+     * @param bucketOffsetsPath the file path where the bucket offsets is stored
      * @param logEndOffsets the log end offsets for each bucket
      * @param logMaxTieredTimestamps the max tiered timestamps for each bucket
      * @return the commit request
      */
     private CommitLakeTableSnapshotRequest toCommitLakeTableSnapshotRequest(
             long tableId,
             long snapshotId,
-            String lakeSnapshotPath,
+            String bucketOffsetsPath,
             Map<TableBucket, Long> logEndOffsets,
             Map<TableBucket, Long> logMaxTieredTimestamps) {
         CommitLakeTableSnapshotRequest commitLakeTableSnapshotRequest =
@@ -220,8 +220,8 @@ private CommitLakeTableSnapshotRequest toCommitLakeTableSnapshotRequest(
         pbLakeTableSnapshotMetadata.setSnapshotId(snapshotId);
         pbLakeTableSnapshotMetadata.setTableId(tableId);
         // tiered snapshot file path is equal to readable snapshot currently
-        pbLakeTableSnapshotMetadata.setTieredSnapshotFilePath(lakeSnapshotPath);
-        pbLakeTableSnapshotMetadata.setReadableSnapshotFilePath(lakeSnapshotPath);
+        pbLakeTableSnapshotMetadata.setTieredBucketOffsetsFilePath(bucketOffsetsPath);
+        pbLakeTableSnapshotMetadata.setReadableBucketOffsetsFilePath(bucketOffsetsPath);
 
         // Add PbLakeTableSnapshotInfo for metrics reporting (to notify tablet servers about
         // synchronized log end offsets and max timestamps)
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java
@@ -57,7 +57,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
 import static org.apache.fluss.utils.Preconditions.checkState;
 
 /**
@@ -230,22 +230,21 @@ private Committable commitWriteResults(
                             ? null
                             : flussCurrentLakeSnapshot.getSnapshotId());
 
-            // get the lake snapshot file storing the log end offsets
-            String lakeSnapshotMetadataFile =
+            // get the lake bucket offsets file storing the log end offsets
+            String lakeBucketOffsetsFile =
                     flussTableLakeSnapshotCommitter.prepareCommit(
                             tableId, tablePath, logEndOffsets);
 
-            // record the lake snapshot metadata file to snapshot property
+            // record the lake snapshot bucket offsets file to snapshot property
             long committedSnapshotId =
                     lakeCommitter.commit(
                             committable,
                             Collections.singletonMap(
-                                    FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
-                                    lakeSnapshotMetadataFile));
+                                    FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, lakeBucketOffsetsFile));
             flussTableLakeSnapshotCommitter.commit(
                     tableId,
                     committedSnapshotId,
-                    lakeSnapshotMetadataFile,
+                    lakeBucketOffsetsFile,
                     logEndOffsets,
                     logMaxTieredTimestamps);
             return committable;
@@ -303,7 +302,7 @@ private void checkFlussNotMissingLakeSnapshot(
             // since this code path should be rare, we do not consider backward compatibility
             // and throw IllegalStateException directly
             String trimmedPath = lakeSnapshotOffsetPath.trim();
-            if (trimmedPath.startsWith("{")) {
+            if (trimmedPath.contains("{")) {
                 throw new IllegalStateException(
                         String.format(
                                 "The %s field in snapshot property is a JSON string (tiered by v0.8), "
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
@@ -617,7 +617,7 @@ void testPartitionsExpiredInFlussButExistInLake(
                                 new TableBucket(tableId, hybridPartitionId, 1), lakeEndOffset,
                                 new TableBucket(tableId, hybridPartitionId, 2), lakeEndOffset));
         LakeTableHelper lakeTableHelper = new LakeTableHelper(zooKeeperClient, tempDir.toString());
-        lakeTableHelper.upsertLakeTableV1(tableId, lakeTableSnapshot);
+        lakeTableHelper.upsertLakeTable(tableId, lakeTableSnapshot);
 
         // Create PartitionInfo for lake partitions
         List<PartitionInfo> lakePartitionInfos = new ArrayList<>();
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java
@@ -55,7 +55,7 @@
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
 import static org.apache.fluss.record.TestData.DATA1_PARTITIONED_TABLE_DESCRIPTOR;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java
@@ -78,7 +78,7 @@
 import java.util.TreeSet;
 
 import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL;
-import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
 import static org.apache.fluss.lake.iceberg.utils.IcebergConversions.toIceberg;
 import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
 import static org.apache.fluss.testutils.DataTestUtils.row;
diff --git a/fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringITCase.java b/fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringITCase.java
@@ -45,7 +45,7 @@
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
 import static org.apache.fluss.testutils.DataTestUtils.row;
 import static org.assertj.core.api.Assertions.assertThat;
 
diff --git a/fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java b/fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java
@@ -64,7 +64,7 @@
 import java.util.Map;
 import java.util.stream.Stream;
 
-import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** The UT for tiering to Lance via {@link LanceLakeTieringFactory}. */
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/FlinkPaimonTieringTestBase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/FlinkPaimonTieringTestBase.java
@@ -71,7 +71,7 @@
 import java.util.Set;
 
 import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL;
-import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
 import static org.apache.fluss.testutils.DataTestUtils.row;
 import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
 import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil;
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java
@@ -71,7 +71,7 @@
 import java.util.Map;
 import java.util.stream.Stream;
 
-import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
 import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
 import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
 import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java
@@ -211,12 +211,6 @@ CompletableFuture<DescribeClusterConfigsResponse> describeClusterConfigs(
      *   <li>Returns the file path where the snapshot is stored
      * </ul>
      *
-     * <p>The returned file path points to a file that stores the bucket log end offset information
-     * for the table. This file path will be used in the subsequent commit phase to reference the
-     * stored snapshot. If any error occurs during processing for a specific table, an error
-     * response will be set for that table in the response, while other tables will continue to be
-     * processed.
-     *
      * @param request the request containing lake table snapshot information for one or more tables
      * @return a future that completes with a response containing the file paths where snapshots
      *     (containing bucket log end offset information) are stored, or error information for
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -958,8 +958,8 @@ message PbDescribeConfig {
 message PbLakeTableSnapshotMetadata {
   required int64 table_id = 1;
   required int64 snapshot_id = 2;
-  required string tiered_snapshot_file_path = 3;
-  optional string readable_snapshot_file_path = 4;
+  required string tiered_bucket_offsets_file_path = 3;
+  optional string readable_bucket_offsets_file_path = 4;
 }
 
 message PbLakeTableSnapshotInfo {
@@ -982,7 +982,7 @@ message PbLakeTableOffsetForBucket {
 }
 
 message PbPrepareCommitLakeTableRespForTable {
-  optional string lake_table_snapshot_file_path = 1;
+  optional string lake_table_bucket_offsets_path = 1;
   optional int32 error_code = 2;
   optional string error_message = 3;
 }
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
@@ -633,7 +633,7 @@ public CompletableFuture<PrepareCommitLakeTableSnapshotResponse> prepareCommitLa
                                 FsPath fsPath =
                                         lakeTableHelper.storeLakeTableBucketOffsets(
                                                 tablePath, tableBucketOffsets);
-                                pbPrepareCommitLakeTableRespForTable.setLakeTableSnapshotFilePath(
+                                pbPrepareCommitLakeTableRespForTable.setLakeTableBucketOffsetsPath(
                                         fsPath.toString());
                             } catch (Exception e) {
                                 Errors error = ApiError.fromThrowable(e).error();
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -1305,7 +1305,7 @@ private void handleCommitLakeTableSnapshotV1(
                                 }
 
                                 // this involves IO operation (ZK), so we do it in ioExecutor
-                                lakeTableHelper.upsertLakeTableV1(
+                                lakeTableHelper.upsertLakeTable(
                                         tableId, lakeTableSnapshotEntry.getValue());
                             } catch (Exception e) {
                                 ApiError error = ApiError.fromThrowable(e);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -173,7 +173,6 @@
 import org.apache.fluss.utils.json.DataTypeJsonSerde;
 import org.apache.fluss.utils.json.JsonSerdeUtils;
 import org.apache.fluss.utils.json.TableBucketOffsets;
-import org.apache.fluss.utils.types.Tuple2;
 
 import javax.annotation.Nullable;
 
@@ -190,7 +189,6 @@
 import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.Set;
-import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -1556,17 +1554,28 @@ public static CommitLakeTableSnapshotData getCommitLakeTableSnapshotData(
         Map<TableBucket, Long> tableBucketsMaxTimestamp = new HashMap<>();
         for (PbLakeTableSnapshotInfo pbLakeTableSnapshotInfo : request.getTablesReqsList()) {
             long tableId = pbLakeTableSnapshotInfo.getTableId();
-            Consumer<Tuple2<TableBucket, PbLakeTableOffsetForBucket>>
-                    pbLakeTableOffsetForBucketConsumer =
-                            (tableBucketAndOffset -> {
-                                if (tableBucketAndOffset.f1.hasMaxTimestamp()) {
-                                    tableBucketsMaxTimestamp.put(
-                                            tableBucketAndOffset.f0,
-                                            tableBucketAndOffset.f1.getMaxTimestamp());
-                                }
-                            });
+            long snapshotId = pbLakeTableSnapshotInfo.getSnapshotId();
+            Map<TableBucket, Long> bucketLogEndOffset = new HashMap<>();
+            for (PbLakeTableOffsetForBucket lakeTableOffsetForBucket :
+                    pbLakeTableSnapshotInfo.getBucketsReqsList()) {
+                Long partitionId =
+                        lakeTableOffsetForBucket.hasPartitionId()
+                                ? lakeTableOffsetForBucket.getPartitionId()
+                                : null;
+                int bucketId = lakeTableOffsetForBucket.getBucketId();
+                TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId);
+                Long logEndOffset =
+                        lakeTableOffsetForBucket.hasLogEndOffset()
+                                ? lakeTableOffsetForBucket.getLogEndOffset()
+                                : null;
+                if (lakeTableOffsetForBucket.hasMaxTimestamp()) {
+                    tableBucketsMaxTimestamp.put(
+                            tableBucket, lakeTableOffsetForBucket.getMaxTimestamp());
+                }
+                bucketLogEndOffset.put(tableBucket, logEndOffset);
+            }
             LakeTableSnapshot lakeTableSnapshot =
-                    toLakeSnapshot(pbLakeTableSnapshotInfo, pbLakeTableOffsetForBucketConsumer);
+                    new LakeTableSnapshot(snapshotId, bucketLogEndOffset);
             lakeTableInfoByTableId.put(tableId, lakeTableSnapshot);
         }
 
@@ -1578,11 +1587,12 @@ public static CommitLakeTableSnapshotData getCommitLakeTableSnapshotData(
                     pbLakeTableSnapshotMetadata.getTableId(),
                     new LakeTable.LakeSnapshotMetadata(
                             pbLakeTableSnapshotMetadata.getSnapshotId(),
-                            new FsPath(pbLakeTableSnapshotMetadata.getTieredSnapshotFilePath()),
-                            pbLakeTableSnapshotMetadata.hasReadableSnapshotFilePath()
+                            new FsPath(
+                                    pbLakeTableSnapshotMetadata.getTieredBucketOffsetsFilePath()),
+                            pbLakeTableSnapshotMetadata.hasReadableBucketOffsetsFilePath()
                                     ? new FsPath(
                                             pbLakeTableSnapshotMetadata
-                                                    .getReadableSnapshotFilePath())
+                                                    .getReadableBucketOffsetsFilePath())
                                     : null));
         }
         return new CommitLakeTableSnapshotData(
@@ -1606,36 +1616,6 @@ public static TableBucketOffsets toTableBucketOffsets(
         return new TableBucketOffsets(tableId, bucketOffsets);
     }
 
-    private static LakeTableSnapshot toLakeSnapshot(
-            PbLakeTableSnapshotInfo pbLakeTableSnapshotInfo,
-            @Nullable
-                    Consumer<Tuple2<TableBucket, PbLakeTableOffsetForBucket>>
-                            pbLakeTableOffsetForBucketConsumer) {
-        long tableId = pbLakeTableSnapshotInfo.getTableId();
-        long snapshotId = pbLakeTableSnapshotInfo.getSnapshotId();
-        Map<TableBucket, Long> bucketLogEndOffset = new HashMap<>();
-        for (PbLakeTableOffsetForBucket lakeTableOffsetForBucket :
-                pbLakeTableSnapshotInfo.getBucketsReqsList()) {
-            Long partitionId =
-                    lakeTableOffsetForBucket.hasPartitionId()
-                            ? lakeTableOffsetForBucket.getPartitionId()
-                            : null;
-            int bucketId = lakeTableOffsetForBucket.getBucketId();
-
-            TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId);
-            Long logEndOffset =
-                    lakeTableOffsetForBucket.hasLogEndOffset()
-                            ? lakeTableOffsetForBucket.getLogEndOffset()
-                            : null;
-            if (pbLakeTableOffsetForBucketConsumer != null) {
-                pbLakeTableOffsetForBucketConsumer.accept(
-                        Tuple2.of(tableBucket, lakeTableOffsetForBucket));
-            }
-            bucketLogEndOffset.put(tableBucket, logEndOffset);
-        }
-        return new LakeTableSnapshot(snapshotId, bucketLogEndOffset);
-    }
-
     public static PbNotifyLakeTableOffsetReqForBucket makeNotifyLakeTableOffsetForBucket(
             TableBucket tableBucket,
             LakeTableSnapshot lakeTableSnapshot,
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java
@@ -120,7 +120,7 @@ public LakeTableSnapshot getLakeTableSnapshot() {
      *
      * @return the LakeTableSnapshot
      */
-    public LakeTableSnapshot getLatestTableSnapshot() throws Exception {
+    public LakeTableSnapshot getLatestTableSnapshot() throws IOException {
         if (lakeTableSnapshot != null) {
             return lakeTableSnapshot;
         }
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableJsonSerde.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableJsonSerde.java
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableHelperTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableHelperTest.java
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableJsonSerdeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableJsonSerdeTest.java
