hubcio commented on code in PR #3957:
URL: https://github.com/apache/iggy/pull/3957#discussion_r3996880954
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -695,7 +1015,9 @@ impl PostgresSource {
));
}
- if let Some(processed_col) = &self.config.processed_column {
+ if !self.config.delete_after_read.unwrap_or(false)
Review Comment:
critical: with `delete_after_read=true` and `processed_column` together,
previously excluded processed rows are now emitted and deleted. keep the
processed-row selection filter while preserving delete precedence.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -230,38 +297,77 @@ impl Source for PostgresSource {
}
};
- let state = self.state.lock().await;
+ let processed_rows = self.state.lock().await.processed_rows;
if self.verbose {
info!(
"PostgreSQL source connector ID: {} produced {} messages.
Total processed: {}",
self.id,
- messages.len(),
- state.processed_rows
+ polled.messages.len(),
+ processed_rows
);
} else {
debug!(
"PostgreSQL source connector ID: {} produced {} messages.
Total processed: {}",
self.id,
- messages.len(),
- state.processed_rows
+ polled.messages.len(),
+ 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);
+ let persisted_state = polled
+ .pending
+ .as_ref()
+ .map(|pending| {
+ self.serialize_state(&pending.state).ok_or_else(|| {
Review Comment:
warning: a restart after checkpoint save loses pending cleanup, leaving
unchanged rows behind the saved cursor. persist cleanup intent that identifies
original rows, replay and retire it safely, and retain compatibility with older
checkpoints.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -604,6 +604,8 @@ pub(crate) async fn source_forwarding_loop(
context.metrics.inc_errors_with_labels(&labels.counter);
context.sources.set_error(&plugin_key, &error_msg).await;
}
+ } else if batch_result == SourceBatchResult::Ack {
Review Comment:
warning: empty or fully filtered batches can clear a delivery error without
contacting iggy. require `sent_count > 0` before restoring `Running`, while
retaining empty-batch ACK and checkpoint handling.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -417,43 +533,55 @@ impl PostgresSource {
(!self.config.tables.is_empty()).then_some(self.config.tables.as_slice());
let batch_size = self.config.batch_size.unwrap_or(1000) as i32;
+ let wal_flush_lsn = with_retry(
+ || sqlx::query_scalar("SELECT
pg_current_wal_flush_lsn()::text").fetch_one(pool),
Review Comment:
warning: `pg_current_wal_flush_lsn()` fails during recovery, so builtin CDC
cannot poll a hot standby. use `pg_last_wal_replay_lsn()` during recovery and
the flush LSN otherwise, keeping this read before the peek.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -669,6 +956,39 @@ impl PostgresSource {
self.config.max_retries.unwrap_or(DEFAULT_MAX_RETRIES)
}
+ fn should_process_rows(&self) -> bool {
+ self.config.delete_after_read.unwrap_or(false) ||
self.config.processed_column.is_some()
+ }
+
+ fn processing_boundary(&self, max_offset: Option<String>) ->
Option<String> {
Review Comment:
warning: ordered custom queries using `$offset` keep the old cursor after
ACK, repeating the first page indefinitely. advance the acknowledged cursor for
ascending, unique tracking values and keep cleanup boundaries separate.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -733,18 +1057,20 @@ impl PostgresSource {
batch_size: u32,
) -> String {
let offset_value = last_offset
- .clone()
- .or_else(|| self.config.initial_offset.clone())
+ .as_deref()
+ .or(self.config.initial_offset.as_deref())
.unwrap_or_default();
+ let formatted_offset = format_offset_value(offset_value);
let now = Utc::now();
query
.replace("$table", table)
- .replace("$offset", &offset_value)
+ .replace("'$offset'", &formatted_offset)
+ .replace("$offset", &formatted_offset)
Review Comment:
critical: quoted `$offset` templates can expand a restored offset twice,
turning data into SQL and selecting rows outside the filter for cleanup.
substitute original tokens once, keeping replacement values opaque and
preserving SQL literal context.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -284,12 +390,18 @@ impl PostgresSource {
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());
+ let connect_options =
+
PgConnectOptions::from_str(self.config.connection_string.expose_secret())
+ .map_err(|e| {
+ Error::InitError(format!("Invalid PostgreSQL connection
string: {e}"))
+ })?
+ .options([("statement_timeout",
STATEMENT_TIMEOUT.as_millis())]);
Review Comment:
warning: the pool-wide `statement_timeout` also cancels valid slow polling
queries and CDC peeks, so they never make progress. apply the short timeout
only to ACK statements and restore connection settings safely after
cancellation.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -707,6 +1029,8 @@ impl PostgresSource {
};
let order_clause = format!(" ORDER BY {quoted_tracking} ASC");
+ // TODO: Add a unique tiebreaker to pagination. Advancing only the
tracking-column
Review Comment:
critical: this existing scalar cursor skips unsent rows tied at the page
boundary, even with a unique cleanup key. use a stable unique composite cursor
with legacy checkpoint handling, or reject non-unique tracking columns.
##########
core/connectors/runtime/src/stream.rs:
##########
@@ -24,6 +24,11 @@ use crate::error::RuntimeError;
const TOKEN_FILE_PREFIX: &str = "file:";
+fn append_query_parameters(connection_string: &str, address: &str, parameters:
&str) -> String {
Review Comment:
nit: the reason for checking `address` rather than the full connection
string is undocumented. explain that credentials may contain `?` and that
`address` must match the connection string's suffix.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -230,38 +297,77 @@ impl Source for PostgresSource {
}
};
- let state = self.state.lock().await;
+ let processed_rows = self.state.lock().await.processed_rows;
if self.verbose {
info!(
"PostgreSQL source connector ID: {} produced {} messages.
Total processed: {}",
self.id,
- messages.len(),
- state.processed_rows
+ polled.messages.len(),
+ processed_rows
);
} else {
debug!(
"PostgreSQL source connector ID: {} produced {} messages.
Total processed: {}",
self.id,
- messages.len(),
- state.processed_rows
+ polled.messages.len(),
+ 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);
+ let persisted_state = polled
+ .pending
+ .as_ref()
+ .map(|pending| {
+ self.serialize_state(&pending.state).ok_or_else(|| {
+ Error::Serialization("failed to serialize PostgreSQL
source state".to_string())
+ })
+ })
+ .transpose()?;
+ *self.pending_batch.lock().await = polled.pending;
Ok(ProducedMessages {
schema,
- messages,
+ messages: polled.messages,
state: persisted_state,
})
}
+ async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(),
Error> {
+ if result == SourceBatchResult::Nack {
+ let mut pending = self.pending_batch.lock().await;
+ if pending
+ .as_ref()
+ .is_some_and(|pending| !pending.acknowledged)
+ {
+ pending.take();
+ }
+ return Ok(());
+ }
+
+ let Some(pending) = self.pending_batch.lock().await.as_ref().cloned()
else {
+ return Ok(());
+ };
+
+ let PendingBatch {
+ state, operations, ..
+ } = pending;
+ let deadline = tokio::time::Instant::now() + ACK_BATCH_TIMEOUT;
+ let failed_operations = self.apply_pending_operations(operations,
deadline).await?;
Review Comment:
critical: existing buffered ACKs on a single node can precede disk flush, so
a crash after source cleanup can lose messages. require durable destination
storage before checkpoint and source ACK, with fsync for power-loss safety.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -230,38 +297,77 @@ impl Source for PostgresSource {
}
};
- let state = self.state.lock().await;
+ let processed_rows = self.state.lock().await.processed_rows;
if self.verbose {
info!(
"PostgreSQL source connector ID: {} produced {} messages.
Total processed: {}",
self.id,
- messages.len(),
- state.processed_rows
+ polled.messages.len(),
+ processed_rows
);
} else {
debug!(
"PostgreSQL source connector ID: {} produced {} messages.
Total processed: {}",
self.id,
- messages.len(),
- state.processed_rows
+ polled.messages.len(),
+ 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);
+ let persisted_state = polled
+ .pending
+ .as_ref()
+ .map(|pending| {
+ self.serialize_state(&pending.state).ok_or_else(|| {
+ Error::Serialization("failed to serialize PostgreSQL
source state".to_string())
+ })
+ })
+ .transpose()?;
+ *self.pending_batch.lock().await = polled.pending;
Ok(ProducedMessages {
schema,
- messages,
+ messages: polled.messages,
state: persisted_state,
})
}
+ async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(),
Error> {
+ if result == SourceBatchResult::Nack {
+ let mut pending = self.pending_batch.lock().await;
+ if pending
+ .as_ref()
+ .is_some_and(|pending| !pending.acknowledged)
+ {
+ pending.take();
+ }
+ return Ok(());
+ }
+
+ let Some(pending) = self.pending_batch.lock().await.as_ref().cloned()
else {
+ return Ok(());
+ };
+
+ let PendingBatch {
+ state, operations, ..
+ } = pending;
+ let deadline = tokio::time::Instant::now() + ACK_BATCH_TIMEOUT;
Review Comment:
warning: the existing SDK deadline can expire after ACK is claimed but
before cleanup finishes, permanently stopping polling. track ACK receipt
separately from bounded hook completion, while keeping one batch in flight.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -995,12 +1313,11 @@ fn extract_column_value(
.unwrap_or(serde_json::Value::Null))
}
"NUMERIC" => {
- let value: Option<String> = row
+ let value: Option<BigDecimal> = row
Review Comment:
warning: valid `NUMERIC` values such as `NaN` still fail `BigDecimal`
decoding and block polling. handle supported special values as strings,
preserving `NULL`, real decode errors, and quoted tracking literals.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -733,18 +1057,20 @@ impl PostgresSource {
batch_size: u32,
) -> String {
let offset_value = last_offset
- .clone()
- .or_else(|| self.config.initial_offset.clone())
+ .as_deref()
+ .or(self.config.initial_offset.as_deref())
.unwrap_or_default();
+ let formatted_offset = format_offset_value(offset_value);
let now = Utc::now();
query
.replace("$table", table)
- .replace("$offset", &offset_value)
+ .replace("'$offset'", &formatted_offset)
+ .replace("$offset", &formatted_offset)
.replace("$limit", &batch_size.to_string())
Review Comment:
warning: the existing `$limit` and `$now` replacements rewrite those tokens
inside inserted offsets, changing predicates. scan only the original template
and leave replacement values untouched.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +680,207 @@ impl PostgresSource {
messages.push(processed.message);
total_processed += 1;
+ table_processed += 1;
}
- // Database I/O without holding the lock
- if !processed_ids.is_empty() {
- self.mark_or_delete_processed_rows(pool, table, pk_column,
&processed_ids)
- .await?;
+ let tracking_boundary = self.processing_boundary(max_offset);
+
+ if self.should_process_rows() && !processed_ids.is_empty() {
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ tracking_boundary: tracking_boundary.clone(),
+ });
}
- // Collect offset update for later
- if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ if let Some(offset) = tracking_boundary {
+ candidate_state
+ .tracking_offsets
+ .insert(table.clone(), offset);
}
if self.verbose {
- info!("Fetched {} rows from table '{table}'", messages.len());
+ info!("Fetched {table_processed} rows from table '{table}'");
} else {
- debug!("Fetched {} rows from table '{table}'", messages.len());
+ debug!("Fetched {table_processed} rows from table '{table}'");
}
}
- // Apply all state updates with a single lock acquisition
- {
- let mut state = self.state.lock().await;
- state.processed_rows += total_processed;
- for (table, offset) in state_updates {
- state.tracking_offsets.insert(table, offset);
+ let pending = if total_processed > 0 {
+ candidate_state.processed_rows += total_processed;
+ candidate_state.last_poll_time = Utc::now();
+ Some(PendingBatch {
+ state: candidate_state,
+ operations,
+ acknowledged: false,
+ })
+ } else {
+ None
+ };
+
+ Ok(PolledBatch { messages, pending })
+ }
+
+ async fn advance_replication_slot(&self, lsn: &str) -> Result<(), Error> {
+ let slot_name = self.replication_slot();
+ let pool = self.get_pool()?;
+ with_retry(
+ || async {
+ let result = sqlx::query("SELECT
pg_replication_slot_advance($1, $2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(pool)
+ .await;
+
+ if let Err(error) = result {
+ match replication_slot_reached_target(pool, slot_name,
lsn).await {
+ Ok(true) => {
+ warn!(
+ "PostgreSQL source connector ID: {} confirmed
replication slot \
+ '{slot_name}' is already at or beyond {lsn}
after advance failed: \
+ {error}",
+ self.id
+ );
+ return Ok(());
+ }
+ Ok(false) => {}
+ Err(read_error) => warn!(
+ "PostgreSQL source connector ID: {} failed to read
replication slot \
+ '{slot_name}' after advance failed: {read_error}",
+ self.id
+ ),
+ }
+ return Err(error);
+ }
+
+ Ok(())
+ },
+ self.get_max_retries(),
+ self.retry_delay.as_millis() as u64,
+ )
+ .await
+ .map_err(|e| {
+ Error::Connection(format!(
+ "failed to advance replication slot '{slot_name}' to {lsn}:
{e}"
+ ))
+ })?;
+ Ok(())
+ }
+
+ async fn apply_pending_operations(
+ &self,
+ operations: Vec<PendingOperation>,
+ deadline: tokio::time::Instant,
+ ) -> Result<Vec<PendingOperation>, Error> {
+ let mut failed_operations = Vec::new();
+ for operation in operations {
+ if !self.apply_pending_operation(&operation, deadline).await? {
+ failed_operations.push(operation);
+ }
+ }
+ Ok(failed_operations)
+ }
+
+ async fn apply_pending_operation(
+ &self,
+ operation: &PendingOperation,
+ deadline: tokio::time::Instant,
+ ) -> Result<bool, Error> {
+ let (operation_name, advances_slot) = match operation {
+ PendingOperation::ProcessRows { .. } => ("process rows", false),
+ PendingOperation::AdvanceReplicationSlot { .. } => ("advance
replication slot", true),
+ };
+
+ let result =
+ tokio::time::timeout_at(deadline,
self.execute_pending_operation(operation)).await;
+
+ match result {
+ Ok(Ok(())) => {
+ if advances_slot {
+ self.consecutive_advance_failures
+ .store(0, Ordering::Relaxed);
+ }
+ Ok(true)
+ }
+ Ok(Err(error)) => {
+ self.record_pending_operation_failure(operation_name,
advances_slot, &error)?;
+ Ok(false)
+ }
+ Err(_) => {
+ self.record_pending_operation_failure(
+ operation_name,
+ advances_slot,
+ "operation exceeded the shared 10s ACK batch budget",
+ )?;
+ Ok(false)
+ }
+ }
+ }
+
+ async fn execute_pending_operation(&self, operation: &PendingOperation) ->
Result<(), Error> {
+ match operation {
+ PendingOperation::ProcessRows {
+ table,
+ ids,
+ tracking_boundary,
+ } => {
+ let pool = self.get_pool()?;
+ self.mark_or_delete_processed_rows(pool, table, ids,
tracking_boundary.as_deref())
+ .await
+ }
+ PendingOperation::AdvanceReplicationSlot { lsn } => {
+ self.advance_replication_slot(lsn).await
}
- state.last_poll_time = Utc::now();
}
+ }
- Ok(messages)
+ fn record_pending_operation_failure(
+ &self,
+ operation_name: &str,
+ advances_slot: bool,
+ error: impl std::fmt::Display,
+ ) -> Result<(), Error> {
+ if advances_slot {
+ let consecutive_failures = self
+ .consecutive_advance_failures
+ .fetch_add(1, Ordering::Relaxed)
+ .saturating_add(1);
+ error!(
+ "Failed to {operation_name} for PostgreSQL source connector
ID: {}. \
+ Consecutive replication slot advance failures:
{consecutive_failures}. {error}",
+ self.id
+ );
+ if consecutive_failures >= MAX_CONSECUTIVE_ADVANCE_FAILURES {
+ return Err(Error::Connection(format!(
+ "stopping PostgreSQL source connector ID {} after
{consecutive_failures} \
+ consecutive replication slot advance failures",
+ self.id
+ )));
+ }
+ } else {
Review Comment:
warning: permanent cleanup failures already stall the source while it
reports `Running`; staged ACK retries remain unbounded. cap `ProcessRows`
failures and return `Err` from `on_batch_result`, preserving pending work.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +680,207 @@ impl PostgresSource {
messages.push(processed.message);
total_processed += 1;
+ table_processed += 1;
}
- // Database I/O without holding the lock
- if !processed_ids.is_empty() {
- self.mark_or_delete_processed_rows(pool, table, pk_column,
&processed_ids)
- .await?;
+ let tracking_boundary = self.processing_boundary(max_offset);
+
+ if self.should_process_rows() && !processed_ids.is_empty() {
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ tracking_boundary: tracking_boundary.clone(),
+ });
}
- // Collect offset update for later
- if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ if let Some(offset) = tracking_boundary {
+ candidate_state
+ .tracking_offsets
+ .insert(table.clone(), offset);
}
if self.verbose {
- info!("Fetched {} rows from table '{table}'", messages.len());
+ info!("Fetched {table_processed} rows from table '{table}'");
} else {
- debug!("Fetched {} rows from table '{table}'", messages.len());
+ debug!("Fetched {table_processed} rows from table '{table}'");
}
}
- // Apply all state updates with a single lock acquisition
- {
- let mut state = self.state.lock().await;
- state.processed_rows += total_processed;
- for (table, offset) in state_updates {
- state.tracking_offsets.insert(table, offset);
+ let pending = if total_processed > 0 {
+ candidate_state.processed_rows += total_processed;
+ candidate_state.last_poll_time = Utc::now();
+ Some(PendingBatch {
+ state: candidate_state,
+ operations,
+ acknowledged: false,
+ })
+ } else {
+ None
+ };
+
+ Ok(PolledBatch { messages, pending })
+ }
+
+ async fn advance_replication_slot(&self, lsn: &str) -> Result<(), Error> {
+ let slot_name = self.replication_slot();
+ let pool = self.get_pool()?;
+ with_retry(
+ || async {
+ let result = sqlx::query("SELECT
pg_replication_slot_advance($1, $2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(pool)
+ .await;
+
+ if let Err(error) = result {
+ match replication_slot_reached_target(pool, slot_name,
lsn).await {
+ Ok(true) => {
+ warn!(
+ "PostgreSQL source connector ID: {} confirmed
replication slot \
+ '{slot_name}' is already at or beyond {lsn}
after advance failed: \
+ {error}",
+ self.id
+ );
+ return Ok(());
+ }
+ Ok(false) => {}
+ Err(read_error) => warn!(
+ "PostgreSQL source connector ID: {} failed to read
replication slot \
+ '{slot_name}' after advance failed: {read_error}",
+ self.id
+ ),
+ }
+ return Err(error);
+ }
+
+ Ok(())
+ },
+ self.get_max_retries(),
+ self.retry_delay.as_millis() as u64,
+ )
+ .await
+ .map_err(|e| {
+ Error::Connection(format!(
+ "failed to advance replication slot '{slot_name}' to {lsn}:
{e}"
+ ))
+ })?;
+ Ok(())
+ }
+
+ async fn apply_pending_operations(
+ &self,
+ operations: Vec<PendingOperation>,
+ deadline: tokio::time::Instant,
+ ) -> Result<Vec<PendingOperation>, Error> {
+ let mut failed_operations = Vec::new();
+ for operation in operations {
+ if !self.apply_pending_operation(&operation, deadline).await? {
+ failed_operations.push(operation);
+ }
+ }
+ Ok(failed_operations)
+ }
+
+ async fn apply_pending_operation(
+ &self,
+ operation: &PendingOperation,
+ deadline: tokio::time::Instant,
+ ) -> Result<bool, Error> {
+ let (operation_name, advances_slot) = match operation {
+ PendingOperation::ProcessRows { .. } => ("process rows", false),
+ PendingOperation::AdvanceReplicationSlot { .. } => ("advance
replication slot", true),
+ };
+
+ let result =
+ tokio::time::timeout_at(deadline,
self.execute_pending_operation(operation)).await;
+
+ match result {
+ Ok(Ok(())) => {
+ if advances_slot {
+ self.consecutive_advance_failures
+ .store(0, Ordering::Relaxed);
+ }
+ Ok(true)
+ }
+ Ok(Err(error)) => {
+ self.record_pending_operation_failure(operation_name,
advances_slot, &error)?;
+ Ok(false)
+ }
+ Err(_) => {
+ self.record_pending_operation_failure(
+ operation_name,
+ advances_slot,
+ "operation exceeded the shared 10s ACK batch budget",
+ )?;
+ Ok(false)
+ }
+ }
+ }
+
+ async fn execute_pending_operation(&self, operation: &PendingOperation) ->
Result<(), Error> {
+ match operation {
+ PendingOperation::ProcessRows {
+ table,
+ ids,
+ tracking_boundary,
+ } => {
+ let pool = self.get_pool()?;
+ self.mark_or_delete_processed_rows(pool, table, ids,
tracking_boundary.as_deref())
+ .await
+ }
+ PendingOperation::AdvanceReplicationSlot { lsn } => {
+ self.advance_replication_slot(lsn).await
}
- state.last_poll_time = Utc::now();
}
+ }
- Ok(messages)
+ fn record_pending_operation_failure(
+ &self,
+ operation_name: &str,
+ advances_slot: bool,
+ error: impl std::fmt::Display,
+ ) -> Result<(), Error> {
+ if advances_slot {
+ let consecutive_failures = self
+ .consecutive_advance_failures
+ .fetch_add(1, Ordering::Relaxed)
+ .saturating_add(1);
+ error!(
+ "Failed to {operation_name} for PostgreSQL source connector
ID: {}. \
+ Consecutive replication slot advance failures:
{consecutive_failures}. {error}",
+ self.id
+ );
+ if consecutive_failures >= MAX_CONSECUTIVE_ADVANCE_FAILURES {
+ return Err(Error::Connection(format!(
+ "stopping PostgreSQL source connector ID {} after
{consecutive_failures} \
+ consecutive replication slot advance failures",
+ self.id
+ )));
+ }
+ } else {
+ error!(
+ "Failed to {operation_name} for PostgreSQL source connector
ID: {}. {error}",
+ self.id
+ );
+ }
+ Ok(())
}
async fn mark_or_delete_processed_rows(
&self,
pool: &Pool<Postgres>,
table: &str,
- pk_column: &str,
ids: &[String],
+ tracking_boundary: Option<&str>,
) -> Result<(), Error> {
if ids.is_empty() {
return Ok(());
}
let quoted_table = quote_qualified_identifier(table)?;
- let quoted_pk = quote_identifier(pk_column)?;
+ let quoted_pk = quote_identifier(self.primary_key_column())?;
Review Comment:
critical: this existing fallback can delete or mark rows that were never
emitted when tracking values are shared. require the resolved cleanup key to
identify rows uniquely, including when its column is configured explicitly.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -230,38 +297,77 @@ impl Source for PostgresSource {
}
};
- let state = self.state.lock().await;
+ let processed_rows = self.state.lock().await.processed_rows;
if self.verbose {
info!(
"PostgreSQL source connector ID: {} produced {} messages.
Total processed: {}",
self.id,
- messages.len(),
- state.processed_rows
+ polled.messages.len(),
+ processed_rows
);
} else {
debug!(
"PostgreSQL source connector ID: {} produced {} messages.
Total processed: {}",
self.id,
- messages.len(),
- state.processed_rows
+ polled.messages.len(),
+ 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);
+ let persisted_state = polled
Review Comment:
warning: idle CDC polls still save unchanged state, repeatedly writing and
syncing the checkpoint file. return no state when its contents are unchanged,
while preserving slot advancement and empty-batch ACK handling.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1460,6 +1777,59 @@ fn format_offset_value(value: &str) -> String {
}
}
+fn build_tracking_condition(
+ tracking_column: &str,
+ tracking_boundary: Option<&str>,
+) -> Result<String, Error> {
+ let Some(boundary) = tracking_boundary else {
+ return Ok(String::new());
+ };
+ let quoted_tracking = quote_identifier(tracking_column)?;
+ Ok(format!(
+ " AND ({quoted_tracking} <= {} OR {quoted_tracking} IS NULL)",
+ format_offset_value(boundary)
+ ))
+}
+
+fn replication_slot_target_lsn(last_change_lsn: Option<String>, wal_flush_lsn:
String) -> String {
+ last_change_lsn.unwrap_or(wal_flush_lsn)
+}
+
+async fn replication_slot_reached_target(
+ pool: &Pool<Postgres>,
+ slot_name: &str,
+ target_lsn: &str,
+) -> Result<bool, sqlx::Error> {
+ Ok(sqlx::query_scalar::<_, bool>(
+ "SELECT COALESCE(confirmed_flush_lsn >= $2::pg_lsn, FALSE) \
+ FROM pg_replication_slots WHERE slot_name = $1",
+ )
+ .bind(slot_name)
+ .bind(target_lsn)
+ .fetch_optional(pool)
+ .await?
+ .unwrap_or(false))
+}
+
+fn extract_tracking_value(
+ row: &sqlx::postgres::PgRow,
+ column_index: usize,
+ value: &serde_json::Value,
+) -> Result<Option<String>, Error> {
+ if row.columns()[column_index].type_info().name() == "NUMERIC" {
Review Comment:
simplification: `extract_tracking_value` decodes `NUMERIC` again after
`extract_column_value` already produced the exact string. reuse that value and
remove the duplicate branch, row/index parameters, and `Result` wrapper.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -837,19 +1163,11 @@ impl PostgresSource {
data.insert(column_name.clone(), value.clone());
Review Comment:
simplification: row conversion still clones the owned column name and JSON
value before extracting tracking fields. extract tracking and key values first,
then move both into `data.insert(column_name, value)`.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1979,6 +2436,24 @@ mod tests {
assert!(validate_capture_operations(None).is_ok());
}
+ #[test]
+ fn given_postgres_10_should_fail_cdc_version_validation() {
+ let error = validate_cdc_server_version_num(100_000).unwrap_err();
+
+ assert!(matches!(error, Error::InitError(_)));
+ }
+
+ #[test]
+ fn given_postgres_11_or_newer_should_pass_cdc_version_validation() {
+ assert!(validate_cdc_server_version_num(110_000).is_ok());
+ assert!(validate_cdc_server_version_num(170_000).is_ok());
+ }
+
+ #[test]
+ fn statement_timeout_should_expire_before_ack_backstop() {
Review Comment:
simplification: this test only compares compile-time timeout constants.
replace it with `const _: () = assert!(STATEMENT_TIMEOUT.as_nanos() <
ACK_BATCH_TIMEOUT.as_nanos());` beside the constants.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1460,6 +1777,59 @@ fn format_offset_value(value: &str) -> String {
}
}
+fn build_tracking_condition(
+ tracking_column: &str,
+ tracking_boundary: Option<&str>,
+) -> Result<String, Error> {
+ let Some(boundary) = tracking_boundary else {
+ return Ok(String::new());
+ };
+ let quoted_tracking = quote_identifier(tracking_column)?;
+ Ok(format!(
+ " AND ({quoted_tracking} <= {} OR {quoted_tracking} IS NULL)",
+ format_offset_value(boundary)
+ ))
+}
+
+fn replication_slot_target_lsn(last_change_lsn: Option<String>, wal_flush_lsn:
String) -> String {
Review Comment:
simplification: this helper only wraps `Option::unwrap_or`, and its tests
only check that expression. inline `last_lsn.unwrap_or(wal_flush_lsn)` at the
caller and remove the helper-only tests.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -417,43 +533,55 @@ impl PostgresSource {
(!self.config.tables.is_empty()).then_some(self.config.tables.as_slice());
let batch_size = self.config.batch_size.unwrap_or(1000) as i32;
+ let wal_flush_lsn = with_retry(
+ || sqlx::query_scalar("SELECT
pg_current_wal_flush_lsn()::text").fetch_one(pool),
+ self.get_max_retries(),
+ self.retry_delay.as_millis() as u64,
+ )
+ .await
+ .map_err(|e| Error::Connection(format!("failed to read current WAL
flush LSN: {e}")))?;
+
// Database I/O without holding the lock. upto_nchanges is only
// checked at transaction-commit boundaries (a single huge transaction
// can still exceed it), so this isn't a hard per-call cap - but it
// stops the backlog from growing unbounded across many transactions
// the way NULL (no limit at all) did.
- let rows =
- sqlx::query("SELECT lsn, xid, data FROM
pg_logical_slot_get_changes($1, NULL, $2)")
+ let rows = with_retry(
+ || {
+ sqlx::query(
+ "SELECT lsn::text AS lsn, data FROM
pg_logical_slot_peek_changes($1, NULL, $2)",
+ )
.bind(slot_name)
.bind(batch_size)
.fetch_all(pool)
- .await
- .map_err(|e| {
- error!("Failed to fetch CDC changes: {e}");
- Error::InvalidRecord
- })?;
+ },
+ self.get_max_retries(),
+ self.retry_delay.as_millis() as u64,
+ )
+ .await
+ .map_err(|e| Error::Connection(format!("failed to fetch CDC changes:
{e}")))?;
let mut messages = Vec::new();
+ let mut last_lsn = None;
for row in rows {
- let data: String = match row.try_get("data") {
- Ok(data) => data,
- Err(e) => {
- error!("Skipping CDC row with unreadable data column:
{e}");
- continue;
- }
- };
+ let lsn: String = row.try_get("lsn").map_err(|e| {
Review Comment:
simplification: CDC decodes an owned LSN for every row but only keeps the
last. decode `rows.last()` once, preserving the pre-peek fallback, all payload
checks, and progress through filtered rows.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +680,207 @@ impl PostgresSource {
messages.push(processed.message);
total_processed += 1;
+ table_processed += 1;
}
- // Database I/O without holding the lock
- if !processed_ids.is_empty() {
- self.mark_or_delete_processed_rows(pool, table, pk_column,
&processed_ids)
- .await?;
+ let tracking_boundary = self.processing_boundary(max_offset);
+
+ if self.should_process_rows() && !processed_ids.is_empty() {
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ tracking_boundary: tracking_boundary.clone(),
+ });
}
- // Collect offset update for later
- if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ if let Some(offset) = tracking_boundary {
+ candidate_state
+ .tracking_offsets
+ .insert(table.clone(), offset);
}
if self.verbose {
- info!("Fetched {} rows from table '{table}'", messages.len());
+ info!("Fetched {table_processed} rows from table '{table}'");
} else {
- debug!("Fetched {} rows from table '{table}'", messages.len());
+ debug!("Fetched {table_processed} rows from table '{table}'");
}
}
- // Apply all state updates with a single lock acquisition
- {
- let mut state = self.state.lock().await;
- state.processed_rows += total_processed;
- for (table, offset) in state_updates {
- state.tracking_offsets.insert(table, offset);
+ let pending = if total_processed > 0 {
+ candidate_state.processed_rows += total_processed;
+ candidate_state.last_poll_time = Utc::now();
+ Some(PendingBatch {
+ state: candidate_state,
+ operations,
+ acknowledged: false,
+ })
+ } else {
+ None
+ };
+
+ Ok(PolledBatch { messages, pending })
+ }
+
+ async fn advance_replication_slot(&self, lsn: &str) -> Result<(), Error> {
+ let slot_name = self.replication_slot();
+ let pool = self.get_pool()?;
+ with_retry(
+ || async {
+ let result = sqlx::query("SELECT
pg_replication_slot_advance($1, $2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(pool)
+ .await;
+
+ if let Err(error) = result {
+ match replication_slot_reached_target(pool, slot_name,
lsn).await {
+ Ok(true) => {
+ warn!(
+ "PostgreSQL source connector ID: {} confirmed
replication slot \
+ '{slot_name}' is already at or beyond {lsn}
after advance failed: \
+ {error}",
+ self.id
+ );
+ return Ok(());
+ }
+ Ok(false) => {}
+ Err(read_error) => warn!(
+ "PostgreSQL source connector ID: {} failed to read
replication slot \
+ '{slot_name}' after advance failed: {read_error}",
+ self.id
+ ),
+ }
+ return Err(error);
+ }
+
+ Ok(())
+ },
+ self.get_max_retries(),
+ self.retry_delay.as_millis() as u64,
+ )
+ .await
+ .map_err(|e| {
+ Error::Connection(format!(
+ "failed to advance replication slot '{slot_name}' to {lsn}:
{e}"
+ ))
+ })?;
+ Ok(())
+ }
+
+ async fn apply_pending_operations(
+ &self,
+ operations: Vec<PendingOperation>,
+ deadline: tokio::time::Instant,
+ ) -> Result<Vec<PendingOperation>, Error> {
+ let mut failed_operations = Vec::new();
+ for operation in operations {
+ if !self.apply_pending_operation(&operation, deadline).await? {
+ failed_operations.push(operation);
+ }
+ }
+ Ok(failed_operations)
+ }
+
+ async fn apply_pending_operation(
+ &self,
+ operation: &PendingOperation,
+ deadline: tokio::time::Instant,
+ ) -> Result<bool, Error> {
+ let (operation_name, advances_slot) = match operation {
+ PendingOperation::ProcessRows { .. } => ("process rows", false),
+ PendingOperation::AdvanceReplicationSlot { .. } => ("advance
replication slot", true),
+ };
+
+ let result =
+ tokio::time::timeout_at(deadline,
self.execute_pending_operation(operation)).await;
+
+ match result {
+ Ok(Ok(())) => {
+ if advances_slot {
+ self.consecutive_advance_failures
+ .store(0, Ordering::Relaxed);
+ }
+ Ok(true)
+ }
+ Ok(Err(error)) => {
+ self.record_pending_operation_failure(operation_name,
advances_slot, &error)?;
+ Ok(false)
+ }
+ Err(_) => {
+ self.record_pending_operation_failure(
+ operation_name,
+ advances_slot,
+ "operation exceeded the shared 10s ACK batch budget",
+ )?;
+ Ok(false)
+ }
+ }
+ }
+
+ async fn execute_pending_operation(&self, operation: &PendingOperation) ->
Result<(), Error> {
+ match operation {
+ PendingOperation::ProcessRows {
+ table,
+ ids,
+ tracking_boundary,
+ } => {
+ let pool = self.get_pool()?;
+ self.mark_or_delete_processed_rows(pool, table, ids,
tracking_boundary.as_deref())
+ .await
+ }
+ PendingOperation::AdvanceReplicationSlot { lsn } => {
+ self.advance_replication_slot(lsn).await
}
- state.last_poll_time = Utc::now();
}
+ }
- Ok(messages)
+ fn record_pending_operation_failure(
Review Comment:
simplification: `operation_name` and `advances_slot` encode the same enum
variant. pass `&PendingOperation` to the failure handler and match there,
preserving the slot-only counter and success reset.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +680,207 @@ impl PostgresSource {
messages.push(processed.message);
total_processed += 1;
+ table_processed += 1;
}
- // Database I/O without holding the lock
- if !processed_ids.is_empty() {
- self.mark_or_delete_processed_rows(pool, table, pk_column,
&processed_ids)
- .await?;
+ let tracking_boundary = self.processing_boundary(max_offset);
+
+ if self.should_process_rows() && !processed_ids.is_empty() {
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ tracking_boundary: tracking_boundary.clone(),
+ });
}
- // Collect offset update for later
- if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ if let Some(offset) = tracking_boundary {
+ candidate_state
+ .tracking_offsets
+ .insert(table.clone(), offset);
}
if self.verbose {
- info!("Fetched {} rows from table '{table}'", messages.len());
+ info!("Fetched {table_processed} rows from table '{table}'");
} else {
- debug!("Fetched {} rows from table '{table}'", messages.len());
+ debug!("Fetched {table_processed} rows from table '{table}'");
}
}
- // Apply all state updates with a single lock acquisition
- {
- let mut state = self.state.lock().await;
- state.processed_rows += total_processed;
- for (table, offset) in state_updates {
- state.tracking_offsets.insert(table, offset);
+ let pending = if total_processed > 0 {
+ candidate_state.processed_rows += total_processed;
+ candidate_state.last_poll_time = Utc::now();
+ Some(PendingBatch {
+ state: candidate_state,
+ operations,
+ acknowledged: false,
+ })
+ } else {
+ None
+ };
+
+ Ok(PolledBatch { messages, pending })
+ }
+
+ async fn advance_replication_slot(&self, lsn: &str) -> Result<(), Error> {
+ let slot_name = self.replication_slot();
+ let pool = self.get_pool()?;
+ with_retry(
+ || async {
+ let result = sqlx::query("SELECT
pg_replication_slot_advance($1, $2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(pool)
+ .await;
+
+ if let Err(error) = result {
+ match replication_slot_reached_target(pool, slot_name,
lsn).await {
+ Ok(true) => {
+ warn!(
+ "PostgreSQL source connector ID: {} confirmed
replication slot \
+ '{slot_name}' is already at or beyond {lsn}
after advance failed: \
+ {error}",
+ self.id
+ );
+ return Ok(());
+ }
+ Ok(false) => {}
+ Err(read_error) => warn!(
+ "PostgreSQL source connector ID: {} failed to read
replication slot \
+ '{slot_name}' after advance failed: {read_error}",
+ self.id
+ ),
+ }
+ return Err(error);
+ }
+
+ Ok(())
+ },
+ self.get_max_retries(),
+ self.retry_delay.as_millis() as u64,
+ )
+ .await
+ .map_err(|e| {
+ Error::Connection(format!(
+ "failed to advance replication slot '{slot_name}' to {lsn}:
{e}"
+ ))
+ })?;
+ Ok(())
+ }
+
+ async fn apply_pending_operations(
+ &self,
+ operations: Vec<PendingOperation>,
+ deadline: tokio::time::Instant,
+ ) -> Result<Vec<PendingOperation>, Error> {
+ let mut failed_operations = Vec::new();
+ for operation in operations {
+ if !self.apply_pending_operation(&operation, deadline).await? {
+ failed_operations.push(operation);
+ }
+ }
+ Ok(failed_operations)
+ }
+
+ async fn apply_pending_operation(
+ &self,
+ operation: &PendingOperation,
+ deadline: tokio::time::Instant,
+ ) -> Result<bool, Error> {
+ let (operation_name, advances_slot) = match operation {
+ PendingOperation::ProcessRows { .. } => ("process rows", false),
+ PendingOperation::AdvanceReplicationSlot { .. } => ("advance
replication slot", true),
+ };
+
+ let result =
+ tokio::time::timeout_at(deadline,
self.execute_pending_operation(operation)).await;
+
+ match result {
+ Ok(Ok(())) => {
+ if advances_slot {
+ self.consecutive_advance_failures
+ .store(0, Ordering::Relaxed);
+ }
+ Ok(true)
+ }
+ Ok(Err(error)) => {
+ self.record_pending_operation_failure(operation_name,
advances_slot, &error)?;
+ Ok(false)
+ }
+ Err(_) => {
+ self.record_pending_operation_failure(
+ operation_name,
+ advances_slot,
+ "operation exceeded the shared 10s ACK batch budget",
+ )?;
+ Ok(false)
+ }
+ }
+ }
+
+ async fn execute_pending_operation(&self, operation: &PendingOperation) ->
Result<(), Error> {
Review Comment:
simplification: this private helper only dispatches for one caller. inline
its match inside an async block passed to `timeout_at`, keeping pool lookup and
both awaited operations inside the deadline.
##########
core/integration/tests/connectors/postgres/postgres_source.rs:
##########
@@ -127,6 +134,232 @@ async fn json_rows_source_produces_messages_to_iggy(
}
}
+#[iggy_harness(
+ cluster_nodes = 1,
+ server(connectors_runtime(config_path =
"tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn
given_delete_after_read_when_iggy_crashes_should_delete_only_after_redelivery(
+ harness: &mut TestHarness,
+ fixture: PostgresSourceDeleteFixture,
+) {
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
+
+ harness
+ .server_mut()
+ .stop_dependents()
+ .expect("Failed to stop connectors runtime");
+ harness
+ .server_mut()
+ .connectors_runtime_mut()
+ .expect("connectors runtime")
+ // Keep a failed send bounded instead of waiting indefinitely for Iggy
to return.
+ .set_iggy_connection_options("reconnection_retries=0");
+ harness
+ .server_mut()
+ .start_dependents()
+ .await
+ .expect("Failed to restart connectors runtime");
+
+ let api_url = harness
+ .connectors_runtime()
+ .expect("connectors runtime")
+ .http_url();
+ let http = Client::new();
+ let errors_before_failure = source_stats(&http, &api_url)
+ .await
+ .expect("PostgreSQL source stats should be present")
+ .errors;
+
+ harness.kill_node(0).expect("Failed to kill Iggy server");
+
+ for index in 0..TEST_MESSAGE_COUNT {
+ fixture
+ .insert_row(&pool, &format!("row_{index}"), index as i32)
+ .await;
+ }
+
+ let failed_source = wait_for_source_errors(&http, &api_url,
errors_before_failure + 2).await;
+ assert_eq!(failed_source.status, ConnectorStatus::Error);
+ assert_eq!(
+ fixture.count_rows(&pool).await,
+ TEST_MESSAGE_COUNT as i64,
+ "NACKed rows must not be deleted"
+ );
+
+ harness
+ .server_mut()
+ .stop_dependents()
+ .expect("Failed to stop connectors runtime");
+ harness
+ .restart_node(0)
+ .expect("Failed to restart Iggy server");
+ harness
+ .server_mut()
+ .connectors_runtime_mut()
+ .expect("connectors runtime")
+ .clear_iggy_connection_options();
+ harness
+ .server_mut()
+ .start_dependents()
+ .await
+ .expect("Failed to restart connectors runtime");
+
+ let client = harness.root_client().await.unwrap();
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "send_failure_consumer".try_into().unwrap();
+ let mut received = 0;
+
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(polled) = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ 10,
+ true,
+ )
+ .await
+ {
+ received += polled.messages.len();
+ if received >= TEST_MESSAGE_COUNT {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+
+ assert_eq!(
+ received, TEST_MESSAGE_COUNT,
+ "Rows polled during the failed send should be delivered after restart"
+ );
+
+ let mut remaining_rows = fixture.count_rows(&pool).await;
+ for _ in 0..POLL_ATTEMPTS {
+ if remaining_rows == 0 {
+ break;
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ remaining_rows = fixture.count_rows(&pool).await;
+ }
+ assert_eq!(remaining_rows, 0, "ACKed rows should be deleted");
+
+ pool.close().await;
+}
+
+#[iggy_harness(
+ cluster_nodes = 1,
+ server(connectors_runtime(config_path =
"tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn
given_delivery_failure_when_iggy_restarts_should_redeliver_without_runtime_restart(
+ harness: &mut TestHarness,
+ fixture: PostgresSourceDeleteSlowPollFixture,
+) {
+ const REDELIVERY_ATTEMPTS: usize = POLL_ATTEMPTS * 3;
+
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
+
+ harness
+ .server_mut()
+ .stop_dependents()
+ .expect("Failed to stop connectors runtime");
+ harness
+ .server_mut()
+ .connectors_runtime_mut()
+ .expect("connectors runtime")
+ .set_iggy_connection_options("reconnection_retries=0");
+ harness
+ .server_mut()
+ .start_dependents()
+ .await
+ .expect("Failed to restart connectors runtime");
+
+ let api_url = harness
+ .connectors_runtime()
+ .expect("connectors runtime")
+ .http_url();
+ let http = Client::new();
+ let errors_before_failure = source_stats(&http, &api_url)
+ .await
+ .expect("PostgreSQL source stats should be present")
+ .errors;
+
+ harness.kill_node(0).expect("Failed to kill Iggy server");
+ fixture.insert_row(&pool, "single_nack", 1).await;
+
+ let failed_source = wait_for_source_errors(&http, &api_url,
errors_before_failure + 1).await;
+ assert_eq!(failed_source.status, ConnectorStatus::Error);
+ assert_eq!(
+ fixture.count_rows(&pool).await,
+ 1,
+ "NACKed row must not be deleted"
+ );
+
+ harness
+ .restart_node(0)
+ .expect("Failed to restart only the Iggy server");
+
+ let client = harness.root_client().await.unwrap();
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "nack_survival_consumer".try_into().unwrap();
+ let mut received = 0;
+
+ for _ in 0..REDELIVERY_ATTEMPTS {
+ if let Ok(polled) = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ 10,
+ true,
+ )
+ .await
+ {
+ received += polled.messages.len();
+ if received == 1 {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+
+ assert_eq!(received, 1, "NACKed row should be redelivered");
+
+ let mut remaining_rows = fixture.count_rows(&pool).await;
+ for _ in 0..REDELIVERY_ATTEMPTS {
+ if remaining_rows == 0 {
+ break;
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ remaining_rows = fixture.count_rows(&pool).await;
+ }
+ assert_eq!(remaining_rows, 0, "ACKed row should be deleted");
+ wait_for_source_status(&http, &api_url, ConnectorStatus::Running).await;
+
+ pool.close().await;
+}
+
+async fn wait_for_source_status(http: &Client, api_url: &str, expected:
ConnectorStatus) {
Review Comment:
simplification: this status wait duplicates `postgres_source_cdc.rs:530`,
and callers ignore its return value. share a unit-returning helper in
`postgres/mod.rs` that reuses `source_stats`, preserving the retry count, sleep
placement, and failure message.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -610,26 +895,28 @@ impl PostgresSource {
.join(", ");
if self.config.delete_after_read.unwrap_or(false) {
- let delete_query =
- format!("DELETE FROM {quoted_table} WHERE {quoted_pk} IN
({ids_list})");
+ let delete_query = format!(
+ "DELETE FROM {quoted_table} WHERE {quoted_pk} IN
({ids_list}){tracking_condition}"
+ );
if self.verbose {
info!("Deleting {} processed rows from '{table}'", ids.len());
} else {
debug!("Deleting {} processed rows from '{table}'", ids.len());
}
- sqlx::query(sqlx::AssertSqlSafe(delete_query))
- .execute(pool)
- .await
- .map_err(|e| {
- error!("Failed to delete processed rows: {e}");
- Error::InvalidRecord
- })?;
+ with_retry(
+ ||
sqlx::query(sqlx::AssertSqlSafe(delete_query.as_str())).execute(pool),
Review Comment:
warning: dynamic cleanup queries still cache batch-specific SQL, displacing
reusable plans when the cache fills. use `.persistent(false)` for changing
cleanup and polling statements, while retaining caching for stable bound
queries.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -533,10 +659,14 @@ impl PostgresSource {
self.get_max_retries(),
self.retry_delay.as_millis() as u64,
)
- .await?;
+ .await
+ .map_err(|e| {
+ Error::Connection(format!("failed to poll PostgreSQL table
'{table}': {e}"))
+ })?;
let mut max_offset: Option<String> = None;
let mut processed_ids: Vec<String> = Vec::new();
Review Comment:
nit: this existing `processed_ids` vector grows without reserving space even
though the fetched row count is known. initialize it with
`Vec::with_capacity(rows.len())`.
##########
core/integration/tests/connectors/fixtures/postgres/source.rs:
##########
@@ -382,6 +382,141 @@ impl TestFixture for PostgresSourceDeleteFixture {
}
}
+/// The longer poll interval leaves time to restart Iggy before the SDK's
+/// consecutive-NACK limit stops the source.
+pub struct PostgresSourceDeleteSlowPollFixture {
+ inner: PostgresSourceDeleteFixture,
+}
+
+impl PostgresOps for PostgresSourceDeleteSlowPollFixture {
+ fn container(&self) -> &PostgresContainer {
+ self.inner.container()
+ }
+}
+
+impl PostgresSourceOps for PostgresSourceDeleteSlowPollFixture {
+ fn table_name(&self) -> &str {
+ self.inner.table_name()
+ }
+}
+
+impl PostgresSourceDeleteSlowPollFixture {
Review Comment:
simplification: this wrapper and
`core/integration/tests/connectors/fixtures/postgres/cdc.rs:226` repeat
forwarding methods. use the `Deref` pattern from
`core/integration/tests/connectors/fixtures/postgres/sink.rs:164`, retaining
both named types, trait implementations, setup, and interval overrides.
--
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]
