hubcio commented on code in PR #3957:
URL: https://github.com/apache/iggy/pull/3957#discussion_r3842143802
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -552,10 +636,12 @@ impl PostgresSource {
total_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?;
+ operations.push(PendingOperation::ProcessRows {
Review Comment:
the delete/mark now runs a full send+ack after the select, so a row updated
in between gets deleted (or marked processed) and its new version is never
delivered when `tracking_column` is an updated-at column. `AND {tracking} <=
'{max_offset}'` in the where clause would close it; at minimum a readme caveat.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -253,15 +285,59 @@ 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> {
Review Comment:
after a nack the next poll re-reads the same rows, so while iggy is down
every batch nacks - and the sdk stops the connector for good at 5 consecutive
nacks. with the shipped 1s poll interval a short outage is enough. before, the
counter got reset by empty-poll acks, but only because progress was committed
pre-send (the bug being fixed). needs a decision: exempt send-failure nacks
from the stop counter, or document the manual-restart behavior loudly.
##########
core/connectors/sources/postgres_source/README.md:
##########
@@ -209,7 +209,7 @@ LIMIT $limit
### Delete After Read
-Deletes rows from the source table after successful processing:
+Deletes rows from the source table only after Iggy acknowledges the batch:
Review Comment:
worth documenting the window: state is persisted before the delete runs, so
a crash/shutdown/timeout in between leaves rows delivered but never deleted,
and the advanced offset means they are never picked up again.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -570,17 +656,36 @@ impl PostgresSource {
}
}
- // Apply all state updates with a single lock acquisition
- {
- let mut state = self.state.lock().await;
- state.processed_rows += total_processed;
+ let pending = if total_processed > 0 {
+ candidate_state.processed_rows += total_processed;
for (table, offset) in state_updates {
- state.tracking_offsets.insert(table, offset);
+ candidate_state.tracking_offsets.insert(table, offset);
}
- state.last_poll_time = Utc::now();
- }
+ candidate_state.last_poll_time = Utc::now();
+ Some(PendingBatch {
+ state: candidate_state,
+ operations,
+ })
+ } else {
+ None
+ };
- Ok(messages)
+ Ok(PolledBatch { messages, pending })
+ }
+
+ async fn advance_replication_slot(&self, slot_name: &str, lsn: &str) ->
Result<(), Error> {
+ sqlx::query("SELECT pg_replication_slot_advance($1, $2::pg_lsn)")
Review Comment:
`pg_replication_slot_advance` needs postgres 11+, the old `get_changes`
worked on 9.4. on pg <= 10 the connector stops right after its first delivered
batch. worth a version note in the readme cdc requirements.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -570,17 +656,36 @@ impl PostgresSource {
}
}
- // Apply all state updates with a single lock acquisition
- {
- let mut state = self.state.lock().await;
- state.processed_rows += total_processed;
+ let pending = if total_processed > 0 {
+ candidate_state.processed_rows += total_processed;
for (table, offset) in state_updates {
Review Comment:
`state_updates` and this merge loop are leftovers from the old locked-state
design - `candidate_state` is a local clone now, offsets can go straight into
it in the table loop. drops the stale lock-era comments too.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -97,13 +100,38 @@ impl PayloadFormat {
}
}
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
struct State {
last_poll_time: DateTime<Utc>,
tracking_offsets: HashMap<String, String>,
processed_rows: u64,
}
+#[derive(Debug)]
+struct PolledBatch {
+ messages: Vec<ProducedMessage>,
+ pending: Option<PendingBatch>,
+}
+
+#[derive(Debug)]
+struct PendingBatch {
+ state: State,
+ operations: Vec<PendingOperation>,
+}
+
+#[derive(Debug)]
+enum PendingOperation {
Review Comment:
`primary_key_column`/`slot_name` are derivable from config at ack time -
small `slot_name()`/`pk_column()` helpers (like `payload_format()`) would
dedupe the three existing fallback chains and drop these two fields. optional.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -422,38 +498,39 @@ impl PostgresSource {
// 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)")
- .bind(slot_name)
- .bind(batch_size)
- .fetch_all(pool)
- .await
- .map_err(|e| {
- error!("Failed to fetch CDC changes: {e}");
- Error::InvalidRecord
- })?;
+ let rows = sqlx::query(
+ "SELECT lsn::text AS lsn, xid, 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
+ })?;
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:
a row failing here fails the whole poll forever - right call for
at-least-once (can't skip a row and advance the slot past it), but poll errors
are invisible: no stats bump, status stays `Running`, and the stuck slot pins
`restart_lsn` so pg wal grows without bound. worth a readme note about
monitoring slot lag.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -253,15 +285,59 @@ 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 pending = self.pending_batch.lock().await.take();
+ if result == SourceBatchResult::Nack {
+ return Ok(());
+ }
+
+ let Some(pending) = pending else {
+ return Ok(());
+ };
+
+ for operation in pending.operations {
Review Comment:
any error in this loop permanently stops the connector - the sdk treats
`Err` from `on_batch_result` as fatal, and the staged delete/mark ops are
dropped while the state file already advanced past those rows. transient pg
errors (deadlock, failover) should retry here; note `with_retry` can't wrap
these calls directly (wrong error type), it has to go around the inner
`.execute()`s.
##########
core/integration/tests/connectors/postgres/postgres_source.rs:
##########
@@ -127,6 +135,166 @@ 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
+)]
Review Comment:
this covers redelivery for the plain config only - no
`delete_after_read`/`processed_column` fixture and no cdc crash test, so the
ack-time ops never run on the failure path (both new unit tests avoid executing
them too).
##########
core/integration/src/harness/handle/connectors_runtime.rs:
##########
@@ -90,8 +95,12 @@ impl ConnectorsRuntimeHandle {
);
if let Some(addr) = self.iggy_address {
+ let address = self
+ .iggy_connection_options
+ .as_ref()
+ .map_or_else(|| addr.to_string(), |options|
format!("{addr}?{options}"));
Review Comment:
joins with `?` unconditionally - if a tls test ever sets options this
collides with the `?tls=true` the runtime appends later. pick `?` or `&` based
on what's already there.
##########
core/integration/tests/connectors/postgres/postgres_source_cdc.rs:
##########
@@ -340,8 +340,8 @@ async fn delete_source_config_version(http: &Client,
api_url: &str, version: u64
);
}
-// The connector calls pg_logical_slot_get_changes on a fixed poll interval and
-// briefly holds the slot active during each call. A drop landing in that
window
+// The connector peeks changes and advances the slot on a fixed poll interval.
Review Comment:
the comment at lines 379-383 in this file still says the slot-peek/lsn gap
"remains open" - this pr is that work, update it too.
##########
core/connectors/sources/postgres_source/README.md:
##########
@@ -267,15 +267,20 @@ tables = ["users", "orders"]
capture_operations = ["INSERT", "UPDATE", "DELETE"]
```
+The connector peeks at logical changes and advances the replication slot only
Review Comment:
worth one sentence on cost: advancing the slot re-decodes the same wal range
(fast-forward), so each acked batch pays two decode passes.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -253,15 +285,59 @@ 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 pending = self.pending_batch.lock().await.take();
+ if result == SourceBatchResult::Nack {
Review Comment:
`!= Nack` implies ack; random_source gates on `== Ack`. same thing today
with 2 variants, but a future variant would silently fall into the ack path
here.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -422,38 +498,39 @@ impl PostgresSource {
// 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)")
- .bind(slot_name)
- .bind(batch_size)
- .fetch_all(pool)
- .await
- .map_err(|e| {
- error!("Failed to fetch CDC changes: {e}");
- Error::InvalidRecord
- })?;
+ let rows = sqlx::query(
+ "SELECT lsn::text AS lsn, xid, data FROM
pg_logical_slot_peek_changes($1, NULL, $2)",
Review Comment:
`xid` is selected but never read - drop it.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -230,20 +259,23 @@ impl Source for PostgresSource {
}
};
- let state = self.state.lock().await;
+ let processed_rows = match polled.pending.as_ref() {
+ Some(pending) => pending.state.processed_rows,
Review Comment:
this logs the candidate total before the batch is acked - a nacked batch
logs the inflated count, then the retry logs it again. it used to report
committed state.
##########
core/integration/tests/connectors/postgres/postgres_source.rs:
##########
@@ -127,6 +135,166 @@ 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_rows_in_postgres_when_iggy_server_crashes_should_redeliver_after_restart(
+ harness: &mut TestHarness,
+ fixture: PostgresSourceJsonFixture,
+) {
+ 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");
Review Comment:
`reconnection_retries=0` stays set for the verification phase too
(`build_envs` reruns on every start), so the post-restart polls get no
reconnect slack. fine today, just tightens the timing budget for no reason.
##########
core/integration/tests/connectors/postgres/postgres_source.rs:
##########
@@ -15,19 +15,27 @@
// specific language governing permissions and limitations
// under the License.
+use std::time::Duration;
+
+use iggy_common::MessageClient;
+use iggy_common::{Consumer, Identifier, PollingStrategy};
+use iggy_connector_sdk::api::{ConnectorRuntimeStats, ConnectorStatus};
+use integration::harness::seeds;
+use integration::iggy_harness;
+use reqwest::Client;
+use tokio::time::{sleep, timeout};
+
use super::{DatabaseRecord, POLL_ATTEMPTS, POLL_INTERVAL_MS,
TEST_MESSAGE_COUNT};
use crate::connectors::create_test_messages;
use crate::connectors::fixtures::{
PostgresOps, PostgresSourceByteaFixture, PostgresSourceDeleteFixture,
PostgresSourceJsonFixture, PostgresSourceJsonbFixture,
PostgresSourceMarkFixture,
PostgresSourceOps,
};
-use iggy_common::MessageClient;
-use iggy_common::{Consumer, Identifier, PollingStrategy};
-use integration::harness::seeds;
-use integration::iggy_harness;
-use std::time::Duration;
-use tokio::time::sleep;
+
+const API_KEY: &str = "test-api-key";
Review Comment:
`API_KEY`/`SOURCE_KEY` duplicate postgres_source_cdc.rs - `SOURCE_KEY` fits
in postgres/mod.rs next to the other shared consts.
`source_errors`/`wait_for_source_errors` could also share one fetch helper
returning `Option`.
--
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]
