You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "mattp5657 (via GitHub)" <gi...@apache.org> on 2026/07/09 23:32:32 UTC
[PR] Fix/connectors postgres builtin cdc [iggy]
mattp5657 opened a new pull request, #3640:
URL: https://github.com/apache/iggy/pull/3640
## Which issue does this PR address?
Relates to #3582
## Rationale
Builtin CDC mode currently produces zero messages against a real
database - wrong plugin/read-path pairing, and a parser matching a
format no real Postgres output plugin emits. This PR is scoped to
getting CDC actually working and provably correct, nothing more.
Left out of scope here, and why:
- `pgoutput` as the long-term plugin - real binary decoder + streaming
connection is a much bigger lift; `test_decoding` unblocks correctness
today. Target once `cdc_backend = "pg_replicate"` gets picked up.
- At-least-once / crash safety (slot advances before send is confirmed,
no LSN tracked) - needs its own review since it touches the
state-persistence contract. Follow-up: #TODO.
- Cosmetic items (UUID vs PK as message id, processing time vs WAL
commit time, `old_data` unpopulated) - mechanical, separate PR.
Bundling all of the above into one PR would make it unreviewable -
splitting it this way keeps each piece independently checkable.
## What changed?
Before: wrong slot plugin, a parser matching a format no real Postgres
output emits, and invalid setup SQL - CDC produced zero messages, and
several related bugs failed silently too. After: parser rewritten
against real captured output, setup/poll SQL fixed and bounded, and
those failure modes now work correctly or fail loudly. Covered by
corpus-driven unit tests plus integration tests against a live
`wal_level=logical` container.
- Parser rewritten against real `test_decoding` output
- Slot/setup SQL fixed, including the upgrade path and table filtering
- Batch handling hardened (bad rows, unbounded backlog)
- Dead config fields removed
## Local Execution
- Passed
- Pre-commit hooks ran
## AI Usage
Claude was used to help generate and review this PR.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): fix postgres source CDC producing no messages [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #3640:
URL: https://github.com/apache/iggy/pull/3640#issuecomment-4978791955
@mmodzelewski could you please check this one?
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): fix postgres source CDC producing no messages [iggy]
Posted by "mattp5657 (via GitHub)" <gi...@apache.org>.
mattp5657 commented on PR #3640:
URL: https://github.com/apache/iggy/pull/3640#issuecomment-5017715458
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): fix postgres source CDC producing no messages [iggy]
Posted by "mattp5657 (via GitHub)" <gi...@apache.org>.
mattp5657 commented on code in PR #3640:
URL: https://github.com/apache/iggy/pull/3640#discussion_r3611518915
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1497,42 +1491,107 @@ fn to_snake_case(input: &str) -> String {
result
}
-fn parse_record_data(data: &str) -> serde_json::Map<String, serde_json::Value> {
+fn unquote_pg_identifier(segment: &str) -> String {
+ match segment.strip_prefix('"').and_then(|s| s.strip_suffix('"')) {
+ Some(inner) => inner.replace("\"\"", "\""),
+ None => segment.to_string(),
+ }
+}
+
+fn parse_record_columns(data: &str) -> serde_json::Map<String, serde_json::Value> {
let mut result = serde_json::Map::new();
+ let bytes = data.as_bytes();
+ let len = bytes.len();
+ let mut pos = 0;
- for part in data.split_whitespace() {
- if let Some(bracket_pos) = part.find('[')
- && let Some(_close_bracket) = part.find(']')
- && let Some(colon_pos) = part.find(':')
- {
- let column_name = &part[..bracket_pos];
- let value_str = &part[colon_pos + 1..];
+ while pos < len {
+ while pos < len && bytes[pos] == b' ' {
+ pos += 1;
+ }
+ if pos >= len {
+ break;
+ }
- let cleaned_value = if value_str.starts_with('\'') && value_str.ends_with('\'') {
- &value_str[1..value_str.len() - 1]
- } else {
- value_str
- };
+ let Some(bracket_offset) = data[pos..].find('[') else {
+ break;
+ };
+ let name_end = pos + bracket_offset;
+ let column_name = &data[pos..name_end];
Review Comment:
This should be done.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): fix postgres source CDC producing no messages [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #3640:
URL: https://github.com/apache/iggy/pull/3640#issuecomment-4933570933
i think this is somewhat related to https://github.com/apache/iggy/discussions/3313
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): fix postgres source CDC producing no messages [iggy]
Posted by "mattp5657 (via GitHub)" <gi...@apache.org>.
mattp5657 commented on code in PR #3640:
URL: https://github.com/apache/iggy/pull/3640#discussion_r3611518915
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1497,42 +1491,107 @@ fn to_snake_case(input: &str) -> String {
result
}
-fn parse_record_data(data: &str) -> serde_json::Map<String, serde_json::Value> {
+fn unquote_pg_identifier(segment: &str) -> String {
+ match segment.strip_prefix('"').and_then(|s| s.strip_suffix('"')) {
+ Some(inner) => inner.replace("\"\"", "\""),
+ None => segment.to_string(),
+ }
+}
+
+fn parse_record_columns(data: &str) -> serde_json::Map<String, serde_json::Value> {
let mut result = serde_json::Map::new();
+ let bytes = data.as_bytes();
+ let len = bytes.len();
+ let mut pos = 0;
- for part in data.split_whitespace() {
- if let Some(bracket_pos) = part.find('[')
- && let Some(_close_bracket) = part.find(']')
- && let Some(colon_pos) = part.find(':')
- {
- let column_name = &part[..bracket_pos];
- let value_str = &part[colon_pos + 1..];
+ while pos < len {
+ while pos < len && bytes[pos] == b' ' {
+ pos += 1;
+ }
+ if pos >= len {
+ break;
+ }
- let cleaned_value = if value_str.starts_with('\'') && value_str.ends_with('\'') {
- &value_str[1..value_str.len() - 1]
- } else {
- value_str
- };
+ let Some(bracket_offset) = data[pos..].find('[') else {
+ break;
+ };
+ let name_end = pos + bracket_offset;
+ let column_name = &data[pos..name_end];
Review Comment:
This should be done.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): fix postgres source CDC producing no messages [iggy]
Posted by "mmodzelewski (via GitHub)" <gi...@apache.org>.
mmodzelewski commented on code in PR #3640:
URL: https://github.com/apache/iggy/pull/3640#discussion_r3586375899
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1497,42 +1491,107 @@ fn to_snake_case(input: &str) -> String {
result
}
-fn parse_record_data(data: &str) -> serde_json::Map<String, serde_json::Value> {
+fn unquote_pg_identifier(segment: &str) -> String {
+ match segment.strip_prefix('"').and_then(|s| s.strip_suffix('"')) {
+ Some(inner) => inner.replace("\"\"", "\""),
+ None => segment.to_string(),
+ }
+}
+
+fn parse_record_columns(data: &str) -> serde_json::Map<String, serde_json::Value> {
let mut result = serde_json::Map::new();
+ let bytes = data.as_bytes();
+ let len = bytes.len();
+ let mut pos = 0;
- for part in data.split_whitespace() {
- if let Some(bracket_pos) = part.find('[')
- && let Some(_close_bracket) = part.find(']')
- && let Some(colon_pos) = part.find(':')
- {
- let column_name = &part[..bracket_pos];
- let value_str = &part[colon_pos + 1..];
+ while pos < len {
+ while pos < len && bytes[pos] == b' ' {
+ pos += 1;
+ }
+ if pos >= len {
+ break;
+ }
- let cleaned_value = if value_str.starts_with('\'') && value_str.ends_with('\'') {
- &value_str[1..value_str.len() - 1]
- } else {
- value_str
- };
+ let Some(bracket_offset) = data[pos..].find('[') else {
+ break;
+ };
+ let name_end = pos + bracket_offset;
+ let column_name = &data[pos..name_end];
Review Comment:
Quoted column names keep their literal quotes as JSON keys. `test_decoding` applies `quote_identifier` to column names, so a mixed-case or keyword column (`createdAt`, `user`) arrives as `"createdAt"[timestamp with time zone]:'...'` and this parser emits the JSON key `\"createdAt\"` with the double-quote characters included. Table names already go through `unquote_pg_identifier`; column names should get the same treatment, otherwise downstream consumers looking up the plain column name miss. Worth a unit test with a quoted column name fixture.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -303,10 +301,6 @@ impl PostgresSource {
}
async fn setup_cdc(&self) -> Result<(), Error> {
Review Comment:
Not on this line, but related to CDC startup: `open()` validates `mode` but not `cdc_backend`. A typo like `cdc_backend = "built-in"` passes startup, then every `poll_cdc` call returns an error that the SDK loop logs and swallows, leaving a Running connector that produces nothing. That is the same silent-death failure mode this PR fixes for the slot mismatch. Validating the backend in the `"cdc"` arm of `open()`, where the slot mismatch already fails loudly, would close it.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -1497,42 +1491,107 @@ fn to_snake_case(input: &str) -> String {
result
}
-fn parse_record_data(data: &str) -> serde_json::Map<String, serde_json::Value> {
+fn unquote_pg_identifier(segment: &str) -> String {
+ match segment.strip_prefix('"').and_then(|s| s.strip_suffix('"')) {
+ Some(inner) => inner.replace("\"\"", "\""),
+ None => segment.to_string(),
+ }
+}
+
+fn parse_record_columns(data: &str) -> serde_json::Map<String, serde_json::Value> {
let mut result = serde_json::Map::new();
+ let bytes = data.as_bytes();
+ let len = bytes.len();
+ let mut pos = 0;
- for part in data.split_whitespace() {
- if let Some(bracket_pos) = part.find('[')
- && let Some(_close_bracket) = part.find(']')
- && let Some(colon_pos) = part.find(':')
- {
- let column_name = &part[..bracket_pos];
- let value_str = &part[colon_pos + 1..];
+ while pos < len {
+ while pos < len && bytes[pos] == b' ' {
+ pos += 1;
+ }
+ if pos >= len {
+ break;
+ }
- let cleaned_value = if value_str.starts_with('\'') && value_str.ends_with('\'') {
- &value_str[1..value_str.len() - 1]
- } else {
- value_str
- };
+ let Some(bracket_offset) = data[pos..].find('[') else {
+ break;
+ };
+ let name_end = pos + bracket_offset;
+ let column_name = &data[pos..name_end];
+
+ let mut depth = 0;
+ let mut type_end = name_end;
+ loop {
+ match bytes.get(type_end) {
+ Some(b'[') => depth += 1,
+ Some(b']') => {
+ depth -= 1;
+ if depth == 0 {
+ break;
+ }
+ }
+ Some(_) => {}
+ None => return result,
+ }
+ type_end += 1;
+ }
+ if bytes.get(type_end + 1) != Some(&b':') {
+ break;
+ }
- let value = if let Ok(num) = cleaned_value.parse::<i64>() {
- serde_json::Value::Number(serde_json::Number::from(num))
- } else if let Ok(float) = cleaned_value.parse::<f64>() {
- serde_json::Value::Number(
- serde_json::Number::from_f64(float).unwrap_or(serde_json::Number::from(0)),
- )
- } else if cleaned_value.eq_ignore_ascii_case("true") {
- serde_json::Value::Bool(true)
- } else if cleaned_value.eq_ignore_ascii_case("false") {
- serde_json::Value::Bool(false)
- } else {
- serde_json::Value::String(cleaned_value.to_string())
- };
+ let (value, next_pos) = parse_column_value(data, type_end + 2);
+ result.insert(column_name.to_string(), value);
+ pos = next_pos;
+ }
+
+ result
+}
+
+fn parse_column_value(data: &str, start: usize) -> (serde_json::Value, usize) {
Review Comment:
The `unchanged-toast-datum` sentinel leaks into records as a real value. When an UPDATE does not touch a TOASTed column (large text/jsonb), `test_decoding` emits the bare token `unchanged-toast-datum`, and the parser stores `data["col"] = "unchanged-toast-datum"`, which is indistinguishable from a legitimate string. Consumers will silently ingest the sentinel as the column value. Suggest handling it explicitly in `parse_bare_scalar` (map to null, or omit the key) and adding a fixture for it.
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -728,99 +762,59 @@ impl PostgresSource {
&self,
data: &str,
capture_ops: &[&str],
+ captured_tables: Option<&[String]>,
) -> Option<DatabaseRecord> {
if data.starts_with("BEGIN") || data.starts_with("COMMIT") {
return None;
}
- if data.starts_with("INSERT:") && capture_ops.contains(&"INSERT") {
- return self.parse_insert_message(data);
- }
-
- if data.starts_with("UPDATE:") && capture_ops.contains(&"UPDATE") {
- return self.parse_update_message(data);
- }
+ let rest = data.strip_prefix("table ")?;
+ let (qualified_table, rest) = rest.split_once(": ")?;
+ let (operation, rest) = rest.split_once(": ")?;
- if data.starts_with("DELETE:") && capture_ops.contains(&"DELETE") {
- return self.parse_delete_message(data);
- }
-
- None
- }
-
- fn parse_insert_message(&self, data: &str) -> Option<DatabaseRecord> {
- if let Some(table_start) = data.find("table ")
- && let Some(colon_pos) = data[table_start..].find(':')
+ if !matches!(operation, "INSERT" | "UPDATE" | "DELETE") || !capture_ops.contains(&operation)
{
- let table_part = &data[table_start + 6..table_start + colon_pos];
- let table_name = table_part
- .split('.')
- .next_back()
- .unwrap_or(table_part)
- .to_string();
-
- let data_part = &data[table_start + colon_pos + 1..];
- let parsed_data = parse_record_data(data_part);
-
- return Some(DatabaseRecord {
- table_name,
- operation_type: "INSERT".to_string(),
- timestamp: Utc::now(),
- data: serde_json::Value::Object(parsed_data),
- old_data: None,
- });
+ return None;
}
- None
- }
- fn parse_update_message(&self, data: &str) -> Option<DatabaseRecord> {
- if let Some(table_start) = data.find("table ")
- && let Some(colon_pos) = data[table_start..].find(':')
+ let table_name = unquote_pg_identifier(
+ qualified_table
+ .rsplit('.')
+ .next()
+ .unwrap_or(qualified_table),
+ );
+ let unquoted_qualified_table = qualified_table
+ .split('.')
+ .map(unquote_pg_identifier)
+ .collect::<Vec<_>>()
+ .join(".");
+
+ // test_decoding has no server-side table filter, so Postgres already
+ // sent us every table's changes - config.tables scoping happens here.
+ if let Some(tables) = captured_tables
+ && !tables.iter().any(|t| {
+ if t.contains('.') {
+ t == &unquoted_qualified_table
+ } else {
+ t == &table_name
+ }
+ })
{
- let table_part = &data[table_start + 6..table_start + colon_pos];
- let table_name = table_part
- .split('.')
- .next_back()
- .unwrap_or(table_part)
- .to_string();
-
- let data_part = &data[table_start + colon_pos + 1..];
- let parsed_data = parse_record_data(data_part);
-
- return Some(DatabaseRecord {
- table_name,
- operation_type: "UPDATE".to_string(),
- timestamp: Utc::now(),
- data: serde_json::Value::Object(parsed_data),
- old_data: None,
- });
+ return None;
}
- None
- }
- fn parse_delete_message(&self, data: &str) -> Option<DatabaseRecord> {
- if let Some(table_start) = data.find("table ")
- && let Some(colon_pos) = data[table_start..].find(':')
- {
- let table_part = &data[table_start + 6..table_start + colon_pos];
- let table_name = table_part
- .split('.')
- .next_back()
- .unwrap_or(table_part)
- .to_string();
-
- let data_part = &data[table_start + colon_pos + 1..];
- let parsed_data = parse_record_data(data_part);
-
- return Some(DatabaseRecord {
- table_name,
- operation_type: "DELETE".to_string(),
- timestamp: Utc::now(),
- data: serde_json::Value::Object(parsed_data),
- old_data: None,
- });
- }
- None
+ let rest = rest
Review Comment:
The old-key columns are parsed past and discarded here, so `DatabaseRecord.old_data` is always `None` even though the struct has the field and the data is sitting in the row. For a PK-changing UPDATE (and for DELETE under `REPLICA IDENTITY FULL`) consumers only see the new tuple and cannot learn which row it replaced, which breaks e.g. keyed upsert/delete downstream. Issue #3582 lists unpopulated `old_data` as well. Suggest running `parse_record_columns` over the old-key section and populating `old_data` instead of dropping it.
##########
core/connectors/sources/postgres_source/README.md:
##########
@@ -41,9 +41,7 @@ primary_key_column = "id"
# Custom query (optional)
custom_query = "SELECT * FROM $table WHERE id > $offset ORDER BY id LIMIT $limit"
-# CDC options (optional)
-enable_wal_cdc = false
-publication_name = "iggy_publication"
+# CDC options (only used when mode = "cdc")
replication_slot = "iggy_slot"
Review Comment:
Two operational hazards of the new slot handling deserve README coverage:
1. Slot sharing: `setup_cdc` accepts any pre-existing `test_decoding` slot, so two CDC connectors against the same database with the default `replication_slot = "iggy_slot"` silently share one slot. `pg_logical_slot_get_changes` consumes on read, so each connector steals a subset of the other's changes. The README should state that every CDC connector needs a unique slot name.
2. WAL retention: a slot orphaned after a connector is decommissioned retains WAL indefinitely (default `max_slot_wal_keep_size = -1`) until the disk fills. The README should document dropping the slot (`SELECT pg_drop_replication_slot('iggy_slot')`) as part of decommissioning.
##########
core/integration/tests/connectors/postgres/postgres_source_cdc.rs:
##########
@@ -0,0 +1,399 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::{POLL_ATTEMPTS, POLL_INTERVAL_MS};
+use crate::connectors::create_test_messages;
+use crate::connectors::fixtures::{PostgresOps, PostgresSourceCdcFixture, PostgresSourceOps};
+use iggy::prelude::IggyClient;
+use iggy_common::MessageClient;
+use iggy_common::{Consumer, Identifier, PollingStrategy};
+use iggy_connector_sdk::api::{ConnectorStatus, SourceInfoResponse};
+use integration::harness::seeds;
+use integration::iggy_harness;
+use reqwest::Client;
+use serde::Deserialize;
+use std::time::Duration;
+use tokio::time::sleep;
+
+const API_KEY: &str = "test-api-key";
+const SOURCE_KEY: &str = "postgres";
+const DEFAULT_SLOT: &str = "iggy_slot";
+
+#[derive(Debug, Deserialize)]
+struct CdcRecord {
+ table_name: String,
+ operation_type: String,
+ data: serde_json::Value,
+}
+
+async fn poll_cdc_records(
+ client: &IggyClient,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ consumer_id: &Identifier,
+ want: usize,
+) -> Vec<CdcRecord> {
+ let mut received = Vec::new();
+ 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
+ {
+ for msg in polled.messages {
+ if let Ok(record) = serde_json::from_slice::<CdcRecord>(&msg.payload) {
+ received.push(record);
+ }
+ }
+ if received.len() >= want {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ received
+}
+
+// End-to-end CDC coverage against a real wal_level=logical container:
+// INSERT, UPDATE, PK-changing UPDATE, DELETE, a rolled-back transaction
+// (must produce nothing), an untracked table (must be filtered out), a
+// 25-row sustained batch, and a final INSERT proving the slot/connector
+// are still healthy after all of the above.
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn cdc_source_captures_insert_update_delete(
+ harness: &TestHarness,
+ fixture: PostgresSourceCdcFixture,
+) {
+ const BATCH_SIZE: usize = 25;
+
+ let client = harness.root_client().await.unwrap();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
+ fixture.create_untracked_table(&pool).await;
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "cdc_test_consumer".try_into().unwrap();
+
+ // INSERT: basic capture, full column fidelity (id/name/count/amount/active).
+ let [msg] = create_test_messages(1).try_into().unwrap();
+ fixture
+ .insert_row(
+ &pool,
+ msg.id as i32,
+ &msg.name,
+ msg.count as i32,
+ msg.amount,
+ msg.active,
+ msg.timestamp,
+ )
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id, &consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 INSERT message");
+ assert_eq!(received[0].table_name, fixture.table_name());
+ assert_eq!(received[0].operation_type, "INSERT");
+ assert_eq!(received[0].data["id"], serde_json::json!(msg.id));
+ assert_eq!(received[0].data["name"], serde_json::json!(msg.name));
+ assert_eq!(received[0].data["count"], serde_json::json!(msg.count));
+ assert_eq!(received[0].data["amount"].as_f64(), Some(msg.amount));
+ assert_eq!(received[0].data["active"], serde_json::json!(msg.active));
+
+ // UPDATE: non-key column change.
+ let updated_count = msg.count + 1;
+ fixture
+ .update_row(&pool, msg.id as i32, updated_count as i32)
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id, &consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 UPDATE message");
+ assert_eq!(received[0].operation_type, "UPDATE");
+ assert_eq!(received[0].data["id"], serde_json::json!(msg.id));
+ assert_eq!(received[0].data["count"], serde_json::json!(updated_count));
+
+ // UPDATE: primary key change - test_decoding emits "old-key: ... new-tuple: ...",
+ // the parser must report the new-tuple's id, not the old one.
+ let new_id = msg.id as i32 + 1000;
+ fixture
+ .update_primary_key(&pool, msg.id as i32, new_id)
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id, &consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 UPDATE message for PK change");
+ assert_eq!(received[0].operation_type, "UPDATE");
+ assert_eq!(
+ received[0].data["id"],
+ serde_json::json!(new_id),
+ "PK-changing UPDATE must report the new-tuple id, not the old-key one"
+ );
+ assert_eq!(received[0].data["name"], serde_json::json!(msg.name));
+
+ // DELETE: only replica-identity columns (here, just id) are present.
+ fixture.delete_row(&pool, new_id).await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id, &consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 DELETE message");
+ assert_eq!(received[0].operation_type, "DELETE");
+ assert_eq!(received[0].data["id"], serde_json::json!(new_id));
+
+ // Negative cases, verified together with the batch below: a rolled-back
+ // transaction must never reach test_decoding, and a table outside
+ // config.tables must be dropped by the client-side filter.
+ fixture
+ .insert_row_rolled_back(&pool, 9000, "should_not_appear")
+ .await;
+ fixture.insert_untracked_row(&pool, "not_configured").await;
+
+ // Sustained batch: BATCH_SIZE separate INSERT rows, asserting order and
+ // full data fidelity hold under a longer run, and that neither the
+ // rolled-back row nor the untracked-table row leaked into the output.
+ let batch = create_test_messages(BATCH_SIZE);
+ for row in &batch {
+ fixture
+ .insert_row(
+ &pool,
+ row.id as i32,
+ &row.name,
+ row.count as i32,
+ row.amount,
+ row.active,
+ row.timestamp,
+ )
+ .await;
+ }
+
+ let received = poll_cdc_records(&client, &stream_id, &topic_id, &consumer_id, BATCH_SIZE).await;
+ assert_eq!(
+ received.len(),
+ BATCH_SIZE,
+ "Expected exactly the {BATCH_SIZE} batch of INSERT rows - the rolled-back row and the \
+ untracked-table row must not appear"
+ );
+ for (i, record) in received.iter().enumerate() {
+ assert_eq!(record.operation_type, "INSERT");
+ assert_eq!(record.table_name, fixture.table_name());
+ assert_eq!(
+ record.data["id"],
+ serde_json::json!(batch[i].id),
+ "Out-of-order or missing row at index {i}"
+ );
+ assert_eq!(record.data["name"], serde_json::json!(batch[i].name));
+ }
+
+ // Final sanity check: the slot/connector must still be healthy after the
+ // earlier PK-changing UPDATE and the sustained batch, not left in some
+ // degraded state that silently swallows further changes.
+ let final_id = new_id + BATCH_SIZE as i32 + 1;
+ fixture
+ .insert_row(
+ &pool,
+ final_id,
+ "after_pk_change",
+ 1,
+ 1.0,
+ true,
+ msg.timestamp,
+ )
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id, &consumer_id, 1).await;
+ assert_eq!(
+ received.len(),
+ 1,
+ "Expected the final INSERT after the PK change to still be captured"
+ );
+ assert_eq!(received[0].operation_type, "INSERT");
+ assert_eq!(received[0].data["id"], serde_json::json!(final_id));
+ assert_eq!(
+ received[0].data["name"],
+ serde_json::json!("after_pk_change")
+ );
+
+ pool.close().await;
+}
+
+async fn wait_for_source_status(
+ http: &Client,
+ api_url: &str,
+ expected: ConnectorStatus,
+) -> SourceInfoResponse {
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(resp) = http
+ .get(format!("{api_url}/sources/{SOURCE_KEY}"))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ && let Ok(info) = resp.json::<SourceInfoResponse>().await
+ && info.status == expected
+ {
+ return info;
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ panic!("Source connector did not reach {expected:?} status in time");
+}
+
+// Two restart scenarios against one container, run in sequence:
+// 1. Simulates upgrading from the old broken version, which left a slot
+// behind created with the pgoutput plugin. setup_cdc must refuse to
+// silently reuse a mismatched slot (which used to fail forever on
+// every poll instead) - it should fail loudly at startup, and recover
+// cleanly once the operator drops the bad slot and restarts.
+// 2. Changes written while the connector is down (the slot retains WAL
+// regardless of consumer state) - not the at-least-once crash window
+// where the slot has already been consumed but send/state-persist
+// hasn't happened yet. That gap remains open until the slot-peek/LSN
+// work lands.
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn cdc_source_recovers_from_slot_mismatch_and_restart(
+ harness: &mut TestHarness,
+ fixture: PostgresSourceCdcFixture,
+) {
+ let client = harness.root_client().await.unwrap();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "cdc_restart_consumer".try_into().unwrap();
+
+ let api_url = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
+ let http = Client::new();
+ wait_for_source_status(&http, &api_url, ConnectorStatus::Running).await;
+
+ sqlx::query("SELECT pg_drop_replication_slot($1)")
Review Comment:
Flake risk: this first `pg_drop_replication_slot` races the still-running connector, which calls `pg_logical_slot_get_changes` every 50ms and briefly holds the slot on each call. If the drop lands inside that window, Postgres raises `ERROR 55006: replication slot "iggy_slot" is active for PID ...` and the `expect` panics. Rough odds are a few percent per run, which is enough to flake CI. Wrapping the drop in a short retry loop fixes it. The second drop below is fine since the failed restart leaves no poller running.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): fix postgres source CDC producing no messages [iggy]
Posted by "mattp5657 (via GitHub)" <gi...@apache.org>.
mattp5657 commented on PR #3640:
URL: https://github.com/apache/iggy/pull/3640#issuecomment-5017715256
These should be fixed, let me know if you have any other feedback :)
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): fix postgres source CDC producing no messages [iggy]
Posted by "mmodzelewski (via GitHub)" <gi...@apache.org>.
mmodzelewski merged PR #3640:
URL: https://github.com/apache/iggy/pull/3640
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): fix postgres source CDC producing no messages [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3640:
URL: https://github.com/apache/iggy/pull/3640#issuecomment-4930557146
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3640?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `50.00000%` with `269 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 39.73%. Comparing base ([`6ae792a`](https://app.codecov.io/gh/apache/iggy/commit/6ae792a37d2d470fadba7f18b780442f582bd258?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`af5bbc4`](https://app.codecov.io/gh/apache/iggy/commit/af5bbc4fe41b7d168f807a1f86a08a0c84fb6eb6?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3640?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [core/connectors/sources/postgres\_source/src/lib.rs](https://app.codecov.io/gh/apache/iggy/pull/3640?src=pr&el=tree&filepath=core%2Fconnectors%2Fsources%2Fpostgres_source%2Fsrc%2Flib.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25uZWN0b3JzL3NvdXJjZXMvcG9zdGdyZXNfc291cmNlL3NyYy9saWIucnM=) | 50.00% | [265 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/3640?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3640 +/- ##
=============================================
- Coverage 73.53% 39.73% -33.81%
Complexity 937 937
=============================================
Files 1283 1281 -2
Lines 138809 121901 -16908
Branches 114699 97791 -16908
=============================================
- Hits 102077 48440 -53637
- Misses 33563 70878 +37315
+ Partials 3169 2583 -586
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/3640/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3640/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `31.25% <50.00%> (-42.53%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3640/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.44% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3640/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.10% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3640/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.65% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3640/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.29% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3640/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.35% <ø> (ø)` | |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3640/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `42.28% <ø> (ø)` | |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3640?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/connectors/sources/postgres\_source/src/lib.rs](https://app.codecov.io/gh/apache/iggy/pull/3640?src=pr&el=tree&filepath=core%2Fconnectors%2Fsources%2Fpostgres_source%2Fsrc%2Flib.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25uZWN0b3JzL3NvdXJjZXMvcG9zdGdyZXNfc291cmNlL3NyYy9saWIucnM=) | `48.94% <50.00%> (-7.74%)` | :arrow_down: |
... and [500 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3640/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(connectors): fix postgres source CDC producing no messages [iggy]
Posted by "mattp5657 (via GitHub)" <gi...@apache.org>.
mattp5657 commented on code in PR #3640:
URL: https://github.com/apache/iggy/pull/3640#discussion_r3611524832
##########
core/connectors/sources/postgres_source/src/lib.rs:
##########
@@ -303,10 +301,6 @@ impl PostgresSource {
}
async fn setup_cdc(&self) -> Result<(), Error> {
Review Comment:
I resolved this but also handled two other edge cases.
1. `capture_operations` - an entry outside `INSERT`/`UPDATE`/`DELETE` (e.g. a typo like `INSRT`) previously passed startup silently and would just never match, quietly dropping that operation type forever.
2. `payload_format` - an unsupported value (e.g. `btea` instead of `bytea`) would similarly pass startup and only surface as a failure deep in row processing on the first poll, not at startup.
Both now validate in `open()` alongside `cdc_backend`, before `setup_cdc()` runs, so any of the three fails loudly at startup instead of leaving a `Running` connector that silently drops or mishandles data.
Then I rewrote the integration test to make sure it covered this as well.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org