[PR] MSSQL Source Connector [iggy]

["lsabi (via GitHub)" <gi...@apache.org>]

lsabi opened a new pull request, #3851:
URL: https://github.com/apache/iggy/pull/3851

   ## Which issue does this PR address?
   
   Partially Closes # [https://github.com/apache/iggy/issues/3573](https://github.com/apache/iggy/issues/3573)
   Relates to #
   
   ## Rationale
   
   This is just the implementation of the source connector for the `cdc`. I'm also working on the table polling, but it'll require more time.
   
   I don't have a database instance to test it, but tests are passing. If someone, who has a test `MSSQL` instance can try it out, it would be great.
   
   ## What changed?
   
   Added the
   
   ## Local Execution
   
   - Unit Tests pass. Not tested with a real database instance, as I don't have an instance with dummy data to ruin (only production)
   - Pre-commit hooks not ran
   
   ## AI Usage
   
   None
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] MSSQL Source Connector [iggy]

["lukaszzborek (via GitHub)" <gi...@apache.org>]

lukaszzborek commented on PR #3851:
URL: https://github.com/apache/iggy/pull/3851#issuecomment-5233636531

   @lsabi please make it draft or close it and add test
   there is docker image [microsoft/mssql-server](https://hub.docker.com/r/microsoft/mssql-server) which you can use


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] MSSQL Source Connector [iggy]

["lsabi (via GitHub)" <gi...@apache.org>]

lsabi commented on PR #3851:
URL: https://github.com/apache/iggy/pull/3851#issuecomment-5267746873

   @lukaszzborek I'm afraid but I can't test it on my locale. I've spent a day understanding the docs, as it's not super clear how one should do in order to build their own connector.
   
   I've then let my laptop run for almost a day and it did not finish compiling, just excessive heating.
   
   I managed to compile `mssql_source` in a couple of hours, but not the `runtime` nor even the whole project. If you want, you can test it, but I've discovered that `iggy` and `runtime` are separate and require two different processes in order to run, which is what I did not want (I though the dynamic runtime could start up libraries on the fly and was part of the "core").


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] MSSQL Source Connector [iggy]

["github-actions[bot] (via GitHub)" <gi...@apache.org>]

github-actions[bot] commented on PR #3851:
URL: https://github.com/apache/iggy/pull/3851#issuecomment-5233562753

   <!-- iggy-pr-triage-welcome -->
   Thanks for the PR. It is labeled `S-waiting-on-review` and queued for review.
   
   Slash commands (own line, regular comment) move it around the queue:
   
   - `/ready` - back to `S-waiting-on-review` after addressing feedback
   - `/author` - flip to `S-waiting-on-author` while you finish changes
   - `/request-review @user-or-team` - request a reviewer
   
   See [CONTRIBUTING.md](https://github.com/apache/iggy/blob/master/CONTRIBUTING.md#pr-triage-commands) for details.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] MSSQL Source Connector [iggy]

["hubcio (via GitHub)" <gi...@apache.org>]

hubcio commented on PR #3851:
URL: https://github.com/apache/iggy/pull/3851#issuecomment-5757837923

   closing this, new assignment in https://github.com/apache/iggy/issues/3573


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] MSSQL Source Connector [iggy]

["hubcio (via GitHub)" <gi...@apache.org>]

hubcio commented on code in PR #3851:
URL: https://github.com/apache/iggy/pull/3851#discussion_r3783928113


