hubcio commented on code in PR #3957:
URL: https://github.com/apache/iggy/pull/3957#discussion_r4003529905
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,86 +795,257 @@ 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 cleanup_boundary = self.cleanup_boundary(max_offset.clone());
+ let tracking_cursor = self.tracking_cursor(max_offset);
+
+ if self.should_process_rows() && !processed_ids.is_empty() {
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ tracking_boundary: cleanup_boundary,
+ });
}
- // Collect offset update for later
- if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ if let Some(offset) = tracking_cursor {
+ 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();
+ candidate_state.pending_operations = operations;
+ Some(PendingBatch {
+ state: candidate_state,
+ acknowledged: false,
+ retiring_operations: false,
+ operations_checkpointed: true,
+ })
+ } 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 mut transaction = begin_ack_transaction(pool).await?;
+ let result = sqlx::query("SELECT
pg_replication_slot_advance($1, $2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(&mut *transaction)
+ .await
+ .map(|_| ());
+ let result = finish_ack_transaction(transaction, result).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 result = tokio::time::timeout_at(deadline, async {
+ 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
+ }
+ }
+ })
+ .await;
+
+ match result {
+ Ok(Ok(())) => {
+ match operation {
+ PendingOperation::ProcessRows { .. } => {
+ self.consecutive_process_failures
+ .store(0, Ordering::Relaxed);
+ }
+ PendingOperation::AdvanceReplicationSlot { .. } => {
+ self.consecutive_advance_failures
+ .store(0, Ordering::Relaxed);
+ }
+ }
+ Ok(true)
+ }
+ Ok(Err(error)) => {
+ self.record_pending_operation_failure(operation, &error)?;
+ Ok(false)
+ }
+ Err(_) => {
+ self.record_pending_operation_failure(
+ operation,
+ "operation exceeded the shared 10s ACK batch budget",
+ )?;
+ Ok(false)
}
- state.last_poll_time = Utc::now();
}
+ }
- Ok(messages)
+ fn record_pending_operation_failure(
+ &self,
+ operation: &PendingOperation,
+ error: impl std::fmt::Display,
+ ) -> Result<(), Error> {
+ match operation {
+ PendingOperation::ProcessRows { .. } => {
+ let consecutive_failures = self
+ .consecutive_process_failures
+ .fetch_add(1, Ordering::Relaxed)
+ .saturating_add(1);
+ error!(
+ "Failed to process rows for PostgreSQL source connector
ID: {}. \
+ Consecutive row processing failures:
{consecutive_failures}. {error}",
+ self.id,
+ );
+ if consecutive_failures >= MAX_CONSECUTIVE_PROCESS_FAILURES {
+ return Err(Error::Connection(format!(
+ "stopping PostgreSQL source connector ID {} after
{consecutive_failures} \
+ consecutive row processing failures",
+ self.id
+ )));
+ }
+ }
+ PendingOperation::AdvanceReplicationSlot { .. } => {
+ let consecutive_failures = self
+ .consecutive_advance_failures
+ .fetch_add(1, Ordering::Relaxed)
+ .saturating_add(1);
+ error!(
+ "Failed to advance replication slot 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
+ )));
+ }
+ }
+ }
+ 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.cleanup_key_column())?;
+ let tracking_condition =
+ build_tracking_condition(self.tracking_column(),
tracking_boundary)?;
- let ids_list = ids
- .iter()
- .map(|id| {
- if id.parse::<i64>().is_ok() {
- id.clone()
- } else {
- format!("'{}'", id.replace('\'', "''"))
- }
- })
- .collect::<Vec<_>>()
- .join(", ");
+ let ids_list = format_cleanup_ids(ids);
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}"
Review Comment:
critical: with a custom query without `$offset`, a crash after cleanup but
before retirement lets replay delete a replacement row reusing the key. bind
cleanup to the acknowledged row version or a transactional operation receipt.
##########
core/connectors/sdk/src/source.rs:
##########
@@ -357,39 +370,54 @@ async fn handle_messages<T, F>(
continue;
}
+ let mut result_receiver = Box::pin(result_receiver);
let (completion, shutting_down) = tokio::select! {
biased;
- result = result_receiver => {
+ result = &mut result_receiver => {
(result.unwrap_or(BatchCompletion::Stop), false)
},
_ = shutdown.changed() => {
- let completion = if
clear_pending_batch(&pending_batch, batch_id, plugin_id) {
- apply_batch_result(
+ let completion = match clear_pending_batch(
+ &pending_batch,
+ batch_id,
+ plugin_id,
+ ) {
+ PendingBatchClearResult::Cleared =>
apply_batch_result(
&source,
&consecutive_nacks,
SourceBatchResult::Nack,
plugin_id,
policy.max_consecutive_nacks,
- ).await
- } else {
- BatchCompletion::Stop
+ ).await,
+ PendingBatchClearResult::ResultReceived =>
result_receiver
+ .as_mut()
+ .await
+ .unwrap_or(BatchCompletion::Stop),
+ PendingBatchClearResult::Unavailable =>
BatchCompletion::Stop,
};
(completion, true)
},
_ = tokio::time::sleep(policy.result_timeout) => {
warn!(
"Timed out waiting for batch result for source
connector with ID: {plugin_id}, batch ID: {batch_id}"
);
- let completion = if
clear_pending_batch(&pending_batch, batch_id, plugin_id) {
- apply_batch_result(
+ let completion = match clear_pending_batch(
+ &pending_batch,
+ batch_id,
+ plugin_id,
+ ) {
+ PendingBatchClearResult::Cleared =>
apply_batch_result(
&source,
&consecutive_nacks,
SourceBatchResult::Nack,
plugin_id,
policy.max_consecutive_nacks,
- ).await
- } else {
- BatchCompletion::Stop
+ ).await,
+ PendingBatchClearResult::ResultReceived =>
result_receiver
+ .as_mut()
+ .await
+ .unwrap_or(BatchCompletion::Stop),
+ PendingBatchClearResult::Unavailable =>
BatchCompletion::Stop,
Review Comment:
warning: the callback can take the pending batch before sending completion,
letting timeout observe `Unavailable` and stop polling during a successful ACK.
await the retained receiver in this branch, preserving explicit stop results
and sender-drop handling.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -493,6 +500,59 @@ pub(crate) async fn setup_source_producer(
Ok((producer, encoder, transforms))
}
+async fn ensure_durable_source_topic(
+ client: &IggyClient,
+ stream_name: &str,
+ topic_name: &str,
+) -> Result<(), RuntimeError> {
+ let stream_id = Identifier::try_from(stream_name)?;
+ if client.get_stream(&stream_id).await?.is_none() {
+ client.create_stream(stream_name).await?;
+ }
+
+ let topic_id = Identifier::try_from(topic_name)?;
+ let topic = match client.get_topic(&stream_id, &topic_id).await? {
+ Some(topic) => topic,
+ None => {
+ client
+ .create_topic(
+ &stream_id,
+ topic_name,
+ &TopicCreateOptions {
+ partitions_count: Some(1),
+ durability: Durability::Persisted,
+ messages_required_to_save:
Some(SOURCE_TOPIC_MESSAGES_REQUIRED_TO_SAVE),
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await?
+ }
+ };
+
+ validate_source_topic_durability(
+ stream_name,
+ topic_name,
+ TopicRuntimeOptions::from_resource_options(&topic.options),
+ )
+}
+
+fn validate_source_topic_durability(
+ stream_name: &str,
+ topic_name: &str,
+ options: TopicRuntimeOptions,
+) -> Result<(), RuntimeError> {
+ if options.durability == Durability::Persisted
+ && options.messages_required_to_save ==
Some(SOURCE_TOPIC_MESSAGES_REQUIRED_TO_SAVE)
Review Comment:
warning: persisted topics with other save thresholds are rejected even
though their send acknowledgments already wait for durable storage. require
`Durability::Persisted` without requiring `messages_required_to_save=1`.
##########
core/connectors/runtime/src/source.rs:
##########
@@ -493,6 +500,59 @@ pub(crate) async fn setup_source_producer(
Ok((producer, encoder, transforms))
}
+async fn ensure_durable_source_topic(
+ client: &IggyClient,
+ stream_name: &str,
+ topic_name: &str,
+) -> Result<(), RuntimeError> {
+ let stream_id = Identifier::try_from(stream_name)?;
+ if client.get_stream(&stream_id).await?.is_none() {
+ client.create_stream(stream_name).await?;
+ }
+
+ let topic_id = Identifier::try_from(topic_name)?;
+ let topic = match client.get_topic(&stream_id, &topic_id).await? {
+ Some(topic) => topic,
+ None => {
+ client
+ .create_topic(
+ &stream_id,
+ topic_name,
+ &TopicCreateOptions {
+ partitions_count: Some(1),
+ durability: Durability::Persisted,
+ messages_required_to_save:
Some(SOURCE_TOPIC_MESSAGES_REQUIRED_TO_SAVE),
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await?
+ }
+ };
+
+ validate_source_topic_durability(
+ stream_name,
+ topic_name,
+ TopicRuntimeOptions::from_resource_options(&topic.options),
+ )
+}
+
+fn validate_source_topic_durability(
+ stream_name: &str,
+ topic_name: &str,
+ options: TopicRuntimeOptions,
+) -> Result<(), RuntimeError> {
+ if options.durability == Durability::Persisted
Review Comment:
warning: both source quick-start guides create replicated topics that this
gate rejects, so their documented setup fails. update their topic creation
commands to request persisted durability and keep the storage safety check.
##########
core/connectors/runtime/src/manager/source.rs:
##########
@@ -103,6 +103,16 @@ impl SourceManager {
}
}
+ pub async fn recover_from_error(&self, key: &str) {
+ if let Some(source) = self.sources.get(key) {
+ let mut source = source.lock().await;
+ if source.info.status == ConnectorStatus::Error {
+ source.info.status = ConnectorStatus::Running;
Review Comment:
warning: recovery sets `Running` without restoring the gauge decremented by
`set_error`, so later failures or stops can make counts negative. route
recovery through `apply_status` with metrics.
##########
core/connectors/sources/postgres_source/README.md:
##########
@@ -54,26 +54,38 @@ cdc_backend = "builtin"
| `connection_string` | string | required | PostgreSQL connection string |
| `mode` | string | required | `polling` or `cdc` |
| `tables` | array | required | List of tables to monitor |
-| `poll_interval` | string | `1s` | How often to poll (e.g., `1s`, `5m`) |
+| `poll_interval` | string | `10s` | How often to poll (e.g., `1s`, `5m`) |
| `batch_size` | u32 | `1000` | Max rows per poll |
-| `tracking_column` | string | `id` | Column for incremental updates |
+| `tracking_column` | string | `id` | Unique, non-null column for incremental
updates |
| `initial_offset` | string | none | Starting value for tracking column |
| `max_connections` | u32 | `10` | Max database connections |
| `snake_case_columns` | bool | `false` | Convert column names to snake_case |
| `include_metadata` | bool | `true` | Wrap results with metadata |
| `payload_column` | string | none | Column to extract as payload |
| `payload_format` | string | `bytea` | Format of payload_column: `bytea`,
`text`, or `json_direct` |
-| `delete_after_read` | bool | `false` | Delete rows after reading |
-| `processed_column` | string | none | Boolean column to mark as processed |
-| `primary_key_column` | string | tracking_column | PK for delete/mark
operations |
+| `delete_after_read` | bool | `false` | Delete rows after reading; takes
precedence over `processed_column` |
+| `processed_column` | string | none | Boolean column to mark as processed
when `delete_after_read` is false |
+| `primary_key_column` | string | tracking_column | Unique, non-null key for
delete/mark operations |
| `custom_query` | string | none | Custom SQL with parameter substitution |
| `replication_slot` | string | `iggy_slot` | Replication slot name (only used
when `mode = "cdc"`) |
| `capture_operations` | array | `["INSERT","UPDATE","DELETE"]` | CDC
operations to capture |
| `cdc_backend` | string | `builtin` | `builtin` or `pg_replicate` |
| `verbose_logging` | bool | `false` | Log at info level instead of debug |
-| `max_retries` | u32 | `3` | Max retry attempts for transient errors |
+| `max_retries` | u32 | `3` | Max attempts for transient errors; `0` and `1`
both perform one attempt |
| `retry_delay` | string | `1s` | Base delay between retries (e.g., `500ms`,
`2s`) |
+## Delivery Failures
+
+Delivery is at-least-once, so consumers must tolerate duplicates. A failed send
Review Comment:
warning: polling can permanently skip a transaction that commits below an
already acknowledged cursor, even with a unique serial key. scope the delivery
guarantee to selected batches and document the visibility-order requirement for
complete polling capture.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -221,7 +286,39 @@ impl Source for PostgresSource {
let poll_interval = self.poll_interval;
tokio::time::sleep(poll_interval).await;
- let messages = match self.config.mode.as_str() {
+ let schema = match self.payload_format() {
+ PayloadFormat::Bytea => Schema::Raw,
+ PayloadFormat::Text => Schema::Text,
+ PayloadFormat::JsonDirect | PayloadFormat::Json => Schema::Json,
+ };
+
+ {
+ let pending = self.pending_batch.lock().await;
+ if let Some(pending) = pending.as_ref() {
+ let state = if pending.retiring_operations {
Review Comment:
warning: retirement checkpoints wait through `poll_interval`, then the next
data poll waits again, slowing batches with cleanup or CDC work. return ready
retirement checkpoints before sleeping, while preserving normal poll and retry
pacing.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,86 +795,257 @@ 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 cleanup_boundary = self.cleanup_boundary(max_offset.clone());
+ let tracking_cursor = self.tracking_cursor(max_offset);
+
+ if self.should_process_rows() && !processed_ids.is_empty() {
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ tracking_boundary: cleanup_boundary,
+ });
}
- // Collect offset update for later
- if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ if let Some(offset) = tracking_cursor {
+ 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();
+ candidate_state.pending_operations = operations;
+ Some(PendingBatch {
+ state: candidate_state,
+ acknowledged: false,
+ retiring_operations: false,
+ operations_checkpointed: true,
+ })
+ } 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 mut transaction = begin_ack_transaction(pool).await?;
+ let result = sqlx::query("SELECT
pg_replication_slot_advance($1, $2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(&mut *transaction)
+ .await
+ .map(|_| ());
+ let result = finish_ack_transaction(transaction, result).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 {
Review Comment:
warning: after one cleanup exhausts the shared deadline, later I/O
operations consume failure counts immediately and can stop the source in that
ACK. defer remaining operations and count failures only for operations started
with budget.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -550,86 +795,257 @@ 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 cleanup_boundary = self.cleanup_boundary(max_offset.clone());
+ let tracking_cursor = self.tracking_cursor(max_offset);
+
+ if self.should_process_rows() && !processed_ids.is_empty() {
+ operations.push(PendingOperation::ProcessRows {
+ table: table.clone(),
+ ids: processed_ids,
+ tracking_boundary: cleanup_boundary,
+ });
}
- // Collect offset update for later
- if let Some(offset) = max_offset {
- state_updates.push((table.clone(), offset));
+ if let Some(offset) = tracking_cursor {
+ 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();
+ candidate_state.pending_operations = operations;
+ Some(PendingBatch {
+ state: candidate_state,
+ acknowledged: false,
+ retiring_operations: false,
+ operations_checkpointed: true,
+ })
+ } 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 mut transaction = begin_ack_transaction(pool).await?;
+ let result = sqlx::query("SELECT
pg_replication_slot_advance($1, $2::pg_lsn)")
+ .bind(slot_name)
+ .bind(lsn)
+ .execute(&mut *transaction)
+ .await
+ .map(|_| ());
+ let result = finish_ack_transaction(transaction, result).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 result = tokio::time::timeout_at(deadline, async {
+ 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
+ }
+ }
+ })
+ .await;
+
+ match result {
+ Ok(Ok(())) => {
+ match operation {
+ PendingOperation::ProcessRows { .. } => {
+ self.consecutive_process_failures
+ .store(0, Ordering::Relaxed);
+ }
+ PendingOperation::AdvanceReplicationSlot { .. } => {
+ self.consecutive_advance_failures
+ .store(0, Ordering::Relaxed);
+ }
+ }
+ Ok(true)
+ }
+ Ok(Err(error)) => {
+ self.record_pending_operation_failure(operation, &error)?;
+ Ok(false)
+ }
+ Err(_) => {
+ self.record_pending_operation_failure(
+ operation,
+ "operation exceeded the shared 10s ACK batch budget",
+ )?;
+ Ok(false)
}
- state.last_poll_time = Utc::now();
}
+ }
- Ok(messages)
+ fn record_pending_operation_failure(
+ &self,
+ operation: &PendingOperation,
+ error: impl std::fmt::Display,
+ ) -> Result<(), Error> {
+ match operation {
+ PendingOperation::ProcessRows { .. } => {
+ let consecutive_failures = self
+ .consecutive_process_failures
+ .fetch_add(1, Ordering::Relaxed)
+ .saturating_add(1);
+ error!(
+ "Failed to process rows for PostgreSQL source connector
ID: {}. \
+ Consecutive row processing failures:
{consecutive_failures}. {error}",
+ self.id,
+ );
+ if consecutive_failures >= MAX_CONSECUTIVE_PROCESS_FAILURES {
+ return Err(Error::Connection(format!(
+ "stopping PostgreSQL source connector ID {} after
{consecutive_failures} \
+ consecutive row processing failures",
+ self.id
+ )));
+ }
+ }
+ PendingOperation::AdvanceReplicationSlot { .. } => {
+ let consecutive_failures = self
+ .consecutive_advance_failures
+ .fetch_add(1, Ordering::Relaxed)
+ .saturating_add(1);
+ error!(
+ "Failed to advance replication slot 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
+ )));
+ }
+ }
+ }
+ 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.cleanup_key_column())?;
Review Comment:
critical: with custom queries, replaying saved ids after changing the
cleanup key can delete undelivered rows through the new column. bind pending
cleanup to its original target and reject incompatible configuration.
--
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]
