From e706297c0a0eabf87c075071da26206a1e23bf1c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Collignon-Ducret=20R=C3=A9mi?=
 <miton18@users.noreply.github.com>
Date: Thu, 3 Oct 2024 17:59:17 +0200
Subject: [PATCH] Fix pulsar uri scheme (#5470)

---
 quickwit/quickwit-config/src/source_config/mod.rs | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/quickwit/quickwit-config/src/source_config/mod.rs b/quickwit/quickwit-config/src/source_config/mod.rs
index c5a255d08b3..de0790004d3 100644
--- a/quickwit/quickwit-config/src/source_config/mod.rs
+++ b/quickwit/quickwit-config/src/source_config/mod.rs
@@ -28,6 +28,7 @@ use quickwit_common::is_false;
 use quickwit_common::uri::Uri;
 use quickwit_proto::metastore::SourceType;
 use quickwit_proto::types::SourceId;
+use regex::Regex;
 use serde::de::Error;
 use serde::{Deserialize, Deserializer, Serialize};
 use serde_json::Value as JsonValue;
@@ -498,8 +499,9 @@ pub enum PulsarSourceAuth {
 fn pulsar_uri<'de, D>(deserializer: D) -> Result<String, D::Error>
 where D: Deserializer<'de> {
     let uri: String = Deserialize::deserialize(deserializer)?;
+    let re: Regex = Regex::new(r"pulsar(\+ssl)?://.*").expect("regular expression should compile");
 
-    if uri.strip_prefix("pulsar://").is_none() {
+    if !re.is_match(uri.as_str()) {
         return Err(Error::custom(format!(
             "invalid Pulsar uri provided, must be in the format of `pulsar://host:port/path`. \
              got: `{uri}`"