feat(iceberg): add 'glue.id' to iceberg sink config

src/connector/src/connector_common/connection.rs

@@ -135,6 +135,10 @@ pub struct IcebergConnection {
     /// Path of iceberg warehouse, only applicable in storage catalog.
     #[serde(rename = "warehouse.path")]
     pub warehouse_path: Option<String>,
+    /// Catalog id, can be omitted for storage catalog or when
+    /// caller's AWS account ID matches glue id
+    #[serde(rename = "glue.id")]
+    pub glue_id: Option<String>,
     /// Catalog name, can be omitted for storage catalog, but
     /// must be set for other catalogs.
     #[serde(rename = "catalog.name")]
@@ -255,6 +259,7 @@ impl Connection for IcebergConnection {
             secret_key: self.secret_key.clone(),
             gcs_credential: self.gcs_credential.clone(),
             warehouse_path: self.warehouse_path.clone(),
+            glue_id: self.glue_id.clone(),
             catalog_name: self.catalog_name.clone(),
             catalog_uri: self.catalog_uri.clone(),
             credential: self.credential.clone(),

src/connector/src/connector_common/iceberg/mod.rs

@@ -57,6 +57,10 @@ pub struct IcebergCommon {
     /// Path of iceberg warehouse, only applicable in storage catalog.
     #[serde(rename = "warehouse.path")]
     pub warehouse_path: Option<String>,
+    /// AWS Client id, can be omitted for storage catalog or when
+    /// caller's AWS account ID matches glue id
+    #[serde(rename = "glue.id")]
+    pub glue_id: Option<String>,
     /// Catalog name, can be omitted for storage catalog, but
     /// must be set for other catalogs.
     #[serde(rename = "catalog.name")]
@@ -274,6 +278,10 @@ impl IcebergCommon {
                             format!("https://glue.{}.amazonaws.com", region),
                         );
                     }
+
+                    if let Some(glue_id) = self.glue_id.as_deref() {
+                        java_catalog_configs.insert("glue.id".to_owned(), glue_id.to_owned());
+                    }
                 }
                 _ => {}
             }

src/connector/src/sink/iceberg/mod.rs

@@ -1432,6 +1432,7 @@ mod test {
                 secret_key: Some("hummockadmin".to_owned()),
                 gcs_credential: None,
                 catalog_type: Some("jdbc".to_owned()),
+                glue_id: None,
                 catalog_name: Some("demo".to_owned()),
                 database_name: Some("demo_db".to_owned()),
                 table_name: "demo_table".to_owned(),

src/connector/with_options_sink.yaml

@@ -447,6 +447,12 @@ IcebergConfig:
     field_type: String
     comments: Path of iceberg warehouse, only applicable in storage catalog.
     required: false
+  - name: glue.id
+    field_type: String
+    comments: |-
+      AWS Client id, can be omitted for storage catalog or when
+      caller's AWS account ID matches glue id
+    required: false
   - name: catalog.name
     field_type: String
     comments: |-

src/connector/with_options_source.yaml

@@ -103,6 +103,12 @@ IcebergProperties:
     field_type: String
     comments: Path of iceberg warehouse, only applicable in storage catalog.
     required: false
+  - name: glue.id
+    field_type: String
+    comments: |-
+      AWS Client id, can be omitted for storage catalog or when
+      caller's AWS account ID matches glue id
+    required: false
   - name: catalog.name
     field_type: String
     comments: |-