fix base url and integreation test

Author: bingquanzhao

src/sinks/doris/client.rs

@@ -37,7 +37,7 @@ pub type ThreadSafeDorisSinkClient = Arc<DorisSinkClient>;
 #[derive(Clone, Debug)]
 pub struct DorisSinkClient {
     http_client: HttpClient,
-    base_url: String,
+    base_url: Uri,
     auth: Option<Auth>,
     compression: Compression,
     label_prefix: String,
@@ -47,7 +47,7 @@ pub struct DorisSinkClient {
 impl DorisSinkClient {
     pub async fn new(
         http_client: HttpClient,
-        base_url: String,
+        base_url: Uri,
         auth: Option<Auth>,
         compression: Compression,
         label_prefix: String,
@@ -108,9 +108,13 @@ impl DorisSinkClient {
                 StreamLoadError::InvalidRedirectUri { source }
             })?
         } else {
-            // Build original URL
-            let stream_load_url =
-                format!("{}/api/{}/{}/_stream_load", self.base_url, database, table);
+            // Build original URL using Uri components to avoid trailing slash issues
+            let scheme = self.base_url.scheme_str().unwrap_or("http");
+            let authority = self.base_url.authority().map(|a| a.as_str()).unwrap_or("");
+            let stream_load_url = format!(
+                "{}://{}/api/{}/{}/_stream_load",
+                scheme, authority, database, table
+            );
 
             stream_load_url.parse::<Uri>().map_err(|source| {
                 debug!(
@@ -293,11 +297,11 @@ impl DorisSinkClient {
         })
     }
 
-    pub async fn healthcheck_fenode(&self, endpoint: String) -> crate::Result<()> {
+    pub async fn healthcheck_fenode(&self, endpoint: &Uri) -> crate::Result<()> {
         // Use Doris bootstrap API endpoint for health check, GET method
-        let query_path = "/api/bootstrap";
-        let endpoint_str = endpoint.trim_end_matches('/');
-        let uri_str = format!("{}{}", endpoint_str, query_path);
+        let scheme = endpoint.scheme_str().unwrap_or("http");
+        let authority = endpoint.authority().map(|a| a.as_str()).unwrap_or("");
+        let uri_str = format!("{}://{}/api/bootstrap", scheme, authority);
 
         let uri = uri_str.parse::<Uri>().map_err(|source| {
             debug!(

src/sinks/doris/common.rs

@@ -10,11 +10,12 @@ use crate::{
     },
     tls::TlsSettings,
 };
+use http::Uri;
 use vector_lib::codecs::encoding::Framer;
 
 #[derive(Debug, Clone)]
 pub struct DorisCommon {
-    pub base_url: String,
+    pub base_url: Uri,
     pub auth: Option<Auth>,
     pub request_builder: DorisRequestBuilder,
     pub tls_settings: TlsSettings,
@@ -30,7 +31,7 @@ impl DorisCommon {
 
         // basic auth must be some for now
         let auth = config.auth.choose_one(&endpoint.auth)?;
-        let base_url = endpoint.uri.to_string().trim_end_matches('/').to_owned();
+        let base_url = endpoint.uri.clone();
         let tls_settings = TlsSettings::from_options(config.tls.as_ref())?;
 
         // Build encoder from the encoding configuration
@@ -61,6 +62,6 @@ impl DorisCommon {
     }
 
     pub async fn healthcheck(&self, client: ThreadSafeDorisSinkClient) -> crate::Result<()> {
-        client.healthcheck_fenode(self.base_url.clone()).await
+        client.healthcheck_fenode(&self.base_url).await
     }
 }

src/sinks/doris/config.rs

@@ -161,21 +161,22 @@ impl SinkConfig for DorisConfig {
 
         let services_futures = commons
             .iter()
-            .cloned()
             .map(|common| {
                 let client_clone = client.clone();
                 let compression = self.compression;
                 let label_prefix = self.label_prefix.clone();
                 let headers = self.headers.clone();
                 let log_request = self.log_request;
+                let base_url = common.base_url.clone();
+                let auth = common.auth.clone();
 
                 async move {
-                    let endpoint = common.base_url.clone();
+                    let endpoint = base_url.to_string();
 
                     let doris_client = DorisSinkClient::new(
                         client_clone,
-                        common.base_url.clone(),
-                        common.auth.clone(),
+                        base_url,
+                        auth,
                         compression,
                         label_prefix,
                         headers,

src/sinks/doris/integration_test.rs

@@ -84,7 +84,7 @@ struct DorisAuth {
 fn config_auth() -> DorisAuth {
     DorisAuth {
         user: "root".to_string(),
-        password: "123456".to_string(),
+        password: "".to_string(),
     }
 }
 
@@ -202,16 +202,20 @@ impl DorisTestClient {
 
         // Configure MySQL connection parameters - For Doris specifically adjusted
         // Disable these options to not send `SET @@sql_mode=CONCAT(@@sql_mode, {})` which is not supported on Doris.
-        let connect_options = MySqlConnectOptions::new()
+        let mut connect_options = MySqlConnectOptions::new()
             .host(&host)
             .port(port)
             .username(&auth.user)
-            .password(&auth.password)
             .no_engine_substitution(false) // Keep false to avoid SET statement
             .pipes_as_concat(false) // Keep false to avoid SET statement
             .ssl_mode(sqlx::mysql::MySqlSslMode::Disabled)
             .disable_statement_logging();
 
+        // Only set password if it's not empty (Doris root user has no password by default)
+        if !auth.password.is_empty() {
+            connect_options = connect_options.password(&auth.password);
+        }
+
         info!(
             message = "DorisTestClient initialized successfully.",
             internal_log_rate_limit = true

tests/integration/doris/config/compose.yaml

@@ -3,6 +3,7 @@ services:
   doris:
     image: docker.io/yagagagaga/doris-standalone:${CONFIG_VERSION}
     container_name: doris
+    hostname: doris
     restart: unless-stopped
     privileged: true
     ports:
@@ -11,6 +12,12 @@ services:
       - "8040"
     environment:
       - SKIP_CHECK_ULIMIT=true
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:8030/api/bootstrap"]
+      interval: 10s
+      timeout: 5s
+      retries: 30
+      start_period: 60s
 
 networks:
   default:

tests/integration/doris/config/test.yaml

@@ -5,7 +5,7 @@ features:
 test_filter: 'sinks::doris::integration_test::'
 
 env:
-  DORIS_ADDRESS: http://doris:8030
+  DORIS_ADDRESS: http://doris:8040
 
 matrix:
   version: ['2.1.7']