From c7e2f4c290e39aa075c5ed22abb8dffa0b4848b8 Mon Sep 17 00:00:00 2001
From: xxhZs <1060434431@qq.com>
Date: Mon, 27 Jan 2025 10:58:29 +0800
Subject: [PATCH] fix comm

---
 .../elasticsearch_opensearch_config.rs            | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_config.rs b/src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_config.rs
index 01c547ffdc476..50e9e48650182 100644
--- a/src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_config.rs
+++ b/src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_config.rs
@@ -112,6 +112,19 @@ impl ElasticSearchOpenSearchConfig {
     }
 
     pub fn build_client(&self, connector: &str) -> Result<ElasticSearchOpenSearchClient> {
+        let check_username_password = || -> Result<()> {
+            if self.username.is_some() && self.password.is_none() {
+                return Err(SinkError::Config(anyhow!(
+                    "please set the password when the username is set."
+                )));
+            }
+            if self.username.is_none() && self.password.is_some() {
+                return Err(SinkError::Config(anyhow!(
+                    "please set the username when the password is set."
+                )));
+            }
+            Ok(())
+        };
         let url =
             Url::parse(&self.url).map_err(|e| SinkError::ElasticSearchOpenSearch(anyhow!(e)))?;
         if connector.eq(ES_SINK) {
@@ -125,6 +138,7 @@ impl ElasticSearchOpenSearchConfig {
                     elasticsearch::auth::Credentials::Basic(username.clone(), password.clone()),
                 );
             }
+            check_username_password()?;
             let transport = transport_builder
                 .build()
                 .map_err(|e| SinkError::ElasticSearchOpenSearch(anyhow!(e)))?;
@@ -142,6 +156,7 @@ impl ElasticSearchOpenSearchConfig {
                     password.clone(),
                 ));
             }
+            check_username_password()?;
             let transport = transport_builder
                 .build()
                 .map_err(|e| SinkError::ElasticSearchOpenSearch(anyhow!(e)))?;