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

[nbeeee]

Search before asking

• [X] I had searched in the issues 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

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

${SEATUNNEL_HOME}/bin/start-seatunnel-spark-3-connector-v2.sh --config seatunnel_199817_605442.conf --deploy-mode client --master yarn

Error Exception

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

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

[Hisoka-X]

cc @dailai and @TaoZex

[github-actions[bot]]

This issue has been automatically marked as stale because it has not had recent activity for 30 days. It will be closed in next 7 days if no further activity occurs.