tusharagrahari commented on code in PR #3568: URL: https://github.com/apache/iggy/pull/3568#discussion_r4302519539
########## core/connectors/sources/mysql_source/src/lib.rs: ########## @@ -0,0 +1,4071 @@ +// 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 async_trait::async_trait; +use base64::Engine; +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::SourceBatchResult, source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use sqlx::mysql::{MySqlDatabaseError, MySqlRow}; +use sqlx::{Column, MySql, Pool, Row, TypeInfo, mysql::MySqlPoolOptions}; +use std::cmp::Ordering as CmpOrdering; +use std::collections::{HashMap, HashSet}; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tracing::{debug, error, info, warn}; +use uuid::Uuid; + +source_connector!(MySqlSource); + +const DEFAULT_MAX_RETRIES: u32 = 3; +const DEFAULT_RETRY_DELAY: &str = "1s"; + +/// Fixed namespace so a given (table, key) always hashes to the same message id +/// across restarts. This is what makes a replayed row idempotent for downstream +/// dedup, so the value must never change once connectors are in the field. +const MESSAGE_ID_NAMESPACE: Uuid = Uuid::from_u128(0x8f3b1e6a4c9d4f2a9b7c0d1e2f3a4b5c); + +/// A disabled table is skipped before anything is logged, so without a reminder the +/// only evidence a connector went dark is the single line written when it happened. +/// At the default 10s poll interval this repeats it every ten minutes. +const DISABLED_TABLES_LOG_INTERVAL: u64 = 60; + +/// How many consecutive polls a table may end in `CursorRegression` before the repeat +/// stops reading as transient. Nothing is published in the meantime, so past this it +/// is a stall the operator has to see. +const CURSOR_REGRESSION_ERROR_THRESHOLD: u32 = 3; + +/// Upper bound on applying a batch's staged mark/delete after its Ack. The SDK waits +/// `BATCH_RESULT_TIMEOUT` (30s) for the result, so this has to finish well inside it. +const ACK_BATCH_TIMEOUT: Duration = Duration::from_secs(10); + +/// MySQL rejects a prepared statement with more placeholders than this (error 1390), and +/// the mark/delete binds one per row of the batch. +const MAX_CLEANUP_BATCH_SIZE: u32 = 65_535; + +/// How many mark/delete statements in a row may fail before the connector stops. The +/// operations stay in the persisted state and run again on restart. +const MAX_CONSECUTIVE_CLEANUP_FAILURES: u32 = 3; + +#[derive(Debug)] +pub struct MySqlSource { + pub id: u32, + pool: Option<Pool<MySql>>, + config: MySqlSourceConfig, + state: Mutex<State>, + pending_batch: Mutex<Option<PendingBatch>>, + consecutive_cleanup_failures: AtomicU32, + verified_cleanup_keys: Mutex<HashSet<String>>, + verbose: bool, + retry_delay: Duration, + poll_interval: Duration, + last_batch_full: AtomicBool, + poisoned_tables: Mutex<HashSet<String>>, + tracking_kinds: Mutex<HashMap<String, OffsetKind>>, + polls: AtomicU64, + cursor_regressions: Mutex<HashMap<String, u32>>, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MySqlSourceConfig { + #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")] + pub connection_string: SecretString, + 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 custom_query: Option<String>, + pub snake_case_columns: Option<bool>, + pub include_metadata: Option<bool>, + 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>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum PayloadFormat { + #[default] + Json, + Bytea, + Text, + JsonDirect, +} + +struct ProcessedRow { + message: ProducedMessage, + tracking_value: Option<String>, + row_pk: Option<PkValue>, +} + +/// Why a table's batch could not be produced. +enum FetchError { + Ordering(String), + /// The batch opened below the cursor it was told to resume from, so the query's + /// filter and the cursor disagree about what "past the offset" means. Retried + /// rather than latched, because unlike a within-batch decrease it depends on the + /// rows as much as on the query: a `custom_query` whose WHERE admits rows the + /// cursor does not gate (an OR branch is the usual one) only fails while such a + /// row is there, so it can clear without a config change. + CursorRegression(String), + MissingTrackingColumn(String), + UnusableTrackingValue(String), + StaleOffsetKind(String), + Other(Error), +} + +impl From<Error> for FetchError { + fn from(error: Error) -> Self { + FetchError::Other(error) + } +} + +/// How MySQL ordered a tracking column, which offsets alone cannot reveal: they are +/// carried as strings, so an `INT` 10 and a `VARCHAR` "10" are indistinguishable by +/// content while the server orders them differently. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum OffsetKind { + Numeric, + Lexical, + Unknown, +} + +/// Maps an `information_schema.columns.data_type` to the order MySQL sorts it in. +/// `decimal` belongs with the numbers even though `extract_column_value` renders it +/// as a string to preserve precision. +fn offset_kind_for_data_type(data_type: &str) -> OffsetKind { + const NUMERIC: &[&str] = &[ + "tinyint", + "smallint", + "mediumint", + "int", + "integer", + "bigint", + "decimal", + "numeric", + "float", + "double", + "real", + "bit", + "year", + ]; + if NUMERIC + .iter() + .any(|numeric| data_type.eq_ignore_ascii_case(numeric)) + { + OffsetKind::Numeric + } else { + OffsetKind::Lexical + } +} + +/// Whether MySQL orders the column by its collation. The other lexically compared +/// types are the temporal ones, which render fixed-width and so compare the same way +/// as text no matter what the collation is. +fn is_collation_ordered(data_type: &str) -> bool { + const TEXT: &[&str] = &[ + "char", + "varchar", + "tinytext", + "text", + "mediumtext", + "longtext", + ]; + TEXT.iter().any(|text| data_type.eq_ignore_ascii_case(text)) +} + +fn offset_kind_for_type_name(type_name: &str) -> OffsetKind { + match type_name { + "TINYINT" | "SMALLINT" | "MEDIUMINT" | "INT" | "BIGINT" | "TINYINT UNSIGNED" + | "SMALLINT UNSIGNED" | "MEDIUMINT UNSIGNED" | "INT UNSIGNED" | "BIGINT UNSIGNED" + | "FLOAT" | "DOUBLE" | "DECIMAL" | "BIT" | "YEAR" => OffsetKind::Numeric, + "DATE" | "TIME" | "DATETIME" | "TIMESTAMP" | "CHAR" | "VARCHAR" | "TINYTEXT" | "TEXT" + | "MEDIUMTEXT" | "LONGTEXT" => OffsetKind::Lexical, + _ => OffsetKind::Unknown, + } +} + +fn unusable_tracking_type(data_type: &str) -> Option<&'static str> { + const BINARY: &[&str] = &[ + "binary", + "varbinary", + "tinyblob", + "blob", + "mediumblob", + "longblob", + ]; + + if data_type.eq_ignore_ascii_case("json") { + return Some("it yields no ordered scalar cursor"); + } + // Judged on `data_type` rather than on the `tinyint(1)` display width that spells + // BOOLEAN, because that width is not reliably reported: MySQL 8.0.19 and later drop + // it for unsigned columns while MariaDB keeps it. + if data_type.eq_ignore_ascii_case("tinyint") { + return Some("it has too few distinct values to be unique per row"); + } + if BINARY + .iter() + .any(|binary| data_type.eq_ignore_ascii_case(binary)) + { + return Some("binary values are compared as base64 text, which does not keep byte order"); + } + if data_type.eq_ignore_ascii_case("enum") || data_type.eq_ignore_ascii_case("set") { + return Some("MySQL orders these by declaration index, not by label"); + } + None +} + +/// What a `custom_query` may begin with. `TABLE` is MySQL 8.0.19's table statement, +/// and `(` opens a parenthesized query such as `(SELECT ...) UNION (SELECT ...)`. +const READ_STATEMENT_STARTS: &[&str] = &["SELECT", "WITH", "TABLE", "("]; + +/// The token that decides which statement a query is, skipping leading comments in +/// all three MySQL spellings. Returns a single character when the statement does not +/// open with a word, so a parenthesized query is still recognisable. +fn leading_statement_token(query: &str) -> Option<String> { + let mut rest = query.trim_start(); + loop { + rest = if let Some(after) = rest.strip_prefix("/*") { + after.split_once("*/")?.1.trim_start() + } else if rest.starts_with("--") || rest.starts_with('#') { + rest.split_once('\n')?.1.trim_start() + } else { + break; + }; + } + + let word: String = rest + .chars() + .take_while(|c| c.is_ascii_alphanumeric() || *c == '_') + .collect(); + if word.is_empty() { + return rest.chars().next().map(String::from); + } + Some(word.to_uppercase()) +} + +/// The order the tracking column sorts in according to the result set that just +/// came back, or `None` when the query did not project the column under that name. +fn observed_tracking_kind(row: &MySqlRow, tracking_column: &str) -> Option<OffsetKind> { + row.columns() + .iter() + .find(|column| column.name().eq_ignore_ascii_case(tracking_column)) + .map(|column| offset_kind_for_type_name(column.type_info().name())) + .filter(|kind| *kind != OffsetKind::Unknown) +} + +/// `Clone` so the passing case, which is every batch, walks the names without collecting. +fn ensure_tracking_column_projected<'a>( + projected: impl IntoIterator<Item = &'a str> + Clone, + tracking_column: &str, +) -> Result<(), FetchError> { + if projected + .clone() + .into_iter() + .any(|column| column.eq_ignore_ascii_case(tracking_column)) + { + return Ok(()); + } + Err(FetchError::MissingTrackingColumn(format!( + "the result set has no column '{tracking_column}', it returned [{}]", + projected.into_iter().collect::<Vec<_>>().join(", ") + ))) +} + +fn unusable_tracking_value(tracking_column: &str, pk: &str) -> FetchError { + FetchError::UnusableTrackingValue(format!( + "row (pk {pk}) has no usable value for tracking_column '{tracking_column}': it is NULL \ + or not an ordered scalar (tinyint(1), JSON)" + )) +} + +/// Verifies rows arrive non-decreasing in the tracking column, which is what makes +/// "cursor = last row's value" correct. The built polling query orders ascending by +/// construction, so this only ever fires on a `custom_query`, where the operator +/// owns the `ORDER BY`. +struct OrderingGuard<'a> { + tracking_column: &'a str, + kind: OffsetKind, + last: Option<String>, + /// True while `last` still holds the cursor carried in from the previous poll + /// rather than a value from this batch. Distinguishes the two violations, and + /// keeps an empty batch from re-persisting the offset it started with. + at_cursor: bool, +} + +enum OrderingViolation { + WithinBatch(String), + BelowCursor(String), +} + +impl<'a> OrderingGuard<'a> { + fn new(tracking_column: &'a str, kind: OffsetKind, cursor: Option<String>) -> Self { + OrderingGuard { + tracking_column, + kind, + at_cursor: cursor.is_some(), + last: cursor, + } + } + + fn accept(&mut self, value: String) -> Result<(), OrderingViolation> { + if let Some(previous) = &self.last + && offset_decreased(previous, &value, self.kind) + { + let column = self.tracking_column; + return Err(if self.at_cursor { + OrderingViolation::BelowCursor(format!( + "row '{value}' is below the stored offset '{previous}' for tracking_column \ + '{column}'. Filter the query with `{column} > $offset`" + )) + } else { + OrderingViolation::WithinBatch(format!( + "rows arrived out of order by tracking_column '{column}': '{previous}' \ + preceded '{value}'. Order the query ascending by '{column}'" + )) + }); + } + self.last = Some(value); + self.at_cursor = false; + Ok(()) + } + + /// `None` when no row was accepted, so an empty batch leaves the stored offset + /// untouched rather than rewriting the cursor it was seeded with. + fn into_cursor(self) -> Option<String> { + if self.at_cursor { None } else { self.last } + } +} + +struct TableBatch { + table: String, + messages: Vec<ProducedMessage>, + processed_ids: Vec<PkValue>, + max_offset: Option<String>, +} + +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") => PayloadFormat::JsonDirect, + _ => PayloadFormat::Json, + } + } +} + +/// A primary-key value in a form the mark/delete `WHERE pk IN (...)` can bind +/// faithfully. `Text` covers ints/strings/dates/decimals (MySQL implicitly +/// converts the bound string to the column type). `Bytes` carries raw binary +/// (BINARY/VARBINARY/BLOB) so a UUID stored as bytes matches the row instead of +/// being compared as its base64 text, which never matches. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +enum PkValue { + Text(String), + Bytes(Vec<u8>), +} + +impl PkValue { + fn as_key(&self) -> String { + match self { + PkValue::Text(text) => text.clone(), + PkValue::Bytes(bytes) => base64::engine::general_purpose::STANDARD.encode(bytes), + } + } +} + +#[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, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct State { + tracking_offsets: HashMap<String, String>, + processed_rows: u64, + /// Mark/delete work for a batch the runtime has already sent. It is persisted with + /// the batch's offsets so a crash between the send and the cleanup reruns it. + #[serde(default)] + pending_operations: Vec<PendingOperation>, +} + +/// The rows of one table to mark or delete once their batch is acknowledged. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +struct PendingOperation { + table: String, + ids: Vec<PkValue>, + target: CleanupTarget, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +struct CleanupTarget { + key_column: String, + action: CleanupAction, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +enum CleanupAction { + Delete, + MarkProcessed { column: String }, +} + +/// The candidate state of the batch most recently handed to the runtime. Live state +/// only takes it over on Ack, so a Nack leaves offsets and rows where they were. +#[derive(Debug, Clone)] +struct PendingBatch { + state: State, + /// The runtime has delivered the batch but its cleanup has not finished yet. + acknowledged: bool, + /// Cleanup finished and the next poll must persist `state` without the operations, + /// so a restart does not run them again. + retiring: bool, +} + +struct PolledBatch { + messages: Vec<ProducedMessage>, + pending: Option<PendingBatch>, +} + +enum CleanupKeyCheck { + Unique, + NotUnique, + TableMissing, +} + +const CONNECTOR_NAME: &str = "MySQL source"; + +#[async_trait] +impl Source for MySqlSource { + async fn open(&mut self) -> Result<(), Error> { + info!( + "Opening MySQL source connector with ID: {}, Tables: {:?}", + self.id, self.config.tables + ); + + if self.config.tables.is_empty() { + return Err(Error::InitError( + "tables must not be empty, list the table even when custom_query hardcodes it" + .to_string(), + )); + } + + if self.config.batch_size == Some(0) { + return Err(Error::InitError( + "batch_size must be greater than 0; omit it to use the default of 1000".to_string(), + )); + } + + if let Some(batch_size) = self.config.batch_size + && batch_size > MAX_CLEANUP_BATCH_SIZE + && self.cleanup_target().is_some() + { + return Err(Error::InitError(format!( + "batch_size {batch_size} exceeds {MAX_CLEANUP_BATCH_SIZE} with delete_after_read or \ + processed_column set, MySQL allows at most {MAX_CLEANUP_BATCH_SIZE} placeholders \ + in the statement that marks or deletes a batch" + ))); + } + + if self.config.batch_size == Some(1) + && self + .config + .custom_query + .as_deref() + .is_some_and(|query| query.contains("$offset")) + { + return Err(Error::InitError( + "batch_size must be greater than 1 when custom_query filters on $offset" + .to_string(), + )); + } + + if self + .config + .tracking_column + .as_deref() + .is_none_or(str::is_empty) + { + let guidance = if self.config.custom_query.is_some() { + "name the column custom_query orders by and filters with $offset" + } else { + "use a unique, monotonically increasing column" + }; + return Err(Error::InitError(format!( + "tracking_column must be set, {guidance}" + ))); + } + + if let Some(ref col) = self.config.payload_column + && !col.is_empty() + && PayloadFormat::from_config(self.config.payload_format.as_deref()) + == PayloadFormat::Json + { + return Err(Error::InitError( + "payload_format must be 'bytea', 'text', or 'json_direct' when payload_column is set" + .to_string(), + )); + } + + // custom_query never changes after startup, so validate it once here + if let Some(ref query) = self.config.custom_query { + self.validate_custom_query(query)?; + + if query.contains("$offset") && self.config.initial_offset.is_none() { + let state = self.state.lock().await; + let missing: Vec<&str> = self + .config + .tables + .iter() + .filter(|t| !state.tracking_offsets.contains_key(*t)) + .map(String::as_str) + .collect(); + if !missing.is_empty() { + return Err(Error::InitError(format!( + "custom_query uses $offset but initial_offset is not set and no stored offset exists for table(s): {}", + missing.join(", ") + ))); + } + } + } + + if self.config.delete_after_read.unwrap_or(false) && self.config.processed_column.is_some() + { + warn!( + "both delete_after_read and processed_column are set; delete_after_read takes precedence, so rows are deleted and processed_column only acts as a poll filter" + ); + } + + self.validate_pending_cleanup()?; + self.connect().await?; + *self.tracking_kinds.get_mut() = self.resolve_tracking_columns().await?; + self.validate_cleanup_keys().await?; + + info!( + "MySQL source connector with ID: {} opened successfully", + self.id + ); + Ok(()) + } + + async fn poll(&self) -> Result<ProducedMessages, Error> { + let schema = match self.payload_format() { + PayloadFormat::Bytea => Schema::Raw, + PayloadFormat::Text => Schema::Text, + PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json, + }; + + if let Some(state) = self.retiring_state().await { + return Ok(ProducedMessages { + schema, + messages: Vec::new(), + state: Some(self.serialize_pending_state(&state)?), + }); + } + + // Skip the pacing delay while draining a backlog, but never while a cleanup + // retry is pending, or a failing mark/delete would spin without a pause. + let cleanup_pending = self.pending_batch.lock().await.is_some(); + if cleanup_pending || !self.last_batch_full.load(Ordering::Relaxed) { + tokio::time::sleep(self.poll_interval).await; + } + + { + let pending = self.pending_batch.lock().await; + if let Some(pending) = pending.as_ref() { + if !pending.acknowledged { + error!( + "{CONNECTOR_NAME} connector ID: {} was polled while a batch was still in flight", + self.id + ); + return Err(Error::InvalidState); + } + // An empty batch whose Ack retries the cleanup the last one left behind. + return Ok(ProducedMessages { + schema, + messages: Vec::new(), + state: None, + }); + } + } + + let polled = self.poll_tables().await?; + + let processed_rows = match &polled.pending { + Some(pending) => pending.state.processed_rows, + None => self.state.lock().await.processed_rows, + }; + if self.verbose { + info!( + "MySQL source connector ID: {} produced {} messages. Total processed: {}", + self.id, + polled.messages.len(), + processed_rows + ); + } else { + debug!( + "MySQL source connector ID: {} produced {} messages. Total processed: {}", + self.id, + polled.messages.len(), + processed_rows + ); + } + + // Idle cycles stage nothing, so return None to let the runtime skip the fsync. + let persisted_state = polled + .pending + .as_ref() + .map(|pending| self.serialize_pending_state(&pending.state)) + .transpose()?; + *self.pending_batch.lock().await = polled.pending; + + Ok(ProducedMessages { + schema, + messages: polled.messages, + state: persisted_state, + }) + } + + async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(), Error> { + if result == SourceBatchResult::Nack { + // Dropping the candidate leaves the live offsets where they were, so the + // next poll reads the same rows again. Delivered batches keep their cleanup. + let mut pending = self.pending_batch.lock().await; + if pending + .as_ref() + .is_some_and(|pending| !pending.acknowledged && !pending.retiring) + { + pending.take(); + } + return Ok(()); + } + + let Some(PendingBatch { mut state, .. }) = self.pending_batch.lock().await.clone() else { + return Ok(()); + }; + + let operations = std::mem::take(&mut state.pending_operations); + let had_operations = !operations.is_empty(); + let deadline = tokio::time::Instant::now() + ACK_BATCH_TIMEOUT; + let failed_operations = self.apply_pending_operations(operations, deadline).await?; + + if failed_operations.is_empty() { + *self.state.lock().await = state.clone(); + let mut pending = self.pending_batch.lock().await; + if had_operations { + *pending = Some(PendingBatch { + state, + acknowledged: false, + retiring: true, + }); + } else { + pending.take(); + } + } else { + state.pending_operations = failed_operations; + *self.pending_batch.lock().await = Some(PendingBatch { + state, + acknowledged: true, + retiring: false, + }); + } + + Ok(()) + } + + async fn close(&mut self) -> Result<(), Error> { + if let Some(pool) = self.pool.take() { + pool.close().await; + info!("MySQL connection pool closed for connector ID: {}", self.id); + } + + let state = self.state.lock().await; + info!( + "MySQL source connector ID: {} closed. Total rows processed: {}", + self.id, state.processed_rows + ); + Ok(()) + } +} + +impl MySqlSource { + pub fn new(id: u32, config: MySqlSourceConfig, 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)); + let state = restored_state.unwrap_or(State { + tracking_offsets: HashMap::new(), + processed_rows: 0, + pending_operations: Vec::new(), + }); + // The runtime only persists a batch's operations after sending it, so restored + // operations belong to a delivered batch and only the cleanup is left to do. + let pending_batch = (!state.pending_operations.is_empty()).then(|| PendingBatch { + state: state.clone(), + acknowledged: true, + retiring: false, + }); + MySqlSource { + id, + pool: None, + config, + state: Mutex::new(state), + pending_batch: Mutex::new(pending_batch), + consecutive_cleanup_failures: AtomicU32::new(0), + verified_cleanup_keys: Mutex::new(HashSet::new()), + verbose, + retry_delay, + poll_interval, + last_batch_full: AtomicBool::new(false), + poisoned_tables: Mutex::new(HashSet::new()), + tracking_kinds: Mutex::new(HashMap::new()), + polls: AtomicU64::new(0), + cursor_regressions: Mutex::new(HashMap::new()), + } + } + + async fn connect(&mut self) -> Result<(), Error> { + let max_connections = self.config.max_connections.unwrap_or(10); + let redacted = redact_connection_string(self.config.connection_string.expose_secret()); + + info!("Connecting to MySQL with max {max_connections} connections: {redacted}"); + + let pool = MySqlPoolOptions::new() + .max_connections(max_connections) + // `format_offset_value` escapes backslashes, which under NO_BACKSLASH_ESCAPES + // would reach MySQL as literal characters and filter on the wrong cursor. + .after_connect(|connection, _| { + Box::pin(async move { + sqlx::query( + "SET SESSION sql_mode = REPLACE(@@SESSION.sql_mode, 'NO_BACKSLASH_ESCAPES', '')", + ) + .execute(&mut *connection) + .await?; + Ok(()) + }) + }) + .connect(self.config.connection_string.expose_secret()) + .await + .map_err(|e| Error::InitError(format!("Failed to connect to MySQL: {e}")))?; + + sqlx::query("SELECT 1") + .execute(&pool) + .await + .map_err(|e| Error::InitError(format!("Database connectivity test failed: {e}")))?; + + self.pool = Some(pool); + info!("Connected to MySQL database with {max_connections} max connections"); + Ok(()) + } + + fn tracking_column(&self) -> &str { + self.config.tracking_column.as_deref().unwrap_or("id") + } + + /// Fail fast if a tracking column can't yield a stable, ordered scalar + /// cursor. `value_as_string` drops NULL, tinyint(1) (decoded as BOOLEAN), + /// and JSON to `None`, which silently stalls offset advancement and re-reads + /// rows every cycle. + async fn resolve_tracking_columns(&self) -> Result<HashMap<String, OffsetKind>, Error> { + // Whether the connector owns the query, and can therefore hold the schema to + // what that query will do with it. + let strict = self.config.custom_query.is_none(); + let pool = self.get_pool()?; + let tracking_column = self.tracking_column(); + let mut kinds = HashMap::with_capacity(self.config.tables.len()); + + for table in &self.config.tables { + let (schema, table_name) = table + .split_once('.') + .map(|(schema, name)| (schema.to_string(), name.to_string())) + .unwrap_or_else(|| (String::new(), table.clone())); + + let row = sqlx::query( + "SELECT is_nullable, data_type, column_type \ + FROM information_schema.columns \ + WHERE table_schema = IF(? = '', DATABASE(), ?) \ + AND table_name = ? AND column_name = ?", + ) + .bind(&schema) + .bind(&schema) + .bind(&table_name) + .bind(tracking_column) + .fetch_optional(pool) + .await + .map_err(|e| { + Error::InitError(format!( + "failed to probe schema for tracking_column '{tracking_column}' on table '{table}': {e}" + )) + })?; + + let Some(row) = row else { + // A missing columns row means either the table doesn't exist yet + // or the column is absent. Only the latter is a misconfiguration: + // a table created after the connector starts is legitimate, so + // defer validation rather than permanently disabling the source. + let table_exists = sqlx::query( + "SELECT 1 FROM information_schema.tables \ + WHERE table_schema = IF(? = '', DATABASE(), ?) \ + AND table_name = ?", + ) + .bind(&schema) + .bind(&schema) + .bind(&table_name) + .fetch_optional(pool) + .await + .map_err(|e| { + Error::InitError(format!("failed to probe existence of table '{table}': {e}")) + })? + .is_some(); + + if !table_exists { Review Comment: Fixed. A table that doesn't exist at startup is now checked on the first poll after it appears, using the same information_schema probe open() runs. A rejected type (BINARY/BLOB, JSON, TINYINT, ENUM/SET) disables that table before anything is read, so the cursor never moves. The probe also resolves the column's comparison order before the first query, which fixes a related stall: a late text table was filtered with a bare code > 0, which matches no non-numeric value. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
