add IT cases and fix checkstle

Author: LsomeYeah

paimon-common/src/main/java/org/apache/paimon/fs/WeightedExternalPathProvider.java

@@ -9,12 +9,11 @@
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
  */
 
 package org.apache.paimon.fs;

paimon-common/src/test/java/org/apache/paimon/fs/WeightedExternalPathProviderTest.java

@@ -9,12 +9,11 @@
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
  */
 
 package org.apache.paimon.fs;

paimon-core/src/main/java/org/apache/paimon/utils/FileStorePathFactory.java

@@ -218,7 +218,10 @@ private ExternalPathProvider createExternalPathProvider(BinaryRow partition, int
             return null;
         }
         return ExternalPathProvider.create(
-                strategy, externalPaths, relativeBucketPath(partition, bucket), externalPathWeights);
+                strategy,
+                externalPaths,
+                relativeBucketPath(partition, bucket),
+                externalPathWeights);
     }
 
     public List<Path> getExternalPaths() {

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendOnlyTableITCase.java

@@ -34,7 +34,10 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
+import java.io.IOException;
+import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.ArrayList;
@@ -225,6 +228,49 @@ public void testReadWriteWithExternalPathSpecificFSStrategy() {
                         Row.of(1, "AAA"), Row.of(2, "BBB"), Row.of(3, "CCC"), Row.of(4, "DDD"));
     }
 
+    @Test
+    public void testReadWriteWithExternalPathWeightRobinStrategy() throws IOException {
+        String externalPaths =
+                TraceableFileIO.SCHEME
+                        + "://"
+                        + tempExternalPath1.toString()
+                        + ","
+                        + LocalFileIOLoader.SCHEME
+                        + "://"
+                        + tempExternalPath2.toString();
+        batchSql(
+                "ALTER TABLE append_table SET ("
+                        + "'data-file.external-paths' = '"
+                        + externalPaths
+                        + "', "
+                        + "'data-file.external-paths.strategy' = 'weight-robin', "
+                        + "'data-file.external-paths.weights' = '1,3', "
+                        + "'write-only' = 'true'"
+                        + ")");
+
+        int fileNum = 50;
+        for (int i = 1; i <= fileNum; i++) {
+            batchSql("INSERT INTO append_table VALUES (" + i + ", 'AAA')");
+        }
+
+        List<Row> rows = batchSql("SELECT * FROM append_table");
+        assertThat(rows.size()).isEqualTo(fileNum);
+
+        // Verify file distribution based on weights
+        long filesInPath1 =
+                Files.list(Paths.get(tempExternalPath1.toString() + "/bucket-0")).count();
+        long filesInPath2 =
+                Files.list(Paths.get(tempExternalPath2.toString() + "/bucket-0")).count();
+        long totalFiles = filesInPath1 + filesInPath2;
+
+        // Since the file sample size is small in IT case, we only verify that higher-weighted path
+        // has more files
+        assertThat(filesInPath1).isGreaterThan(0);
+        assertThat(filesInPath2).isGreaterThan(0);
+        assertThat(filesInPath2).isGreaterThan(filesInPath1);
+        assertThat(totalFiles).isEqualTo(fileNum);
+    }
+
     @Test
     public void testReadWriteWithExternalPathNoneStrategy() {
         String externalPaths = TraceableFileIO.SCHEME + "://" + tempExternalPath1.toString();

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java

@@ -52,6 +52,7 @@
 
 import java.io.IOException;
 import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -306,6 +307,63 @@ public void testTableReadWriteWithExternalPathRoundRobin() throws Exception {
         assertThat(actual).containsExactlyInAnyOrder("+I[1, A]", "+I[2, B]", "+I[3, C]");
     }
 
+    @Test
+    public void testTableReadWriteWithExternalPathWeightRobin() throws Exception {
+        TableEnvironment sEnv =
+                tableEnvironmentBuilder()
+                        .streamingMode()
+                        .checkpointIntervalMs(ThreadLocalRandom.current().nextInt(900) + 100)
+                        .parallelism(1)
+                        .build();
+
+        sEnv.executeSql(createCatalogSql("testCatalog", path + "/warehouse"));
+        sEnv.executeSql("USE CATALOG testCatalog");
+        String externalPaths =
+                TraceableFileIO.SCHEME
+                        + "://"
+                        + externalPath1.toString()
+                        + ","
+                        + LocalFileIOLoader.SCHEME
+                        + "://"
+                        + externalPath2.toString();
+        sEnv.executeSql(
+                "CREATE TABLE T2 ( k INT, v STRING, PRIMARY KEY (k) NOT ENFORCED ) "
+                        + "WITH ( "
+                        + "'bucket' = '1',"
+                        + "'write-only' = 'true',"
+                        + "'data-file.external-paths' = '"
+                        + externalPaths
+                        + "',"
+                        + "'data-file.external-paths.strategy' = 'weight-robin',"
+                        + "'data-file.external-paths.weights' = '10,5'"
+                        + ")");
+
+        CloseableIterator<Row> it = collect(sEnv.executeSql("SELECT * FROM T2"));
+
+        int fileNum = 50;
+        for (int i = 1; i <= fileNum; i++) {
+            sEnv.executeSql("INSERT INTO T2 VALUES (" + i + ", 'data" + i + "')").await();
+        }
+
+        List<String> actual = new ArrayList<>();
+        for (int i = 0; i < fileNum; i++) {
+            actual.add(it.next().toString());
+        }
+        // Verify all data is readable
+        assertThat(actual).hasSize(fileNum);
+
+        long filesInPath1 = Files.list(Paths.get(externalPath1.toString() + "/bucket-0")).count();
+        long filesInPath2 = Files.list(Paths.get(externalPath2.toString() + "/bucket-0")).count();
+        long totalFiles = filesInPath1 + filesInPath2;
+
+        // Since the file sample size is small in IT case, we only verify that higher-weighted path
+        // has more files
+        assertThat(filesInPath1).isGreaterThan(0);
+        assertThat(filesInPath2).isGreaterThan(0);
+        assertThat(filesInPath1).isGreaterThan(filesInPath2);
+        assertThat(totalFiles).isEqualTo(fileNum);
+    }
+
     @Test
     public void testDropTableWithExternalPaths() throws Exception {
         TableEnvironment sEnv =