##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");
+		query.bind(&cdc_tables);
+		let non_capture_tables = self.query(query).await?;
+
+		if !non_capture_tables.is_empty() {
+			warn!("Not all desired tables are monitored by CDC!");
+			if !self.config.enable_cdc.unwrap_or(false) {
+				error!("Update config to enable_cdc=true in order to monitor them!");
+				return Ok(());
+			}
+			warn!("Adding tables to the CDC monitor");
+			let schema = self.config.cdc_schema.clone().unwrap_or("dbo".to_string());
+			for row in non_capture_tables {
+				let table = row.get("name").unwrap_or("");
+				let cdc_role = match &self.config.cdc_role {
+					Some(x) => format!("N'{x}'"),
+					None => "NULL".to_string()
+				};
+				let query = format!(r#"
+					EXEC sys.sp_cdc_enable_table
+					@source_schema = N'{schema}',
+					@source_name   = N'{table}',
+					@role_name	 = {cdc_role}"#);
+
+				self.query(Query::new(query)).await?;
+			}
+		}
+
+		// For each monitored table, get the capture columns
+		let table_columns = self.query(Query::new(
+			format!(r#"
+				SELECT OBJECT_NAME(ct.object_id, DB_ID('{database}')) as table_name, c.object_id, ct.capture_instance, ct.start_lsn, c.columns FROM [{database}].[cdc].[change_tables] as ct JOIN (SELECT object_id, STRING_AGG(column_name, ',') as columns FROM [{database}].[cdc].[captured_columns] GROUP BY object_id) as c ON c.object_id = ct.object_id;"#

Review Comment:
   `OBJECT_NAME(ct.object_id, DB_ID(...))` names the change table (`dbo_users_CT`), not the source table - in `cdc.change_tables`, `object_id` is the change table's id and `source_object_id` is the source table's. so `capture_table_columns` ends up keyed by change-table names and a user-provided entry never matches the lookup on line 411. use `ct.source_object_id`.



##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");
+		query.bind(&cdc_tables);
+		let non_capture_tables = self.query(query).await?;
+
+		if !non_capture_tables.is_empty() {
+			warn!("Not all desired tables are monitored by CDC!");
+			if !self.config.enable_cdc.unwrap_or(false) {
+				error!("Update config to enable_cdc=true in order to monitor them!");
+				return Ok(());
+			}
+			warn!("Adding tables to the CDC monitor");
+			let schema = self.config.cdc_schema.clone().unwrap_or("dbo".to_string());
+			for row in non_capture_tables {
+				let table = row.get("name").unwrap_or("");
+				let cdc_role = match &self.config.cdc_role {
+					Some(x) => format!("N'{x}'"),
+					None => "NULL".to_string()
+				};
+				let query = format!(r#"
+					EXEC sys.sp_cdc_enable_table
+					@source_schema = N'{schema}',
+					@source_name   = N'{table}',
+					@role_name	 = {cdc_role}"#);
+
+				self.query(Query::new(query)).await?;
+			}
+		}
+
+		// For each monitored table, get the capture columns
+		let table_columns = self.query(Query::new(
+			format!(r#"
+				SELECT OBJECT_NAME(ct.object_id, DB_ID('{database}')) as table_name, c.object_id, ct.capture_instance, ct.start_lsn, c.columns FROM [{database}].[cdc].[change_tables] as ct JOIN (SELECT object_id, STRING_AGG(column_name, ',') as columns FROM [{database}].[cdc].[captured_columns] GROUP BY object_id) as c ON c.object_id = ct.object_id;"#
+			)
+		)).await?;
+
+		for row in table_columns {
+			let table_name = row.get::<&str, _>("table_name").unwrap().into();
+			let	first_lsn: Lsn = row.get::<&[u8], _>("start_lsn").unwrap().try_into().unwrap();
+			let	data_table = row.get::<&str, _>("capture_instance").unwrap().to_string();
+			let	columns = row.get::<&str, _>("columns").unwrap().to_string();
+
+			let columns = match self.config.capture_table_columns.get(&table_name) {
+				Some(x) => x,
+				None => &columns
+			};
+			self.config.capture_table_columns.insert(table_name, columns.clone());
+		}
+
+		info!("MSSQL CDC setup completed");
+		Ok(())
+	}
+
+
+	async fn poll_cdc(&self) -> Result<Vec<ProducedMessage>, Error> {
+
+		let q = replace_all_instances(
+			&"#db",
+			&"SELECT [#db].sys.fn_cdc_get_max_lsn() as lsn;",
+			&self.config.database
+		);
+
+		let new_lsn: Lsn = self.query(Query::new(q))
+			.await?[0]
+			.get::<&[u8], _>("lsn")
+			.unwrap()
+			.try_into()
+			.unwrap();
+		let mut state = self.state.lock().await;
+		if state.last_poll_lsn.is_some() && state.last_poll_lsn.unwrap() == new_lsn {
+			return Ok(vec![]);
+		}
+
+		let n_lsn = lsn_to_hex(&new_lsn);
+		let query_filter = if state.last_poll_lsn.is_some() {
+			let last_poll_lsn = lsn_to_hex(&state.last_poll_lsn.unwrap());
+			format!("[__$start_lsn] > {last_poll_lsn} AND [__$start_lsn] <= {n_lsn}")
+		} else {
+			format!("[__$start_lsn] <= {n_lsn}")
+		};
+
+		let capture_ops = self
+			.config
+			.capture_operations
+			.as_ref()
+			.map(|ops| ops.iter().map(|s| s.as_str()).collect::<Vec<_>>())
+			.unwrap_or_else(|| vec!["INSERT", "UPDATE", "DELETE"])
+			.into_iter()
+			.map(|e| match e {
+				"INSERT" => vec![2],
+				"UPDATE" => vec![3,4], // 3 is for delete and 4 for insert (in atomic operation UPDATE)
+				"DELETE" => vec![1],
+				e => {
+					warn!("Invalid capture operation {e}. This will be ignored");
+					vec![]
+				}
+			})
+			.flatten()
+			.collect::<Vec<_>>();
+
+		let tables: Vec<String> = self.config.capture_table_columns.values().cloned().collect();

Review Comment:
   this iterates the map's values (the column lists) as table names, so the query becomes `FROM [testdb].cdc.[id,name,active]` and fails with error 208 "invalid object name".



##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");
+		query.bind(&cdc_tables);
+		let non_capture_tables = self.query(query).await?;
+
+		if !non_capture_tables.is_empty() {
+			warn!("Not all desired tables are monitored by CDC!");
+			if !self.config.enable_cdc.unwrap_or(false) {
+				error!("Update config to enable_cdc=true in order to monitor them!");
+				return Ok(());
+			}
+			warn!("Adding tables to the CDC monitor");
+			let schema = self.config.cdc_schema.clone().unwrap_or("dbo".to_string());
+			for row in non_capture_tables {
+				let table = row.get("name").unwrap_or("");
+				let cdc_role = match &self.config.cdc_role {
+					Some(x) => format!("N'{x}'"),
+					None => "NULL".to_string()
+				};
+				let query = format!(r#"
+					EXEC sys.sp_cdc_enable_table
+					@source_schema = N'{schema}',
+					@source_name   = N'{table}',
+					@role_name	 = {cdc_role}"#);
+
+				self.query(Query::new(query)).await?;
+			}
+		}
+
+		// For each monitored table, get the capture columns
+		let table_columns = self.query(Query::new(
+			format!(r#"
+				SELECT OBJECT_NAME(ct.object_id, DB_ID('{database}')) as table_name, c.object_id, ct.capture_instance, ct.start_lsn, c.columns FROM [{database}].[cdc].[change_tables] as ct JOIN (SELECT object_id, STRING_AGG(column_name, ',') as columns FROM [{database}].[cdc].[captured_columns] GROUP BY object_id) as c ON c.object_id = ct.object_id;"#
+			)
+		)).await?;
+
+		for row in table_columns {
+			let table_name = row.get::<&str, _>("table_name").unwrap().into();
+			let	first_lsn: Lsn = row.get::<&[u8], _>("start_lsn").unwrap().try_into().unwrap();
+			let	data_table = row.get::<&str, _>("capture_instance").unwrap().to_string();
+			let	columns = row.get::<&str, _>("columns").unwrap().to_string();
+
+			let columns = match self.config.capture_table_columns.get(&table_name) {
+				Some(x) => x,
+				None => &columns
+			};
+			self.config.capture_table_columns.insert(table_name, columns.clone());
+		}
+
+		info!("MSSQL CDC setup completed");
+		Ok(())
+	}
+
+
+	async fn poll_cdc(&self) -> Result<Vec<ProducedMessage>, Error> {
+
+		let q = replace_all_instances(
+			&"#db",
+			&"SELECT [#db].sys.fn_cdc_get_max_lsn() as lsn;",
+			&self.config.database
+		);
+
+		let new_lsn: Lsn = self.query(Query::new(q))
+			.await?[0]
+			.get::<&[u8], _>("lsn")
+			.unwrap()
+			.try_into()
+			.unwrap();
+		let mut state = self.state.lock().await;
+		if state.last_poll_lsn.is_some() && state.last_poll_lsn.unwrap() == new_lsn {
+			return Ok(vec![]);
+		}
+
+		let n_lsn = lsn_to_hex(&new_lsn);
+		let query_filter = if state.last_poll_lsn.is_some() {
+			let last_poll_lsn = lsn_to_hex(&state.last_poll_lsn.unwrap());
+			format!("[__$start_lsn] > {last_poll_lsn} AND [__$start_lsn] <= {n_lsn}")
+		} else {
+			format!("[__$start_lsn] <= {n_lsn}")
+		};
+
+		let capture_ops = self
+			.config
+			.capture_operations
+			.as_ref()
+			.map(|ops| ops.iter().map(|s| s.as_str()).collect::<Vec<_>>())
+			.unwrap_or_else(|| vec!["INSERT", "UPDATE", "DELETE"])
+			.into_iter()
+			.map(|e| match e {
+				"INSERT" => vec![2],
+				"UPDATE" => vec![3,4], // 3 is for delete and 4 for insert (in atomic operation UPDATE)
+				"DELETE" => vec![1],
+				e => {
+					warn!("Invalid capture operation {e}. This will be ignored");
+					vec![]
+				}
+			})
+			.flatten()
+			.collect::<Vec<_>>();
+
+		let tables: Vec<String> = self.config.capture_table_columns.values().cloned().collect();
+		let db = self.config.database.clone();
+
+		let mut result = Vec::new();
+		let default_columns = "*".to_string();
+		for table in tables {
+			let columns = match self.config.capture_table_columns.get::<String>(&table) {
+				Some(x) => x,
+				None => &default_columns
+			};
+			let changes_query = format!(
+				r#"SELECT 
+						TODATETIMEOFFSET(
+						[#db].sys.fn_cdc_map_lsn_to_time([__$start_lsn]),
+						DATEPART(TZOFFSET, SYSDATETIMEOFFSET())
+					) as event_timestamp,
+					-- '{table}' as table,
+					[__$start_lsn]
+					,[__$end_lsn]
+					,[__$seqval]
+					,[__$update_mask]
+					,min([__$operation]) as [__$operation]
+					,[__$command_id]
+					,(
+						SELECT
+							{columns}
+						FROM [{db}].[cdc].[{table}] t_json
+						WHERE t.[__$start_lsn] = t_json.[__$start_lsn]
+							AND t.[__$update_mask] = t_json.[__$update_mask]
+							AND t.[__$command_id]= t_json.[__$command_id]
+						ORDER BY t_json.[__$command_id] ASC
+						FOR JSON AUTO
+					) as data
+				FROM [{db}].cdc.[{table}] t
+				WHERE {query_filter}
+				GROUP BY [__$start_lsn], [__$command_id], [__$seqval], [__$end_lsn], [__$update_mask]
+				ORDER BY [__$start_lsn] ASC, [__$command_id] ASC;";

Review Comment:
   leftover `;";` inside the sql string - every poll fails with error 105 "unclosed quotation mark after the character string". ran the connector against sql server 2022 with cdc enabled: as shipped it never produces a single message.



##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");
+		query.bind(&cdc_tables);
+		let non_capture_tables = self.query(query).await?;
+
+		if !non_capture_tables.is_empty() {
+			warn!("Not all desired tables are monitored by CDC!");
+			if !self.config.enable_cdc.unwrap_or(false) {
+				error!("Update config to enable_cdc=true in order to monitor them!");
+				return Ok(());
+			}
+			warn!("Adding tables to the CDC monitor");
+			let schema = self.config.cdc_schema.clone().unwrap_or("dbo".to_string());
+			for row in non_capture_tables {
+				let table = row.get("name").unwrap_or("");
+				let cdc_role = match &self.config.cdc_role {
+					Some(x) => format!("N'{x}'"),
+					None => "NULL".to_string()
+				};
+				let query = format!(r#"
+					EXEC sys.sp_cdc_enable_table
+					@source_schema = N'{schema}',
+					@source_name   = N'{table}',
+					@role_name	 = {cdc_role}"#);
+
+				self.query(Query::new(query)).await?;
+			}
+		}
+
+		// For each monitored table, get the capture columns
+		let table_columns = self.query(Query::new(
+			format!(r#"
+				SELECT OBJECT_NAME(ct.object_id, DB_ID('{database}')) as table_name, c.object_id, ct.capture_instance, ct.start_lsn, c.columns FROM [{database}].[cdc].[change_tables] as ct JOIN (SELECT object_id, STRING_AGG(column_name, ',') as columns FROM [{database}].[cdc].[captured_columns] GROUP BY object_id) as c ON c.object_id = ct.object_id;"#
+			)
+		)).await?;
+
+		for row in table_columns {
+			let table_name = row.get::<&str, _>("table_name").unwrap().into();
+			let	first_lsn: Lsn = row.get::<&[u8], _>("start_lsn").unwrap().try_into().unwrap();
+			let	data_table = row.get::<&str, _>("capture_instance").unwrap().to_string();
+			let	columns = row.get::<&str, _>("columns").unwrap().to_string();
+
+			let columns = match self.config.capture_table_columns.get(&table_name) {
+				Some(x) => x,
+				None => &columns
+			};
+			self.config.capture_table_columns.insert(table_name, columns.clone());
+		}
+
+		info!("MSSQL CDC setup completed");
+		Ok(())
+	}
+
+
+	async fn poll_cdc(&self) -> Result<Vec<ProducedMessage>, Error> {
+
+		let q = replace_all_instances(
+			&"#db",
+			&"SELECT [#db].sys.fn_cdc_get_max_lsn() as lsn;",
+			&self.config.database
+		);
+
+		let new_lsn: Lsn = self.query(Query::new(q))
+			.await?[0]
+			.get::<&[u8], _>("lsn")
+			.unwrap()
+			.try_into()
+			.unwrap();
+		let mut state = self.state.lock().await;
+		if state.last_poll_lsn.is_some() && state.last_poll_lsn.unwrap() == new_lsn {
+			return Ok(vec![]);
+		}
+
+		let n_lsn = lsn_to_hex(&new_lsn);
+		let query_filter = if state.last_poll_lsn.is_some() {
+			let last_poll_lsn = lsn_to_hex(&state.last_poll_lsn.unwrap());
+			format!("[__$start_lsn] > {last_poll_lsn} AND [__$start_lsn] <= {n_lsn}")
+		} else {
+			format!("[__$start_lsn] <= {n_lsn}")
+		};
+
+		let capture_ops = self
+			.config
+			.capture_operations
+			.as_ref()
+			.map(|ops| ops.iter().map(|s| s.as_str()).collect::<Vec<_>>())
+			.unwrap_or_else(|| vec!["INSERT", "UPDATE", "DELETE"])
+			.into_iter()
+			.map(|e| match e {
+				"INSERT" => vec![2],
+				"UPDATE" => vec![3,4], // 3 is for delete and 4 for insert (in atomic operation UPDATE)
+				"DELETE" => vec![1],
+				e => {
+					warn!("Invalid capture operation {e}. This will be ignored");
+					vec![]
+				}
+			})
+			.flatten()
+			.collect::<Vec<_>>();
+
+		let tables: Vec<String> = self.config.capture_table_columns.values().cloned().collect();
+		let db = self.config.database.clone();
+
+		let mut result = Vec::new();
+		let default_columns = "*".to_string();
+		for table in tables {
+			let columns = match self.config.capture_table_columns.get::<String>(&table) {
+				Some(x) => x,
+				None => &default_columns
+			};
+			let changes_query = format!(
+				r#"SELECT 
+						TODATETIMEOFFSET(
+						[#db].sys.fn_cdc_map_lsn_to_time([__$start_lsn]),

Review Comment:
   `#db` is only substituted in the max-lsn query; here it reaches the server literally and fails with error 4121. the surrounding query already interpolates `{db}` via `format!`.



##########
core/connectors/sources/mssql_source/Cargo.toml:
##########
@@ -0,0 +1,60 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "iggy_connector_mssql_source"
+version = "0.1.1-edge.1"
+description = "Iggy MSSQL source connector supporting CDC and table polling for message streaming platform"
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming", "mssql", "microsoft", "cdc"]
+categories = ["command-line-utilities", "database", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+publish = false
+
+[package.metadata.cargo-machete]
+ignored = ["dashmap", "futures", "simd-json"]
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+# [features]
+# Placeholder feature for future CDC using Supabase's ETL framework
+# cdc_pg_replicate = []
+
+[dependencies]
+async-trait = { workspace = true }
+base64 = { workspace = true }

Review Comment:
   `base64` is only referenced inside commented-out code - dead dependency, drop it. machete won't catch this one: its heuristic matches the crate name inside comments, so the dead code keeps it looking used.



##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");

Review Comment:
   binding the comma-joined list as one `@P1` makes `name IN (@P1)` compare against the literal string `"t1,t2"` - matches nothing for more than one table, so the auto-enable path never fires on multi-table configs.



##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");
+		query.bind(&cdc_tables);
+		let non_capture_tables = self.query(query).await?;
+
+		if !non_capture_tables.is_empty() {
+			warn!("Not all desired tables are monitored by CDC!");
+			if !self.config.enable_cdc.unwrap_or(false) {
+				error!("Update config to enable_cdc=true in order to monitor them!");
+				return Ok(());
+			}
+			warn!("Adding tables to the CDC monitor");
+			let schema = self.config.cdc_schema.clone().unwrap_or("dbo".to_string());
+			for row in non_capture_tables {
+				let table = row.get("name").unwrap_or("");
+				let cdc_role = match &self.config.cdc_role {
+					Some(x) => format!("N'{x}'"),
+					None => "NULL".to_string()
+				};
+				let query = format!(r#"
+					EXEC sys.sp_cdc_enable_table
+					@source_schema = N'{schema}',
+					@source_name   = N'{table}',
+					@role_name	 = {cdc_role}"#);
+
+				self.query(Query::new(query)).await?;
+			}
+		}
+
+		// For each monitored table, get the capture columns
+		let table_columns = self.query(Query::new(
+			format!(r#"
+				SELECT OBJECT_NAME(ct.object_id, DB_ID('{database}')) as table_name, c.object_id, ct.capture_instance, ct.start_lsn, c.columns FROM [{database}].[cdc].[change_tables] as ct JOIN (SELECT object_id, STRING_AGG(column_name, ',') as columns FROM [{database}].[cdc].[captured_columns] GROUP BY object_id) as c ON c.object_id = ct.object_id;"#
+			)
+		)).await?;
+
+		for row in table_columns {
+			let table_name = row.get::<&str, _>("table_name").unwrap().into();
+			let	first_lsn: Lsn = row.get::<&[u8], _>("start_lsn").unwrap().try_into().unwrap();
+			let	data_table = row.get::<&str, _>("capture_instance").unwrap().to_string();
+			let	columns = row.get::<&str, _>("columns").unwrap().to_string();
+
+			let columns = match self.config.capture_table_columns.get(&table_name) {
+				Some(x) => x,
+				None => &columns
+			};
+			self.config.capture_table_columns.insert(table_name, columns.clone());
+		}
+
+		info!("MSSQL CDC setup completed");
+		Ok(())
+	}
+
+
+	async fn poll_cdc(&self) -> Result<Vec<ProducedMessage>, Error> {
+
+		let q = replace_all_instances(
+			&"#db",
+			&"SELECT [#db].sys.fn_cdc_get_max_lsn() as lsn;",
+			&self.config.database
+		);
+
+		let new_lsn: Lsn = self.query(Query::new(q))
+			.await?[0]
+			.get::<&[u8], _>("lsn")
+			.unwrap()
+			.try_into()
+			.unwrap();
+		let mut state = self.state.lock().await;
+		if state.last_poll_lsn.is_some() && state.last_poll_lsn.unwrap() == new_lsn {
+			return Ok(vec![]);
+		}
+
+		let n_lsn = lsn_to_hex(&new_lsn);
+		let query_filter = if state.last_poll_lsn.is_some() {
+			let last_poll_lsn = lsn_to_hex(&state.last_poll_lsn.unwrap());
+			format!("[__$start_lsn] > {last_poll_lsn} AND [__$start_lsn] <= {n_lsn}")
+		} else {
+			format!("[__$start_lsn] <= {n_lsn}")
+		};
+
+		let capture_ops = self
+			.config
+			.capture_operations
+			.as_ref()
+			.map(|ops| ops.iter().map(|s| s.as_str()).collect::<Vec<_>>())
+			.unwrap_or_else(|| vec!["INSERT", "UPDATE", "DELETE"])
+			.into_iter()
+			.map(|e| match e {
+				"INSERT" => vec![2],
+				"UPDATE" => vec![3,4], // 3 is for delete and 4 for insert (in atomic operation UPDATE)
+				"DELETE" => vec![1],
+				e => {
+					warn!("Invalid capture operation {e}. This will be ignored");
+					vec![]
+				}
+			})
+			.flatten()
+			.collect::<Vec<_>>();
+
+		let tables: Vec<String> = self.config.capture_table_columns.values().cloned().collect();
+		let db = self.config.database.clone();
+
+		let mut result = Vec::new();
+		let default_columns = "*".to_string();
+		for table in tables {
+			let columns = match self.config.capture_table_columns.get::<String>(&table) {
+				Some(x) => x,
+				None => &default_columns
+			};
+			let changes_query = format!(
+				r#"SELECT 
+						TODATETIMEOFFSET(
+						[#db].sys.fn_cdc_map_lsn_to_time([__$start_lsn]),
+						DATEPART(TZOFFSET, SYSDATETIMEOFFSET())
+					) as event_timestamp,
+					-- '{table}' as table,
+					[__$start_lsn]
+					,[__$end_lsn]
+					,[__$seqval]
+					,[__$update_mask]
+					,min([__$operation]) as [__$operation]
+					,[__$command_id]
+					,(
+						SELECT
+							{columns}
+						FROM [{db}].[cdc].[{table}] t_json
+						WHERE t.[__$start_lsn] = t_json.[__$start_lsn]
+							AND t.[__$update_mask] = t_json.[__$update_mask]
+							AND t.[__$command_id]= t_json.[__$command_id]
+						ORDER BY t_json.[__$command_id] ASC
+						FOR JSON AUTO
+					) as data
+				FROM [{db}].cdc.[{table}] t
+				WHERE {query_filter}
+				GROUP BY [__$start_lsn], [__$command_id], [__$seqval], [__$end_lsn], [__$update_mask]
+				ORDER BY [__$start_lsn] ASC, [__$command_id] ASC;";
+				"#);
+
+			let rows = self.query(Query::new(changes_query)).await?;
+			result.push((table, rows));
+		}
+
+		let mut messages = Vec::new();
+		for (table_name, row) in result {
+			for r in row {
+				let data: String = match r.get::<&str, _>("data") {
+					Some(x) => x.to_string(),
+					None => return Err(Error::InvalidRecord)
+				};
+
+				let operation_type = match r.get::<i64, _>("[__$operation]").unwrap_or(0) {

Review Comment:
   this panics instead of erroring: the result column is `__$operation` (no brackets) and it's an `int`, so both the name and the `i64` type are wrong. tiberius `get` is `try_get(..).unwrap()`, and `try_get` errors on an unknown column name as well as on a failed conversion - so `unwrap_or(0)` is unreachable. the panic lands inside the task the sdk spawns for `poll()`, killing the polling task silently: the connector still reports as running and never produces again. use `try_get` with `i32` and return an error.



##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");
+		query.bind(&cdc_tables);
+		let non_capture_tables = self.query(query).await?;
+
+		if !non_capture_tables.is_empty() {
+			warn!("Not all desired tables are monitored by CDC!");
+			if !self.config.enable_cdc.unwrap_or(false) {
+				error!("Update config to enable_cdc=true in order to monitor them!");
+				return Ok(());
+			}
+			warn!("Adding tables to the CDC monitor");
+			let schema = self.config.cdc_schema.clone().unwrap_or("dbo".to_string());
+			for row in non_capture_tables {
+				let table = row.get("name").unwrap_or("");
+				let cdc_role = match &self.config.cdc_role {
+					Some(x) => format!("N'{x}'"),
+					None => "NULL".to_string()
+				};
+				let query = format!(r#"
+					EXEC sys.sp_cdc_enable_table
+					@source_schema = N'{schema}',
+					@source_name   = N'{table}',
+					@role_name	 = {cdc_role}"#);
+
+				self.query(Query::new(query)).await?;
+			}
+		}
+
+		// For each monitored table, get the capture columns
+		let table_columns = self.query(Query::new(
+			format!(r#"
+				SELECT OBJECT_NAME(ct.object_id, DB_ID('{database}')) as table_name, c.object_id, ct.capture_instance, ct.start_lsn, c.columns FROM [{database}].[cdc].[change_tables] as ct JOIN (SELECT object_id, STRING_AGG(column_name, ',') as columns FROM [{database}].[cdc].[captured_columns] GROUP BY object_id) as c ON c.object_id = ct.object_id;"#
+			)
+		)).await?;
+
+		for row in table_columns {
+			let table_name = row.get::<&str, _>("table_name").unwrap().into();
+			let	first_lsn: Lsn = row.get::<&[u8], _>("start_lsn").unwrap().try_into().unwrap();
+			let	data_table = row.get::<&str, _>("capture_instance").unwrap().to_string();
+			let	columns = row.get::<&str, _>("columns").unwrap().to_string();
+
+			let columns = match self.config.capture_table_columns.get(&table_name) {
+				Some(x) => x,
+				None => &columns
+			};
+			self.config.capture_table_columns.insert(table_name, columns.clone());
+		}
+
+		info!("MSSQL CDC setup completed");
+		Ok(())
+	}
+
+
+	async fn poll_cdc(&self) -> Result<Vec<ProducedMessage>, Error> {
+
+		let q = replace_all_instances(
+			&"#db",
+			&"SELECT [#db].sys.fn_cdc_get_max_lsn() as lsn;",
+			&self.config.database
+		);
+
+		let new_lsn: Lsn = self.query(Query::new(q))
+			.await?[0]
+			.get::<&[u8], _>("lsn")
+			.unwrap()
+			.try_into()
+			.unwrap();
+		let mut state = self.state.lock().await;
+		if state.last_poll_lsn.is_some() && state.last_poll_lsn.unwrap() == new_lsn {
+			return Ok(vec![]);
+		}
+
+		let n_lsn = lsn_to_hex(&new_lsn);
+		let query_filter = if state.last_poll_lsn.is_some() {
+			let last_poll_lsn = lsn_to_hex(&state.last_poll_lsn.unwrap());
+			format!("[__$start_lsn] > {last_poll_lsn} AND [__$start_lsn] <= {n_lsn}")
+		} else {
+			format!("[__$start_lsn] <= {n_lsn}")
+		};
+
+		let capture_ops = self
+			.config
+			.capture_operations
+			.as_ref()
+			.map(|ops| ops.iter().map(|s| s.as_str()).collect::<Vec<_>>())
+			.unwrap_or_else(|| vec!["INSERT", "UPDATE", "DELETE"])
+			.into_iter()
+			.map(|e| match e {
+				"INSERT" => vec![2],
+				"UPDATE" => vec![3,4], // 3 is for delete and 4 for insert (in atomic operation UPDATE)
+				"DELETE" => vec![1],
+				e => {
+					warn!("Invalid capture operation {e}. This will be ignored");
+					vec![]
+				}
+			})
+			.flatten()
+			.collect::<Vec<_>>();
+
+		let tables: Vec<String> = self.config.capture_table_columns.values().cloned().collect();
+		let db = self.config.database.clone();
+
+		let mut result = Vec::new();
+		let default_columns = "*".to_string();
+		for table in tables {
+			let columns = match self.config.capture_table_columns.get::<String>(&table) {
+				Some(x) => x,
+				None => &default_columns
+			};
+			let changes_query = format!(
+				r#"SELECT 
+						TODATETIMEOFFSET(
+						[#db].sys.fn_cdc_map_lsn_to_time([__$start_lsn]),
+						DATEPART(TZOFFSET, SYSDATETIMEOFFSET())
+					) as event_timestamp,
+					-- '{table}' as table,
+					[__$start_lsn]
+					,[__$end_lsn]
+					,[__$seqval]
+					,[__$update_mask]
+					,min([__$operation]) as [__$operation]
+					,[__$command_id]
+					,(
+						SELECT
+							{columns}
+						FROM [{db}].[cdc].[{table}] t_json
+						WHERE t.[__$start_lsn] = t_json.[__$start_lsn]
+							AND t.[__$update_mask] = t_json.[__$update_mask]
+							AND t.[__$command_id]= t_json.[__$command_id]
+						ORDER BY t_json.[__$command_id] ASC
+						FOR JSON AUTO
+					) as data
+				FROM [{db}].cdc.[{table}] t
+				WHERE {query_filter}
+				GROUP BY [__$start_lsn], [__$command_id], [__$seqval], [__$end_lsn], [__$update_mask]
+				ORDER BY [__$start_lsn] ASC, [__$command_id] ASC;";
+				"#);
+
+			let rows = self.query(Query::new(changes_query)).await?;
+			result.push((table, rows));
+		}
+
+		let mut messages = Vec::new();
+		for (table_name, row) in result {
+			for r in row {
+				let data: String = match r.get::<&str, _>("data") {
+					Some(x) => x.to_string(),
+					None => return Err(Error::InvalidRecord)
+				};
+
+				let operation_type = match r.get::<i64, _>("[__$operation]").unwrap_or(0) {
+						1 => "DELETE".to_string(),
+						2 => "INSERT".to_string(),
+						3 => "UPDATE".to_string(),
+						4 => "UPDATE".to_string(),
+						e => {
+							error!("Operation type not recognized! {e}");
+							return Err(Error::InvalidRecord)
+						}
+					};
+
+				let timestamp = match r.get::<&str, _>("event_timestamp") {

Review Comment:
   another panic: `event_timestamp` is a `datetimeoffset`, so `get::<&str>` blows up. even with a string conversion the format doesn't match `%Y-%m-%d %H:%M:%S %:z`, and the parsed value is discarded anyway - `DatabaseRecord.timestamp` below uses `Utc::now()`.



##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");
+		query.bind(&cdc_tables);
+		let non_capture_tables = self.query(query).await?;
+
+		if !non_capture_tables.is_empty() {
+			warn!("Not all desired tables are monitored by CDC!");
+			if !self.config.enable_cdc.unwrap_or(false) {
+				error!("Update config to enable_cdc=true in order to monitor them!");
+				return Ok(());
+			}
+			warn!("Adding tables to the CDC monitor");
+			let schema = self.config.cdc_schema.clone().unwrap_or("dbo".to_string());
+			for row in non_capture_tables {
+				let table = row.get("name").unwrap_or("");
+				let cdc_role = match &self.config.cdc_role {
+					Some(x) => format!("N'{x}'"),
+					None => "NULL".to_string()
+				};
+				let query = format!(r#"
+					EXEC sys.sp_cdc_enable_table
+					@source_schema = N'{schema}',
+					@source_name   = N'{table}',
+					@role_name	 = {cdc_role}"#);
+
+				self.query(Query::new(query)).await?;
+			}
+		}
+
+		// For each monitored table, get the capture columns
+		let table_columns = self.query(Query::new(
+			format!(r#"
+				SELECT OBJECT_NAME(ct.object_id, DB_ID('{database}')) as table_name, c.object_id, ct.capture_instance, ct.start_lsn, c.columns FROM [{database}].[cdc].[change_tables] as ct JOIN (SELECT object_id, STRING_AGG(column_name, ',') as columns FROM [{database}].[cdc].[captured_columns] GROUP BY object_id) as c ON c.object_id = ct.object_id;"#
+			)
+		)).await?;
+
+		for row in table_columns {
+			let table_name = row.get::<&str, _>("table_name").unwrap().into();
+			let	first_lsn: Lsn = row.get::<&[u8], _>("start_lsn").unwrap().try_into().unwrap();
+			let	data_table = row.get::<&str, _>("capture_instance").unwrap().to_string();
+			let	columns = row.get::<&str, _>("columns").unwrap().to_string();
+
+			let columns = match self.config.capture_table_columns.get(&table_name) {
+				Some(x) => x,
+				None => &columns
+			};
+			self.config.capture_table_columns.insert(table_name, columns.clone());
+		}
+
+		info!("MSSQL CDC setup completed");
+		Ok(())
+	}
+
+
+	async fn poll_cdc(&self) -> Result<Vec<ProducedMessage>, Error> {
+
+		let q = replace_all_instances(
+			&"#db",
+			&"SELECT [#db].sys.fn_cdc_get_max_lsn() as lsn;",
+			&self.config.database
+		);
+
+		let new_lsn: Lsn = self.query(Query::new(q))
+			.await?[0]
+			.get::<&[u8], _>("lsn")
+			.unwrap()
+			.try_into()
+			.unwrap();
+		let mut state = self.state.lock().await;
+		if state.last_poll_lsn.is_some() && state.last_poll_lsn.unwrap() == new_lsn {
+			return Ok(vec![]);
+		}
+
+		let n_lsn = lsn_to_hex(&new_lsn);
+		let query_filter = if state.last_poll_lsn.is_some() {
+			let last_poll_lsn = lsn_to_hex(&state.last_poll_lsn.unwrap());
+			format!("[__$start_lsn] > {last_poll_lsn} AND [__$start_lsn] <= {n_lsn}")
+		} else {
+			format!("[__$start_lsn] <= {n_lsn}")
+		};
+
+		let capture_ops = self
+			.config
+			.capture_operations
+			.as_ref()
+			.map(|ops| ops.iter().map(|s| s.as_str()).collect::<Vec<_>>())
+			.unwrap_or_else(|| vec!["INSERT", "UPDATE", "DELETE"])
+			.into_iter()
+			.map(|e| match e {
+				"INSERT" => vec![2],
+				"UPDATE" => vec![3,4], // 3 is for delete and 4 for insert (in atomic operation UPDATE)
+				"DELETE" => vec![1],
+				e => {
+					warn!("Invalid capture operation {e}. This will be ignored");
+					vec![]
+				}
+			})
+			.flatten()
+			.collect::<Vec<_>>();
+
+		let tables: Vec<String> = self.config.capture_table_columns.values().cloned().collect();
+		let db = self.config.database.clone();
+
+		let mut result = Vec::new();
+		let default_columns = "*".to_string();
+		for table in tables {
+			let columns = match self.config.capture_table_columns.get::<String>(&table) {
+				Some(x) => x,
+				None => &default_columns
+			};
+			let changes_query = format!(
+				r#"SELECT 
+						TODATETIMEOFFSET(
+						[#db].sys.fn_cdc_map_lsn_to_time([__$start_lsn]),
+						DATEPART(TZOFFSET, SYSDATETIMEOFFSET())
+					) as event_timestamp,
+					-- '{table}' as table,
+					[__$start_lsn]
+					,[__$end_lsn]
+					,[__$seqval]
+					,[__$update_mask]
+					,min([__$operation]) as [__$operation]
+					,[__$command_id]
+					,(
+						SELECT
+							{columns}
+						FROM [{db}].[cdc].[{table}] t_json
+						WHERE t.[__$start_lsn] = t_json.[__$start_lsn]
+							AND t.[__$update_mask] = t_json.[__$update_mask]
+							AND t.[__$command_id]= t_json.[__$command_id]
+						ORDER BY t_json.[__$command_id] ASC
+						FOR JSON AUTO
+					) as data
+				FROM [{db}].cdc.[{table}] t
+				WHERE {query_filter}
+				GROUP BY [__$start_lsn], [__$command_id], [__$seqval], [__$end_lsn], [__$update_mask]
+				ORDER BY [__$start_lsn] ASC, [__$command_id] ASC;";
+				"#);
+
+			let rows = self.query(Query::new(changes_query)).await?;
+			result.push((table, rows));
+		}
+
+		let mut messages = Vec::new();
+		for (table_name, row) in result {
+			for r in row {
+				let data: String = match r.get::<&str, _>("data") {
+					Some(x) => x.to_string(),
+					None => return Err(Error::InvalidRecord)
+				};
+
+				let operation_type = match r.get::<i64, _>("[__$operation]").unwrap_or(0) {
+						1 => "DELETE".to_string(),
+						2 => "INSERT".to_string(),
+						3 => "UPDATE".to_string(),
+						4 => "UPDATE".to_string(),
+						e => {
+							error!("Operation type not recognized! {e}");
+							return Err(Error::InvalidRecord)
+						}
+					};
+
+				let timestamp = match r.get::<&str, _>("event_timestamp") {
+					Some(x) => match DateTime::parse_from_str(x, "%Y-%m-%d %H:%M:%S %:z") {
+						Ok(y) => y.with_timezone(&Utc),
+						Err(y) => {
+							warn!("Cannot parse datetime! {y}");
+							Utc::now()
+						}
+					}
+					None => Utc::now()
+				};
+
+				let parsed_data = parse_record_data(&data);

Review Comment:
   `data` is json (`FOR JSON AUTO`) but this runs it through the text-format splitter, producing mangled output like `"data":{"":"1,\"name\":\"alice\",...}]"}`. parse it with `serde_json::from_str`. update events also arrive with before+after images concatenated in one array while `old_data` stays null.



##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {

Review Comment:
   "polling" is accepted here but `poll()` returns `Err(InvalidConfig)` forever since `poll_tables` is commented out - reject the mode in `open()` until it exists.



##########
Cargo.toml:
##########
@@ -50,6 +50,7 @@ members = [
     "core/connectors/sources/elasticsearch_source",
     "core/connectors/sources/influxdb_source",
     "core/connectors/sources/postgres_source",
+    "core/connectors/sources/mssql_source",

Review Comment:
   `Cargo.lock` isn't part of the PR - building adds ~238 lines (tiberius, async-native-tls, connection-string, ...). commit the lock update.



##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");
+		query.bind(&cdc_tables);
+		let non_capture_tables = self.query(query).await?;
+
+		if !non_capture_tables.is_empty() {
+			warn!("Not all desired tables are monitored by CDC!");
+			if !self.config.enable_cdc.unwrap_or(false) {
+				error!("Update config to enable_cdc=true in order to monitor them!");
+				return Ok(());
+			}
+			warn!("Adding tables to the CDC monitor");
+			let schema = self.config.cdc_schema.clone().unwrap_or("dbo".to_string());
+			for row in non_capture_tables {
+				let table = row.get("name").unwrap_or("");
+				let cdc_role = match &self.config.cdc_role {
+					Some(x) => format!("N'{x}'"),
+					None => "NULL".to_string()
+				};
+				let query = format!(r#"
+					EXEC sys.sp_cdc_enable_table
+					@source_schema = N'{schema}',
+					@source_name   = N'{table}',
+					@role_name	 = {cdc_role}"#);
+
+				self.query(Query::new(query)).await?;
+			}
+		}
+
+		// For each monitored table, get the capture columns
+		let table_columns = self.query(Query::new(
+			format!(r#"
+				SELECT OBJECT_NAME(ct.object_id, DB_ID('{database}')) as table_name, c.object_id, ct.capture_instance, ct.start_lsn, c.columns FROM [{database}].[cdc].[change_tables] as ct JOIN (SELECT object_id, STRING_AGG(column_name, ',') as columns FROM [{database}].[cdc].[captured_columns] GROUP BY object_id) as c ON c.object_id = ct.object_id;"#
+			)
+		)).await?;
+
+		for row in table_columns {
+			let table_name = row.get::<&str, _>("table_name").unwrap().into();
+			let	first_lsn: Lsn = row.get::<&[u8], _>("start_lsn").unwrap().try_into().unwrap();
+			let	data_table = row.get::<&str, _>("capture_instance").unwrap().to_string();
+			let	columns = row.get::<&str, _>("columns").unwrap().to_string();
+
+			let columns = match self.config.capture_table_columns.get(&table_name) {
+				Some(x) => x,
+				None => &columns
+			};
+			self.config.capture_table_columns.insert(table_name, columns.clone());
+		}
+
+		info!("MSSQL CDC setup completed");
+		Ok(())
+	}
+
+
+	async fn poll_cdc(&self) -> Result<Vec<ProducedMessage>, Error> {
+
+		let q = replace_all_instances(
+			&"#db",
+			&"SELECT [#db].sys.fn_cdc_get_max_lsn() as lsn;",
+			&self.config.database
+		);
+
+		let new_lsn: Lsn = self.query(Query::new(q))
+			.await?[0]
+			.get::<&[u8], _>("lsn")
+			.unwrap()
+			.try_into()
+			.unwrap();
+		let mut state = self.state.lock().await;
+		if state.last_poll_lsn.is_some() && state.last_poll_lsn.unwrap() == new_lsn {
+			return Ok(vec![]);
+		}
+
+		let n_lsn = lsn_to_hex(&new_lsn);
+		let query_filter = if state.last_poll_lsn.is_some() {
+			let last_poll_lsn = lsn_to_hex(&state.last_poll_lsn.unwrap());
+			format!("[__$start_lsn] > {last_poll_lsn} AND [__$start_lsn] <= {n_lsn}")
+		} else {
+			format!("[__$start_lsn] <= {n_lsn}")
+		};
+
+		let capture_ops = self
+			.config
+			.capture_operations
+			.as_ref()
+			.map(|ops| ops.iter().map(|s| s.as_str()).collect::<Vec<_>>())
+			.unwrap_or_else(|| vec!["INSERT", "UPDATE", "DELETE"])
+			.into_iter()
+			.map(|e| match e {
+				"INSERT" => vec![2],
+				"UPDATE" => vec![3,4], // 3 is for delete and 4 for insert (in atomic operation UPDATE)
+				"DELETE" => vec![1],
+				e => {
+					warn!("Invalid capture operation {e}. This will be ignored");
+					vec![]
+				}
+			})
+			.flatten()
+			.collect::<Vec<_>>();
+
+		let tables: Vec<String> = self.config.capture_table_columns.values().cloned().collect();
+		let db = self.config.database.clone();
+
+		let mut result = Vec::new();
+		let default_columns = "*".to_string();
+		for table in tables {
+			let columns = match self.config.capture_table_columns.get::<String>(&table) {
+				Some(x) => x,
+				None => &default_columns
+			};
+			let changes_query = format!(
+				r#"SELECT 
+						TODATETIMEOFFSET(
+						[#db].sys.fn_cdc_map_lsn_to_time([__$start_lsn]),
+						DATEPART(TZOFFSET, SYSDATETIMEOFFSET())
+					) as event_timestamp,
+					-- '{table}' as table,
+					[__$start_lsn]
+					,[__$end_lsn]
+					,[__$seqval]
+					,[__$update_mask]
+					,min([__$operation]) as [__$operation]
+					,[__$command_id]
+					,(
+						SELECT
+							{columns}
+						FROM [{db}].[cdc].[{table}] t_json
+						WHERE t.[__$start_lsn] = t_json.[__$start_lsn]
+							AND t.[__$update_mask] = t_json.[__$update_mask]
+							AND t.[__$command_id]= t_json.[__$command_id]
+						ORDER BY t_json.[__$command_id] ASC
+						FOR JSON AUTO
+					) as data
+				FROM [{db}].cdc.[{table}] t
+				WHERE {query_filter}
+				GROUP BY [__$start_lsn], [__$command_id], [__$seqval], [__$end_lsn], [__$update_mask]
+				ORDER BY [__$start_lsn] ASC, [__$command_id] ASC;";
+				"#);
+
+			let rows = self.query(Query::new(changes_query)).await?;
+			result.push((table, rows));
+		}
+
+		let mut messages = Vec::new();
+		for (table_name, row) in result {
+			for r in row {
+				let data: String = match r.get::<&str, _>("data") {
+					Some(x) => x.to_string(),
+					None => return Err(Error::InvalidRecord)
+				};
+
+				let operation_type = match r.get::<i64, _>("[__$operation]").unwrap_or(0) {
+						1 => "DELETE".to_string(),
+						2 => "INSERT".to_string(),
+						3 => "UPDATE".to_string(),
+						4 => "UPDATE".to_string(),
+						e => {
+							error!("Operation type not recognized! {e}");
+							return Err(Error::InvalidRecord)
+						}
+					};
+
+				let timestamp = match r.get::<&str, _>("event_timestamp") {
+					Some(x) => match DateTime::parse_from_str(x, "%Y-%m-%d %H:%M:%S %:z") {
+						Ok(y) => y.with_timezone(&Utc),
+						Err(y) => {
+							warn!("Cannot parse datetime! {y}");
+							Utc::now()
+						}
+					}
+					None => Utc::now()
+				};
+
+				let parsed_data = parse_record_data(&data);
+				let change_record = DatabaseRecord {
+					table_name: table_name.to_string(),
+					operation_type,
+					timestamp: Utc::now(),
+					data: serde_json::Value::Object(parsed_data),
+					old_data: None
+				};
+				let payload =
+					simd_json::to_vec(&change_record).map_err(|_| Error::InvalidRecord)?;
+
+				let message = ProducedMessage {
+					id: Some(Uuid::new_v4().as_u128()),
+					headers: None,
+					checksum: None,
+					timestamp: Some(Utc::now().timestamp_millis() as u64),
+					origin_timestamp: Some(Utc::now().timestamp_millis() as u64),
+					payload
+				};
+				messages.push(message);
+			}
+		}
+
+		// Update state with minimal lock time
+		if !messages.is_empty() {
+			// let mut state = self.state.lock().await;
+			state.processed_rows += messages.len() as u64;
+			state.last_poll_lsn = Some(new_lsn);
+		}
+
+		if self.verbose {
+			info!("CDC: Fetched {} change records", messages.len());
+		} else {
+			debug!("CDC: Fetched {} change records", messages.len());
+		}
+		Ok(messages)
+	}
+
+
+	/*

Review Comment:
   over half the file is commented-out postgres source code (poll_tables, sqlx type mapping, nearly all tests) - please delete it, git keeps history. the crate also fails `cargo fmt --check` (tabs) and `cargo clippy -- -D warnings` (21 errors), both are ci gates.



##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");
+		query.bind(&cdc_tables);
+		let non_capture_tables = self.query(query).await?;
+
+		if !non_capture_tables.is_empty() {
+			warn!("Not all desired tables are monitored by CDC!");
+			if !self.config.enable_cdc.unwrap_or(false) {
+				error!("Update config to enable_cdc=true in order to monitor them!");
+				return Ok(());
+			}
+			warn!("Adding tables to the CDC monitor");
+			let schema = self.config.cdc_schema.clone().unwrap_or("dbo".to_string());
+			for row in non_capture_tables {
+				let table = row.get("name").unwrap_or("");
+				let cdc_role = match &self.config.cdc_role {
+					Some(x) => format!("N'{x}'"),
+					None => "NULL".to_string()
+				};
+				let query = format!(r#"
+					EXEC sys.sp_cdc_enable_table
+					@source_schema = N'{schema}',
+					@source_name   = N'{table}',
+					@role_name	 = {cdc_role}"#);
+
+				self.query(Query::new(query)).await?;
+			}
+		}
+
+		// For each monitored table, get the capture columns
+		let table_columns = self.query(Query::new(

Review Comment:
   no filter on the configured tables here, so the connector streams every cdc-enabled table in the database - with `tables = ["users"]` configured, rows from an unconfigured `orders` table came through.



##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");
+		query.bind(&cdc_tables);
+		let non_capture_tables = self.query(query).await?;
+
+		if !non_capture_tables.is_empty() {
+			warn!("Not all desired tables are monitored by CDC!");
+			if !self.config.enable_cdc.unwrap_or(false) {
+				error!("Update config to enable_cdc=true in order to monitor them!");
+				return Ok(());
+			}
+			warn!("Adding tables to the CDC monitor");
+			let schema = self.config.cdc_schema.clone().unwrap_or("dbo".to_string());
+			for row in non_capture_tables {
+				let table = row.get("name").unwrap_or("");
+				let cdc_role = match &self.config.cdc_role {
+					Some(x) => format!("N'{x}'"),
+					None => "NULL".to_string()
+				};
+				let query = format!(r#"
+					EXEC sys.sp_cdc_enable_table
+					@source_schema = N'{schema}',
+					@source_name   = N'{table}',
+					@role_name	 = {cdc_role}"#);
+
+				self.query(Query::new(query)).await?;
+			}
+		}
+
+		// For each monitored table, get the capture columns
+		let table_columns = self.query(Query::new(
+			format!(r#"
+				SELECT OBJECT_NAME(ct.object_id, DB_ID('{database}')) as table_name, c.object_id, ct.capture_instance, ct.start_lsn, c.columns FROM [{database}].[cdc].[change_tables] as ct JOIN (SELECT object_id, STRING_AGG(column_name, ',') as columns FROM [{database}].[cdc].[captured_columns] GROUP BY object_id) as c ON c.object_id = ct.object_id;"#
+			)
+		)).await?;
+
+		for row in table_columns {
+			let table_name = row.get::<&str, _>("table_name").unwrap().into();
+			let	first_lsn: Lsn = row.get::<&[u8], _>("start_lsn").unwrap().try_into().unwrap();
+			let	data_table = row.get::<&str, _>("capture_instance").unwrap().to_string();
+			let	columns = row.get::<&str, _>("columns").unwrap().to_string();
+
+			let columns = match self.config.capture_table_columns.get(&table_name) {
+				Some(x) => x,
+				None => &columns
+			};
+			self.config.capture_table_columns.insert(table_name, columns.clone());
+		}
+
+		info!("MSSQL CDC setup completed");
+		Ok(())
+	}
+
+
+	async fn poll_cdc(&self) -> Result<Vec<ProducedMessage>, Error> {
+
+		let q = replace_all_instances(
+			&"#db",
+			&"SELECT [#db].sys.fn_cdc_get_max_lsn() as lsn;",
+			&self.config.database
+		);
+
+		let new_lsn: Lsn = self.query(Query::new(q))
+			.await?[0]
+			.get::<&[u8], _>("lsn")
+			.unwrap()
+			.try_into()
+			.unwrap();
+		let mut state = self.state.lock().await;
+		if state.last_poll_lsn.is_some() && state.last_poll_lsn.unwrap() == new_lsn {
+			return Ok(vec![]);
+		}
+
+		let n_lsn = lsn_to_hex(&new_lsn);
+		let query_filter = if state.last_poll_lsn.is_some() {
+			let last_poll_lsn = lsn_to_hex(&state.last_poll_lsn.unwrap());
+			format!("[__$start_lsn] > {last_poll_lsn} AND [__$start_lsn] <= {n_lsn}")
+		} else {
+			format!("[__$start_lsn] <= {n_lsn}")
+		};
+
+		let capture_ops = self

Review Comment:
   `capture_ops` is built but never used - the `capture_operations` filter is silently ignored and every operation type gets streamed.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] MSSQL Source Connector [iggy]

["codecov[bot] (via GitHub)" <gi...@apache.org>]

codecov[bot] commented on PR #3851:
URL: https://github.com/apache/iggy/pull/3851#issuecomment-5237718203

   ## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3851?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :white_check_mark: All modified and coverable lines are covered by tests.
   :white_check_mark: Project coverage is 76.14%. Comparing base ([`01a64b2`](https://app.codecov.io/gh/apache/iggy/commit/01a64b2e00eb3037d45a85e4cdefe09eed6e2e07?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`3c54da0`](https://app.codecov.io/gh/apache/iggy/commit/3c54da01b9b672c6b2c6f62048b3898e7b8e5e15?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   :warning: Report is 1 commits behind head on master.
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   ```diff
   @@             Coverage Diff              @@
   ##             master    #3851      +/-   ##
   ============================================
   - Coverage     76.58%   76.14%   -0.44%     
     Complexity     1046     1046              
   ============================================
     Files          1347     1345       -2     
     Lines        171018   170768     -250     
     Branches     142372   142316      -56     
   ============================================
   - Hits         130967   130032     -935     
   - Misses        36233    36915     +682     
   - Partials       3818     3821       +3     
   ```
   
   | [Components](https://app.codecov.io/gh/apache/iggy/pull/3851/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3851/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.78% <ø> (ø)` | |
   | [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3851/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `63.67% <ø> (ø)` | |
   | [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3851/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.28% <ø> (ø)` | |
   | [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3851/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `88.70% <ø> (ø)` | |
   | [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3851/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `82.97% <ø> (ø)` | |
   | [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3851/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `89.68% <ø> (-6.61%)` | :arrow_down: |
   | [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3851/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `69.18% <ø> (ø)` | |
   [see 52 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3851/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   </details>
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] MSSQL Source Connector [iggy]

["lsabi (via GitHub)" <gi...@apache.org>]

lsabi commented on code in PR #3851:
URL: https://github.com/apache/iggy/pull/3851#discussion_r3789014420


##########
core/connectors/sources/mssql_source/src/lib.rs:
##########
@@ -0,0 +1,2103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use regex::Regex;
+use async_trait::async_trait;
+use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
+use humantime::Duration as HumanDuration;
+use iggy_common::{DateTime, Utc};
+use iggy_connector_sdk::{
+	ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use tiberius::{Client, Config, AuthMethod, Query, Row};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+
+source_connector!(MSSQLSource);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+
+
+// Define a new type for the client with the generic type incorporated
+pub type DBClient = Client<Compat<TcpStream>>;
+
+pub(crate) type Lsn = [u8; 10];
+
+/// Convert LSN parameter into a binary string
+pub(crate) fn lsn_to_hex(lsn: &Lsn) -> String {
+
+	let mut hex_string = String::with_capacity(lsn.len() * 2);
+	// Append the leading 0s and push a hex upper case representation
+	hex_string.push_str("0x");
+	for byte in lsn.into_iter() {
+		hex_string.push_str(&format!("{:02X}", byte));
+	}
+	hex_string
+}
+
+
+#[derive(Debug)]
+pub struct MSSQLSource {
+	pub id: u32,
+	client: Mutex<Option<DBClient>>,
+	config: MSSQLSourceConfig,
+	state: Mutex<State>,
+	verbose: bool,
+	retry_delay: Duration,
+	poll_interval: Duration,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MSSQLSourceConfig {
+	// Example jdbc:sqlserver://localhost;encrypt=true;user=MyUserName;password=<password>;
+	// See https://learn.microsoft.com/en-us/sql/connect/jdbc/building-the-connection-url?view=sql-server-ver15
+	#[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+	pub connection_string: SecretString,
+	pub database: String,
+	pub mode: String,
+	pub tables: Vec<String>,
+	pub poll_interval: Option<String>,
+	pub batch_size: Option<u32>,
+	pub tracking_column: Option<String>,
+	pub initial_offset: Option<String>,
+	// pub max_connections: Option<u32>,
+	pub enable_cdc: Option<bool>,
+	pub custom_query: Option<String>,
+	pub snake_case_columns: Option<bool>,
+	pub include_metadata: Option<bool>,
+	pub capture_operations: Option<Vec<String>>,
+	pub delete_after_read: Option<bool>,
+	pub processed_column: Option<String>,
+	pub primary_key_column: Option<String>,
+	pub payload_column: Option<String>,
+	pub payload_format: Option<String>,
+	pub verbose_logging: Option<bool>,
+	pub max_retries: Option<u32>,
+	pub retry_delay: Option<String>,
+	pub cdc_schema: Option<String>,
+	pub cdc_role: Option<String>,
+	pub capture_table_columns: HashMap<String, String>
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+	#[default]
+	Json,
+	Bytea,
+	Text,
+	JsonDirect,
+}
+
+impl PayloadFormat {
+	fn from_config(s: Option<&str>) -> Self {
+		match s.map(|s| s.to_lowercase()).as_deref() {
+			Some("bytea") | Some("raw") => PayloadFormat::Bytea,
+			Some("text") => PayloadFormat::Text,
+			Some("json_direct") | Some("jsonb") | Some("jsonb_direct") => PayloadFormat::JsonDirect,
+			_ => PayloadFormat::Json,
+		}
+	}
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+	last_poll_lsn: Option<Lsn>,
+	start_lsn: Option<Lsn>,
+	end_lsn: Option<Lsn>,
+	last_poll_time: DateTime<Utc>,
+	tracking_offsets: HashMap<String, String>,
+	processed_rows: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseRecord {
+	pub table_name: String,
+	pub operation_type: String,
+	pub timestamp: DateTime<Utc>,
+	pub data: serde_json::Value,
+	pub old_data: Option<serde_json::Value>,
+}
+
+#[derive(Clone, Copy)]
+struct RowProcessingConfig<'a> {
+	table: &'a str,
+	tracking_column: &'a str,
+	pk_column: &'a str,
+	payload_format: PayloadFormat,
+	payload_col: &'a str,
+	snake_case_columns: bool,
+	include_metadata: bool,
+}
+
+struct ProcessedRow {
+	message: ProducedMessage,
+	max_offset: Option<String>,
+	row_pk: Option<String>,
+}
+
+const CONNECTOR_NAME: &str = "MSSQL source";
+
+impl MSSQLSource {
+	pub fn new(id: u32, config: MSSQLSourceConfig, state: Option<ConnectorState>) -> Self {
+		let verbose = config.verbose_logging.unwrap_or(false);
+		let restored_state = state
+			.and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+			.inspect(|s| {
+				info!(
+					"Restored state for {CONNECTOR_NAME} connector with ID: {id}. \
+					 Tracking offsets: {:?}, processed rows: {}",
+					s.tracking_offsets, s.processed_rows
+				);
+			});
+
+		let delay_str = config.retry_delay.as_deref().unwrap_or(DEFAULT_RETRY_DELAY);
+		let retry_delay = HumanDuration::from_str(delay_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(1));
+		let interval_str = config.poll_interval.as_deref().unwrap_or("10s");
+		let poll_interval = HumanDuration::from_str(interval_str)
+			.map(|duration| duration.into())
+			.unwrap_or_else(|_| Duration::from_secs(10));
+		MSSQLSource {
+			id,
+			client: Mutex::new(None),
+			config,
+			state: Mutex::new(restored_state.unwrap_or(State {
+				last_poll_lsn: None,
+				start_lsn: None,
+				end_lsn: None,
+				last_poll_time: Utc::now(),
+				tracking_offsets: HashMap::new(),
+				processed_rows: 0
+			})),
+			verbose,
+			retry_delay,
+			poll_interval
+		}
+	}
+
+	fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+		ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+	}
+}
+
+#[async_trait]
+impl Source for MSSQLSource {
+	async fn open(&mut self) -> Result<(), Error> {
+		info!(
+			"Opening MSSQL source connector with ID: {}. Mode: {}, Tables: {:?}",
+			self.id, self.config.mode, self.config.tables
+		);
+
+		self.connect().await?;
+
+		match self.config.mode.as_str() {
+			"cdc" => {
+				self.setup_cdc().await?;
+				info!(
+					"MSSQL CDC mode enabled for connector ID: {}",
+					self.id
+				);
+			}
+			"polling" => {
+				info!(
+					"MSSQL polling mode not enabled for connector ID: {}",
+					self.id
+				);
+				info!("Poll interval: {:?}", self.poll_interval);
+			}
+			_ => {
+				return Err(Error::InitError(format!(
+					"Invalid mode '{}'. Supported modes: 'polling', 'cdc'",
+					self.config.mode
+				)));
+			}
+		}
+
+		info!(
+			"MSSQL source connector with ID: {} opened successfully",
+			self.id
+		);
+		Ok(())
+	}
+
+	async fn poll(&self) -> Result<ProducedMessages, Error> {
+		let poll_interval = self.poll_interval;
+		tokio::time::sleep(poll_interval).await;
+
+		let messages = match self.config.mode.as_str() {
+			// "polling" => self.poll_tables().await?,
+			"cdc" => self.poll_cdc().await?,
+			_ => {
+				error!("Invalid mode: {}", self.config.mode);
+				return Err(Error::InvalidConfig);
+			}
+		};
+
+		let state = self.state.lock().await;
+		if self.verbose {
+			info!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		} else {
+			debug!(
+				"MSSQL source connector ID: {} produced {} messages. Total processed: {}",
+				self.id,
+				messages.len(),
+				state.processed_rows
+			);
+		}
+
+		let schema = match self.payload_format() {
+			PayloadFormat::Bytea => Schema::Raw,
+			PayloadFormat::Text => Schema::Text,
+			PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+		};
+
+		let persisted_state = self.serialize_state(&state);
+
+		Ok(ProducedMessages {
+			schema,
+			messages,
+			state: persisted_state,
+		})
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		let mut lock = self.client.lock().await;
+
+		let mut client = lock.take();
+
+		let _ = match client {
+			Some(x) => x.close().await,
+			None => Ok(())
+		};
+		info!(
+			"MSSQL connection closed for connector ID: {}",
+			self.id
+		);
+
+		let state = self.state.lock().await;
+		info!(
+			"MSSQL source connector ID: {} closed. Total rows processed: {}",
+			self.id, state.processed_rows
+		);
+		Ok(())
+	}
+}
+
+impl MSSQLSource {
+	async fn connect(&mut self) -> Result<(), Error> {
+
+		let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+		info!("Connecting to MSSQL with 1 connection: {redacted}");
+		let config = Config::from_jdbc_string(self.config.connection_string.expose_secret())
+			.map_err(|e| Error::InitError(format!("Invalid connection string configuration: {e}")))?;
+
+		let tcp = TcpStream::connect(config.get_addr())
+			.await
+			.map_err(|e| Error::InitError(
+				format!("Cannot connect to TCP stream: {e}")))?;
+
+		if tcp.set_nodelay(true).is_ok() { } else {
+			warn!("Cannot set no delay on the TCP socket!");
+		}
+
+		// To be able to use Tokio's tcp, we're using the `compat_write` from
+		// the `TokioAsyncWriteCompatExt` to get a stream compatible with the
+		// traits from the `futures` crate.
+		let mut client = Client::connect(
+			config,
+			tcp.compat_write()
+		).await
+		.map_err(|e| Error::InitError(format!("Failed to connect to MSSQL server: {e}")))?;
+
+		client.query("SELECT 1", &[])
+			.await
+			.map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?;
+
+		{
+			let mut lock = self.client.lock().await;
+			*lock = Some(client);
+		}
+
+		info!("Connected to MSSQL database");
+		Ok(())
+	}
+
+	async fn setup_cdc(&mut self) -> Result<(), Error> {
+
+		let database = self.config.database.clone();
+		let cdc_tables = if self.config.tables.is_empty() {
+			let res = self.query(Query::new("SELECT name FROM sys.tables WHERE is_tracked_by_cdc = 1;"))
+				.await?;
+			res
+				.into_iter()
+				.map(|r| r.get::<&str, _>("name").unwrap().to_string())
+				.collect()
+		} else {
+		   self.config.tables.clone()
+		}.join(",");
+
+		let mut query = Query::new("SELECT name, is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 0 AND name IN (@P1);");
+		query.bind(&cdc_tables);
+		let non_capture_tables = self.query(query).await?;
+
+		if !non_capture_tables.is_empty() {
+			warn!("Not all desired tables are monitored by CDC!");
+			if !self.config.enable_cdc.unwrap_or(false) {
+				error!("Update config to enable_cdc=true in order to monitor them!");
+				return Ok(());
+			}
+			warn!("Adding tables to the CDC monitor");
+			let schema = self.config.cdc_schema.clone().unwrap_or("dbo".to_string());
+			for row in non_capture_tables {
+				let table = row.get("name").unwrap_or("");
+				let cdc_role = match &self.config.cdc_role {
+					Some(x) => format!("N'{x}'"),
+					None => "NULL".to_string()
+				};
+				let query = format!(r#"
+					EXEC sys.sp_cdc_enable_table
+					@source_schema = N'{schema}',
+					@source_name   = N'{table}',
+					@role_name	 = {cdc_role}"#);
+
+				self.query(Query::new(query)).await?;
+			}
+		}
+
+		// For each monitored table, get the capture columns
+		let table_columns = self.query(Query::new(
+			format!(r#"
+				SELECT OBJECT_NAME(ct.object_id, DB_ID('{database}')) as table_name, c.object_id, ct.capture_instance, ct.start_lsn, c.columns FROM [{database}].[cdc].[change_tables] as ct JOIN (SELECT object_id, STRING_AGG(column_name, ',') as columns FROM [{database}].[cdc].[captured_columns] GROUP BY object_id) as c ON c.object_id = ct.object_id;"#

Review Comment:
   Since MSSQL has the source table and the change table with different names (by default `_ct` get's appended), either the user inserts the change table name or, there's need to make an internal conversion (while keeping a mapping of the names). Which one is the preferred one? 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] MSSQL Source Connector [iggy]

["lukaszzborek (via GitHub)" <gi...@apache.org>]

lukaszzborek commented on PR #3851:
URL: https://github.com/apache/iggy/pull/3851#issuecomment-5286928623

   @lsabi, quite intresting. Did you join to our discord? there are many people which can help. You can find discord invite in readme. Also you can merge current master to yours, because you have old version of server. And please write on discord what machine do you have (where you compile code)


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] MSSQL Source Connector [iggy]

["hubcio (via GitHub)" <gi...@apache.org>]

hubcio closed pull request #3851: MSSQL Source Connector
URL: https://github.com/apache/iggy/pull/3851


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org