hubcio commented on code in PR #3957:
URL: https://github.com/apache/iggy/pull/3957#discussion_r3957387433
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1521,6 +1772,50 @@ fn validate_capture_operations(capture_operations:
Option<&[String]>) -> Result<
Ok(())
}
+fn validate_retry_budget(max_retries: u32, retry_delay: Duration) ->
Result<(), Error> {
Review Comment:
critical: this rejects configs that booted before - `max_retries=3` with
`retry_delay="2s"`, both straight from the readme, needs 12s and fails startup,
and n>=5 never passes at all. bound only the ACK retry loop, then document the
range.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -253,15 +290,40 @@ impl Source for PostgresSource {
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> {
+ let (SourceBatchResult::Ack, Some(pending)) =
+ (result, self.pending_batch.lock().await.take())
+ else {
+ return Ok(());
+ };
+
+ let PendingBatch { state, operations } = pending;
+ for operation in operations {
+ self.apply_pending_operation(operation).await;
+ }
+
+ *self.state.lock().await = state;
Review Comment:
critical: state commits even when the loop above failed.
`apply_pending_operation` returns `()` and only logs, so failed rows still
advance the offset and never come back. keep the failed op staged, or return
`Err`.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +621,165 @@ 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?;
+ if self.should_process_rows() && !processed_ids.is_empty() {
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ tracking_boundary:
self.processing_boundary(max_offset.clone()),
+ });
}
- // Collect offset update for later
if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ 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,
+ })
+ } 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 {
+ match sqlx::query("SELECT pg_replication_slot_advance($1,
$2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(pool)
+ .await
+ {
+ Err(error) if is_invalid_parameter_value(&error) => {
+ warn!(
+ "PostgreSQL source connector ID: {} received
invalid_parameter_value \
+ while advancing replication slot '{slot_name}' to
{lsn}; treating the \
+ target as already applied: {error}",
+ self.id
+ );
+ Ok(())
+ }
+ result => result.map(|_| ()),
+ }
+ },
+ 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_operation(&self, operation: PendingOperation) {
+ let (operation_name, advances_slot) = match &operation {
+ PendingOperation::ProcessRows { .. } => ("process rows", false),
+ PendingOperation::AdvanceReplicationSlot { .. } => ("advance
replication slot", true),
+ };
+
+ let result = tokio::time::timeout(
+ ACK_OPERATION_TIMEOUT,
+ self.execute_pending_operation(operation),
+ )
+ .await;
+
+ match result {
+ Ok(Ok(())) => {
+ if advances_slot {
+ self.consecutive_advance_failures
+ .store(0, Ordering::Relaxed);
+ }
}
- state.last_poll_time = Utc::now();
+ Ok(Err(error)) => {
+ self.log_pending_operation_failure(operation_name,
advances_slot, &error)
+ }
+ Err(_) => self.log_pending_operation_failure(
+ operation_name,
+ advances_slot,
+ "operation exceeded the 10s ACK budget",
+ ),
}
+ }
- Ok(messages)
+ 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
+ }
+ }
+ }
+
+ fn log_pending_operation_failure(
Review Comment:
critical: a failed slot advance only logs. the batch is already ACKed, so
nothing stops it - the same changes reship every poll and WAL grows forever.
return `Err` past a threshold and let the SDK stop the source.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -995,11 +1209,11 @@ fn extract_column_value(
.unwrap_or(serde_json::Value::Null))
}
"NUMERIC" => {
- let value: Option<String> = row
+ let value: Option<BigDecimal> = row
.try_get(column_index)
.map_err(|_| Error::InvalidRecord)?;
Ok(value
- .and_then(|s| s.parse::<f64>().ok())
+ .and_then(|value| value.to_string().parse::<f64>().ok())
Review Comment:
warning: the payload still collapses NUMERIC to f64 while the ACK boundary
keeps full precision. emit the decimal as a string, and add a payload assertion
next to `numeric_tracking_source_preserves_exact_ack_boundary`.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -289,6 +351,7 @@ impl PostgresSource {
let pool = PgPoolOptions::new()
.max_connections(max_connections)
+ .acquire_timeout(POOL_ACQUIRE_TIMEOUT)
.connect(self.config.connection_string.expose_secret())
Review Comment:
warning: no statement timeout anywhere. after acquire, `fetch_all` and
`execute` block without limit, and a wedged backend stalls the source with no
error and no status change. set `statement_timeout` on the pool.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -389,7 +448,7 @@ impl PostgresSource {
Ok(())
Review Comment:
warning: nothing probes `server_version`, yet the readme now claims PG 11+.
every check here works on 9.4, and a missing `pg_replication_slot_advance`
raises 42883 which the ACK path only logs. probe the version and fail `open()`.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +621,165 @@ 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?;
+ if self.should_process_rows() && !processed_ids.is_empty() {
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ tracking_boundary:
self.processing_boundary(max_offset.clone()),
+ });
}
- // Collect offset update for later
if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ 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,
+ })
+ } 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 {
+ match sqlx::query("SELECT pg_replication_slot_advance($1,
$2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(pool)
+ .await
+ {
+ Err(error) if is_invalid_parameter_value(&error) => {
+ warn!(
+ "PostgreSQL source connector ID: {} received
invalid_parameter_value \
+ while advancing replication slot '{slot_name}' to
{lsn}; treating the \
+ target as already applied: {error}",
+ self.id
+ );
+ Ok(())
+ }
+ result => result.map(|_| ()),
+ }
+ },
+ 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_operation(&self, operation: PendingOperation) {
+ let (operation_name, advances_slot) = match &operation {
+ PendingOperation::ProcessRows { .. } => ("process rows", false),
+ PendingOperation::AdvanceReplicationSlot { .. } => ("advance
replication slot", true),
+ };
+
+ let result = tokio::time::timeout(
+ ACK_OPERATION_TIMEOUT,
+ self.execute_pending_operation(operation),
+ )
+ .await;
Review Comment:
warning: `timeout` drops the client future, it does not abort the statement
- a DELETE or a slot advance can still commit server-side after this logs a
failure. set `statement_timeout` and keep this as a backstop.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -739,6 +955,8 @@ impl PostgresSource {
let now = Utc::now();
+ // TODO: Substitute `$now_unix` before `$now` so the longer
placeholder remains intact.
+ // TODO: Bind or quote `$offset` according to its PostgreSQL type
instead of inserting raw data.
Review Comment:
critical: pre-existing, but this is a TODO in place of a one-line fix.
`$offset` comes from row data, is spliced raw under `AssertSqlSafe`, and
survives restarts in connector state. quote it through `format_offset_value`.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -837,11 +1055,7 @@ impl PostgresSource {
data.insert(column_name.clone(), value.clone());
if column.name() == config.tracking_column {
- if let serde_json::Value::String(ref s) = value {
- max_offset = Some(s.clone());
- } else if let serde_json::Value::Number(ref n) = value {
- max_offset = Some(n.to_string());
- }
+ max_offset = extract_tracking_value(row, i, &value)?;
}
if column.name() == config.pk_column {
Review Comment:
warning: `row_pk` still goes through the f64 path while the tracking
boundary is now exact. with pk defaulting to a NUMERIC tracking column,
9007199254740993.25 becomes '9007199254740994.0' and the DELETE matches
nothing. route this through `extract_tracking_value`.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1521,6 +1772,50 @@ fn validate_capture_operations(capture_operations:
Option<&[String]>) -> Result<
Ok(())
}
+fn validate_retry_budget(max_retries: u32, retry_delay: Duration) ->
Result<(), Error> {
+ if max_retries == 0 {
+ return Err(Error::InitError(
+ "max_retries must be greater than zero".to_string(),
+ ));
+ }
+
+ let attempts = u128::from(max_retries);
+ let retry_delay_steps = attempts * (attempts - 1) / 2;
+ let acquisition_budget = POOL_ACQUIRE_TIMEOUT
+ .as_nanos()
+ .checked_mul(attempts)
+ .ok_or_else(retry_budget_overflow)?;
+ let retry_delay_budget = retry_delay
+ .as_nanos()
+ .checked_mul(retry_delay_steps)
+ .ok_or_else(retry_budget_overflow)?;
+ let retry_budget = acquisition_budget
+ .checked_add(retry_delay_budget)
+ .ok_or_else(retry_budget_overflow)?;
+
+ if retry_budget >= ACK_OPERATION_TIMEOUT.as_nanos() {
+ return Err(Error::InitError(format!(
+ "max_retries ({max_retries}) and retry_delay ({retry_delay:?})
require an ACK retry \
+ budget greater than or equal to {ACK_OPERATION_TIMEOUT:?}"
+ )));
+ }
+
+ Ok(())
+}
+
+fn retry_budget_overflow() -> Error {
+ Error::InitError("configured ACK retry budget is too large".to_string())
+}
+
+fn validate_row_processing_config(config: &PostgresSourceConfig) -> Result<(),
Error> {
Review Comment:
warning: this pair was legal before, delete simply won over the processed
column. the readme quick start ships both keys together, so flipping
`delete_after_read` to true now breaks startup. warn and keep the precedence,
or document the exclusion.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1521,6 +1772,50 @@ fn validate_capture_operations(capture_operations:
Option<&[String]>) -> Result<
Ok(())
}
+fn validate_retry_budget(max_retries: u32, retry_delay: Duration) ->
Result<(), Error> {
+ if max_retries == 0 {
Review Comment:
warning: `max_retries = 0` was legal and meant one attempt - `with_retry`
tests `attempts >= max_retries` after incrementing, so 0 and 1 behave the same.
keep 0 working, or document the break.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -669,6 +855,36 @@ 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: the boundary is skipped for custom queries, but line 635 still
commits the cursor for them, and `max_offset` is last-wins rather than max. an
unordered custom query then skips rows for good. guard the cursor too, or
require `ORDER BY`.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1677,16 +1976,28 @@ fn is_transient_error(e: &sqlx::Error) -> bool {
sqlx::Error::PoolTimedOut => true,
sqlx::Error::PoolClosed => false,
sqlx::Error::Protocol(_) => false,
- sqlx::Error::Database(db_err) => db_err.code().is_some_and(|code| {
- matches!(
- code.as_ref(),
- "40001" | "40P01" | "57P01" | "57P02" | "57P03" | "08000" |
"08003" | "08006"
- )
- }),
+ sqlx::Error::Database(db_err) => db_err
+ .code()
+ .is_some_and(|code| is_transient_sqlstate(code.as_ref())),
_ => false,
}
}
+fn is_transient_sqlstate(code: &str) -> bool {
+ matches!(
+ code,
+ "40001" | "40P01" | "55006" | "57P01" | "57P02" | "57P03" | "08000" |
"08003" | "08006"
+ )
+}
+
+fn is_invalid_parameter_value(error: &sqlx::Error) -> bool {
+ matches!(error, sqlx::Error::Database(database_error) if
database_error.code().is_some_and(|code|
is_invalid_parameter_value_sqlstate(code.as_ref())))
+}
+
+fn is_invalid_parameter_value_sqlstate(code: &str) -> bool {
Review Comment:
warning: `pg_replication_slot_advance` raises 55000 for an already-applied
target, never 22023, so this branch is dead on PG 12-17 and the benign case
falls through as a hard error. read `confirmed_flush_lsn` back instead -
remapping to 55000 would swallow an invalidated slot.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +621,165 @@ 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?;
+ if self.should_process_rows() && !processed_ids.is_empty() {
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ tracking_boundary:
self.processing_boundary(max_offset.clone()),
+ });
}
- // Collect offset update for later
if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ 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,
+ })
+ } 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 {
+ match sqlx::query("SELECT pg_replication_slot_advance($1,
$2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(pool)
+ .await
+ {
+ Err(error) if is_invalid_parameter_value(&error) => {
+ warn!(
+ "PostgreSQL source connector ID: {} received
invalid_parameter_value \
+ while advancing replication slot '{slot_name}' to
{lsn}; treating the \
+ target as already applied: {error}",
+ self.id
+ );
+ Ok(())
+ }
+ result => result.map(|_| ()),
+ }
+ },
+ 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_operation(&self, operation: PendingOperation) {
+ let (operation_name, advances_slot) = match &operation {
+ PendingOperation::ProcessRows { .. } => ("process rows", false),
+ PendingOperation::AdvanceReplicationSlot { .. } => ("advance
replication slot", true),
+ };
+
+ let result = tokio::time::timeout(
Review Comment:
critical: 10s is per operation, not per batch - `poll_tables` stages one
`ProcessRows` per table, so N tables burn N*10s inside the SDK's 30s window.
blowing it stops the source for good, because `clear_pending_batch` then
returns false. use one shared deadline sized under 30s.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -289,6 +351,7 @@ impl PostgresSource {
let pool = PgPoolOptions::new()
.max_connections(max_connections)
+ .acquire_timeout(POOL_ACQUIRE_TIMEOUT)
Review Comment:
warning: 2s replaces sqlx's 30s default for every acquire, pool construction
included - a slow WAN or TLS handshake now fails `open()`. make it
configurable, or scope the short timeout to the ACK path.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -739,6 +955,8 @@ impl PostgresSource {
let now = Utc::now();
+ // TODO: Substitute `$now_unix` before `$now` so the longer
placeholder remains intact.
Review Comment:
warning: `$now` runs before `$now_unix`, so `$now_unix` becomes
`<rfc3339>_unix` and splices unquoted timestamp text. swapping the two
`.replace` lines is cheaper than the TODO.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -468,31 +535,37 @@ impl PostgresSource {
}
}
- // Update state with minimal lock time
- if !messages.is_empty() {
- let mut state = self.state.lock().await;
- state.processed_rows += messages.len() as u64;
- }
-
if self.verbose {
info!("CDC: Fetched {} change records", messages.len());
} else {
debug!("CDC: Fetched {} change records", messages.len());
}
- Ok(messages)
+ let lsn = replication_slot_target_lsn(last_lsn, wal_flush_lsn);
+ let pending = if messages.is_empty() {
+ self.advance_replication_slot(&lsn).await?;
Review Comment:
warning: this advances the slot inside `poll()` with no ACK gate, and it
fires when every peeked row is filtered out too. that contradicts your own
readme and SKILL.md pitfall 4. stage it as an empty `PendingBatch`.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,52 +621,165 @@ 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?;
+ if self.should_process_rows() && !processed_ids.is_empty() {
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ tracking_boundary:
self.processing_boundary(max_offset.clone()),
+ });
}
- // Collect offset update for later
if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ 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,
+ })
+ } 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 {
+ match sqlx::query("SELECT pg_replication_slot_advance($1,
$2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(pool)
+ .await
+ {
+ Err(error) if is_invalid_parameter_value(&error) => {
+ warn!(
+ "PostgreSQL source connector ID: {} received
invalid_parameter_value \
+ while advancing replication slot '{slot_name}' to
{lsn}; treating the \
+ target as already applied: {error}",
+ self.id
+ );
+ Ok(())
+ }
+ result => result.map(|_| ()),
+ }
+ },
+ 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_operation(&self, operation: PendingOperation) {
+ let (operation_name, advances_slot) = match &operation {
+ PendingOperation::ProcessRows { .. } => ("process rows", false),
+ PendingOperation::AdvanceReplicationSlot { .. } => ("advance
replication slot", true),
+ };
+
+ let result = tokio::time::timeout(
+ ACK_OPERATION_TIMEOUT,
+ self.execute_pending_operation(operation),
+ )
+ .await;
+
+ match result {
+ Ok(Ok(())) => {
+ if advances_slot {
+ self.consecutive_advance_failures
+ .store(0, Ordering::Relaxed);
+ }
}
- state.last_poll_time = Utc::now();
+ Ok(Err(error)) => {
+ self.log_pending_operation_failure(operation_name,
advances_slot, &error)
+ }
+ Err(_) => self.log_pending_operation_failure(
+ operation_name,
+ advances_slot,
+ "operation exceeded the 10s ACK budget",
+ ),
}
+ }
- Ok(messages)
+ 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
+ }
+ }
+ }
+
+ fn log_pending_operation_failure(
+ &self,
+ operation_name: &str,
+ advances_slot: bool,
+ error: impl std::fmt::Display,
+ ) {
+ 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
+ );
+ } else {
+ error!(
+ "Failed to {operation_name} for PostgreSQL source connector
ID: {}. {error}",
+ self.id
+ );
+ }
}
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:
warning: pre-existing, worth its own issue - pk defaults to the tracking
column, so rows tied at the `LIMIT` cut get deleted without ever reaching iggy.
the new tracking `AND` is a no-op in that config.
--
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]
