[Bug] [Connector-Iceberg] Fail to create iceberg sink table with primary-keys

[litiliu]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

I checked the metadata of the generated iceberg sink table and found the primary keys are not set although the iceberg.table.primary-keys has been set.

SeaTunnel Version

2.3.5-release

SeaTunnel Config

{
  "env": {
    "execution.parallelism": 1,
    "job.mode": "BATCH"
  },
  "source": [{
    "plugin_name": "FakeSource",
    "result_table_name": "s3_source",
    "schema": {
      "fields": {
        "update_time": "string",
        "code": "string",
        "create_time": "string",
        "name": "string",
        "id": "string"
      }
    }

  }],
  "sink": [{
    "plugin_name": "Iceberg",
    "source_table_name": "s3_source",
    "iceberg.catalog.config": {
      "warehouse": "*",
      "catalog-impl": "org.apache.iceberg.rest.RESTCatalog",
      "uri": "*",
      "io-impl": "org.apache.iceberg.aws.s3.S3FileIO",
      "s3.write.storage-class": "INTELLIGENT_TIERING",
      "client.region":"*"
    },
    "iceberg.table.primary-keys":"id,name",
    "iceberg.table.partition-keys":"name",
    "namespace": "ns",
    "table": "t21",
    "iceberg.table.write-props": {
      "write.format.default": "parquet",
      "write.target-file-size-bytes": 10,
      "write.distribution-mode": "hash"
    },
    "case_sensitive": true

  }
  ]
}

Running Command

run the SeaTunnelApiExample.java

Error Exception

{
  "format-version" : 2,
  "table-uuid" : "d87e5641-6661-45e6-be58-fa1d5b4da9c3",
  "location" : "s3://*/t21",
  "last-sequence-number" : 0,
  "last-updated-ms" : 1716446620137,
  "last-column-id" : 5,
  "current-schema-id" : 0,
  "schemas" : [ {
    "type" : "struct",
    "schema-id" : 0,
    "fields" : [ {
      "id" : 1,
      "name" : "update_time",
      "required" : false,
      "type" : "string"
    }, {
      "id" : 2,
      "name" : "code",
      "required" : false,
      "type" : "string"
    }, {
      "id" : 3,
      "name" : "create_time",
      "required" : false,
      "type" : "string"
    }, {
      "id" : 4,
      "name" : "name",
      "required" : false,
      "type" : "string"
    }, {
      "id" : 5,
      "name" : "id",
      "required" : false,
      "type" : "string"
    } ]
  } ],
  "default-spec-id" : 0,
  "partition-specs" : [ {
    "spec-id" : 0,
    "fields" : [ {
      "name" : "name",
      "transform" : "identity",
      "source-id" : 4,
      "field-id" : 1000
    } ]
  } ],
  "last-partition-id" : 1000,
  "default-sort-order-id" : 0,
  "sort-orders" : [ {
    "order-id" : 0,
    "fields" : [ ]
  } ],
  "properties" : {
    "write.parquet.compression-codec" : "zstd"
  },
  "current-snapshot-id" : -1,
  "refs" : { },
  "snapshots" : [ ],
  "statistics" : [ ],
  "partition-statistics" : [ ],
  "snapshot-log" : [ ],
  "metadata-log" : [ ]
}

Zeta or Flink or Spark Version

Flink 1.15.4

Java or Scala Version

Java 8

Screenshots

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct