refactor(connector): extract IcebergCommon config

src/batch/src/executor/iceberg_scan.rs

@@ -90,7 +90,12 @@ impl IcebergScanExecutor {
     async fn do_execute(mut self: Box<Self>) {
         let table = self
             .iceberg_config
-            .load_table_v2_with_metadata(self.table_meta)
+            .common
+            .load_table_v2_with_metadata(
+                self.table_meta,
+                &self.iceberg_config.path_style_access,
+                &self.iceberg_config.java_catalog_props,
+            )
             .await?;
         let data_types = self.schema.data_types();
         let executor_schema_names = self.schema.names();

src/connector/src/connector_common/common.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::borrow::Cow;
 use std::collections::BTreeMap;
 use std::io::Write;
 use std::time::Duration;
@@ -38,8 +37,6 @@ use crate::deserialize_duration_from_string;
 use crate::error::ConnectorResult;
 use crate::sink::SinkError;
 use crate::source::nats::source::NatsOffset;
-// The file describes the common abstractions for each connector and can be used in both source and
-// sink.
 
 pub const PRIVATE_LINK_BROKER_REWRITE_MAP_KEY: &str = "broker.rewrite.endpoints";
 pub const PRIVATE_LINK_TARGETS_KEY: &str = "privatelink.targets";
@@ -563,13 +560,6 @@ impl KinesisCommon {
         Ok(KinesisClient::from_conf(builder.build()))
     }
 }
-#[derive(Debug, Deserialize)]
-pub struct UpsertMessage<'a> {
-    #[serde(borrow)]
-    pub primary_key: Cow<'a, [u8]>,
-    #[serde(borrow)]
-    pub record: Cow<'a, [u8]>,
-}
 
 #[serde_as]
 #[derive(Deserialize, Debug, Clone, WithOptions)]