use checkargs and add unit test

Author: polyzos

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

@@ -185,10 +185,6 @@ public FlinkSink<InputT> build() {
         List<String> bucketKeys = tableInfo.getBucketKeys();
         List<String> partitionKeys = tableInfo.getPartitionKeys();
 
-        if (!isUpsert && partialUpdateColumns != null) {
-            LOG.error("Partial updates are not supported in append mode.");
-        }
-
         if (isUpsert) {
             LOG.info("Initializing Fluss upsert sink writer ...");
             writerBuilder =
@@ -231,5 +227,9 @@ private void validateConfiguration() {
 
         checkNotNull(tableName, "Table name is required but not provided.");
         checkArgument(!tableName.isEmpty(), "Table name cannot be empty.");
+
+        checkArgument(
+                isUpsert && partialUpdateColumns == null,
+                "Partial updates are not supported in append mode.");
     }
 }

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

@@ -249,6 +249,22 @@ void testBootstrapServersSetting() throws Exception {
         assertThat(bootstrapServers).isEqualTo(this.bootstrapServers);
     }
 
+    @Test
+    void testPartialUpdateColumnsNotAllowedInAppendMode() {
+        FlussSinkBuilder<Order> builder = new FlussSinkBuilder<>();
+        builder.setBootstrapServers("localhost:9123")
+                .setDatabase("testDb")
+                .setTable("testTable")
+                .setRowType(orderRowType)
+                .setSerializationSchema(new OrderSerializationSchema())
+                .setPartialUpdateColumns(new int[] {0, 1, 2})
+                .useAppend();
+
+        assertThatThrownBy(builder::build)
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("Partial updates are not supported in append mode.");
+    }
+
     @Test
     void testFluentChaining() {
         // Test that all methods can be chained