address comments

Author: polyzos

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/sink/FlussSinkBuilder.java

@@ -19,6 +19,7 @@
 import com.alibaba.fluss.client.Connection;
 import com.alibaba.fluss.client.ConnectionFactory;
 import com.alibaba.fluss.client.admin.Admin;
+import com.alibaba.fluss.config.ConfigOptions;
 import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.flink.sink.serializer.FlussSerializationSchema;
 import com.alibaba.fluss.flink.sink.writer.FlinkSinkWriter;
@@ -53,6 +54,7 @@
  * FlinkSink<Order> sink = new FlussSinkBuilder<Order>()
  *          .setBootstrapServers(bootstrapServers)
  *          .setTable(tableName)
+ *          .setDatabase("testDb")
  *          .setRowType(orderRowType)
  *          .setSerializationSchema(new OrderSerializationSchema())
  *          .build())
@@ -111,17 +113,15 @@ public FlussSinkBuilder<InputT> setPartialUpdateColumns(int[] partialUpdateColum
         return this;
     }
 
-    /** Set target column indexes. */
-    public FlussSinkBuilder<InputT> setDataLakeFormat(DataLakeFormat lakeFormat) {
-        this.lakeFormat = lakeFormat;
-        return this;
-    }
+//    /** Set target column indexes. */
+//    public FlussSinkBuilder<InputT> setDataLakeFormat(DataLakeFormat lakeFormat) {
+//        this.lakeFormat = lakeFormat;
+//        return this;
+//    }
 
     /** Set shuffle by bucket id. */
     public FlussSinkBuilder<InputT> setShuffleByBucketId(boolean shuffleByBucketId) {
-        if (!shuffleByBucketId) {
-            this.shuffleByBucketId = false;
-        }
+        this.shuffleByBucketId = shuffleByBucketId;
         return this;
     }
 
@@ -157,15 +157,15 @@ public FlussSinkBuilder<InputT> setSerializationSchema(
     }
 
     /** Build the FlussSink. */
-    public FlinkSink<InputT> build() {
+    public FlussSink<InputT> build() {
         validateConfiguration();
 
         Configuration flussConfig = Configuration.fromMap(configOptions);
 
         FlinkSink.SinkWriterBuilder<? extends FlinkSinkWriter<InputT>, InputT> writerBuilder;
 
         TablePath tablePath = new TablePath(database, tableName);
-        flussConfig.setString("bootstrap.servers", bootstrapServers);
+        flussConfig.setString(ConfigOptions.BOOTSTRAP_SERVERS.key(), bootstrapServers);
 
         TableInfo tableInfo;
         try (Connection connection = ConnectionFactory.createConnection(flussConfig);
@@ -185,6 +185,10 @@ public FlinkSink<InputT> build() {
         int numBucket = tableInfo.getNumBuckets();
         List<String> bucketKeys = tableInfo.getBucketKeys();
         List<String> partitionKeys = tableInfo.getPartitionKeys();
+        tableInfo
+                .getTableConfig()
+                .getDataLakeFormat()
+                .ifPresent(format -> lakeFormat = format);
 
         if (isUpsert) {
             LOG.info("Initializing Fluss upsert sink writer ...");

fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/sink/FlussSinkBuilderTest.java

@@ -37,9 +37,9 @@
 
 /** Tests for {@link FlussSinkBuilder} configuration and argument handling. */
 class FlussSinkBuilderTest {
-    private String bootstrapServers = "localhost:9123";
-    private String databaseName = "testDb";
-    private String tableName = "testTable";
+    private final String bootstrapServers = "localhost:9123";
+    private final String databaseName = "testDb";
+    private final String tableName = "testTable";
 
     private FlussSinkBuilder<Order> builder;
     private RowType orderRowType;
@@ -190,8 +190,7 @@ void testDataLakeFormat() throws Exception {
         assertThat(lakeFormat).isNull();
 
         // Test setting format
-        builder.setDataLakeFormat(DataLakeFormat.PAIMON);
-        lakeFormat = getFieldValue(builder, "lakeFormat");
+        lakeFormat = getFieldValue(builder, null);
         assertThat(lakeFormat).isEqualTo(DataLakeFormat.PAIMON);
     }
 
@@ -279,7 +278,6 @@ void testFluentChaining() {
                         .useUpsert()
                         .setOption("key1", "value1")
                         .setOptions(new HashMap<>())
-                        .setDataLakeFormat(DataLakeFormat.PAIMON)
                         .setShuffleByBucketId(false);
 
         // Verify the builder instance is returned