fix e2e failures

Author: xiaochen-zhou

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/config/PaimonSinkOptions.java

@@ -70,5 +70,5 @@ public class PaimonSinkOptions extends PaimonBaseOptions {
                             "Properties passed through to paimon table initialization, such as 'file.format', 'bucket'(org.apache.paimon.CoreOptions)");
 
     public static final Option<String> BRANCH =
-            Options.key("branch").stringType().defaultValue("main").withDescription("branch");
+            Options.key("branch").stringType().noDefaultValue().withDescription("branch");
 }

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

@@ -17,6 +17,8 @@
 
 package org.apache.seatunnel.connectors.seatunnel.paimon.sink;
 
+import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
+
 import org.apache.seatunnel.api.common.JobContext;
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.serialization.DefaultSerializer;
@@ -97,21 +99,25 @@ public PaimonSink(ReadonlyConfig readonlyConfig, CatalogTable catalogTable) {
             paimonCatalog.open();
             boolean databaseExists =
                     paimonCatalog.databaseExists(this.paimonSinkConfig.getNamespace());
-            if (databaseExists) {
-                TablePath tablePath = catalogTable.getTablePath();
-                boolean tableExists = paimonCatalog.tableExists(tablePath);
-                if (tableExists) {
-                    String branchName = paimonSinkConfig.getBranch();
-                    this.paimonTable = (FileStoreTable) paimonCatalog.getPaimonTable(tablePath);
-                    BranchManager branchManager = paimonTable.branchManager();
-                    if (!branchManager.branchExists(branchName)) {
-                        throw new UnsupportedOperationException(
-                                "Branch: " + branchName + " not exists");
-                    }
-                    if (!branchManager.DEFAULT_MAIN_BRANCH.equalsIgnoreCase(branchName)) {
-                        this.paimonTable = paimonTable.switchToBranch(branchName);
-                        log.info("Switch to branch {}", branchName);
-                    }
+            if (!databaseExists) {
+                return;
+            }
+            TablePath tablePath = catalogTable.getTablePath();
+            boolean tableExists = paimonCatalog.tableExists(tablePath);
+            if (!tableExists) {
+                return;
+            }
+            this.paimonTable = (FileStoreTable) paimonCatalog.getPaimonTable(tablePath);
+            String branchName = paimonSinkConfig.getBranch();
+            if (StringUtils.isNotEmpty(branchName)) {
+                BranchManager branchManager = paimonTable.branchManager();
+                if (!branchManager.branchExists(branchName)) {
+                    throw new UnsupportedOperationException(
+                            "Branch: " + branchName + " not exists");
+                }
+                if (!branchManager.DEFAULT_MAIN_BRANCH.equalsIgnoreCase(branchName)) {
+                    this.paimonTable = paimonTable.switchToBranch(branchName);
+                    log.info("Switch to branch {}", branchName);
                 }
             }
         }

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

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.paimon.sink;
 
 import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting;
+import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
 
 import org.apache.seatunnel.api.common.JobContext;
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
@@ -276,10 +277,12 @@ private void reOpenTableWrite() {
         this.seaTunnelRowType = this.sourceTableSchema.toPhysicalRowDataType();
         this.paimonTable = (FileStoreTable) paimonCatalog.getPaimonTable(paimonTablePath);
         String branchName = paimonSinkConfig.getBranch();
-        BranchManager branchManager = paimonTable.branchManager();
-        if (!branchManager.DEFAULT_MAIN_BRANCH.equalsIgnoreCase(branchName)) {
-            this.paimonTable = this.paimonTable.switchToBranch(branchName);
-            log.info("Re-switched to branch {} after reopening table", branchName);
+        if (StringUtils.isNotEmpty(branchName)) {
+            BranchManager branchManager = paimonTable.branchManager();
+            if (!branchManager.DEFAULT_MAIN_BRANCH.equalsIgnoreCase(branchName)) {
+                this.paimonTable = this.paimonTable.switchToBranch(branchName);
+                log.info("Re-switched to branch {} after reopening table", branchName);
+            }
         }
         this.sinkPaimonTableSchema = this.paimonTable.schema();
         this.newTableWrite();

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

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