[I] [Bug] [connector-paimon] NullPointerException when sink to paimon table [seatunnel]

[via GitHub]

nbeeee opened a new issue, #7307:
URL: https://github.com/apache/seatunnel/issues/7307

   ### Search before asking
   
   - [X] I had searched in the 
[issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22)
 and found no similar issues.
   
   
   ### What happened
   
   Reading data from Kudu and writing it to Paimon table reports an error, 
indicating an abnormal commit NullPointerException
   
   
   ### SeaTunnel Version
   
   2.3.6
   
   ### SeaTunnel Config
   
   ```conf
   env {
     parallelism = 3
     job.mode = "BATCH"
     spark.app.name = "t_sale_pay_kudu2paimon"
     spark.executor.instances = 2
     spark.executor.cores = 3
     spark.executor.memory = "6g"
     spark.driver-memory = "2g"
   }
   
   source {
     kudu {
       kudu_masters = "node1:7051,node2:7051,node3:7051"
       table_name = "presto::transdb.t_sale_pay"
       result_table_name = "t_sale_pay"
     }
   }
   
   transform {
      Sql {
       source_table_name = "t_sale_pay"
       result_table_name = "v_sale_pay"
       query = """
           SELECT saleno, paytype, cardno, cast(netsum as decimal(20,4)) as 
netsum, ext_str1, ext_str2, ext_str3, group_id ,substring( accdate,1,10) as 
cdate
           FROM t_sale_pay 
           where time_stamp > 1722442480000
       """
     }
      Filter {
       source_table_name = "v_sale_pay"
       result_table_name = "f_sale_pay"
       fields = [saleno, paytype, cardno, netsum, ext_str1, ext_str2, ext_str3, 
group_id,cdate]
     }
     
   }
   
   sink {
     Paimon {
       source_table_name = "f_sale_pay"
       catalog_name="paimon"
       warehouse="oss://bigdata/paimon/"
       database="hydee_dw"
       table="ods_dts_trade_sale_pay_di_test"
       paimon.table.write-props ={
         bucket = 1
       }
       paimon.hadoop.conf = {
         fs.oss.endpoint ="xxxxx"
         fs.oss.impl ="org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem"
         fs.oss.accessKeyId ="xxxxxx"
         fs.oss.accessKeySecret ="xxxx"
       }
     }
   }
   ```
   
   
   ### Running Command
   
   ```shell
   ${SEATUNNEL_HOME}/bin/start-seatunnel-spark-3-connector-v2.sh --config 
seatunnel_199817_605442.conf --deploy-mode client --master yarn
   ```
   
   
   ### Error Exception
   
   ```log
   Caused by: 
org.apache.seatunnel.connectors.seatunnel.paimon.exception.PaimonConnectorException:
 ErrorCode:[PAIMON-01], ErrorDescription:[Paimon write commit failed] - Flink 
table store commit operation failed
                at 
org.apache.seatunnel.connectors.seatunnel.paimon.sink.commit.PaimonAggregatedCommitter.commit(PaimonAggregatedCommitter.java:95)
                at 
org.apache.seatunnel.translation.spark.sink.SeaTunnelBatchWrite.commit(SeaTunnelBatchWrite.java:88)
                at 
org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.writeWithV2(WriteToDataSourceV2Exec.scala:392)
                ... 49 more
        Caused by: java.lang.NullPointerException
                at 
org.apache.paimon.operation.FileStoreCommitImpl.collectChanges(FileStoreCommitImpl.java:597)
                at 
org.apache.paimon.operation.FileStoreCommitImpl.commit(FileStoreCommitImpl.java:245)
                at 
org.apache.paimon.table.sink.TableCommitImpl.commitMultiple(TableCommitImpl.java:206)
                at 
org.apache.paimon.table.sink.TableCommitImpl.commit(TableCommitImpl.java:200)
                at 
org.apache.paimon.table.sink.TableCommitImpl.commit(TableCommitImpl.java:179)
                at 
org.apache.paimon.table.sink.TableCommitImpl.commit(TableCommitImpl.java:163)
                at 
org.apache.seatunnel.connectors.seatunnel.paimon.sink.commit.PaimonAggregatedCommitter.lambda$commit$0(PaimonAggregatedCommitter.java:86)
                at 
org.apache.paimon.security.SecurityContext.runSecured(SecurityContext.java:67)
                at 
org.apache.seatunnel.connectors.seatunnel.paimon.sink.commit.PaimonAggregatedCommitter.commit(PaimonAggregatedCommitter.java:82)
   ```
   
   
   ### Zeta or Flink or Spark Version
   
   spark3.3.1
   
   ### Java or Scala Version
   
   java jdk1.8.0_291
   scala 2.12
   
   ### Screenshots
   
   
![image](https://github.com/user-attachments/assets/54cd52f3-31db-470f-865b-ca690775beba)
   
   
   ### Are you willing to submit PR?
   
   - [ ] Yes I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [X] I agree to follow this project's [Code of 
Conduct](https://www.apache.org/foundation/policies/conduct)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@seatunnel.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org