fix comments

Author: xx789633

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

@@ -25,16 +25,21 @@
 import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.VectorSchemaRoot;
 
-/** An Arrow writer for InternalRow. */
+/** An Arrow writer for {@link InternalRow}. */
 public class ArrowWriter {
-    /**
-     * An array of writers which are responsible for the serialization of each column of the rows.
-     */
+    private final VectorSchemaRoot root;
+
     private final ArrowFieldWriter<InternalRow>[] fieldWriters;
 
     private int recordsCount;
-    private VectorSchemaRoot root;
 
+    /**
+     * Writer which serializes the Fluss rows to Arrow record batches.
+     *
+     * @param fieldWriters An array of writers which are responsible for the serialization of each
+     *     column of the rows
+     * @param root Container that holds a set of vectors for the rows
+     */
     public ArrowWriter(ArrowFieldWriter<InternalRow>[] fieldWriters, VectorSchemaRoot root) {
         this.fieldWriters = fieldWriters;
         this.root = root;

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

@@ -61,7 +61,7 @@ public LanceCommittable deserialize(int version, byte[] serialized) throws IOExc
         try {
             return new LanceCommittable((List<FragmentMetadata>) ois.readObject());
         } catch (ClassNotFoundException e) {
-            throw new RuntimeException(e);
+            throw new IOException("Couldn't deserialize LanceCommittable", e);
         }
     }
 }

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

@@ -74,9 +74,7 @@ public long commit(LanceCommittable committable, Map<String, String> snapshotPro
     }
 
     @Override
-    public void abort(LanceCommittable committable) throws IOException {
-        throw new UnsupportedOperationException();
-    }
+    public void abort(LanceCommittable committable) throws IOException {}
 
     @SuppressWarnings("checkstyle:LocalVariableName")
     @Nullable

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

@@ -38,7 +38,7 @@
 /** Implementation of {@link LakeWriter} for Lance. */
 public class LanceLakeWriter implements LakeWriter<LanceWriteResult> {
     private final LanceArrowWriter arrowWriter;
-    FutureTask<List<FragmentMetadata>> fragmentCreationTask;
+    private final FutureTask<List<FragmentMetadata>> fragmentCreationTask;
 
     public LanceLakeWriter(Configuration options, WriterInitContext writerInitContext)
             throws IOException {

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

@@ -50,7 +50,7 @@ public LanceWriteResult deserialize(int version, byte[] serialized) throws IOExc
         try {
             return (LanceWriteResult) ois.readObject();
         } catch (ClassNotFoundException e) {
-            throw new RuntimeException(e);
+            throw new IOException("Couldn't deserialize LanceWriteResult", e);
         }
     }
 }

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

@@ -28,22 +28,19 @@
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
 import org.apache.arrow.vector.VectorSchemaRoot;
-import org.apache.arrow.vector.VectorUnloader;
-import org.apache.arrow.vector.ipc.message.ArrowRecordBatch;
 import org.junit.jupiter.api.Test;
 
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
 
 import static com.alibaba.fluss.record.ChangeType.APPEND_ONLY;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** The UT for Lance Arrow Writer. */
 public class LanceArrowWriterTest {
     @Test
-    public void test() throws Exception {
+    public void testLanceArrowWriter() throws Exception {
         try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) {
             List<DataField> fields = Arrays.asList(new DataField("column1", DataTypes.INT()));
 
@@ -56,7 +53,6 @@ public void test() throws Exception {
                             allocator, LanceArrowUtils.toArrowSchema(rowType), batchSize, rowType);
             AtomicInteger rowsWritten = new AtomicInteger(0);
             AtomicInteger rowsRead = new AtomicInteger(0);
-            AtomicLong expectedBytesRead = new AtomicLong(0);
 
             Thread writerThread =
                     new Thread(
@@ -89,11 +85,6 @@ public void test() throws Exception {
                                         VectorSchemaRoot root = arrowWriter.getVectorSchemaRoot();
                                         int rowCount = root.getRowCount();
                                         rowsRead.addAndGet(rowCount);
-                                        try (ArrowRecordBatch recordBatch =
-                                                new VectorUnloader(root).getRecordBatch()) {
-                                            expectedBytesRead.addAndGet(
-                                                    recordBatch.computeBodyLength());
-                                        }
                                         for (int i = 0; i < rowCount; i++) {
                                             int value =
                                                     (int) root.getVector("column1").getObject(i);

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

@@ -124,9 +124,6 @@ private void checkDataInLanceAppendOnlyTable(LanceConfig config, List<InternalRo
                         LanceConfig.genReadOptionFromConfig(config))) {
             ArrowReader reader = dataset.newScan().scanBatches();
             VectorSchemaRoot readerRoot = reader.getVectorSchemaRoot();
-            //        while (reader.loadNextBatch()) {
-            //            System.out.print(readerRoot.contentToTSVString());
-            //        }
             reader.loadNextBatch();
             Iterator<InternalRow> flussRowIterator = expectedRows.iterator();
             int rowCount = readerRoot.getRowCount();

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

@@ -171,9 +171,6 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
                         LanceConfig.genReadOptionFromConfig(config))) {
             ArrowReader reader = dataset.newScan().scanBatches();
             VectorSchemaRoot readerRoot = reader.getVectorSchemaRoot();
-            //        while (reader.loadNextBatch()) {
-            //            System.out.print(readerRoot.contentToTSVString());
-            //        }
 
             // then, check data
             for (int bucket = 0; bucket < 3; bucket++) {
@@ -311,13 +308,10 @@ private Schema createTable(LanceConfig config) throws Exception {
         columns.add(new Schema.Column("c3", DataTypes.STRING()));
         Schema.Builder schemaBuilder = Schema.newBuilder().fromColumns(columns);
         Schema schema = schemaBuilder.build();
-        doCreateLanceTable(schema, config);
-        return schema;
-    }
-
-    private void doCreateLanceTable(Schema schema, LanceConfig config) throws Exception {
         WriteParams params = LanceConfig.genWriteParamsFromConfig(config);
         LanceDatasetAdapter.createDataset(
                 config.getDatasetUri(), LanceArrowUtils.toArrowSchema(schema.getRowType()), params);
+
+        return schema;
     }
 }

fluss-lake/fluss-lake-paimon/src/test/java/com/alibaba/fluss/lake/paimon/testutils/FlinkPaimonTieringTestBase.java

@@ -126,7 +126,6 @@ public void beforeEach() {
         execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
         execEnv.setRuntimeMode(RuntimeExecutionMode.STREAMING);
         execEnv.setParallelism(2);
-        execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
     }
 
     protected JobClient buildTieringJob(StreamExecutionEnvironment execEnv) throws Exception {