fix failures tests

Author: xiaochen-zhou

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkWriter.java

@@ -176,7 +176,7 @@ public PaimonSinkWriter(
             log.warn("Append only table currently do not support dynamic bucket");
         }
         if (dynamicBucket) {
-            paimonBucketAssignerFactory.init(paimonTablePath, paimonTable, parallelism);
+            paimonBucketAssignerFactory.init(paimonTablePath, paimonFileStoretable, parallelism);
         }
         PaimonSecurityContext.shouldEnableKerberos(paimonHadoopConfiguration);
     }

seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonPrivilegeCatalogTest.java

@@ -105,6 +105,7 @@
 public class PaimonPrivilegeCatalogTest {
 
     private PaimonCatalog authorizedCatalog;
+    private PaimonCatalog authorizedSelectCatalog;
     private PaimonCatalog unAuthorizedCatalog;
     private PaimonCatalog rootUserPaimonCatalog;
     private String CATALOG_NAME = "paimon_catalog";
@@ -117,9 +118,11 @@ public class PaimonPrivilegeCatalogTest {
     private String rootPassword = "123456";
     private String bucketKey = "f0";
     private String authorizeUser = "paimon";
+    private String authorizeSelectUser = "authorize_select_user";
     private String authorizeUserPassword = "123456";
     private String unAuthorizeUser = "unauthorized_paimon";
     private String unAuthorizeUserPassword = "123456";
+    private String authorizeSelectUserPassword = "123456";
 
     private int writeRows = 0;
 
@@ -129,6 +132,8 @@ public void before() {
         initPrivilege();
         rootUserPaimonCatalog = createPaimonCatalog(rootUser, rootPassword);
         authorizedCatalog = createPaimonCatalog(authorizeUser, authorizeUserPassword);
+        authorizedSelectCatalog =
+                createPaimonCatalog(authorizeSelectUser, authorizeSelectUserPassword);
         unAuthorizedCatalog = createPaimonCatalog(unAuthorizeUser, unAuthorizeUserPassword);
 
         createUser(authorizeUser, authorizeUserPassword);
@@ -141,6 +146,12 @@ public void before() {
                     PrivilegeType.INSERT
                 });
         createUser(unAuthorizeUser, unAuthorizeUserPassword);
+        createUser(authorizeSelectUser, authorizeSelectUserPassword);
+        grantPrivilege(
+                authorizeSelectUser,
+                new PrivilegeType[] {
+                    PrivilegeType.SELECT,
+                });
 
         createDatabase();
         catalogTable = buildTable(TABLE_NAME);
@@ -348,14 +359,14 @@ public void testWriteTable() throws IOException {
                 NoPrivilegeException.class,
                 () -> {
                     try {
-                        writeTable(unAuthorizedCatalog, rows);
+                        writeTable(authorizedSelectCatalog, rows);
                     } catch (NoPrivilegeException e) {
                         assertTrue(
                                 e.getMessage()
                                         .contains(
                                                 String.format(
                                                         "User %s doesn't have privilege INSERT on table",
-                                                        unAuthorizeUser)));
+                                                        authorizeSelectUser)));
                         throw e;
                     }
                 });