add test for txn property check

Author: xx789633

fluss-lake/fluss-lake-lance/src/main/java/com/alibaba/fluss/lake/lance/tiering/LanceLakeCommitter.java

@@ -104,7 +104,7 @@ public CommittedLakeSnapshot getMissingLakeSnapshot(@Nullable Long latestLakeSna
                 new CommittedLakeSnapshot(latestLakeSnapshotIdOfLake.get());
         String flussOffsetProperties =
                 LanceDatasetAdapter.getTransactionProperties(
-                                config, Math.toIntExact(latestLakeSnapshotIdOfLake.get() + 1))
+                                config, latestLakeSnapshotIdOfLake.get() + 1)
                         .get(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY);
         for (JsonNode node : OBJECT_MAPPER.readTree(flussOffsetProperties)) {
             BucketOffset bucketOffset = BucketOffsetJsonSerde.INSTANCE.deserialize(node);

fluss-lake/fluss-lake-lance/src/main/java/com/alibaba/fluss/lake/lance/utils/LanceDatasetAdapter.java

@@ -36,6 +36,8 @@
 import org.apache.arrow.vector.ipc.ArrowReader;
 import org.apache.arrow.vector.types.pojo.Schema;
 
+import javax.annotation.Nullable;
+
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -95,9 +97,12 @@ public static long commitAppend(
         }
     }
 
-    public static Map<String, String> getTransactionProperties(LanceConfig config, int version) {
+    public static Map<String, String> getTransactionProperties(
+            LanceConfig config, @Nullable Long version) {
         ReadOptions.Builder builder = new ReadOptions.Builder();
-        builder.setVersion(version);
+        if (version != null) {
+            builder.setVersion(Math.toIntExact(version));
+        }
         builder.setStorageOptions(LanceConfig.genStorageOptions(config));
         try (Dataset dataset = Dataset.open(allocator, config.getDatasetUri(), builder.build())) {
             Transaction transaction = dataset.readTransaction().orElse(null);

fluss-lake/fluss-lake-lance/src/test/java/com/alibaba/fluss/lake/lance/tiering/LanceTieringITCase.java

@@ -20,6 +20,7 @@
 import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.lake.lance.LanceConfig;
 import com.alibaba.fluss.lake.lance.testutils.FlinkLanceTieringTestBase;
+import com.alibaba.fluss.lake.lance.utils.LanceDatasetAdapter;
 import com.alibaba.fluss.metadata.TableBucket;
 import com.alibaba.fluss.metadata.TablePath;
 import com.alibaba.fluss.row.InternalRow;
@@ -37,9 +38,12 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 
+import static com.alibaba.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
 import static com.alibaba.fluss.testutils.DataTestUtils.row;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -80,20 +84,38 @@ void testTiering() throws Exception {
         // note: we can't update log start offset for unaware bucket mode log table
         assertReplicaStatus(t1Bucket, 30);
 
+        LanceConfig config =
+                LanceConfig.from(
+                        lanceConf.toMap(),
+                        Collections.emptyMap(),
+                        t1.getDatabaseName(),
+                        t1.getTableName());
+
         // check data in lance
-        checkDataInLanceAppendOnlyTable(t1, flussRows);
+        checkDataInLanceAppendOnlyTable(config, flussRows);
+        // check snapshot property in lance
+        Map<String, String> properties =
+                new HashMap<String, String>() {
+                    {
+                        put(
+                                FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
+                                "[{\"bucket_id\":0,\"log_offset\":30}]");
+                    }
+                };
+        checkSnapshotPropertyInLance(config, properties);
 
         jobClient.cancel().get();
     }
 
-    private void checkDataInLanceAppendOnlyTable(
-            TablePath tablePath, List<InternalRow> expectedRows) throws Exception {
-        LanceConfig config =
-                LanceConfig.from(
-                        lanceConf.toMap(),
-                        Collections.emptyMap(),
-                        tablePath.getDatabaseName(),
-                        tablePath.getTableName());
+    private void checkSnapshotPropertyInLance(
+            LanceConfig config, Map<String, String> expectedProperties) throws Exception {
+        Map<String, String> transactionProperties =
+                LanceDatasetAdapter.getTransactionProperties(config, null);
+        assertThat(transactionProperties).isEqualTo(expectedProperties);
+    }
+
+    private void checkDataInLanceAppendOnlyTable(LanceConfig config, List<InternalRow> expectedRows)
+            throws Exception {
 
         try (Dataset dataset =
                 Dataset.open(