fix: fix http custom headers for rest catalog (#1010)

- Add `extra_headers` to the http client. Previously, we didn't use
`extra_headers`.
~~- Support multiple values for customer headers. Because we use a
HashMap as RestCatalog properties, we can't specify multiple values for
a single HTTP header directly. This PR parses and extracts multiple
values from the properties for each header.~~

---------

Co-authored-by: Renjie Liu <[email protected]>

Author: chenzl25

crates/catalog/rest/src/client.rs

@@ -57,12 +57,15 @@ impl Debug for HttpClient {
 impl HttpClient {
     /// Create a new http client.
     pub fn new(cfg: &RestCatalogConfig) -> Result<Self> {
+        let extra_headers = cfg.extra_headers()?;
         Ok(HttpClient {
-            client: Client::new(),
+            client: Client::builder()
+                .default_headers(extra_headers.clone())
+                .build()?,
             token: Mutex::new(cfg.token()),
             token_endpoint: cfg.get_token_endpoint(),
             credential: cfg.credential(),
-            extra_headers: cfg.extra_headers()?,
+            extra_headers,
             extra_oauth_params: cfg.extra_oauth_params(),
         })
     }
@@ -72,8 +75,14 @@ impl HttpClient {
     /// If cfg carries new value, we will use cfg instead.
     /// Otherwise, we will keep the old value.
     pub fn update_with(self, cfg: &RestCatalogConfig) -> Result<Self> {
+        let extra_headers = (!cfg.extra_headers()?.is_empty())
+            .then(|| cfg.extra_headers())
+            .transpose()?
+            .unwrap_or(self.extra_headers);
         Ok(HttpClient {
-            client: self.client,
+            client: Client::builder()
+                .default_headers(extra_headers.clone())
+                .build()?,
             token: Mutex::new(
                 cfg.token()
                     .or_else(|| self.token.into_inner().ok().flatten()),
@@ -82,10 +91,7 @@ impl HttpClient {
                 .then(|| cfg.get_token_endpoint())
                 .unwrap_or(self.token_endpoint),
             credential: cfg.credential().or(self.credential),
-            extra_headers: (!cfg.extra_headers()?.is_empty())
-                .then(|| cfg.extra_headers())
-                .transpose()?
-                .unwrap_or(self.extra_headers),
+            extra_headers,
             extra_oauth_params: (!cfg.extra_oauth_params().is_empty())
                 .then(|| cfg.extra_oauth_params())
                 .unwrap_or(self.extra_oauth_params),