You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/14 17:14:18 UTC
(iggy) branch master updated: fix(connectors): defer source checkpoints and surface sink failures (#4153)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new f239fca1e fix(connectors): defer source checkpoints and surface sink failures (#4153)
f239fca1e is described below
commit f239fca1eb3926078fa2bc56d8104f94198e3c24
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Sep 14 19:14:13 2026 +0200
fix(connectors): defer source checkpoints and surface sink failures (#4153)
---
Cargo.lock | 2 +
.../connectors/sinks/clickhouse_sink/src/binary.rs | 161 +++++++--
.../connectors/sinks/clickhouse_sink/src/client.rs | 4 +-
core/connectors/sinks/doris_sink/src/lib.rs | 1 +
core/connectors/sinks/elasticsearch_sink/README.md | 7 +-
.../sinks/elasticsearch_sink/config.toml | 4 +-
.../connectors/sinks/elasticsearch_sink/src/lib.rs | 391 ++++++++++++++++-----
core/connectors/sinks/iceberg_sink/README.md | 4 +-
core/connectors/sinks/iceberg_sink/src/lib.rs | 41 +++
core/connectors/sinks/iceberg_sink/src/sink.rs | 10 +
core/connectors/sinks/influxdb_sink/src/lib.rs | 20 +-
core/connectors/sinks/postgres_sink/README.md | 21 +-
core/connectors/sinks/postgres_sink/src/lib.rs | 130 +++++--
core/connectors/sinks/quickwit_sink/README.md | 7 +-
core/connectors/sinks/quickwit_sink/config.toml | 2 +-
core/connectors/sinks/quickwit_sink/src/lib.rs | 60 +++-
core/connectors/sinks/s3_sink/README.md | 8 +-
core/connectors/sinks/s3_sink/src/buffer.rs | 10 +
core/connectors/sinks/s3_sink/src/client.rs | 55 ---
core/connectors/sinks/s3_sink/src/formatter.rs | 56 ++-
core/connectors/sinks/s3_sink/src/sink.rs | 391 ++++++++++++++++++++-
core/connectors/sinks/surrealdb_sink/README.md | 2 +-
.../sources/elasticsearch_source/Cargo.toml | 4 +
.../sources/elasticsearch_source/README.md | 335 ++++++------------
.../sources/elasticsearch_source/config.toml | 2 +-
.../sources/elasticsearch_source/src/lib.rs | 337 ++++++++++++++++--
.../elasticsearch_source/src/state_manager.rs | 40 ++-
core/connectors/sources/influxdb_source/README.md | 106 +++---
.../connectors/sources/influxdb_source/config.toml | 10 +-
.../sources/influxdb_source/src/common.rs | 22 +-
core/connectors/sources/influxdb_source/src/lib.rs | 300 +++++++++++++---
core/connectors/sources/influxdb_source/src/row.rs | 112 ++++--
core/connectors/sources/influxdb_source/src/v2.rs | 16 +-
core/connectors/sources/influxdb_source/src/v3.rs | 114 ++++--
core/connectors/sources/postgres_source/README.md | 233 ++++++++----
.../connectors/sources/postgres_source/config.toml | 4 +-
core/connectors/sources/postgres_source/src/lib.rs | 26 +-
.../connectors/elasticsearch/elasticsearch_sink.rs | 2 +-
core/integration/tests/connectors/fixtures/mod.rs | 11 +-
.../tests/connectors/fixtures/postgres/mod.rs | 7 +-
.../tests/connectors/fixtures/postgres/sink.rs | 36 ++
.../tests/connectors/fixtures/postgres/source.rs | 48 +++
.../tests/connectors/mongodb/mongodb_sink.rs | 8 +-
.../tests/connectors/postgres/postgres_sink.rs | 136 ++++++-
.../tests/connectors/postgres/postgres_source.rs | 349 +++++++++++++++++-
core/integration/tests/connectors/s3/s3_sink.rs | 18 +-
.../tests/connectors/surrealdb/surrealdb_sink.rs | 4 +-
47 files changed, 2828 insertions(+), 839 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 0580a25af..4c5cc600c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7178,8 +7178,10 @@ dependencies = [
"serde",
"serde_json",
"simd-json",
+ "tempfile",
"tokio",
"tracing",
+ "wiremock",
]
[[package]]
diff --git a/core/connectors/sinks/clickhouse_sink/src/binary.rs b/core/connectors/sinks/clickhouse_sink/src/binary.rs
index 227eba0e6..fb3fcb27a 100644
--- a/core/connectors/sinks/clickhouse_sink/src/binary.rs
+++ b/core/connectors/sinks/clickhouse_sink/src/binary.rs
@@ -294,6 +294,11 @@ pub(crate) fn serialize_value(
// ── Decimal ──────────────────────────────────────────────────────────
ChType::Decimal(precision, scale) => {
let int_val = coerce_decimal(value, *precision, *scale)?;
+ let precision_limit = 10i128.pow(u32::from(*precision));
+ if int_val <= -precision_limit || int_val >= precision_limit {
+ error!("Decimal value out of range for precision {precision}");
+ return Err(Error::InvalidRecord);
+ }
if *precision <= 9 {
buf.extend_from_slice(
&i32::try_from(int_val)
@@ -632,25 +637,24 @@ fn parse_decimal_str(s: &str, scale: u8) -> Result<i128, ()> {
let mut result = int_val.checked_mul(10i128.pow(scale)).ok_or(())?;
if !frac_str.is_empty() {
- // Cap at 38 digits: Decimal128(38) is the widest ClickHouse type, and
- // 10^38 < i128::MAX, so no intermediate value overflows.
- let frac_str = if frac_str.len() > 38 {
- &frac_str[..38]
- } else {
- frac_str
- };
- let frac_len = frac_str.len() as u32;
- let frac_val: i128 = frac_str.parse().map_err(|_| ())?;
- let frac_scaled = if frac_len <= scale {
- frac_val
- .checked_mul(10i128.pow(scale - frac_len))
- .ok_or(())?
+ // Parse only retained digits so long fractions cannot overflow before rounding.
+ let retained_digits = frac_str.len().min(scale as usize);
+ let frac_val: i128 = if retained_digits == 0 {
+ 0
} else {
- // More digits than scale: round half-up.
- let divisor = 10i128.pow(frac_len - scale);
- (frac_val + divisor / 2) / divisor
+ frac_str[..retained_digits].parse().map_err(|_| ())?
};
+ let frac_scaled = frac_val
+ .checked_mul(10i128.pow(scale - retained_digits as u32))
+ .ok_or(())?;
result = result.checked_add(frac_scaled).ok_or(())?;
+ if frac_str
+ .as_bytes()
+ .get(scale as usize)
+ .is_some_and(|digit| *digit >= b'5')
+ {
+ result = result.checked_add(1).ok_or(())?;
+ }
}
Ok(if negative { -result } else { result })
@@ -718,7 +722,12 @@ fn coerce_to_unix_seconds_f64(value: &OwnedValue) -> Result<f64, Error> {
/// Returns Unix seconds as i64 (truncates fractional seconds).
fn coerce_to_unix_seconds(value: &OwnedValue) -> Result<i64, Error> {
- Ok(coerce_to_unix_seconds_f64(value)? as i64)
+ let seconds = coerce_to_unix_seconds_f64(value)?;
+ if !seconds.is_finite() || seconds < 0.0 {
+ error!("DateTime requires finite, non-negative Unix seconds");
+ return Err(Error::InvalidRecord);
+ }
+ Ok(seconds as i64)
}
/// Parse "YYYY-MM-DDThh:mm:ss[.frac][Z|±hh:mm]" into Unix seconds (f64).
@@ -893,6 +902,111 @@ mod tests {
}
// ── primitives ───────────────────────────────────────────────────────────
+ #[test]
+ fn decimal_serialization_enforces_declared_precision_after_scaling() {
+ for precision in [1, 9, 10, 18, 19, 38] {
+ let limit = 10i128.pow(u32::from(precision));
+ for scale in [0, precision] {
+ for sign in [1, -1] {
+ let boundary = if scale == 0 {
+ (sign * (limit - 1)).to_string()
+ } else {
+ format!(
+ "{}0.{}",
+ if sign < 0 { "-" } else { "" },
+ "9".repeat(precision as usize)
+ )
+ };
+ let mut buffer = Vec::new();
+ serialize_value(
+ &json_str(&boundary),
+ &ChType::Decimal(precision, scale),
+ &mut buffer,
+ )
+ .expect("The last value within precision must serialize");
+ let expected = (sign * (limit - 1)).to_le_bytes();
+ let width = if precision <= 9 {
+ 4
+ } else if precision <= 18 {
+ 8
+ } else {
+ 16
+ };
+ assert_eq!(
+ buffer,
+ expected[..width],
+ "Decimal({precision}, {scale}): {boundary}"
+ );
+
+ let overflow = if scale == 0 {
+ (sign * limit).to_string()
+ } else {
+ sign.to_string()
+ };
+ let rounded_overflow =
+ format!("{boundary}{}", if scale == 0 { ".5" } else { "5" });
+ for value in [json_str(&overflow), json_str(&rounded_overflow)] {
+ let mut buffer = Vec::new();
+ assert!(
+ serialize_value(
+ &value,
+ &ChType::Decimal(precision, scale),
+ &mut buffer
+ )
+ .is_err(),
+ "Decimal({precision}, {scale}) must reject {value}"
+ );
+ assert!(buffer.is_empty());
+ }
+ }
+ }
+ }
+ for value in [
+ json_i64(10_000_000),
+ json_u64(10_000_000),
+ json_f64(10_000_000.0),
+ ] {
+ assert!(
+ serialize_value(&value, &ChType::Decimal(9, 2), &mut Vec::new()).is_err(),
+ "{value}"
+ );
+ }
+ assert!(
+ serialize_value(
+ &json_str(&i128::MIN.to_string()),
+ &ChType::Decimal(38, 0),
+ &mut Vec::new()
+ )
+ .is_err()
+ );
+ }
+
+ #[test]
+ fn datetime_rejects_non_finite_and_pre_epoch_values() {
+ for value in [
+ json_f64(f64::NAN),
+ json_f64(f64::INFINITY),
+ json_f64(f64::NEG_INFINITY),
+ json_f64(-0.5),
+ json_str("1969-12-31T23:59:59.5Z"),
+ ] {
+ let mut buffer = Vec::new();
+ assert!(
+ serialize_value(&value, &ChType::DateTime, &mut buffer).is_err(),
+ "{value}"
+ );
+ assert!(buffer.is_empty());
+ }
+ for seconds in [0.0, 0.5, u32::MAX as f64 + 0.5] {
+ let mut buffer = Vec::new();
+ serialize_value(&json_f64(seconds), &ChType::DateTime, &mut buffer).unwrap();
+ assert_eq!(buffer, (seconds as u32).to_le_bytes());
+ }
+ let mut buffer = Vec::new();
+ serialize_value(&json_f64(-0.5), &ChType::DateTime64(3), &mut buffer).unwrap();
+ assert_eq!(buffer, (-500i64).to_le_bytes());
+ }
+
#[test]
fn given_boundary_floats_when_serializing_numbers_should_preserve_values() {
for (column_type, value, expected) in [
@@ -906,11 +1020,6 @@ mod tests {
(1u64 << 63) as f64,
(1u64 << 63).to_le_bytes().to_vec(),
),
- (
- ChType::Decimal(38, 0),
- i128::MIN as f64,
- i128::MIN.to_le_bytes().to_vec(),
- ),
] {
let mut buffer = Vec::new();
serialize_value(&json_f64(value), &column_type, &mut buffer).unwrap();
@@ -941,7 +1050,13 @@ mod tests {
#[test]
fn given_overflowing_float_when_serializing_decimal_should_reject() {
- for value in [i128::MAX as f64, f64::MAX, -f64::MAX, f64::NAN] {
+ for value in [
+ i128::MIN as f64,
+ i128::MAX as f64,
+ f64::MAX,
+ -f64::MAX,
+ f64::NAN,
+ ] {
let mut buffer = Vec::new();
let result = serialize_value(&json_f64(value), &ChType::Decimal(38, 0), &mut buffer);
assert!(
diff --git a/core/connectors/sinks/clickhouse_sink/src/client.rs b/core/connectors/sinks/clickhouse_sink/src/client.rs
index 017ae05fa..9da425fc3 100644
--- a/core/connectors/sinks/clickhouse_sink/src/client.rs
+++ b/core/connectors/sinks/clickhouse_sink/src/client.rs
@@ -160,8 +160,8 @@ impl ClickHouseClient {
/// Insert `body` into `table` using the given ClickHouse FORMAT string.
///
- /// Retries up to `max_retries` times on transient errors (network errors,
- /// HTTP 429, HTTP 5xx). Does not retry on HTTP 4xx (data errors).
+ /// Makes up to `max_retries` total attempts (at least one) on network errors
+ /// and HTTP 408, 429 or 5xx. Other HTTP errors stop retries.
///
/// # At-least-once semantics
///
diff --git a/core/connectors/sinks/doris_sink/src/lib.rs b/core/connectors/sinks/doris_sink/src/lib.rs
index c28594d13..9ed094876 100644
--- a/core/connectors/sinks/doris_sink/src/lib.rs
+++ b/core/connectors/sinks/doris_sink/src/lib.rs
@@ -156,6 +156,7 @@ pub struct DorisSinkConfig {
/// Stream Load output format: `"json"` (default) or `"csv"`. CSV is opt-in
/// for throughput; because Doris CSV is positional (JSON is name-mapped), it
/// requires `columns` to pin the column order, else `open()` fails.
+ /// A field named `format` would lose its environment override to `PLUGIN_CONFIG_FORMAT`.
pub output_format: Option<Format>,
/// Total per-request HTTP timeout as a human-readable duration, e.g. "30s"
/// (default 30s). Matches the `timeout` field on the http/influxdb sinks.
diff --git a/core/connectors/sinks/elasticsearch_sink/README.md b/core/connectors/sinks/elasticsearch_sink/README.md
index f70d1ecd3..8b52b63aa 100644
--- a/core/connectors/sinks/elasticsearch_sink/README.md
+++ b/core/connectors/sinks/elasticsearch_sink/README.md
@@ -9,13 +9,18 @@ A sink connector that consumes messages from Iggy streams and indexes them to El
- `username/password`: Optional authentication credentials
- `batch_size`: Accepted but unused; the stream-level `batch_length` determines each incoming batch
- `timeout_seconds`: Client-wide HTTP timeout for `open()` and bulk `consume()` (default: 30s). Values of `0` are clamped to 1s. Raise for slow bulk workloads; a timed-out bulk fails the batch after the poll offset is already committed
+- `max_retries`: Total bulk attempts for explicitly rejected items, including the first (default: 3). Values of `0` and `1` disable retries
+- `retry_delay`: Initial backoff delay in humantime format (default: `"1s"`)
+- `retry_max_delay`: Maximum backoff delay in humantime format (default: `"5s"`)
- `create_index_if_not_exists`: Automatically create index (default: true)
- `index_mapping`: Index mapping configuration
+Retry delays allow zero; invalid duration strings log a warning and fall back to `1s`. The HTTP timeout applies per request, so a consume call can span multiple requests and backoff delays; there is no separate total time budget.
+
## Features
- Bulk indexing optimization
- Automatic index creation
-- Request errors and bulk requests with no successfully indexed documents fail the batch without connector retries. Partial bulk failures log the rejected count for the batch and are counted in closing statistics without failing the batch
+- Items rejected with HTTP 429 or 5xx receive up to `max_retries` attempts with exponential backoff and jitter. Only rejected items are retried; accepted documents are not resubmitted. Request failures or malformed responses fail the batch without replaying an ambiguous write. Any remaining rejection fails the callback, including partial failures, while successful items remain indexed
- Metadata field injection
- Support for multiple data formats
diff --git a/core/connectors/sinks/elasticsearch_sink/config.toml b/core/connectors/sinks/elasticsearch_sink/config.toml
index 171e0eec6..63e9df7a5 100644
--- a/core/connectors/sinks/elasticsearch_sink/config.toml
+++ b/core/connectors/sinks/elasticsearch_sink/config.toml
@@ -35,5 +35,7 @@ consumer_group = "elasticsearch_sink_cg"
url = "http://localhost:9200"
index = "iggy_messages"
create_index_if_not_exists = true
-batch_size = 100
timeout_seconds = 30
+max_retries = 3
+retry_delay = "1s"
+retry_max_delay = "5s"
diff --git a/core/connectors/sinks/elasticsearch_sink/src/lib.rs b/core/connectors/sinks/elasticsearch_sink/src/lib.rs
index 0b1ace79d..438131fca 100644
--- a/core/connectors/sinks/elasticsearch_sink/src/lib.rs
+++ b/core/connectors/sinks/elasticsearch_sink/src/lib.rs
@@ -23,13 +23,12 @@ use elasticsearch::{
http::{Url, request::JsonBody, transport::TransportBuilder},
};
use iggy_common::IggyTimestamp;
+use iggy_connector_sdk::retry::{RetryPolicy, parse_duration};
use iggy_connector_sdk::{
- ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
- convert::owned_value_to_serde_json, sink_connector,
+ ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector,
};
use secrecy::{ExposeSecret, SecretString};
use serde::{Deserialize, Serialize};
-use serde_json::json;
use simd_json::{OwnedValue, prelude::*};
use std::time::Duration;
use tokio::sync::Mutex;
@@ -38,6 +37,25 @@ use tracing::{error, info, warn};
sink_connector!(ElasticsearchSink);
const DEFAULT_TIMEOUT_SECONDS: u64 = 30;
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+
+#[derive(Deserialize)]
+struct BulkResponse {
+ items: Vec<BulkItem>,
+}
+
+#[derive(Deserialize)]
+struct BulkItem {
+ index: BulkIndexResult,
+}
+
+#[derive(Deserialize)]
+struct BulkIndexResult {
+ status: u16,
+ error: Option<serde_json::Value>,
+}
#[derive(Debug)]
struct State {
@@ -60,6 +78,11 @@ pub struct ElasticsearchSinkConfig {
/// a timeout on consume drops the batch after the poll offset is already
/// committed.
pub timeout_seconds: Option<u64>,
+ /// Total attempts for explicitly rejected bulk items, including the first.
+ /// Defaults to 3; values of 0 and 1 both disable retries.
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
pub create_index_if_not_exists: Option<bool>,
pub index_mapping: Option<serde_json::Value>,
}
@@ -69,15 +92,22 @@ pub struct ElasticsearchSink {
id: u32,
config: ElasticsearchSinkConfig,
client: Option<Elasticsearch>,
+ retry_policy: RetryPolicy,
state: Mutex<State>,
}
impl ElasticsearchSink {
pub fn new(id: u32, config: ElasticsearchSinkConfig) -> Self {
+ let retry_policy = RetryPolicy {
+ max_attempts: config.max_retries.unwrap_or(DEFAULT_MAX_RETRIES).max(1),
+ base_delay: parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY),
+ max_delay: parse_duration(config.retry_max_delay.as_deref(), DEFAULT_RETRY_MAX_DELAY),
+ };
ElasticsearchSink {
id,
config,
client: None,
+ retry_policy,
state: Mutex::new(State {
invocations_count: 0,
documents_indexed: 0,
@@ -175,104 +205,107 @@ impl ElasticsearchSink {
async fn bulk_index_documents(
&self,
client: &Elasticsearch,
- documents: Vec<OwnedValue>,
+ mut documents: Vec<OwnedValue>,
) -> Result<usize, Error> {
if documents.is_empty() {
return Ok(0);
}
- let mut body: Vec<JsonBody<_>> = Vec::with_capacity(documents.len() * 2);
- for doc in documents {
- // Add index action
- body.push(
- json!({
- "index": {
- "_index": self.config.index
- }
- })
- .into(),
- );
- let doc_json: serde_json::Value = owned_value_to_serde_json(&doc);
- body.push(doc_json.into());
- }
-
- let response = client
- .bulk(BulkParts::None)
- .body(body)
- .send()
- .await
- .map_err(|e| Error::Connection(format!("Failed to execute bulk request: {}", e)))?;
+ let action = simd_json::json!({"index": {"_index": self.config.index.as_str()}});
+ let total_documents = documents.len();
+ let mut documents_indexed = 0;
+ let mut permanent_rejections = 0;
+ let mut attempt = 0;
+ loop {
+ attempt += 1;
+ let mut body: Vec<JsonBody<&OwnedValue>> = Vec::with_capacity(documents.len() * 2);
+ for document in &documents {
+ body.push((&action).into());
+ body.push(document.into());
+ }
- if !response.status_code().is_success() {
- let error_text = response
- .text()
+ // Generated document IDs make transport failures ambiguous. Retry only
+ // items the server explicitly rejected, never the whole request.
+ let response = client
+ .bulk(BulkParts::None)
+ .body(body)
+ .send()
.await
- .unwrap_or_else(|_| "Unknown error".to_string());
- return Err(Error::Connection(format!(
- "Bulk indexing failed: {}",
- error_text
- )));
- }
-
- let response_body: serde_json::Value = response
- .json()
- .await
- .map_err(|e| Error::Connection(format!("Failed to parse bulk response: {}", e)))?;
-
- // A 200 without an items array is not a bulk response this connector
- // can account for, so it must not be read as "nothing was indexed".
- let Some(items) = response_body.get("items").and_then(|v| v.as_array()) else {
- return Err(Error::Connection(format!(
- "Elasticsearch bulk response for index '{}' carried no items array",
- self.config.index
- )));
- };
+ .map_err(|error| {
+ Error::Connection(format!("Failed to execute bulk request: {error}"))
+ })?;
+ if !response.status_code().is_success() {
+ let status = response.status_code();
+ let reason = response.text().await.unwrap_or_default();
+ return Err(Error::Connection(format!(
+ "Bulk indexing failed with status {status}: {reason}"
+ )));
+ }
+ let response: BulkResponse = response.json().await.map_err(|error| {
+ Error::Connection(format!("Failed to parse bulk response: {error}"))
+ })?;
+ if response.items.len() != documents.len() {
+ return Err(Error::Connection(format!(
+ "Elasticsearch bulk response carried {} items for {} documents",
+ response.items.len(),
+ documents.len()
+ )));
+ }
- let mut errors = 0;
- let mut retryable_rejection = false;
- for item in items {
- if let Some(index_result) = item.get("index")
- && let Some(error) = index_result.get("error")
+ let mut retry_documents = Vec::new();
+ let mut indexed = 0;
+ let mut rejected = 0;
+ for (document, item) in documents.into_iter().zip(response.items) {
+ let result = item.index;
+ if (200..300).contains(&result.status) && result.error.is_none() {
+ indexed += 1;
+ } else {
+ warn!(
+ "Document indexing error: status {}: {:?}",
+ result.status, result.error
+ );
+ if result.status == 429 || (500..600).contains(&result.status) {
+ retry_documents.push(document);
+ } else {
+ rejected += 1;
+ }
+ }
+ }
+ documents_indexed += indexed;
+ permanent_rejections += rejected;
{
- warn!("Document indexing error: {error}");
- errors += 1;
- retryable_rejection |= index_result
- .get("status")
- .and_then(serde_json::Value::as_u64)
- .is_none_or(|status| status == 429 || status >= 500);
+ let mut state = self.state.lock().await;
+ state.documents_indexed += indexed;
+ state.errors_count += rejected;
}
- }
-
- let documents_indexed = items.len() - errors;
- {
- let mut state = self.state.lock().await;
- state.errors_count += errors;
- state.documents_indexed += documents_indexed;
- }
-
- if errors > 0 {
- // The runtime counts a batch, not a document, so a partial rejection
- // is invisible in `/stats`. This line is the only per-batch record.
- error!(
- "Elasticsearch rejected {errors} of {} documents in index '{}'",
- items.len(),
- self.config.index
+ if retry_documents.is_empty() {
+ break;
+ }
+ if attempt >= self.retry_policy.max_attempts {
+ self.state.lock().await.errors_count += retry_documents.len();
+ return Err(Error::CannotStoreData(format!(
+ "Elasticsearch indexed {documents_indexed} of {total_documents} documents in index '{}'; {} transient rejections remain after {attempt} attempts, {permanent_rejections} permanent rejections",
+ self.config.index,
+ retry_documents.len()
+ )));
+ }
+ let delay = self.retry_policy.backoff(attempt);
+ warn!(
+ "Retrying {} rejected Elasticsearch documents after {delay:?}, attempt {attempt}/{}",
+ retry_documents.len(),
+ self.retry_policy.max_attempts
);
+ tokio::time::sleep(delay).await;
+ documents = retry_documents;
}
- if documents_indexed == 0 {
+ if permanent_rejections > 0 {
let reason = format!(
- "Elasticsearch bulk request indexed no documents in index '{}'",
+ "Elasticsearch rejected {permanent_rejections} of {total_documents} documents in index '{}'; indexed {documents_indexed}",
self.config.index
);
- // Bad data must not look like a connectivity failure to a circuit
- // breaker, but a 429 or 5xx rejection is transient and is not
- // permanent. See `Error::PermanentHttpError`.
- return Err(if retryable_rejection {
- Error::CannotStoreData(reason)
- } else {
- Error::PermanentHttpError(reason)
- });
+ error!("{reason}");
+ return Err(Error::PermanentHttpError(reason));
}
Ok(documents_indexed)
@@ -287,6 +320,11 @@ impl Sink for ElasticsearchSink {
self.id, self.config.url, self.config.index
);
+ if self.config.batch_size.is_some() {
+ warn!(
+ "Elasticsearch plugin_config.batch_size is ignored; use streams[].batch_length to control the input batch size"
+ );
+ }
let client = self.create_client().await?;
self.ensure_index_exists(&client).await?;
self.client = Some(client);
@@ -418,6 +456,7 @@ impl Sink for ElasticsearchSink {
#[cfg(test)]
mod tests {
+ use serde_json::json;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -431,13 +470,59 @@ mod tests {
password: None,
batch_size: None,
timeout_seconds: Some(1),
+ max_retries: None,
+ retry_delay: None,
+ retry_max_delay: None,
create_index_if_not_exists: Some(false),
index_mapping: None,
}
}
#[test]
- fn given_bulk_item_results_when_indexing_should_fail_only_fully_rejected_batches() {
+ fn given_omitted_retry_settings_when_loading_should_preserve_defaults() {
+ let config = serde_json::from_value(json!({
+ "url": "http://localhost:9200",
+ "index": "test"
+ }))
+ .expect("Existing configurations should remain valid");
+ let sink = ElasticsearchSink::new(1, config);
+
+ assert_eq!(sink.retry_policy.max_attempts, 3);
+ assert_eq!(sink.retry_policy.base_delay, Duration::from_secs(1));
+ assert_eq!(sink.retry_policy.max_delay, Duration::from_secs(5));
+ }
+
+ #[test]
+ fn given_configured_retry_delays_when_loading_should_parse_or_fall_back() {
+ for (delay, max_delay, expected_delay, expected_max) in [
+ (
+ "200ms",
+ "2s",
+ Duration::from_millis(200),
+ Duration::from_secs(2),
+ ),
+ ("0s", "0s", Duration::ZERO, Duration::ZERO),
+ (
+ "invalid",
+ "invalid",
+ Duration::from_secs(1),
+ Duration::from_secs(1),
+ ),
+ ] {
+ let config = ElasticsearchSinkConfig {
+ retry_delay: Some(delay.to_string()),
+ retry_max_delay: Some(max_delay.to_string()),
+ ..test_config()
+ };
+ let sink = ElasticsearchSink::new(1, config);
+
+ assert_eq!(sink.retry_policy.base_delay, expected_delay, "{delay}");
+ assert_eq!(sink.retry_policy.max_delay, expected_max, "{max_delay}");
+ }
+ }
+
+ #[test]
+ fn given_bulk_item_results_when_indexing_should_fail_any_rejected_batch() {
let runtime = tokio::runtime::Runtime::new().expect("test runtime should start");
runtime.block_on(async {
let accepted = json!({"index": {"status": 201}});
@@ -474,7 +559,7 @@ mod tests {
)
.await;
- if expected_indexed == 0 {
+ if expected_indexed < items.len() {
assert!(
matches!(result, Err(Error::PermanentHttpError(_))),
"{result:?}"
@@ -504,7 +589,7 @@ mod tests {
"error": {"type": "es_rejected_execution_exception"}
}})],
})))
- .expect(1)
+ .expect(u64::from(DEFAULT_MAX_RETRIES))
.mount(&server)
.await;
let mut config = test_config();
@@ -527,6 +612,47 @@ mod tests {
});
}
+ #[test]
+ fn given_configured_attempts_when_indexing_should_limit_transient_retries() {
+ let runtime = tokio::runtime::Runtime::new().expect("test runtime should start");
+ runtime.block_on(async {
+ for (max_retries, expected_attempts) in [(0, 1), (1, 1), (2, 2), (4, 4)] {
+ let server = MockServer::start().await;
+ Mock::given(method("POST"))
+ .and(path("/_bulk"))
+ .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+ "items": [{"index": {"status": 429}}]
+ })))
+ .expect(expected_attempts)
+ .mount(&server)
+ .await;
+ let mut config = serde_json::to_value(test_config()).expect("serialize config");
+ config["url"] = json!(server.uri());
+ config["max_retries"] = json!(max_retries);
+ config["retry_delay"] = json!("1ms");
+ config["retry_max_delay"] = json!("2ms");
+ let config = serde_json::from_value(config).expect("deserialize config");
+ let sink = ElasticsearchSink::new(1, config);
+ let client = sink
+ .create_client()
+ .await
+ .expect("client should initialize");
+
+ let result = sink
+ .bulk_index_documents(&client, vec![simd_json::json!({"id": 1})])
+ .await;
+
+ assert!(
+ matches!(result, Err(Error::CannotStoreData(_))),
+ "{result:?}"
+ );
+ let state = sink.state.lock().await;
+ assert_eq!(state.documents_indexed, 0);
+ assert_eq!(state.errors_count, 1);
+ }
+ });
+ }
+
#[test]
fn given_a_bulk_response_without_items_when_indexing_should_report_a_connection_error() {
let runtime = tokio::runtime::Runtime::new().expect("test runtime should start");
@@ -557,6 +683,95 @@ mod tests {
});
}
+ #[test]
+ fn given_mixed_bulk_results_when_retrying_should_send_only_transient_rejections() {
+ let runtime = tokio::runtime::Runtime::new().expect("test runtime should start");
+ runtime.block_on(async {
+ for permanent_rejection in [false, true] {
+ let server = MockServer::start().await;
+ let last_status = if permanent_rejection { 400 } else { 201 };
+ Mock::given(method("POST"))
+ .and(path("/_bulk"))
+ .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+ "items": [
+ {"index": {"status": 201}},
+ {"index": {"status": 429, "error": {"type": "es_rejected_execution_exception"}}},
+ {"index": {"status": last_status}},
+ ]
+ })))
+ .up_to_n_times(1)
+ .expect(1)
+ .mount(&server)
+ .await;
+ Mock::given(method("POST"))
+ .and(path("/_bulk"))
+ .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+ "items": [{"index": {"status": 201}}]
+ })))
+ .expect(1)
+ .mount(&server)
+ .await;
+ let mut config = test_config();
+ config.url = server.uri();
+ let sink = ElasticsearchSink::new(1, config);
+ let client = sink.create_client().await.unwrap();
+ let documents = vec![
+ simd_json::json!({"id": 1}),
+ simd_json::json!({"id": 2}),
+ simd_json::json!({"id": 3}),
+ ];
+ let result = sink.bulk_index_documents(&client, documents).await;
+ if permanent_rejection {
+ assert!(matches!(result, Err(Error::PermanentHttpError(_))), "{result:?}");
+ } else {
+ assert_eq!(result, Ok(3));
+ }
+ let requests = server.received_requests().await.unwrap();
+ assert_eq!(requests.len(), 2);
+ let retry_body: Vec<serde_json::Value> = requests[1].body
+ .split(|byte| *byte == b'\n')
+ .filter(|line| !line.is_empty())
+ .map(|line| serde_json::from_slice(line).unwrap())
+ .collect();
+ assert_eq!(retry_body, vec![json!({"index": {"_index": "test"}}), json!({"id": 2})]);
+ let state = sink.state.lock().await;
+ assert_eq!(state.documents_indexed, if permanent_rejection { 2 } else { 3 });
+ assert_eq!(state.errors_count, usize::from(permanent_rejection));
+ }
+ });
+ }
+
+ #[test]
+ fn given_ambiguous_bulk_response_when_indexing_should_fail_without_replaying() {
+ let runtime = tokio::runtime::Runtime::new().expect("test runtime should start");
+ runtime.block_on(async {
+ for (status, body) in [
+ (503, json!({"error": "unavailable"})),
+ (200, json!({"items": []})),
+ (
+ 200,
+ json!({"items": [{"index": {"error": "missing status"}}]}),
+ ),
+ ] {
+ let server = MockServer::start().await;
+ Mock::given(method("POST"))
+ .and(path("/_bulk"))
+ .respond_with(ResponseTemplate::new(status).set_body_json(body))
+ .expect(1)
+ .mount(&server)
+ .await;
+ let mut config = test_config();
+ config.url = server.uri();
+ let sink = ElasticsearchSink::new(1, config);
+ let client = sink.create_client().await.unwrap();
+ let result = sink
+ .bulk_index_documents(&client, vec![simd_json::json!({"id": 1})])
+ .await;
+ assert!(matches!(result, Err(Error::Connection(_))), "{result:?}");
+ }
+ });
+ }
+
#[test]
fn given_empty_batch_when_indexing_should_succeed_without_a_request() {
let runtime = tokio::runtime::Runtime::new().expect("test runtime should start");
diff --git a/core/connectors/sinks/iceberg_sink/README.md b/core/connectors/sinks/iceberg_sink/README.md
index a2f3ec4d3..093cf18d5 100644
--- a/core/connectors/sinks/iceberg_sink/README.md
+++ b/core/connectors/sinks/iceberg_sink/README.md
@@ -47,8 +47,8 @@ store_path_style_access = true
- **store_class**: The storage class to use. **Currently, only S3-compatible storage is supported.**
- **store_path_style_access**: Use path-style S3 URLs (`http://host/bucket/key`). Defaults to `true`, which MinIO-style endpoints require; set to `false` for stores that only accept virtual-hosted-style URLs.
-All options above are required except the credential pair and `store_path_style_access`.
-`tables` and `dynamic_route_field` must be present even when the selected mode does not use them.
+All options above are required except the credential pair, `store_path_style_access`, and the unused routing field.
+Static routing requires a non-empty `tables` list; dynamic routing requires a non-empty `dynamic_route_field`.
## Static Routing
diff --git a/core/connectors/sinks/iceberg_sink/src/lib.rs b/core/connectors/sinks/iceberg_sink/src/lib.rs
index 6c6315968..f61ba1cd8 100644
--- a/core/connectors/sinks/iceberg_sink/src/lib.rs
+++ b/core/connectors/sinks/iceberg_sink/src/lib.rs
@@ -52,11 +52,13 @@ pub struct IcebergSink {
#[derive(Debug, Serialize, Deserialize)]
pub struct IcebergSinkConfig {
+ #[serde(default)]
pub tables: Vec<String>,
pub catalog_type: IcebergSinkTypes,
pub warehouse: String,
pub uri: String,
pub dynamic_routing: bool,
+ #[serde(default)]
pub dynamic_route_field: String,
pub store_url: String,
pub store_access_key_id: Option<String>,
@@ -84,3 +86,42 @@ impl IcebergSink {
IcebergSink { id, config, router }
}
}
+
+#[cfg(test)]
+mod tests {
+ use iggy_connector_sdk::Sink;
+
+ use super::*;
+
+ #[tokio::test]
+ async fn omitted_routing_fields_deserialize_but_the_active_mode_requires_configuration() {
+ for dynamic_routing in [false, true] {
+ let mut json = simd_json::to_vec(&simd_json::json!({
+ "catalog_type": "rest",
+ "warehouse": "warehouse",
+ "uri": "http://localhost:1",
+ "dynamic_routing": dynamic_routing,
+ "store_url": "http://localhost:1",
+ "store_region": "us-east-1",
+ "store_class": "s3"
+ }))
+ .unwrap();
+ let config: IcebergSinkConfig =
+ simd_json::from_slice(&mut json).expect("Unused routing fields must be optional");
+ assert!(config.tables.is_empty());
+ assert!(config.dynamic_route_field.is_empty());
+ let mut sink = IcebergSink::new(1, config);
+ let error = sink
+ .open()
+ .await
+ .expect_err("The active routing mode needs a destination");
+ assert!(matches!(error, Error::InvalidConfigValue(_)), "{error}");
+ let field = if dynamic_routing {
+ "dynamic_route_field"
+ } else {
+ "tables"
+ };
+ assert!(error.to_string().contains(field), "{error}");
+ }
+ }
+}
diff --git a/core/connectors/sinks/iceberg_sink/src/sink.rs b/core/connectors/sinks/iceberg_sink/src/sink.rs
index 7ec7ef0b9..1dd33cd39 100644
--- a/core/connectors/sinks/iceberg_sink/src/sink.rs
+++ b/core/connectors/sinks/iceberg_sink/src/sink.rs
@@ -28,6 +28,16 @@ use tracing::{debug, error, info};
#[async_trait]
impl Sink for IcebergSink {
async fn open(&mut self) -> Result<(), Error> {
+ if self.config.dynamic_routing && self.config.dynamic_route_field.trim().is_empty() {
+ return Err(Error::InvalidConfigValue(
+ "dynamic_route_field must not be empty in dynamic routing mode".to_string(),
+ ));
+ }
+ if !self.config.dynamic_routing && self.config.tables.is_empty() {
+ return Err(Error::InvalidConfigValue(
+ "tables must not be empty in static routing mode".to_string(),
+ ));
+ }
match (
&self.config.store_access_key_id,
&self.config.store_secret_access_key,
diff --git a/core/connectors/sinks/influxdb_sink/src/lib.rs b/core/connectors/sinks/influxdb_sink/src/lib.rs
index cef1fd609..c2e5ec122 100644
--- a/core/connectors/sinks/influxdb_sink/src/lib.rs
+++ b/core/connectors/sinks/influxdb_sink/src/lib.rs
@@ -31,7 +31,7 @@ use iggy_connector_sdk::{
use reqwest::Url;
use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderValue};
use reqwest_middleware::ClientWithMiddleware;
-use secrecy::{ExposeSecret, SecretBox, SecretString};
+use secrecy::{ExposeSecret, SecretString};
use serde::{Deserialize, Serialize};
use std::fmt::Write as _;
use std::sync::Arc;
@@ -348,7 +348,7 @@ impl InfluxDbSinkConfig {
/// `measurement`, `precision`, `include_*`, `batch_size_limit`.
///
/// **Open-time fields** (populated in `open()`, guarded by `Option<T>`):
-/// `client`, `write_url`, `auth_header` — callers must invoke `open()` before
+/// `client`, `write_url` - callers must invoke `open()` before
/// any `process_batch()` call; `get_client()` returns an error otherwise.
#[derive(Debug)]
pub struct InfluxDbSink {
@@ -356,7 +356,6 @@ pub struct InfluxDbSink {
config: InfluxDbSinkConfig,
client: Option<ClientWithMiddleware>,
write_url: Option<Url>,
- auth_header: Option<SecretBox<String>>,
circuit_breaker: Arc<CircuitBreaker>,
messages_attempted: AtomicU64,
write_success: AtomicU64,
@@ -433,7 +432,6 @@ impl InfluxDbSink {
config,
client: None,
write_url: None,
- auth_header: None,
circuit_breaker,
messages_attempted: AtomicU64::new(0),
write_success: AtomicU64::new(0),
@@ -457,7 +455,9 @@ impl InfluxDbSink {
let timeout = parse_duration(self.config.timeout(), DEFAULT_TIMEOUT);
let mut authorization =
HeaderValue::from_str(&self.config.auth_header()).map_err(|error| {
- Error::InvalidConfigValue(format!("Invalid InfluxDB authorization header: {error}"))
+ Error::InvalidConfigValue(format!(
+ "InfluxDB token contains characters that are invalid in an HTTP header: {error}"
+ ))
})?;
authorization.set_sensitive(true);
let mut headers = HeaderMap::new();
@@ -670,17 +670,8 @@ impl InfluxDbSink {
let url = self.write_url.as_ref().ok_or_else(|| {
Error::Connection("write_url not initialized — call open() first".to_string())
})?;
- let auth = self
- .auth_header
- .as_ref()
- .map(|s| s.expose_secret().as_str())
- .ok_or_else(|| {
- Error::Connection("auth_header not initialised — was open() called?".to_string())
- })?;
-
let response = client
.post(url.as_str())
- .header("Authorization", auth)
.header("Content-Type", "text/plain; charset=utf-8")
// into_bytes() hands the Vec<u8> directly to Bytes without copying.
.body(Bytes::from(body.into_bytes()))
@@ -775,7 +766,6 @@ impl Sink for InfluxDbSink {
));
self.write_url = Some(self.config.build_write_url()?);
- self.auth_header = Some(SecretBox::new(Box::new(self.config.auth_header())));
info!("InfluxDB sink ID: {} opened successfully", self.id);
Ok(())
diff --git a/core/connectors/sinks/postgres_sink/README.md b/core/connectors/sinks/postgres_sink/README.md
index 1d20d764a..3d4b14587 100644
--- a/core/connectors/sinks/postgres_sink/README.md
+++ b/core/connectors/sinks/postgres_sink/README.md
@@ -13,7 +13,7 @@ The PostgreSQL sink connector consumes messages from Iggy topics and stores them
## Configuration
-Build from the matching 0.9.0/edge checkout root:
+Build from the matching Iggy checkout root:
```bash
cargo build --release -p iggy_connector_postgres_sink
@@ -59,7 +59,7 @@ payload_format = "bytea"
| ------ | ---- | ------- | ----------- |
| `connection_string` | string | required | PostgreSQL connection string |
| `target_table` | string | required | One quoted table identifier; `schema.table` is a literal name containing a dot |
-| `batch_size` | u32 | `100` | Maximum messages per insert statement; `0` behaves as `1` |
+| `batch_size` | u32 | `100` | Maximum messages per insert statement, capped by PostgreSQL's bind-parameter limit; `0` behaves as `1` |
| `max_connections` | u32 | `10` | Max database connections |
| `auto_create_table` | bool | `false` | Create a missing table; existing tables are not migrated |
| `include_metadata` | bool | `true` | Include Iggy metadata columns |
@@ -187,10 +187,10 @@ CREATE INDEX idx_payload_gin ON iggy_messages USING GIN (payload);
- Adjust `max_connections` based on PostgreSQL's `max_connections` setting
- Use `poll_interval` to control how often the sink checks for new messages
- Use `payload_format = "json"` for JSON data to enable native querying
-- Stay within PostgreSQL's 65,535 bind-parameter limit: each row uses
- `2 + 5 * include_metadata + include_checksum + include_origin_timestamp`
- parameters. With default flags, at most 7,281 rows fit in one statement.
- Larger actual chunks fail; the runtime's `batch_length` also limits poll size.
+
+The sink caps each statement at PostgreSQL's 65,535 bind parameters based on
+the enabled columns. With default flags, at most 7,281 rows fit in one statement.
+The runtime's `batch_length` also limits poll size.
## Example Configs
@@ -273,10 +273,11 @@ loop.
Each current-poll chunk becomes one multi-row `INSERT`. A bad payload, duplicate
primary key, incompatible table or other terminal failure rejects that chunk.
-Later chunks are still attempted. Failures are logged and added to a private
-insertion-error count, but `consume()` returns success and counts every attempted
-message as processed. Runtime processed counts can include unstored messages,
-and runtime error counts do not expose these insert failures.
+Later chunks are still attempted. Failures are logged and added to the sink's
+insertion-error count, and `consume()` returns the last chunk error. The sink
+counts only successfully inserted chunks as processed. The runtime counts
+callbacks, so a partially successful callback is reported as an error even when
+some rows were stored.
The query has no `ON CONFLICT` or upsert clause. An ID already in the table
rejects the whole chunk, including new IDs alongside it. There is no transaction
diff --git a/core/connectors/sinks/postgres_sink/src/lib.rs b/core/connectors/sinks/postgres_sink/src/lib.rs
index 9f0ac08f0..cbaac94eb 100644
--- a/core/connectors/sinks/postgres_sink/src/lib.rs
+++ b/core/connectors/sinks/postgres_sink/src/lib.rs
@@ -34,6 +34,7 @@ sink_connector!(PostgresSink);
const DEFAULT_MAX_RETRIES: u32 = 3;
const DEFAULT_RETRY_DELAY: &str = "1s";
+const MAX_BIND_PARAMETERS: usize = u16::MAX as usize;
#[derive(Debug)]
pub struct PostgresSink {
@@ -231,23 +232,36 @@ impl PostgresSink {
messages: &[ConsumedMessage],
) -> Result<(), Error> {
let pool = self.get_pool()?;
- let batch_size = self.config.batch_size.unwrap_or(100).max(1) as usize;
+ let params_per_row = Self::insert_columns(
+ self.config.include_metadata.unwrap_or(true),
+ self.config.include_checksum.unwrap_or(true),
+ self.config.include_origin_timestamp.unwrap_or(true),
+ )
+ .len();
+ let batch_size = (self.config.batch_size.unwrap_or(100).max(1) as usize)
+ .min(MAX_BIND_PARAMETERS / params_per_row);
+ let mut last_error = None;
+ let mut msg_count = 0;
+ let mut insertion_errors = 0;
for batch in messages.chunks(batch_size) {
- if let Err(e) = self
+ match self
.insert_batch(batch, topic_metadata, messages_metadata, pool)
.await
{
- let mut state = self.state.lock().await;
- state.insertion_errors += batch.len() as u64;
- error!("Failed to insert batch: {e}");
+ Ok(()) => msg_count += batch.len(),
+ Err(error) => {
+ insertion_errors += batch.len();
+ error!("Failed to insert batch: {error}");
+ last_error = Some(error);
+ }
}
}
let mut state = self.state.lock().await;
- state.messages_processed += messages.len() as u64;
+ state.messages_processed += msg_count as u64;
+ state.insertion_errors += insertion_errors as u64;
- let msg_count = messages.len();
let table = &self.config.target_table;
if self.verbose {
info!(
@@ -261,7 +275,7 @@ impl PostgresSink {
);
}
- Ok(())
+ last_error.map_or(Ok(()), Err)
}
async fn insert_batch(
@@ -458,31 +472,13 @@ impl PostgresSink {
row_count: usize,
) -> Result<(String, u32), Error> {
let quoted_table = quote_identifier(table_name)?;
- let mut query = format!("INSERT INTO {quoted_table} (id");
-
- let mut params_per_row: u32 = 1; // id
-
- if include_metadata {
- query.push_str(
- ", iggy_offset, iggy_timestamp, iggy_stream, iggy_topic, iggy_partition_id",
- );
- params_per_row += 5;
- }
-
- if include_checksum {
- query.push_str(", iggy_checksum");
- params_per_row += 1;
- }
-
- if include_origin_timestamp {
- query.push_str(", iggy_origin_timestamp");
- params_per_row += 1;
- }
-
- query.push_str(", payload");
- params_per_row += 1;
-
- query.push_str(") VALUES ");
+ let columns =
+ Self::insert_columns(include_metadata, include_checksum, include_origin_timestamp);
+ let params_per_row = columns.len() as u32;
+ let mut query = format!(
+ "INSERT INTO {quoted_table} ({}) VALUES ",
+ columns.join(", ")
+ );
let mut value_groups = Vec::with_capacity(row_count);
for row_idx in 0..row_count {
@@ -499,6 +495,31 @@ impl PostgresSink {
Ok((query, params_per_row))
}
+
+ fn insert_columns(
+ include_metadata: bool,
+ include_checksum: bool,
+ include_origin_timestamp: bool,
+ ) -> Vec<&'static str> {
+ let mut columns = vec!["id"];
+ if include_metadata {
+ columns.extend([
+ "iggy_offset",
+ "iggy_timestamp",
+ "iggy_stream",
+ "iggy_topic",
+ "iggy_partition_id",
+ ]);
+ }
+ if include_checksum {
+ columns.push("iggy_checksum");
+ }
+ if include_origin_timestamp {
+ columns.push("iggy_origin_timestamp");
+ }
+ columns.push("payload");
+ columns
+ }
}
fn is_transient_error(e: &sqlx::Error) -> bool {
@@ -562,6 +583,49 @@ mod tests {
}
}
+ #[test]
+ fn given_invalid_json_when_consumed_should_report_failure_without_counting_success() {
+ let runtime = tokio::runtime::Runtime::new().expect("Failed to create test runtime");
+ runtime.block_on(async {
+ let mut config = test_config();
+ config.payload_format = Some("json".to_string());
+ let pool = PgPoolOptions::new()
+ .connect_lazy(config.connection_string.expose_secret())
+ .expect("Failed to configure lazy test pool");
+ let mut sink = PostgresSink::new(1, config);
+ sink.pool = Some(pool);
+ let result = sink
+ .consume(
+ &TopicMetadata {
+ stream: "events".to_string(),
+ topic: "messages".to_string(),
+ },
+ MessagesMetadata {
+ partition_id: 0,
+ current_offset: 0,
+ schema: iggy_connector_sdk::Schema::Raw,
+ },
+ vec![ConsumedMessage {
+ id: 1,
+ offset: 0,
+ checksum: 0,
+ timestamp: 0,
+ origin_timestamp: 0,
+ headers: None,
+ payload: iggy_connector_sdk::Payload::Raw(b"invalid JSON".to_vec()),
+ }],
+ )
+ .await;
+ assert!(
+ matches!(result, Err(Error::CannotStoreData(_))),
+ "{result:?}"
+ );
+ let state = sink.state.lock().await;
+ assert_eq!(state.messages_processed, 0);
+ assert_eq!(state.insertion_errors, 1);
+ });
+ }
+
#[test]
fn given_zero_batch_size_should_consume_empty_batch() {
let runtime = tokio::runtime::Runtime::new().expect("Failed to create test runtime");
diff --git a/core/connectors/sinks/quickwit_sink/README.md b/core/connectors/sinks/quickwit_sink/README.md
index fb6f7a30a..40b5971d6 100644
--- a/core/connectors/sinks/quickwit_sink/README.md
+++ b/core/connectors/sinks/quickwit_sink/README.md
@@ -12,7 +12,7 @@ The Quickwit connector sends data to the Quickwit API using HTTP. It checks read
| `max_retries` | `3` | Total HTTP attempts including the first; `0` and `1` both allow one attempt. |
| `retry_delay` | `"1s"` | Base exponential delay for HTTP retries and readiness probes. |
| `retry_max_delay` | `"5s"` | Cap for calculated HTTP retry delays; a valid `Retry-After` on HTTP 429 overrides it. |
-| `max_open_retries` | `10` | Total attempts per readiness check including the first; `0` and `1` both allow one attempt. |
+| `max_open_retries` | `10` | One initial probe per readiness check, with up to `max_open_retries - 1` retries shared between them; `0` and `1` disable retries. |
| `open_retry_max_delay` | `"30s"` | Maximum delay between readiness probes. |
| `timeout` | `"30s"` | Timeout per HTTP attempt; retries and waits can extend the complete operation. |
@@ -28,7 +28,7 @@ verbose_logging = false
max_retries = 3
retry_delay = "1s"
retry_max_delay = "5s"
-# Total attempts per readiness check including the first; 1 disables retries.
+# One initial probe per readiness check; share up to max_open_retries - 1 retries.
max_open_retries = 10
open_retry_max_delay = "30s"
timeout = "30s"
@@ -110,7 +110,8 @@ This is intentional: legacy index metadata can exist before the ingest queue is
The probe uses `commit=auto`, so it adds no documents and does not force a commit; see Quickwit's [ingest implementation](https://github.com/quickwit-oss/quickwit/blob/v0.8.2/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs) and [legacy/V2 routing](https://github.com/quickwit-oss/quickwit/blob/v0.9.0/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs).
Ingest V2 accepts empty requests without checking shard readiness, so this probe does not guarantee that the first data request will succeed.
Index readiness retries HTTP 404, 429, 5xx and network failures.
-Each readiness check uses `max_open_retries` and `open_retry_max_delay`; these probes submit no documents.
+The two checks share up to `max_open_retries - 1` retries, in addition to one initial probe each.
+Values of `0` or `1` disable retries. Delays use `open_retry_max_delay`; these probes submit no documents.
The sink cannot guarantee at-least-once delivery. The runtime commits offsets when polling, logs/counts plugin callback errors and continues without replaying the failed batch. A permanent error or exhausted retry budget is logged, but affected messages are not redelivered. The runtime's processed-message count does not prove successful indexing.
diff --git a/core/connectors/sinks/quickwit_sink/config.toml b/core/connectors/sinks/quickwit_sink/config.toml
index b947e2db6..4e29a09a0 100644
--- a/core/connectors/sinks/quickwit_sink/config.toml
+++ b/core/connectors/sinks/quickwit_sink/config.toml
@@ -39,7 +39,7 @@ verbose_logging = true
max_retries = 2
retry_delay = "250ms"
retry_max_delay = "2s"
-# Total attempts per readiness check including the first; 1 disables retries.
+# One initial probe per readiness check; share up to max_open_retries - 1 retries.
max_open_retries = 8
open_retry_max_delay = "10s"
timeout = "10s"
diff --git a/core/connectors/sinks/quickwit_sink/src/lib.rs b/core/connectors/sinks/quickwit_sink/src/lib.rs
index 7d537c74d..61e013a90 100644
--- a/core/connectors/sinks/quickwit_sink/src/lib.rs
+++ b/core/connectors/sinks/quickwit_sink/src/lib.rs
@@ -20,8 +20,7 @@ use std::time::Duration;
use async_trait::async_trait;
use base64::{Engine as _, engine::general_purpose};
use iggy_connector_sdk::retry::{
- RetryPolicy, build_retry_client, check_connectivity_with_retry, is_transient_status,
- retry_async,
+ RetryPolicy, build_retry_client, check_connectivity, is_transient_status, retry_async,
};
use iggy_connector_sdk::{
ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector,
@@ -73,7 +72,9 @@ pub struct QuickwitSinkConfig {
pub retry_delay: Option<String>,
/// Maximum retry delay cap as a human-readable duration string, e.g. "5s" (default: 5s).
pub retry_max_delay: Option<String>,
- /// Total attempts per readiness check including the first (default: 10). 1 disables retries.
+ /// Startup attempt budget shared by health and ingest readiness (default: 10).
+ /// Each check runs once, with up to max_open_retries - 1 retries shared between them.
+ /// Values of 0 or 1 disable retries.
pub max_open_retries: Option<u32>,
/// Maximum retry delay cap when opening the sink, e.g. "30s" (default: 30s).
pub open_retry_max_delay: Option<String>,
@@ -405,7 +406,7 @@ impl Sink for QuickwitSink {
.timeout(timeout)
.build()
.map_err(|error| Error::InitError(format!("reqwest client: {error}")))?;
- let open_retry_policy = RetryPolicy {
+ let mut open_retry_policy = RetryPolicy {
max_attempts: self
.config
.max_open_retries
@@ -414,14 +415,26 @@ impl Sink for QuickwitSink {
base_delay: retry_delay,
max_delay: open_retry_max_delay,
};
- check_connectivity_with_retry(
- &raw_client,
- endpoint_url(&base_url, &["health", "readyz"])?,
- "Quickwit sink",
- self.id,
+ let health_url = endpoint_url(&base_url, &["health", "readyz"])?;
+ let mut health_attempts = 0;
+ retry_async(
open_retry_policy,
+ &format!("Quickwit sink ID {} startup connectivity", self.id),
+ |_| true,
+ || {
+ health_attempts += 1;
+ check_connectivity(&raw_client, health_url.clone(), "Quickwit sink")
+ },
)
- .await?;
+ .await
+ .map_err(|failure| {
+ error!(
+ "Quickwit sink ID {} startup connectivity: {failure}",
+ self.id
+ );
+ failure.into_error()
+ })?;
+ open_retry_policy.max_attempts -= health_attempts - 1;
self.client = Some(build_retry_client(
raw_client.clone(),
@@ -824,6 +837,33 @@ mod tests {
});
}
+ #[test]
+ fn given_health_retries_when_ingest_is_unavailable_should_share_the_startup_budget() {
+ let runtime = test_runtime();
+ runtime.block_on(async {
+ let (url, server) = start_test_server(vec![
+ ("GET /health/readyz HTTP/1.1", 503, "not ready"),
+ ("GET /health/readyz HTTP/1.1", 200, ""),
+ ("GET /api/v1/indexes/test HTTP/1.1", 200, "{}"),
+ (
+ "POST /api/v1/test/ingest?commit=auto HTTP/1.1",
+ 503,
+ "not ready",
+ ),
+ ("POST /api/v1/test/ingest?commit=auto HTTP/1.1", 200, "{}"),
+ ])
+ .await;
+ let mut config = test_config();
+ config.url = url;
+ config.max_open_retries = Some(2);
+ let mut sink = QuickwitSink::new(1, config);
+ let result = sink.open().await;
+ server.abort();
+ assert!(matches!(result, Err(Error::InitError(_))), "{result:?}");
+ assert!(sink.client.is_none());
+ });
+ }
+
#[test]
fn given_unavailable_ingest_readiness_when_opened_should_stop_within_the_attempt_budget() {
let runtime = test_runtime();
diff --git a/core/connectors/sinks/s3_sink/README.md b/core/connectors/sinks/s3_sink/README.md
index bb25c7bd4..c3bf0c1aa 100644
--- a/core/connectors/sinks/s3_sink/README.md
+++ b/core/connectors/sinks/s3_sink/README.md
@@ -16,7 +16,7 @@ Writes messages from Iggy streams to Amazon S3 and S3-compatible object stores (
### Connector Runtime Config
-Build and start the matching 0.9.0/edge broker and connector runtime from the checkout root. Use the [sink guide](https://iggy.apache.org/docs/connectors/sinks/sink/) for broker credentials and the main runtime configuration. Save the following connector entry in its configured connector directory and append one `[plugin_config]` table. Create the bucket separately before startup.
+Build and start the broker and connector runtime from the matching Iggy checkout root. Use the [sink guide](https://iggy.apache.org/docs/connectors/sinks/sink/) for broker credentials and the main runtime configuration. Save the following connector entry in its configured connector directory and append one `[plugin_config]` table. Create the bucket separately before startup.
```toml
type = "sink"
@@ -111,7 +111,9 @@ The pinned `aws-creds` chain tries these sources in order:
For temporary key pairs, supply the token through the environment or shared credentials file. This is not the AWS SDK credential chain.
-Startup writes an empty object at bucket-root `.iggy-sink-probe`, ignoring `prefix`, and then attempts deletion. It requires `PutObject` on the probe key, not `ListBucket`; non-success HTTP statuses prevent startup. Delete permission is optional and cleanup errors are ignored. Reserve this key: any existing object there is overwritten and, if cleanup succeeds, deleted.
+Startup initiates a multipart upload at `<prefix>/.iggy-sink-probe` (bucket root when the prefix is empty) and aborts it before consuming messages. No parts are uploaded and no object is published or deleted. Both steps must succeed: missing buckets, denied writes or failed cleanup prevent startup. The credentials need `s3:PutObject` and `s3:AbortMultipartUpload` on the probe key, plus `s3:PutObject` on data and loss-marker keys. Neither `ListBucket` nor `DeleteObject` is required.
+
+Each probe step uses `max_attempts` and `retry_delay` for HTTP 408, 429 and 5xx failures. Abort also retries transport failures. Initiation does not add retries for an ambiguous transport failure because the upload ID may be lost; rust-s3 can still retry internally. A crash or lost response can leave an incomplete upload, so configure an `AbortIncompleteMultipartUpload` lifecycle rule. Failed aborts report the upload ID for cleanup.
## Output Example
@@ -162,7 +164,7 @@ secret_access_key = "..."
The runtime auto-commits while polling, logs/counts a failed plugin callback, and continues without replaying that batch. A successful callback may only mean that messages were buffered; processed counts do not establish S3 delivery. A crash loses unflushed buffers.
-The sink retries HTTP 408, 429, 5xx and client errors, using exponential backoff with jitter capped at 60 seconds. Other HTTP statuses stop that upload. `Retry-After` is not parsed. The S3 library also retries transport errors once internally and has a 60-second request timeout, so `max_attempts` counts outer `PutObject` calls, not individual network requests or a total time budget. The startup probe does not use the sink's upload retry loop.
+The sink retries HTTP 408, 429 and 5xx responses, plus rust-s3 request errors, using exponential backoff with jitter capped at 60 seconds. Other HTTP statuses stop that upload. `Retry-After` is not parsed. The S3 library also retries transport errors once internally and has a 60-second request timeout, so `max_attempts` counts outer `PutObject` calls, not individual network requests or a total time budget.
## Known Limitations
diff --git a/core/connectors/sinks/s3_sink/src/buffer.rs b/core/connectors/sinks/s3_sink/src/buffer.rs
index ad49e2651..eebc0df98 100644
--- a/core/connectors/sinks/s3_sink/src/buffer.rs
+++ b/core/connectors/sinks/s3_sink/src/buffer.rs
@@ -62,6 +62,16 @@ impl FileBuffer {
self.boundaries.is_empty()
}
+ pub fn byte_len(&self) -> usize {
+ self.data.len()
+ }
+
+ pub fn take_data(&mut self) -> Vec<u8> {
+ let data = std::mem::take(&mut self.data);
+ self.reset();
+ data
+ }
+
/// Returns an iterator over individual entry slices without copying.
pub fn entries(&self) -> impl Iterator<Item = &[u8]> {
let mut start = 0;
diff --git a/core/connectors/sinks/s3_sink/src/client.rs b/core/connectors/sinks/s3_sink/src/client.rs
index df14e80ad..50edaa7aa 100644
--- a/core/connectors/sinks/s3_sink/src/client.rs
+++ b/core/connectors/sinks/s3_sink/src/client.rs
@@ -78,33 +78,8 @@ pub(crate) async fn create_bucket(config: &S3SinkConfig) -> Result<Box<Bucket>,
Ok(bucket)
}
-/// Verify bucket connectivity with a zero-byte probe write. Only requires
-/// `s3:PutObject` permission, unlike `list_page` which needs `s3:ListBucket`
-/// and fails for write-only IAM policies.
-pub(crate) async fn verify_bucket(bucket: &Bucket) -> Result<(), Error> {
- const PROBE_KEY: &str = ".iggy-sink-probe";
- let response = bucket.put_object(PROBE_KEY, &[]).await.map_err(|e| {
- Error::InitError(format!(
- "S3 bucket '{}' connectivity check failed: {e}",
- bucket.name
- ))
- })?;
- let status = response.status_code();
- if !(200..300).contains(&status) {
- return Err(Error::InitError(format!(
- "S3 bucket '{}' connectivity check returned status {status}",
- bucket.name
- )));
- }
- let _ = bucket.delete_object(PROBE_KEY).await;
- Ok(())
-}
-
#[cfg(test)]
mod tests {
- use wiremock::matchers::method;
- use wiremock::{Mock, MockServer, ResponseTemplate};
-
use super::*;
use crate::{
FileRotation, default_max_file_size, default_output_format, default_path_template,
@@ -131,36 +106,6 @@ mod tests {
}
}
- #[test]
- fn given_probe_response_should_require_success_status() {
- let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
- runtime.block_on(async {
- for status in [200, 403, 404, 503] {
- let server = MockServer::start().await;
- Mock::given(method("PUT"))
- .respond_with(ResponseTemplate::new(status))
- .mount(&server)
- .await;
- let config = S3SinkConfig {
- endpoint: Some(server.uri()),
- access_key_id: Some("test-access-key".into()),
- secret_access_key: Some("test-secret-key".into()),
- ..base_config()
- };
- let bucket = create_bucket(&config).await.expect("Create S3 test client");
- let result = verify_bucket(&bucket).await;
- if status == 200 {
- assert!(result.is_ok(), "Successful probe must pass: {result:?}");
- } else {
- assert!(
- matches!(result, Err(Error::InitError(_))),
- "Failed probe status {status} must prevent initialization: {result:?}"
- );
- }
- }
- });
- }
-
#[test]
fn validate_both_credentials_present() {
let config = S3SinkConfig {
diff --git a/core/connectors/sinks/s3_sink/src/formatter.rs b/core/connectors/sinks/s3_sink/src/formatter.rs
index bec194c90..05b2b08d3 100644
--- a/core/connectors/sinks/s3_sink/src/formatter.rs
+++ b/core/connectors/sinks/s3_sink/src/formatter.rs
@@ -16,6 +16,7 @@
// under the License.
use crate::OutputFormat;
+use crate::buffer::FileBuffer;
use chrono::{DateTime, Utc};
use iggy_connector_sdk::{
ConsumedMessage, Error, MessagesMetadata, Payload, TopicMetadata, owned_value_to_serde_json,
@@ -192,31 +193,24 @@ fn timestamp_to_rfc3339(micros: u64) -> String {
}
/// Finalize buffer entries into the output byte format.
-pub(crate) fn finalize_buffer<'a>(
- entries: impl Iterator<Item = &'a [u8]>,
- format: OutputFormat,
-) -> Vec<u8> {
+pub(crate) fn finalize_buffer(buffer: &mut FileBuffer, format: OutputFormat) -> Vec<u8> {
match format {
OutputFormat::JsonLines => {
- let mut result = Vec::new();
- for entry in entries {
+ let mut result =
+ Vec::with_capacity(buffer.byte_len() + buffer.message_count() as usize);
+ for entry in buffer.entries() {
result.extend_from_slice(entry);
result.push(b'\n');
}
result
}
- OutputFormat::Raw => {
- let mut result = Vec::new();
- for entry in entries {
- result.extend_from_slice(entry);
- }
- result
- }
+ OutputFormat::Raw => buffer.take_data(),
OutputFormat::JsonArray => {
- let mut result = Vec::new();
+ let separators = (buffer.message_count() as usize).saturating_sub(1);
+ let mut result = Vec::with_capacity(buffer.byte_len() + separators + b"[]".len());
result.push(b'[');
let mut first = true;
- for entry in entries {
+ for entry in buffer.entries() {
if !first {
result.push(b',');
}
@@ -340,8 +334,8 @@ mod tests {
fn finalize_json_lines() {
let data = b"{\"a\":1}{\"b\":2}";
let boundaries = [7usize, 14];
- let entries = entries_from_boundaries(data, &boundaries);
- let result = finalize_buffer(entries, OutputFormat::JsonLines);
+ let mut buffer = buffer_from_boundaries(data, &boundaries);
+ let result = finalize_buffer(&mut buffer, OutputFormat::JsonLines);
assert_eq!(result, b"{\"a\":1}\n{\"b\":2}\n");
}
@@ -349,8 +343,14 @@ mod tests {
fn finalize_raw_no_delimiter() {
let data = b"\x00\x01\x02\x0a\xff\xfe";
let boundaries = [4usize, 6];
- let entries = entries_from_boundaries(data, &boundaries);
- let result = finalize_buffer(entries, OutputFormat::Raw);
+ let mut buffer = buffer_from_boundaries(data, &boundaries);
+ let original_allocation = buffer.entries().next().unwrap().as_ptr();
+ let result = finalize_buffer(&mut buffer, OutputFormat::Raw);
+ assert_eq!(
+ result.as_ptr(),
+ original_allocation,
+ "Raw output must reuse the buffer allocation"
+ );
assert_eq!(
result, data,
"Raw must concatenate without inserting delimiters"
@@ -361,8 +361,8 @@ mod tests {
fn finalize_json_array() {
let data = b"{\"a\":1}{\"b\":2}";
let boundaries = [7usize, 14];
- let entries = entries_from_boundaries(data, &boundaries);
- let result = finalize_buffer(entries, OutputFormat::JsonArray);
+ let mut buffer = buffer_from_boundaries(data, &boundaries);
+ let result = finalize_buffer(&mut buffer, OutputFormat::JsonArray);
assert_eq!(result, b"[{\"a\":1},{\"b\":2}]");
}
@@ -379,15 +379,13 @@ mod tests {
assert_eq!(ts, "1970-01-01T00:00:00Z");
}
- fn entries_from_boundaries<'a>(
- data: &'a [u8],
- boundaries: &'a [usize],
- ) -> impl Iterator<Item = &'a [u8]> {
+ fn buffer_from_boundaries(data: &[u8], boundaries: &[usize]) -> FileBuffer {
+ let mut buffer = FileBuffer::new();
let mut start = 0;
- boundaries.iter().map(move |&end| {
- let s = &data[start..end];
+ for (offset, &end) in boundaries.iter().enumerate() {
+ buffer.append(&data[start..end], offset as u64, 0);
start = end;
- s
- })
+ }
+ buffer
}
}
diff --git a/core/connectors/sinks/s3_sink/src/sink.rs b/core/connectors/sinks/s3_sink/src/sink.rs
index 66c990d36..fcf155f60 100644
--- a/core/connectors/sinks/s3_sink/src/sink.rs
+++ b/core/connectors/sinks/s3_sink/src/sink.rs
@@ -20,13 +20,16 @@ use crate::formatter;
use crate::path::{PathContext, render_s3_key};
use crate::{BufferKey, S3Sink};
use async_trait::async_trait;
-use iggy_connector_sdk::retry::retry_backoff;
+use iggy_connector_sdk::retry::{RetryPolicy, retry_async, retry_backoff};
use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata};
+use s3::error::S3Error;
use std::sync::Arc;
use std::time::Duration;
use tracing::{debug, error, info, warn};
const MAX_BACKOFF: Duration = Duration::from_secs(60);
+const PROBE_KEY: &str = ".iggy-sink-probe";
+const PROBE_CONTENT_TYPE: &str = "application/octet-stream";
struct FlushPayload {
data: Vec<u8>,
@@ -44,14 +47,7 @@ impl Sink for S3Sink {
self.validate_and_parse_config()?;
let bucket = crate::client::create_bucket(&self.config).await?;
-
- crate::client::verify_bucket(&bucket).await?;
-
- info!(
- "S3 sink ID: {} connected to bucket '{}' in region '{}'",
- self.id, self.config.bucket, self.config.region
- );
-
+ self.check_write_access(&bucket).await?;
self.bucket = Some(bucket);
info!(
@@ -190,6 +186,62 @@ impl Sink for S3Sink {
}
impl S3Sink {
+ async fn check_write_access(&self, bucket: &s3::Bucket) -> Result<(), Error> {
+ let prefix = self
+ .config
+ .prefix
+ .as_deref()
+ .unwrap_or_default()
+ .trim_matches('/');
+ let probe_key = if prefix.is_empty() {
+ PROBE_KEY.to_string()
+ } else {
+ format!("{prefix}/{PROBE_KEY}")
+ };
+ let resolved = self.resolved();
+ let policy = RetryPolicy {
+ max_attempts: resolved.max_attempts,
+ base_delay: resolved.retry_delay,
+ max_delay: MAX_BACKOFF,
+ };
+
+ // Initiation checks PutObject without publishing an object. A transport
+ // failure can lose the upload ID, so do not retry that ambiguous result.
+ let upload = retry_async(
+ policy,
+ &format!("S3 sink ID: {} initiate write probe", self.id),
+ |error| matches!(error, S3Error::HttpFailWithBody(status, _) if is_retriable_status(*status)),
+ || bucket.initiate_multipart_upload(&probe_key, PROBE_CONTENT_TYPE),
+ )
+ .await
+ .map_err(|error| Error::InitError(format!(
+ "S3 bucket '{}' write probe at '{probe_key}' failed: {error}", bucket.name
+ )))?;
+
+ if upload.upload_id.is_empty() {
+ return Err(Error::InitError(format!(
+ "S3 bucket '{}' write probe at '{probe_key}' returned an empty upload ID",
+ bucket.name
+ )));
+ }
+
+ retry_async(
+ policy,
+ &format!("S3 sink ID: {} abort write probe", self.id),
+ |error| match error {
+ S3Error::HttpFailWithBody(status, _) => is_retriable_status(*status),
+ S3Error::Reqwest(_) => true,
+ _ => false,
+ },
+ || bucket.abort_upload(&probe_key, &upload.upload_id),
+ )
+ .await
+ .map_err(|error| Error::InitError(format!(
+ "S3 bucket '{}' could not abort write probe at '{probe_key}' (upload ID '{}'); s3:AbortMultipartUpload permission is required: {error}",
+ bucket.name, upload.upload_id
+ )))
+ }
+
#[allow(clippy::too_many_arguments)]
async fn process_messages_inner(
&self,
@@ -242,7 +294,6 @@ impl S3Sink {
buffer: &mut FileBuffer,
) -> Result<FlushPayload, Error> {
let resolved = self.resolved();
- let data = formatter::finalize_buffer(buffer.entries(), resolved.output_format);
let ctx = PathContext {
stream: &key.stream,
@@ -264,6 +315,7 @@ impl S3Sink {
let first_offset = buffer.first_offset();
let last_offset = buffer.last_offset();
+ let data = formatter::finalize_buffer(buffer, resolved.output_format);
buffer.reset();
Ok(FlushPayload {
@@ -405,7 +457,7 @@ fn is_retriable_status(status: u16) -> bool {
#[cfg(test)]
mod tests {
use iggy_connector_sdk::{Payload, Schema};
- use wiremock::matchers::{method, path};
+ use wiremock::matchers::{any, method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate};
use super::*;
@@ -414,6 +466,32 @@ mod tests {
S3SinkConfig,
};
+ const TEST_UPLOAD_ID: &str = "test-upload-id";
+
+ fn probe_response(probe_key: &str, upload_id: &str) -> ResponseTemplate {
+ ResponseTemplate::new(200).set_body_string(format!(
+ "<InitiateMultipartUploadResult><Bucket>test-bucket</Bucket><Key>{probe_key}</Key><UploadId>{upload_id}</UploadId></InitiateMultipartUploadResult>"
+ ))
+ }
+
+ async fn mock_write_probe(server: &MockServer, probe_key: &str) {
+ let probe_path = format!("/test-bucket/{probe_key}");
+ Mock::given(method("POST"))
+ .and(path(&probe_path))
+ .and(query_param("uploads", ""))
+ .respond_with(probe_response(probe_key, TEST_UPLOAD_ID))
+ .expect(1)
+ .mount(server)
+ .await;
+ Mock::given(method("DELETE"))
+ .and(path(&probe_path))
+ .and(query_param("uploadId", TEST_UPLOAD_ID))
+ .respond_with(ResponseTemplate::new(204))
+ .expect(1)
+ .mount(server)
+ .await;
+ }
+
fn test_config() -> S3SinkConfig {
S3SinkConfig {
bucket: "test-bucket".to_string(),
@@ -436,19 +514,298 @@ mod tests {
}
#[test]
- fn given_failed_upload_should_count_each_lost_message_once() {
- const TOTAL_MESSAGES: u64 = 3;
- const ROTATION_MESSAGES: u64 = 2;
+ fn given_an_unwritable_bucket_when_opening_should_fail_before_consumption() {
+ let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
+ runtime.block_on(async {
+ for (status, expected_attempts) in [(403, 1), (404, 1), (503, 2)] {
+ let server = MockServer::start().await;
+ Mock::given(any())
+ .respond_with(ResponseTemplate::new(status))
+ .mount(&server)
+ .await;
+ let config = S3SinkConfig {
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ max_attempts: Some(2),
+ retry_delay: Some("1ms".to_string()),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+
+ let result = sink.open().await;
+ assert!(
+ matches!(result, Err(Error::InitError(_))),
+ "S3 status {status} must prevent startup: {result:?}"
+ );
+ assert!(
+ sink.bucket.is_none(),
+ "An unwritable sink must remain unopened"
+ );
+ let requests = server.received_requests().await.expect("record requests");
+ assert_eq!(requests.len(), expected_attempts);
+ assert!(requests.iter().all(|request| request.method == "POST"));
+ let state = sink.state.lock().await;
+ assert_eq!(state.messages_received, 0);
+ assert_eq!(state.messages_lost, 0);
+ }
+ });
+ }
+
+ #[test]
+ fn given_a_multipart_probe_when_opening_should_abort_before_uploading_data() {
let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
runtime.block_on(async {
- for previously_buffered in [0, 1] {
+ let server = MockServer::start().await;
+ mock_write_probe(&server, "allowed/events/.iggy-sink-probe").await;
+ Mock::given(method("PUT"))
+ .respond_with(ResponseTemplate::new(200))
+ .mount(&server)
+ .await;
+ let config = S3SinkConfig {
+ prefix: Some("/allowed/events/".to_string()),
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ path_template: "{topic}".to_string(),
+ file_rotation: FileRotation::Messages,
+ max_messages_per_file: Some(1),
+ output_format: "raw".to_string(),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ sink.open().await.expect("Sink should initialize");
+ let requests = server.received_requests().await.expect("record requests");
+ assert_eq!(requests.len(), 2, "Startup must initiate and abort the probe");
+ assert_eq!(requests[0].method, "POST");
+ assert_eq!(requests[1].method, "DELETE");
+
+ sink.consume(
+ &TopicMetadata {
+ stream: "events".to_string(),
+ topic: "messages".to_string(),
+ },
+ MessagesMetadata {
+ partition_id: 0,
+ current_offset: 0,
+ schema: Schema::Raw,
+ },
+ vec![ConsumedMessage {
+ id: 1,
+ offset: 0,
+ checksum: 0,
+ timestamp: 1_000_000,
+ origin_timestamp: 0,
+ headers: None,
+ payload: Payload::Raw(b"message".to_vec()),
+ }],
+ )
+ .await
+ .expect("Data upload should succeed after the probe is cleaned up");
+ sink.close().await.expect("Sink should close");
+
+ let requests = server.received_requests().await.expect("record requests");
+ assert_eq!(requests.len(), 3, "Only the data object should be written after the probe");
+ assert_eq!(requests[2].method, "PUT");
+ assert_eq!(requests[2].body, b"message");
+ assert_eq!(
+ requests[2].url.path(),
+ "/test-bucket/allowed/events/messages/00000-00000000000000000000-00000000000000000000.bin"
+ );
+ let state = sink.state.lock().await;
+ assert_eq!(state.messages_uploaded, 1);
+ assert_eq!(state.messages_lost, 0);
+ });
+ }
+
+ #[test]
+ fn given_a_probe_prefix_when_opening_should_use_the_configured_write_scope() {
+ let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
+ runtime.block_on(async {
+ for (prefix, probe_key) in [
+ (None, ".iggy-sink-probe"),
+ (Some(""), ".iggy-sink-probe"),
+ (Some("/"), ".iggy-sink-probe"),
+ (Some("/allowed/events/"), "allowed/events/.iggy-sink-probe"),
+ ] {
+ let server = MockServer::start().await;
+ mock_write_probe(&server, probe_key).await;
+ let config = S3SinkConfig {
+ prefix: prefix.map(str::to_string),
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+
+ sink.open()
+ .await
+ .expect("Probe should succeed within the configured prefix");
+ }
+ });
+ }
+
+ #[test]
+ fn given_transient_probe_failures_when_opening_should_retry_the_failed_step() {
+ let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
+ runtime.block_on(async {
+ for operation in ["POST", "DELETE"] {
+ for status in [408, 429, 503] {
+ let server = MockServer::start().await;
+ Mock::given(method(operation))
+ .and(path("/test-bucket/data/.iggy-sink-probe"))
+ .respond_with(ResponseTemplate::new(status))
+ .up_to_n_times(1)
+ .expect(1)
+ .mount(&server)
+ .await;
+ mock_write_probe(&server, "data/.iggy-sink-probe").await;
+ let config = S3SinkConfig {
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ max_attempts: Some(2),
+ retry_delay: Some("1ms".to_string()),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+
+ sink.open()
+ .await
+ .expect("Transient probe failure should recover");
+ }
+ }
+ });
+ }
+
+ #[test]
+ fn given_a_failed_probe_abort_when_opening_should_refuse_startup() {
+ let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
+ runtime.block_on(async {
+ for (status, expected_attempts) in [(403, 1), (503, 2)] {
let server = MockServer::start().await;
+ Mock::given(method("POST"))
+ .and(path("/test-bucket/data/.iggy-sink-probe"))
+ .and(query_param("uploads", ""))
+ .respond_with(probe_response("data/.iggy-sink-probe", TEST_UPLOAD_ID))
+ .expect(1)
+ .mount(&server)
+ .await;
+ Mock::given(method("DELETE"))
+ .and(path("/test-bucket/data/.iggy-sink-probe"))
+ .and(query_param("uploadId", TEST_UPLOAD_ID))
+ .respond_with(ResponseTemplate::new(status))
+ .expect(expected_attempts)
+ .mount(&server)
+ .await;
+ let config = S3SinkConfig {
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ max_attempts: Some(2),
+ retry_delay: Some("1ms".to_string()),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+
+ let result = sink.open().await;
+
+ assert!(matches!(result, Err(Error::InitError(_))), "{result:?}");
+ assert!(sink.bucket.is_none(), "Cleanup must succeed before opening");
+ }
+ });
+ }
+
+ #[test]
+ fn given_an_empty_upload_id_when_probing_should_fail_without_a_delete_request() {
+ let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
+ runtime.block_on(async {
+ let server = MockServer::start().await;
+ Mock::given(method("POST"))
+ .respond_with(probe_response("data/.iggy-sink-probe", ""))
+ .mount(&server)
+ .await;
+ let config = S3SinkConfig {
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+
+ let result = sink.open().await;
+
+ assert!(matches!(result, Err(Error::InitError(_))), "{result:?}");
+ let requests = server.received_requests().await.expect("record requests");
+ assert_eq!(
+ requests.len(),
+ 1,
+ "An abort must identify the multipart upload"
+ );
+ assert_eq!(requests[0].method, "POST");
+ });
+ }
+
+ #[test]
+ fn given_upload_status_when_writing_should_retry_only_transient_failures() {
+ let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
+ runtime.block_on(async {
+ for status in [200, 403, 404, 408, 429, 503] {
+ let server = MockServer::start().await;
+ mock_write_probe(&server, "data/.iggy-sink-probe").await;
+ Mock::given(method("PUT"))
+ .and(path("/test-bucket/data/message.bin"))
+ .respond_with(ResponseTemplate::new(status))
+ .up_to_n_times(1)
+ .expect(1)
+ .mount(&server)
+ .await;
Mock::given(method("PUT"))
- .and(path("/test-bucket/.iggy-sink-probe"))
+ .and(path("/test-bucket/data/message.bin"))
.respond_with(ResponseTemplate::new(200))
+ .expect(u64::from(matches!(status, 408 | 429 | 503)))
.mount(&server)
.await;
+ let config = S3SinkConfig {
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ max_attempts: Some(2),
+ retry_delay: Some("1ms".to_string()),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ sink.open()
+ .await
+ .expect("Sink should initialize before uploading");
+ let bucket = sink.bucket.as_ref().expect("S3 client should be available");
+ let result = sink
+ .upload_with_retry(bucket, "data/message.bin", b"payload")
+ .await;
+ if matches!(status, 403 | 404) {
+ assert!(
+ matches!(result, Err(Error::CannotStoreData(_))),
+ "{result:?}"
+ );
+ } else {
+ assert!(result.is_ok(), "status {status}: {result:?}");
+ }
+ }
+ });
+ }
+
+ #[test]
+ fn given_failed_upload_should_count_each_lost_message_once() {
+ const TOTAL_MESSAGES: u64 = 3;
+ const ROTATION_MESSAGES: u64 = 2;
+
+ let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
+ runtime.block_on(async {
+ for previously_buffered in [0, 1] {
+ let server = MockServer::start().await;
+ mock_write_probe(&server, "data/.iggy-sink-probe").await;
let config = S3SinkConfig {
endpoint: Some(server.uri()),
access_key_id: Some("test-access-key".into()),
@@ -460,7 +817,7 @@ mod tests {
..test_config()
};
let mut sink = S3Sink::new(1, config);
- sink.open().await.expect("Successful startup probe");
+ sink.open().await.expect("Sink should initialize");
let topic = TopicMetadata {
stream: "events".to_string(),
topic: "messages".to_string(),
diff --git a/core/connectors/sinks/surrealdb_sink/README.md b/core/connectors/sinks/surrealdb_sink/README.md
index 9a8c731d0..4a57b93cc 100644
--- a/core/connectors/sinks/surrealdb_sink/README.md
+++ b/core/connectors/sinks/surrealdb_sink/README.md
@@ -14,7 +14,7 @@ the same identities, but do not ensure that a failed batch will be delivered.
## Configuration
-From the matching 0.9.0/edge Iggy checkout root, build the plugin:
+From the matching Iggy checkout root, build the plugin:
```bash
cargo build --release -p iggy_connector_surrealdb_sink
diff --git a/core/connectors/sources/elasticsearch_source/Cargo.toml b/core/connectors/sources/elasticsearch_source/Cargo.toml
index 03c9101b5..c4051aa89 100644
--- a/core/connectors/sources/elasticsearch_source/Cargo.toml
+++ b/core/connectors/sources/elasticsearch_source/Cargo.toml
@@ -49,5 +49,9 @@ simd-json = { workspace = true }
tokio = { workspace = true }
tracing = { workspace = true }
+[dev-dependencies]
+tempfile = { workspace = true }
+wiremock = { workspace = true }
+
[lints]
workspace = true
diff --git a/core/connectors/sources/elasticsearch_source/README.md b/core/connectors/sources/elasticsearch_source/README.md
index a49a9c2bb..10fd3af5f 100644
--- a/core/connectors/sources/elasticsearch_source/README.md
+++ b/core/connectors/sources/elasticsearch_source/README.md
@@ -1,20 +1,30 @@
# Elasticsearch Source Connector with State Management
-This Elasticsearch source connector provides comprehensive state management capabilities to track processing progress and enable fault-tolerant data ingestion.
+This connector polls Elasticsearch documents and returns JSON messages with a
+checkpoint to the Iggy connectors runtime.
## Features
-- **Incremental Data Processing**: Track last processed timestamp to avoid reprocessing data
-- **Cursor-based Pagination**: Support for document ID-based cursors
-- **Scroll-based Pagination**: Support for Elasticsearch scroll API
-- **Error Tracking**: Monitor error counts and last error messages
-- **Processing Statistics**: Track performance metrics and processing times
-- **Persistent State Storage**: Multiple storage backends (file, Elasticsearch, Redis)
-- **Auto-save**: Configurable automatic state persistence
-- **State Recovery**: Resume processing from last known position after restart
+- **Incremental Data Processing**: Use a top-level RFC3339 timestamp string as a watermark, subject to the timestamp limitations below.
+- **Error Tracking**: Monitor error counts and last error messages.
+- **Processing Statistics**: Track fetched documents, payload bytes and successful-poll durations, including empty polls.
+- **Persistent State Storage**: Return MessagePack checkpoints to the runtime's file or HTTP state backend.
+- **State Recovery**: Restore the runtime checkpoint on restart; optional plugin JSON snapshots can override it.
## Configuration
+From the matching 0.9.0/edge Iggy checkout root, build the plugin:
+
+```bash
+cargo build --release -p iggy_connector_elasticsearch_source
+```
+
+Use the [source guide](https://iggy.apache.org/docs/connectors/sources/source/)
+for the broker credentials and main runtime configuration. The
+[Elasticsearch source walkthrough](https://iggy.apache.org/docs/connectors/sources/elasticsearch/)
+provides local backend, index and Iggy CLI commands. Create a matching index
+before starting the connector and start the runtime from the checkout root.
+
### Basic Configuration
```toml
@@ -38,238 +48,105 @@ index = "logs-*"
polling_interval = "30s"
batch_size = 100
timestamp_field = "@timestamp"
-query = {
- "match_all": {}
-}
-```
-
-### State Management Configuration
-```toml
-[plugin_config]
-# ... basic config ...
-state = {
- enabled = true
- storage_type = "file" # "file", "elasticsearch", "redis"
- storage_config = {
- base_path = "./connector_states" # for file storage
- # index = "connector_states" # for elasticsearch storage
- # url = "redis://localhost:6379" # for redis storage
- }
- state_id = "elasticsearch_logs_connector"
- auto_save_interval = "5m"
- tracked_fields = [
- "last_poll_timestamp",
- "last_document_id",
- "total_documents_fetched"
- ]
-}
+[plugin_config.query.match_all]
```
-## State Information
-
-The connector tracks the following state information:
-
-### Processing State
-
-- `last_poll_timestamp`: Last successful poll timestamp
-- `total_documents_fetched`: Total number of documents processed
-- `poll_count`: Number of polling cycles executed
-- `last_document_id`: Last processed document ID (for cursor pagination)
-- `last_scroll_id`: Last scroll ID (for scroll pagination)
-- `last_offset`: Last processed offset
-
-### Error Tracking
-
-- `error_count`: Total number of errors encountered
-- `last_error`: Last error message
-
-### Performance Statistics
+| Field | Default | Behavior |
+| --- | --- | --- |
+| `url` | required | Elasticsearch URL. |
+| `index` | required | Index expression; startup checks that it exists and is accessible. |
+| `username` / `password` | none | Basic authentication is enabled only when both are present. |
+| `query` | `match_all` | Structured Elasticsearch Query DSL object, represented by TOML tables. |
+| `polling_interval` | `10s` | Delay before each poll; invalid strings fall back to `10s`, zero is accepted. |
+| `batch_size` | `100` | Search size per poll; zero returns no hits. Elasticsearch applies its result-window limit. |
+| `timestamp_field` | none | Top-level RFC3339 string field used to advance the watermark. |
+| `scroll_timeout` | none | Accepted but unused; the connector does not use scroll. |
+| `state` | none | Optional, separate plugin JSON snapshot configuration described below. |
+
+The local connector file remains TOML. `plugin_config_format` selects the default
+format of the HTTP API's plugin-config response; it does not change local file
+parsing or the JSON passed through the FFI.
+
+## Polling and Delivery
+
+Each poll sleeps first, then searches with the configured query and batch size,
+sorted ascending by `timestamp_field`, or `@timestamp` when it is unset. The
+index mapping must support that sort. Each hit with `_source` becomes one JSON
+message. Elasticsearch `_id` is not included automatically or used as an Iggy
+message ID. The plugin leaves message headers and timestamps unset.
+
+With a timestamp field, searches filter values strictly greater than the last
+acknowledged watermark. The watermark is extracted only from top-level RFC3339
+strings; numeric dates, date-only strings and nested paths do not advance it.
+Without a usable timestamp, polls repeat the first matching batch.
+
+There is no document-ID tiebreaker, scroll or `search_after` pagination. If a
+timestamp group spans multiple batches, the remaining tied documents are skipped
+after the watermark advances. Late arrivals and updates at or below the watermark
+are also skipped. This is not a complete change-data-capture feed.
+
+The candidate watermark is returned in the checkpoint and committed in memory
+only after the runtime sends the batch, saves the checkpoint and returns `Ack`.
+`Nack` keeps the previous watermark, making those documents eligible for another
+poll. Fetched counters include these retries. Replayed messages have no stable
+Iggy ID from this plugin, so consumers must allow for duplicates.
+
+HTTP/search/JSON failures, `timed_out: true` and nonzero `_shards.failed` return a
+poll error without advancing progress. The next poll retries after the configured
+delay. There is no per-query retry loop or configured HTTP request timeout.
+Poll errors are logged by the SDK; they do not change runtime connector status
+or increment the runtime forwarding-error counter.
-- `total_bytes_processed`: Total bytes processed
-- `avg_batch_processing_time_ms`: Average processing time per batch
-- `last_successful_poll`: Timestamp of last successful poll
-- `empty_polls_count`: Number of polls that returned no documents
-- `successful_polls_count`: Number of successful polls
+## State Information
-## Storage Backends
+The runtime checkpoint contains:
-### File Storage (Default)
+- `last_poll_timestamp`: Timestamp watermark for the returned batch.
+- `total_documents_fetched`: Number of `_source` records fetched, including retries.
+- `poll_count`: Successful search polls, including empty ones.
+- `error_count` and `last_error`: Search failures and the latest error text.
+- `processing_stats`: Payload bytes, average successful-poll elapsed milliseconds, last successful poll, empty polls and successful polls.
-```toml
-state = {
- enabled = true
- storage_type = "file"
- storage_config = {
- base_path = "./connector_states"
- }
-}
-```
+Successful polls include empty polls. Average duration includes the configured
+polling delay, and payload bytes count serialized `_source` values, excluding
+Elasticsearch response metadata. `last_document_id`, `last_scroll_id` and
+`last_offset` remain in snapshots but do not drive polling.
-### Elasticsearch Storage
+The runtime saves the MessagePack checkpoint after sending the batch, including
+empty successful polls. Its default file backend uses
+`local_state/source_<key>.state`; the main runtime `[state]` configuration can
+change that location or select HTTP storage. Invalid MessagePack state warns
+and starts fresh. No plugin state configuration is required for this path.
-```toml
-state = {
- enabled = true
- storage_type = "elasticsearch"
- storage_config = {
- index = "connector_states"
- url = "http://localhost:9200"
- }
-}
-```
+### Optional Plugin JSON Snapshot
-### Redis Storage
+Append this configuration only when the separate plugin snapshot is needed:
```toml
-state = {
- enabled = true
- storage_type = "redis"
- storage_config = {
- url = "redis://localhost:6379"
- key_prefix = "connector_states:"
- }
-}
-```
-
-## Usage Examples
-
-### Basic Usage with State Management
-
-```rust
-use elasticsearch_source::{ElasticsearchSource, StateManagerExt};
-
-// Create connector with state management enabled
-let mut connector = ElasticsearchSource::new(id, config);
-
-// Open connector (automatically loads state if available)
-connector.open().await?;
-
-// Start polling (automatically saves state)
-let messages = connector.poll().await?;
-
-// Close connector (automatically saves final state)
-connector.close().await?;
-```
-
-### Manual State Management
-
-```rust
-use elasticsearch_source::{ElasticsearchSource, StateManagerExt};
-
-let mut connector = ElasticsearchSource::new(id, config);
-
-// Load state manually
-connector.load_state().await?;
-
-// Get current state
-let state = connector.get_state().await?;
-println!("Current state: {:?}", state);
-
-// Export state to JSON
-let state_json = connector.export_state().await?;
-println!("State JSON: {}", serde_json::to_string_pretty(&state_json)?);
-
-// Import state from JSON
-connector.import_state(state_json).await?;
-
-// Reset state
-connector.reset_state().await?;
-```
-
-### State Manager Utilities
-
-```rust
-use elasticsearch_source::{ElasticsearchSource, StateManagerExt};
-
-let connector = ElasticsearchSource::new(id, config);
-
-// Get state manager
-if let Some(state_manager) = connector.get_state_manager() {
- // Get state statistics
- let stats = state_manager.get_state_stats().await?;
- println!("Total states: {}", stats.total_states);
-
- // Clean up old states (older than 30 days)
- let deleted_count = state_manager.cleanup_old_states(30).await?;
- println!("Deleted {} old states", deleted_count);
-}
-```
+[plugin_config.state]
+enabled = true
+storage_type = "file"
+state_id = "elasticsearch_logs_connector"
-## State File Format
-
-State files are stored as JSON with the following structure:
-
-```json
-{
- "id": "elasticsearch_logs_connector",
- "last_updated": "2024-01-15T10:30:00Z",
- "version": 1,
- "data": {
- "last_poll_timestamp": "2024-01-15T10:30:00Z",
- "total_documents_fetched": 15000,
- "poll_count": 150,
- "last_document_id": "doc_12345",
- "last_scroll_id": "scroll_abc123",
- "last_offset": 15000,
- "error_count": 2,
- "last_error": "Connection timeout",
- "processing_stats": {
- "total_bytes_processed": 1048576,
- "avg_batch_processing_time_ms": 125.5,
- "last_successful_poll": "2024-01-15T10:30:00Z",
- "empty_polls_count": 5,
- "successful_polls_count": 145
- }
- },
- "metadata": {
- "connector_type": "elasticsearch_source",
- "connector_id": 1,
- "index": "logs-*",
- "url": "http://localhost:9200"
- }
-}
+[plugin_config.state.storage_config]
+base_path = "./connector_states"
```
-## Best Practices
-
-1. **State ID Uniqueness**: Use unique state IDs for different connector instances
-2. **Auto-save Interval**: Set appropriate auto-save intervals based on your data volume
-3. **Storage Location**: Use persistent storage locations for production deployments
-4. **State Cleanup**: Regularly clean up old state files to prevent disk space issues
-5. **Error Handling**: Monitor error counts and implement appropriate alerting
-6. **Backup**: Regularly backup state files for disaster recovery
-
-## Troubleshooting
-
-### Common Issues
-
-1. **State Not Loading**: Check file permissions and storage path
-2. **State Corruption**: Delete corrupted state files to start fresh
-3. **Performance Issues**: Adjust auto-save interval and batch sizes
-4. **Storage Full**: Implement state cleanup policies
-
-### Monitoring
-
-Monitor the following metrics:
-
-- State save/load success rates
-- Processing statistics
-- Error counts and types
-- Storage usage for state files
-
-## Migration
-
-To migrate from a connector without state management:
-
-1. Add state configuration to your connector config
-2. Set `enabled = true` in state config
-3. Restart the connector
-4. The connector will start tracking state from the next poll cycle
-
-To migrate between storage backends:
-
-1. Export state from current storage
-2. Update storage configuration
-3. Import state to new storage
-4. Restart connector
+The plugin loads this JSON snapshot during `open()` and saves it during `close()`.
+Its loaded values can override the already-restored runtime checkpoint. The
+file name is `<state_id>.json`; without a state ID it uses
+`elasticsearch_source_<numeric_plugin_id>.json`, and the default base path is
+`./connector_states`. Missing directories are created when saving.
+
+Only file storage is implemented. `storage_type = "elasticsearch"`, `"redis"` or
+an unknown type warns and falls back to `./connector_states`, ignoring the
+configured backend location. `auto_save_interval` and `tracked_fields` do not
+affect the runtime plugin. It does not start the separately exported
+`StateManager` background task.
+
+This snapshot uses a direct JSON file write, separate from the runtime's atomic
+checkpoint protocol. Snapshot read/write failures warn and do not fail startup
+or close. To reset progress, stop the runtime and remove its checkpoint and any
+configured plugin snapshot. Removing only the runtime file leaves the plugin
+snapshot available for restoration.
diff --git a/core/connectors/sources/elasticsearch_source/config.toml b/core/connectors/sources/elasticsearch_source/config.toml
index c58b76382..8db73d285 100644
--- a/core/connectors/sources/elasticsearch_source/config.toml
+++ b/core/connectors/sources/elasticsearch_source/config.toml
@@ -20,7 +20,7 @@ key = "elasticsearch"
enabled = true
version = 0
name = "Elasticsearch source"
-path = "../../target/release/libiggy_connector_elasticsearch_source"
+path = "target/release/libiggy_connector_elasticsearch_source"
plugin_config_format = "json"
[[streams]]
diff --git a/core/connectors/sources/elasticsearch_source/src/lib.rs b/core/connectors/sources/elasticsearch_source/src/lib.rs
index f7e18844e..5fe137c5c 100644
--- a/core/connectors/sources/elasticsearch_source/src/lib.rs
+++ b/core/connectors/sources/elasticsearch_source/src/lib.rs
@@ -23,7 +23,8 @@ use elasticsearch::{
};
use iggy_common::{DateTime, Utc};
use iggy_connector_sdk::{
- ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector,
+ ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source,
+ source::SourceBatchResult, source_connector,
};
use secrecy::{ExposeSecret, SecretString};
use serde::{Deserialize, Serialize};
@@ -47,11 +48,11 @@ struct State {
last_poll_timestamp: Option<DateTime<Utc>>,
total_documents_fetched: usize,
poll_count: usize,
- /// Last document ID processed (for cursor-based pagination)
+ /// Retained in snapshots; not used for pagination.
last_document_id: Option<String>,
- /// Last scroll ID (for scroll-based pagination)
+ /// Retained in snapshots; the connector does not use scroll.
last_scroll_id: Option<String>,
- /// Last processed offset
+ /// Retained in snapshots; not used as a polling cursor.
last_offset: Option<u64>,
/// Error count and last error
error_count: usize,
@@ -62,31 +63,47 @@ struct State {
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ProcessingStats {
- /// Total bytes processed
+ /// Serialized source payload bytes fetched, including retries.
total_bytes_processed: u64,
- /// Average processing time per batch
+ /// Mean successful-poll elapsed milliseconds, including the polling delay.
avg_batch_processing_time_ms: f64,
/// Last successful processing timestamp
last_successful_poll: Option<DateTime<Utc>>,
/// Number of empty polls
empty_polls_count: usize,
- /// Number of successful polls
+ /// Number of successful polls, including empty polls.
successful_polls_count: usize,
}
+impl ProcessingStats {
+ fn record_success(&mut self, elapsed: Duration, empty: bool) {
+ self.successful_polls_count += 1;
+ self.last_successful_poll = Some(Utc::now());
+
+ let total_polls = self.successful_polls_count;
+ self.avg_batch_processing_time_ms = (self.avg_batch_processing_time_ms
+ * (total_polls - 1) as f64
+ + elapsed.as_millis() as f64)
+ / total_polls as f64;
+ if empty {
+ self.empty_polls_count += 1;
+ }
+ }
+}
+
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StateConfig {
/// Enable state persistence
pub enabled: bool,
- /// State storage type: "file", "elasticsearch", "redis", etc.
+ /// Only "file" is implemented; other values fall back to the default file directory.
pub storage_type: Option<String>,
/// State storage configuration (depends on storage_type)
pub storage_config: Option<Value>,
/// State ID for this connector instance
pub state_id: Option<String>,
- /// Auto-save state interval (e.g., "30s", "5m")
+ /// Interval for manually started StateManager tasks; unused by the runtime plugin.
pub auto_save_interval: Option<String>,
- /// Fields to track in state (e.g., ["last_timestamp", "last_document_id"])
+ /// Reserved; does not filter the saved state.
pub tracked_fields: Option<Vec<String>>,
}
@@ -101,6 +118,7 @@ pub struct ElasticsearchSourceConfig {
pub polling_interval: Option<String>,
pub batch_size: Option<usize>,
pub timestamp_field: Option<String>,
+ /// Reserved; the connector does not use the scroll API.
pub scroll_timeout: Option<String>,
pub state: Option<StateConfig>,
}
@@ -112,6 +130,7 @@ pub struct ElasticsearchSource {
client: Option<Elasticsearch>,
polling_interval: Duration,
state: Mutex<State>,
+ pending_timestamp: Mutex<Option<DateTime<Utc>>>,
}
impl ElasticsearchSource {
@@ -139,6 +158,7 @@ impl ElasticsearchSource {
config,
client: None,
polling_interval,
+ pending_timestamp: Mutex::new(None),
state: Mutex::new(restored_state.unwrap_or(State {
last_poll_timestamp: None,
total_documents_fetched: 0,
@@ -319,7 +339,7 @@ impl ElasticsearchSource {
async fn search_documents(
&self,
client: &Elasticsearch,
- ) -> Result<Vec<ProducedMessage>, Error> {
+ ) -> Result<(Vec<ProducedMessage>, Option<DateTime<Utc>>), Error> {
let state = self.state.lock().await;
let batch_size = self.config.batch_size.unwrap_or(100);
@@ -387,8 +407,23 @@ impl ElasticsearchSource {
.await
.map_err(|e| Error::Storage(format!("Failed to parse search response: {}", e)))?;
+ if response_body.get("timed_out").and_then(Value::as_bool) == Some(true) {
+ return Err(Error::Storage("Elasticsearch search timed out".to_string()));
+ }
+ if let Some(failed_shards) = response_body
+ .get("_shards")
+ .and_then(|shards| shards.get("failed"))
+ .and_then(Value::as_u64)
+ .filter(|failed| *failed > 0)
+ {
+ return Err(Error::Storage(format!(
+ "Elasticsearch search failed on {failed_shards} shards"
+ )));
+ }
+
let mut messages = Vec::new();
let mut latest_timestamp = None;
+ let mut payload_bytes = 0;
if let Some(hits) = response_body
.get("hits")
@@ -414,6 +449,7 @@ impl ElasticsearchSource {
Error::Serialization(format!("Failed to serialize document: {}", e))
})?;
+ payload_bytes += payload.len() as u64;
let message = ProducedMessage {
id: None,
headers: None,
@@ -430,12 +466,9 @@ impl ElasticsearchSource {
// Update state
let mut state = self.state.lock().await;
state.total_documents_fetched += messages.len();
+ state.processing_stats.total_bytes_processed += payload_bytes;
state.poll_count += 1;
- if let Some(timestamp) = latest_timestamp {
- state.last_poll_timestamp = Some(timestamp);
- }
-
- Ok(messages)
+ Ok((messages, latest_timestamp))
}
}
@@ -500,28 +533,15 @@ impl Source for ElasticsearchSource {
.as_ref()
.ok_or_else(|| Error::Storage("Elasticsearch client not initialized".to_string()))?;
- let messages = match self.search_documents(client).await {
- Ok(msgs) => {
- // Update success statistics
+ let (messages, latest_timestamp) = match self.search_documents(client).await {
+ Ok((msgs, latest_timestamp)) => {
let mut state = self.state.lock().await;
- state.processing_stats.successful_polls_count += 1;
- state.processing_stats.last_successful_poll = Some(Utc::now());
-
- let processing_time = start_time.elapsed().as_millis() as f64;
- let total_polls = state.processing_stats.successful_polls_count
- + state.processing_stats.empty_polls_count;
- state.processing_stats.avg_batch_processing_time_ms =
- (state.processing_stats.avg_batch_processing_time_ms
- * (total_polls - 1) as f64
- + processing_time)
- / total_polls as f64;
-
- if msgs.is_empty() {
- state.processing_stats.empty_polls_count += 1;
- }
+ state
+ .processing_stats
+ .record_success(start_time.elapsed(), msgs.is_empty());
drop(state);
- msgs
+ (msgs, latest_timestamp)
}
Err(e) => {
// Update error statistics
@@ -533,17 +553,33 @@ impl Source for ElasticsearchSource {
}
};
let persisted_state = {
- let state = self.state.lock().await;
- self.serialize_state(&state)
+ let mut candidate_state = self.state.lock().await.clone();
+ if let Some(timestamp) = latest_timestamp {
+ candidate_state.last_poll_timestamp = Some(timestamp);
+ }
+ self.serialize_state(&candidate_state).ok_or_else(|| {
+ Error::Serialization("Failed to serialize Elasticsearch source state".to_string())
+ })?
};
+ *self.pending_timestamp.lock().await = latest_timestamp;
Ok(ProducedMessages {
schema: Schema::Json,
messages,
- state: persisted_state,
+ state: Some(persisted_state),
})
}
+ async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(), Error> {
+ let pending_timestamp = self.pending_timestamp.lock().await.take();
+ if result == SourceBatchResult::Ack
+ && let Some(timestamp) = pending_timestamp
+ {
+ self.state.lock().await.last_poll_timestamp = Some(timestamp);
+ }
+ Ok(())
+ }
+
async fn close(&mut self) -> Result<(), Error> {
let state = self.state.lock().await;
info!(
@@ -575,3 +611,228 @@ impl Source for ElasticsearchSource {
Ok(())
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::{CONNECTOR_NAME, ElasticsearchSource, ElasticsearchSourceConfig, State};
+ use iggy_connector_sdk::{ConnectorState, Source, source::SourceBatchResult};
+ use serde_json::json;
+ use wiremock::matchers::{body_partial_json, method, path};
+ use wiremock::{Mock, MockServer, ResponseTemplate};
+
+ fn test_config(url: &str) -> ElasticsearchSourceConfig {
+ serde_json::from_value(json!({
+ "url": url,
+ "index": "logs",
+ "polling_interval": "0s",
+ "timestamp_field": "timestamp"
+ }))
+ .unwrap()
+ }
+
+ async fn source_backend() -> MockServer {
+ let server = MockServer::start().await;
+ Mock::given(method("HEAD"))
+ .and(path("/logs"))
+ .respond_with(ResponseTemplate::new(200))
+ .mount(&server)
+ .await;
+ Mock::given(method("POST"))
+ .and(path("/logs/_search"))
+ .and(body_partial_json(json!({"query": {"match_all": {}}})))
+ .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+ "hits": {"hits": [{"_source": {"timestamp": "2026-01-01T00:00:00Z", "value": 1}}]}
+ })))
+ .mount(&server)
+ .await;
+ Mock::given(method("POST"))
+ .and(path("/logs/_search"))
+ .and(body_partial_json(json!({"query": {"bool": {"must": [
+ {"match_all": {}}, {"range": {"timestamp": {"gt": "2026-01-01T00:00:00+00:00"}}}
+ ]}}})))
+ .respond_with(ResponseTemplate::new(200).set_body_json(json!({"hits": {"hits": []}})))
+ .mount(&server)
+ .await;
+ server
+ }
+
+ #[tokio::test]
+ async fn given_nack_should_retry_documents_before_advancing_cursor() {
+ let server = source_backend().await;
+ let mut source = ElasticsearchSource::new(1, test_config(&server.uri()), None);
+ source.open().await.unwrap();
+ let first = source.poll().await.unwrap();
+ assert_eq!(first.messages.len(), 1);
+
+ source
+ .on_batch_result(SourceBatchResult::Nack)
+ .await
+ .unwrap();
+ let retry = source.poll().await.unwrap();
+ assert_eq!(
+ retry.messages.len(),
+ 1,
+ "Nack must leave the document eligible for the next poll"
+ );
+ assert_eq!(retry.messages[0].payload, first.messages[0].payload);
+
+ source
+ .on_batch_result(SourceBatchResult::Ack)
+ .await
+ .unwrap();
+ assert!(source.poll().await.unwrap().messages.is_empty());
+ source
+ .on_batch_result(SourceBatchResult::Ack)
+ .await
+ .unwrap();
+ source.close().await.unwrap();
+ }
+
+ #[tokio::test]
+ async fn given_no_state_should_start_fresh() {
+ let source = ElasticsearchSource::new(1, test_config("http://localhost:9200"), None);
+ let state = source.state.lock().await;
+ assert!(state.last_poll_timestamp.is_none());
+ assert_eq!(state.total_documents_fetched, 0);
+ }
+
+ #[tokio::test]
+ async fn given_invalid_state_should_start_fresh() {
+ let source = ElasticsearchSource::new(
+ 1,
+ test_config("http://localhost:9200"),
+ Some(ConnectorState(b"invalid state".to_vec())),
+ );
+ let state = source.state.lock().await;
+ assert!(state.last_poll_timestamp.is_none());
+ assert_eq!(state.total_documents_fetched, 0);
+ }
+
+ #[tokio::test]
+ async fn given_persisted_state_should_restore_cursor() {
+ let source = ElasticsearchSource::new(1, test_config("http://localhost:9200"), None);
+ let mut state = source.state.lock().await.clone();
+ state.last_poll_timestamp = Some("2026-01-01T00:00:00Z".parse().unwrap());
+ state.total_documents_fetched = 12;
+ let restored = ElasticsearchSource::new(
+ 1,
+ test_config("http://localhost:9200"),
+ source.serialize_state(&state),
+ );
+ let restored = restored.state.lock().await;
+ assert_eq!(restored.last_poll_timestamp, state.last_poll_timestamp);
+ assert_eq!(restored.total_documents_fetched, 12);
+ }
+
+ #[tokio::test]
+ async fn state_should_be_serializable_and_deserializable() {
+ let source = ElasticsearchSource::new(1, test_config("http://localhost:9200"), None);
+ let mut state = source.state.lock().await.clone();
+ state.poll_count = 7;
+ state.last_document_id = Some("record_1".to_string());
+ state.error_count = 2;
+ state.last_error = Some("backend unavailable".to_string());
+ state.processing_stats.successful_polls_count = 5;
+ let restored = source
+ .serialize_state(&state)
+ .unwrap()
+ .deserialize::<State>(CONNECTOR_NAME, 1)
+ .unwrap();
+ assert_eq!(
+ serde_json::to_value(restored).unwrap(),
+ serde_json::to_value(state).unwrap()
+ );
+ }
+
+ #[tokio::test]
+ async fn given_empty_and_nonempty_polls_should_average_each_poll_once() {
+ let source = ElasticsearchSource::new(1, test_config("http://localhost:9200"), None);
+ let mut state = source.state.lock().await;
+ let stats = &mut state.processing_stats;
+ for (milliseconds, empty) in [(100, false), (200, true), (300, false), (400, true)] {
+ stats.record_success(std::time::Duration::from_millis(milliseconds), empty);
+ }
+ assert_eq!(stats.successful_polls_count, 4);
+ assert_eq!(stats.empty_polls_count, 2);
+ assert_eq!(stats.avg_batch_processing_time_ms, 250.0);
+ }
+
+ #[tokio::test]
+ async fn given_fetched_documents_should_count_payload_bytes_including_retries() {
+ let server = source_backend().await;
+ let mut source = ElasticsearchSource::new(1, test_config(&server.uri()), None);
+ source.open().await.unwrap();
+ let first = source.poll().await.unwrap();
+ assert_eq!(first.messages.len(), 1);
+ let payload_bytes = first.messages[0].payload.len() as u64;
+ assert_eq!(
+ source
+ .state
+ .lock()
+ .await
+ .processing_stats
+ .total_bytes_processed,
+ payload_bytes
+ );
+
+ source
+ .on_batch_result(SourceBatchResult::Nack)
+ .await
+ .unwrap();
+ let retry = source.poll().await.unwrap();
+ assert_eq!(retry.messages.len(), 1);
+ assert_eq!(
+ source
+ .state
+ .lock()
+ .await
+ .processing_stats
+ .total_bytes_processed,
+ payload_bytes * 2
+ );
+ source
+ .on_batch_result(SourceBatchResult::Ack)
+ .await
+ .unwrap();
+ source.close().await.unwrap();
+ }
+
+ #[tokio::test]
+ async fn given_incomplete_search_should_preserve_cursor_and_retry() {
+ for mut response in [
+ json!({"timed_out": true}),
+ json!({"_shards": {"failed": 1}}),
+ ] {
+ let server = source_backend().await;
+ response["hits"] =
+ json!({"hits": [{"_source": {"timestamp": "2026-01-01T00:00:00Z", "value": 1}}]});
+ Mock::given(method("POST"))
+ .and(path("/logs/_search"))
+ .respond_with(ResponseTemplate::new(200).set_body_json(&response))
+ .with_priority(1)
+ .up_to_n_times(1)
+ .mount(&server)
+ .await;
+ let mut source = ElasticsearchSource::new(1, test_config(&server.uri()), None);
+ source.open().await.unwrap();
+
+ assert!(
+ source.poll().await.is_err(),
+ "Incomplete response must fail: {response}"
+ );
+ let state = source.state.lock().await;
+ assert!(state.last_poll_timestamp.is_none());
+ assert_eq!(state.total_documents_fetched, 0);
+ assert_eq!(state.error_count, 1);
+ drop(state);
+
+ let retry = source.poll().await.unwrap();
+ assert_eq!(retry.messages.len(), 1);
+ source
+ .on_batch_result(SourceBatchResult::Ack)
+ .await
+ .unwrap();
+ source.close().await.unwrap();
+ }
+ }
+}
diff --git a/core/connectors/sources/elasticsearch_source/src/state_manager.rs b/core/connectors/sources/elasticsearch_source/src/state_manager.rs
index b3cce5be9..988c9e397 100644
--- a/core/connectors/sources/elasticsearch_source/src/state_manager.rs
+++ b/core/connectors/sources/elasticsearch_source/src/state_manager.rs
@@ -306,12 +306,9 @@ impl StateStorage for FileStateStorage {
async fn save_source_state(&self, state: &SourceState) -> Result<(), Error> {
use tokio::fs;
- // Ensure directory exists
- if let Some(parent) = self.base_path.parent() {
- fs::create_dir_all(parent)
- .await
- .map_err(|e| Error::Storage(format!("Failed to create state directory: {e}")))?;
- }
+ fs::create_dir_all(&self.base_path)
+ .await
+ .map_err(|e| Error::Storage(format!("Failed to create state directory: {e}")))?;
let path = self.get_state_path(&state.id);
let json = serde_json::to_string_pretty(state)
@@ -386,3 +383,34 @@ impl StateStorage for FileStateStorage {
Ok(states)
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::{FileStateStorage, SourceState, StateStorage};
+ use iggy_common::Utc;
+ use serde_json::json;
+ use tempfile::tempdir;
+
+ #[tokio::test]
+ async fn given_missing_storage_directory_should_save_and_restore_state() {
+ let directory = tempdir().unwrap();
+ let storage = FileStateStorage::new(directory.path().join("nested/states"));
+ let mut state = SourceState {
+ id: "source_1".to_string(),
+ last_updated: Utc::now(),
+ version: 1,
+ data: json!({"last_poll_timestamp": "2026-01-01T00:00:00Z"}),
+ metadata: None,
+ };
+
+ storage.save_source_state(&state).await.unwrap();
+ let restored = storage.load_source_state(&state.id).await.unwrap().unwrap();
+ assert_eq!(restored.data, state.data);
+ assert_eq!(storage.list_states().await.unwrap(), vec![state.id.clone()]);
+
+ state.data = json!({"last_poll_timestamp": "2026-01-02T00:00:00Z"});
+ storage.save_source_state(&state).await.unwrap();
+ let restored = storage.load_source_state(&state.id).await.unwrap().unwrap();
+ assert_eq!(restored.data, state.data);
+ }
+}
diff --git a/core/connectors/sources/influxdb_source/README.md b/core/connectors/sources/influxdb_source/README.md
index 333e4f3d7..337f61e84 100644
--- a/core/connectors/sources/influxdb_source/README.md
+++ b/core/connectors/sources/influxdb_source/README.md
@@ -13,70 +13,66 @@ A source connector that polls InfluxDB and produces messages into Iggy streams.
| Auth header | `Authorization: Token {t}` | `Authorization: Bearer {t}` |
| Cursor semantics | `>= $cursor` (inclusive) | `> '$cursor'` (exclusive) |
| Default cursor field | `_time` | `time` |
-| Config `version` key | `"v2"` (or omit — default) | `"v3"` |
-| Payload value types | All fields as strings (CSV has no types); `parse_scalar` coerces to bool/int/float where possible | Native JSON types: numbers, booleans, and nulls preserved as-is from SQL |
+| Config `version` key | `"v2"` (or omit, the default) | `"v3"` |
+| Payload value types | CSV cells are strings; annotation types/defaults are ignored and payload conversion infers scalars | Native JSON types: numbers, booleans, and nulls preserved as-is from SQL |
-> **Breaking schema note for consumers migrating V2 → V3**: a field that arrives as the string `"42"` under V2 will arrive as the integer `42` under V3. Update consumer deserialization accordingly.
+> **Schema note for consumers migrating V2 to V3**: a string field containing `"42"` becomes the number `42` in V2 whole-row JSON, but remains a string in V3. V2 also uses an envelope while V3 emits a flat row. Update consumer deserialization accordingly.
## Cursor-Based Polling
-Both versions use an RFC 3339 timestamp cursor stored in persistent connector state. On each poll the cursor is substituted into the query template via the `$cursor` and `$limit` placeholders; after a successful batch the cursor advances to the highest timestamp seen.
+Both versions use an RFC3339 timestamp cursor and row position in persistent connector state. Polling stages a candidate checkpoint; `on_batch_result(Ack)` applies it and `Nack` discards it. The runtime acknowledges after forwarding messages and saving the checkpoint. Retries can still duplicate messages after a forwarding/checkpoint failure.
+
+Queries must return ascending timestamps and a stable, unique order within each timestamp group. Include the necessary tag or key columns in the sort. A timestamp watermark does not track deletions and can miss late inserts or updates at or below its saved position. Changing the order of tied rows invalidates skip/offset progress.
### V2 Cursor Semantics
-V2 Flux queries use `>= $cursor` (inclusive). Rows at the cursor timestamp are re-delivered by the first query after a cursor advance and must be skipped. The connector tracks `cursor_row_count` in persisted state and skips exactly that many leading rows to prevent duplicate delivery.
+V2 queries should use an inclusive cursor, such as `range(start: time(v: "$cursor"))`. The connector persists `cursor_row_count` and skips that many matching rows when it queries the saved timestamp again. Flux sorting and limiting operate per table. Group compatible rows into one table before sorting and limiting; the sample selects one numeric field and sorts by time and its only tag, `host`.
-The Flux query **must** sort by the cursor field when using `>=` semantics. The connector enforces this at startup and returns an error if `>=` is present without a `|> sort(columns: ["_time"])` call.
+Startup rejects a `>=` cursor expression without a `sort(` call, but that text check does not validate ordering and does not cover a `range()` expression. `$limit` is the base batch size plus the skip count, with the extra count capped at ten times the base. Dense timestamp groups can exhaust this bound and cause a poll error without advancing the cursor. Using strict `>` bypasses rows that the inclusive skip scheme needs to revisit.
### V3 Cursor Semantics
-V3 SQL uses strict `WHERE time > '$cursor'` — no rows are re-delivered across batch boundaries. The query **must** include `OFFSET $offset` when `stuck_batch_cap_factor > 0` (the default). The connector validates this at startup.
+V3 requires strict `WHERE time > '$cursor'`; inclusive `>=` cursor expressions are rejected. With `stuck_batch_cap_factor > 0`, startup also requires `$offset` and an ascending `ORDER BY`. These checks are textual and cannot establish a stable tiebreaker order.
#### Stuck-Timestamp Handling (V3)
-InfluxDB 3's DataFusion engine does not guarantee stable row ordering for rows that share the same timestamp. Two full-batch scenarios require special handling:
-
-**All rows share the same timestamp** — the cursor cannot advance. The connector:
+A full batch containing one timestamp retains the previous cursor, emits its rows, advances `$offset`, and doubles the next effective batch size up to `stuck_batch_cap_factor` times the base size. Offset pagination still requires stable ordering of tied rows.
-1. Doubles `effective_batch_size` on each stuck poll, up to `stuck_batch_cap_factor × batch_size`.
-2. Uses `OFFSET $offset` to page through the tied rows without re-delivering earlier ones.
-3. Trips the circuit breaker when the cap is reached and resets `effective_batch_size` to `batch_size` after the cool-down so the connector recovers cleanly.
+A full batch with mixed timestamps emits only rows before the maximum timestamp. It advances to the second-highest timestamp, resets the effective size and offset, and re-fetches the deferred timestamp group on the next poll.
-**Mixed timestamps, full batch** — the batch is full and rows span more than one timestamp. Advancing the cursor to the highest timestamp seen would silently discard rows at that timestamp that did not fit in the batch. The connector:
-
-1. Emits only rows whose timestamp is strictly less than the batch maximum (`safe_message_count` rows).
-2. Advances the cursor to the second-highest distinct timestamp seen (`penultimate_cursor`) so the next poll re-fetches all rows at the maximum timestamp cleanly via `WHERE time > penultimate_cursor`.
-3. Resets `effective_batch_size` to `batch_size` (no inflation needed — cursor advanced).
+At the inflation cap, the connector emits no messages, resets the effective size to the base, preserves its offset and records a circuit-breaker failure. That failure opens the breaker only if its configured threshold is reached. Setting the cap factor to `0` disables both guards and can skip unseen ties.
## Configuration
-### V2 — InfluxDB OSS 2.x / Cloud
+Build the plugin and runtime from the same 0.9.0/edge checkout. The following snippets are plugin settings; place them under `[plugin_config]` in the full connector entry below. The example data has a numeric `usage` field and `host` as its only tag. Use an existing bucket/database and a token with query access. Unknown settings and unsupported versions are rejected.
+
+### V2 - InfluxDB OSS 2.x / Cloud
```toml
-version = "v2" # optional — omitting defaults to v2
+version = "v2" # optional; omitting defaults to v2
url = "http://localhost:8086"
org = "my-org"
token = "my-token"
query = '''
from(bucket: "telemetry")
|> range(start: time(v: "$cursor"))
- |> filter(fn: (r) => r._measurement == "cpu")
- |> sort(columns: ["_time"])
+ |> filter(fn: (r) => r._measurement == "cpu" and r._field == "usage")
+ |> group(columns: [])
+ |> sort(columns: ["_time", "host"])
|> limit(n: $limit)
'''
# Optional
poll_interval = "5s" # how often to issue queries (default: "5s")
-batch_size = 500 # max rows per query (default: 500)
+batch_size = 500 # base query size; cursor handling can enlarge it
cursor_field = "_time" # column used as the cursor (default: "_time")
initial_offset = "2024-01-01T00:00:00Z" # starting cursor on first run
-payload_column = "" # extract a single column as payload (default: whole row as JSON)
payload_format = "json" # json | text | raw (default: "json")
include_metadata = true # include all row columns in the payload (default: true)
verbose_logging = false
```
-### V3 — InfluxDB 3.x Core / Enterprise
+### V3 - InfluxDB 3.x Core / Enterprise
```toml
version = "v3"
@@ -86,7 +82,7 @@ token = "my-token"
query = '''
SELECT * FROM cpu
WHERE time > '$cursor'
- ORDER BY time
+ ORDER BY time, host
LIMIT $limit OFFSET $offset
'''
@@ -95,12 +91,11 @@ poll_interval = "5s"
batch_size = 500
cursor_field = "time" # default cursor column for V3 (default: "time")
initial_offset = "2024-01-01T00:00:00Z"
-payload_column = ""
payload_format = "json"
include_metadata = true
stuck_batch_cap_factor = 10 # max effective_batch = 10 × batch_size (default: 10, max: 100)
- # set to 0 to disable stuck-batch detection entirely;
- # $offset is not required in the query when cap = 0
+ # 1 and values above 100 are rejected;
+ # 0 disables the guards and the $offset requirement
verbose_logging = false
```
@@ -108,47 +103,64 @@ verbose_logging = false
```toml
timeout = "10s" # per-request timeout
-max_retries = 3 # total query attempts, including the first (429/5xx)
+max_retries = 3 # total query attempts, including the first (network errors, 429/5xx)
retry_delay = "1s" # initial backoff
retry_max_delay = "5s" # backoff cap
max_open_retries = 10 # total open() health-check attempts, including the first
open_retry_max_delay = "60s" # backoff cap for open() retries
circuit_breaker_threshold = 5 # consecutive failures before circuit trips
-circuit_breaker_cool_down = "30s" # cool-down before half-open probe
+circuit_breaker_cool_down = "30s" # cooldown before queries resume
```
+Attempt counts are clamped to at least one. Query retries use exponential backoff with jitter; integer-seconds `Retry-After` on HTTP 429 can override the delay cap. Other HTTP errors and malformed successful response bodies fail the poll. Startup retries failed authenticated `GET /health` requests. Invalid duration strings warn and fall back to `1s`, while zero durations are accepted. `batch_size = 0` is clamped to `1`. Without `initial_offset`, the cursor starts at `1970-01-01T00:00:00Z`.
+
## Query Template Placeholders
| Placeholder | Substituted with |
| --- | --- |
| `$cursor` | Current cursor value (RFC 3339 timestamp) |
-| `$limit` | Current effective batch size |
+| `$limit` | Base size plus capped skip count (V2), or effective batch size (V3) |
| `$offset` | Row offset within the current cursor group (V3 only; required when `stuck_batch_cap_factor > 0`) |
## Payload Formats
-- **`json`** (default): Behaviour differs by version.
- - **V3**: Each row is a flat JSON object. `include_metadata = false` excludes the cursor column (`time`); all other columns are present. `include_metadata = true` (the default) includes all columns.
- - **V2**: Each row is wrapped in an envelope: `{"measurement":"..","field":"..","timestamp":"..","value":..,"row":{..}}`. The `row` object always includes `_time` and `_value`. `include_metadata = false` additionally excludes `_measurement` and `_field` from `row`; `include_metadata = true` includes them.
-- **`text`**: The value of `payload_column` is written as a UTF-8 string. Requires `payload_column` to be set.
-- **`raw`**: The value of `payload_column` is base64-decoded and written as raw bytes. Requires `payload_column` to be set.
+Without `payload_column`, the connector emits whole-row JSON regardless of `payload_format`:
+
+- **V3**: a flat object with native JSON types. `include_metadata = false` removes only the cursor column.
+- **V2**: an envelope containing `measurement`, `field`, `timestamp`, `value` and `row`. Metadata disabled leaves only `_time` and `_value` in `row`; the envelope remains. CSV cells are inferred as bool, integer or finite float, then string; an empty cell becomes null. Annotation datatype/default values are not applied.
+
+With `payload_column`, the selected format applies:
+
+- **`json`**: V2 parses the cell as JSON. V3 serializes its existing JSON value, so a string containing JSON remains a string.
+- **`text`**: V2 emits the cell as UTF-8. V3 emits strings directly and serializes other JSON values as text.
+- **`raw`**: decode the column's standard base64 text into bytes.
+
+Set the destination `streams.schema` to the intended output format. Missing selected columns, invalid V2 JSON cells and invalid base64 fail the poll without advancing the acknowledged state. Omitting `payload_column` selects whole-row output; an empty column name is not a whole-row selector.
+
+V3 requires a valid string cursor in every row and treats timestamps without a timezone suffix as UTC, preserving nanoseconds. V2 can emit an invalid/missing-cursor row if another row supplies a valid watermark, so retain a valid cursor in every result row. Keep database timestamps in the payload: the runtime does not transfer the plugin's timestamp fields into broker message metadata.
+
+Message IDs combine timestamp nanoseconds with result position. Distinct rows can collide across batches, so these IDs are not unique database-row keys. V2 falls back to a random batch base when a timestamp cannot be represented as nanoseconds.
## Full Configuration Example
+Save one connector entry in the runtime connector directory and start the runtime from the checkout root. Configure the broker connection in the main `connectors.toml` and create the `metrics` stream and `cpu` topic first.
+
```toml
-[[sources]]
-key = "influxdb-source"
+type = "source"
+key = "influxdb"
+version = 0
+name = "InfluxDB source"
enabled = true
path = "target/release/libiggy_connector_influxdb_source"
-[[sources.streams]]
+[[streams]]
stream = "metrics"
topic = "cpu"
schema = "json"
batch_length = 100
linger_time = "10ms"
-[sources.plugin_config]
+[plugin_config]
version = "v3"
url = "http://localhost:8181"
db = "telemetry"
@@ -156,7 +168,7 @@ token = "my-secret-token"
query = """
SELECT * FROM cpu
WHERE time > '$cursor'
- ORDER BY time
+ ORDER BY time, host
LIMIT $limit OFFSET $offset
"""
poll_interval = "10s"
@@ -173,9 +185,11 @@ The source uses a layered design with version-specific modules:
- **`v2` module**: Flux query construction, annotated-CSV response parsing, skip-N deduplication for `>=` cursor semantics.
- **`v3` module**: SQL query construction, JSONL response parsing, stuck-batch detection and cap inflation.
-- **`common` module**: Shared cursor state management, payload format handling, retry/circuit-breaker logic, and the `RowContext` passed into both `process_rows` functions.
+- **`common` module**: Shared configuration/state types, payload-format and timestamp helpers, and the `RowContext` passed into both `process_rows` functions. Retry and circuit-breaker helpers come from the connector SDK.
+
+The runtime checkpoint holds the cursor, processed-row count and version-specific skip/offset state. V2 accepts versioned V2 or legacy unversioned V2 state; V3 accepts only versioned V3 state. Corrupt state, invalid saved timestamps and version mismatches prevent startup. With the default file backend, the entry above uses `local_state/source_influxdb.state`. Stop the runtime before removing that checkpoint to reset progress; the runtime also supports an HTTP state backend.
-Persisted state (via the connector runtime's state store) holds the last cursor value and, for V2, the `cursor_row_count` needed to skip duplicate rows. State is versioned so V2 and V3 states are never confused on a connector restart.
+Successful empty polls return checkpoints. Open-breaker polls return empty messages without a checkpoint, and cooldown expiry permits another query. Query and parsing errors preserve acknowledged progress and are logged by the SDK; they do not increment the runtime's forwarding-error metric or change its running status. V3 treats a database-not-found query response as an empty result. Both response readers cap buffered successful query bodies at 256 MiB.
### V2 vs V3 API Comparison
@@ -184,7 +198,7 @@ Persisted state (via the connector runtime's state store) holds the last cursor
| Write body format | Line Protocol | Line Protocol |
| Query body format | JSON (Flux query payload) | JSON (SQL query payload) |
| Health check | `GET /health` | `GET /health` |
-| Retry triggers | 429 / 5xx | 429 / 5xx |
+| Query retry triggers | Network errors, 429 / 5xx | Network errors, 429 / 5xx |
| Cursor field default | `_time` | `time` |
| Timestamp format | RFC 3339 with TZ | RFC 3339 (TZ appended by connector if absent) |
-| Sort guarantee | Flux <code>\|> sort(...)</code> explicit | DataFusion — no stable order for tied timestamps |
+| Sort guarantee | Group compatible rows, then sort by time and unique tiebreakers | Explicit timestamp and unique tiebreaker order required |
diff --git a/core/connectors/sources/influxdb_source/config.toml b/core/connectors/sources/influxdb_source/config.toml
index 0a726035a..5ca9ee2cd 100644
--- a/core/connectors/sources/influxdb_source/config.toml
+++ b/core/connectors/sources/influxdb_source/config.toml
@@ -20,7 +20,7 @@ key = "influxdb"
enabled = true
version = 0
name = "InfluxDB source"
-path = "../../target/release/libiggy_connector_influxdb_source"
+path = "target/release/libiggy_connector_influxdb_source"
verbose = false
[[streams]]
@@ -36,10 +36,10 @@ org = "iggy"
token = "replace-with-token"
query = '''
from(bucket: "events")
- |> range(start: -1h)
- |> filter(fn: (r) => r._measurement == "iggy_messages")
- |> filter(fn: (r) => r._time > time(v: "$cursor"))
- |> sort(columns: ["_time"])
+ |> range(start: time(v: "$cursor"))
+ |> filter(fn: (r) => r._measurement == "iggy_messages" and r._field == "payload_json")
+ |> group(columns: [])
+ |> sort(columns: ["_time", "message_id"])
|> limit(n: $limit)
'''
poll_interval = "5s"
diff --git a/core/connectors/sources/influxdb_source/src/common.rs b/core/connectors/sources/influxdb_source/src/common.rs
index 109492029..54d5d5e2d 100644
--- a/core/connectors/sources/influxdb_source/src/common.rs
+++ b/core/connectors/sources/influxdb_source/src/common.rs
@@ -110,10 +110,8 @@ pub struct V2SourceConfig {
pub(crate) initial_offset: Option<String>,
pub(crate) payload_column: Option<String>,
pub(crate) payload_format: Option<String>,
- /// When `false`, non-cursor measurement and field columns are excluded from
- /// the emitted JSON payload. Note: the `_time` cursor column is always
- /// present in V2 Flux CSV output and cannot be stripped at query time —
- /// it will appear in the payload regardless of this setting.
+ /// When `false`, whole-row JSON retains only `_time` and `_value` in its
+ /// `row` object. The surrounding measurement/field/timestamp/value envelope remains.
pub(crate) include_metadata: Option<bool>,
pub(crate) verbose_logging: Option<bool>,
pub(crate) max_retries: Option<u32>,
@@ -140,9 +138,8 @@ pub struct V3SourceConfig {
pub(crate) initial_offset: Option<String>,
pub(crate) payload_column: Option<String>,
pub(crate) payload_format: Option<String>,
- /// When `false`, the cursor column (`time` by default) is excluded from the
- /// emitted JSON payload. Useful when consumers don't need the timestamp in
- /// the message body since it's available as message metadata.
+ /// When `false`, the cursor column is excluded from whole-row JSON.
+ /// The runtime does not copy the database timestamp into broker metadata.
pub(crate) include_metadata: Option<bool>,
pub(crate) verbose_logging: Option<bool>,
pub(crate) max_retries: Option<u32>,
@@ -153,11 +150,9 @@ pub struct V3SourceConfig {
pub(crate) retry_max_delay: Option<String>,
pub(crate) circuit_breaker_threshold: Option<u32>,
pub(crate) circuit_breaker_cool_down: Option<String>,
- /// Maximum factor by which batch_size may be inflated before the stuck-timestamp
- /// circuit breaker trips. Defaults to 10 (i.e. up to 10× the configured batch_size).
- /// Maximum accepted value is 100; higher values risk OOM-inducing queries.
- /// Set to `0` to disable stuck-timestamp detection entirely (cursor advances
- /// even when a full batch shares one timestamp, risking re-delivery of tied rows).
+ /// Maximum batch inflation before recording a circuit-breaker failure.
+ /// Defaults to 10; accepts 2 through 100, or 0 to disable the guards.
+ /// Disabling them can skip rows in timestamp groups spanning multiple batches.
pub(crate) stuck_batch_cap_factor: Option<u32>,
}
@@ -209,8 +204,7 @@ impl InfluxDbSourceConfig {
delegate!(opt self.poll_interval)
}
pub fn batch_size(&self) -> u32 {
- // Floor at 1 — callers build LIMIT $limit queries; LIMIT 0 stalls silently.
- // open() also rejects 0 explicitly, but defense-in-depth here costs nothing.
+ // LIMIT 0 would stall polling.
delegate!(unwrap self.batch_size, 500).max(1)
}
pub fn initial_offset(&self) -> Option<&str> {
diff --git a/core/connectors/sources/influxdb_source/src/lib.rs b/core/connectors/sources/influxdb_source/src/lib.rs
index 59a1b5e12..3de07b4be 100644
--- a/core/connectors/sources/influxdb_source/src/lib.rs
+++ b/core/connectors/sources/influxdb_source/src/lib.rs
@@ -33,10 +33,12 @@ use common::{
use iggy_connector_sdk::retry::{
CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
};
+use iggy_connector_sdk::source::SourceBatchResult;
use iggy_connector_sdk::{
ConnectorState, Error, ProducedMessages, Schema, Source, source_connector,
};
use reqwest::Url;
+use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderValue};
use reqwest_middleware::ClientWithMiddleware;
use secrecy::{ExposeSecret, SecretBox};
use std::sync::Arc;
@@ -72,6 +74,7 @@ pub struct InfluxDbSource {
config: InfluxDbSourceConfig,
client: Option<ClientWithMiddleware>,
version_state: VersionState,
+ pending_state: Mutex<Option<PersistedState>>,
payload_format: PayloadFormat,
poll_interval: Duration,
retry_delay: Duration,
@@ -112,6 +115,7 @@ impl InfluxDbSource {
config,
client: None,
version_state,
+ pending_state: Mutex::new(None),
payload_format,
poll_interval,
retry_delay,
@@ -406,7 +410,20 @@ impl Source for InfluxDbSource {
}
let timeout = parse_duration(self.config.timeout(), DEFAULT_TIMEOUT);
+ let token = self.config.token_secret().expose_secret();
+ let auth_header = SecretBox::new(Box::new(match &self.config {
+ InfluxDbSourceConfig::V2(_) => format!("Token {token}"),
+ InfluxDbSourceConfig::V3(_) => format!("Bearer {token}"),
+ }));
+ let mut authorization =
+ HeaderValue::from_str(auth_header.expose_secret()).map_err(|error| {
+ Error::InvalidConfigValue(format!("Invalid InfluxDB authorization header: {error}"))
+ })?;
+ authorization.set_sensitive(true);
+ let mut headers = HeaderMap::new();
+ headers.insert(AUTHORIZATION, authorization);
let raw_client = reqwest::Client::builder()
+ .default_headers(headers)
.timeout(timeout)
.build()
.map_err(|e| Error::InitError(format!("Failed to create HTTP client: {e}")))?;
@@ -440,11 +457,7 @@ impl Source for InfluxDbSource {
"InfluxDB",
));
- let token = self.config.token_secret().expose_secret();
- self.auth_header = Some(SecretBox::new(Box::new(match &self.config {
- InfluxDbSourceConfig::V2(_) => format!("Token {token}"),
- InfluxDbSourceConfig::V3(_) => format!("Bearer {token}"),
- })));
+ self.auth_header = Some(auth_header);
info!(
"{CONNECTOR_NAME} ID: {} opened successfully (version={ver})",
@@ -499,7 +512,7 @@ impl Source for InfluxDbSource {
return Err(Error::InvalidState);
};
- let state_snap = state_mu.lock().await.clone();
+ let mut state_snap = state_mu.lock().await.clone();
match v2::poll(
client,
cfg,
@@ -514,34 +527,26 @@ impl Source for InfluxDbSource {
self.circuit_breaker.record_success();
let messages = result.messages;
let schema = result.schema;
- let state_snap = {
- let mut state = state_mu.lock().await;
- state.processed_rows += messages.len() as u64;
- apply_v2_cursor_advance(
- &mut state,
- result.max_cursor,
- result.rows_at_max_cursor,
- result.skipped,
- );
-
- poll_event!(
- skipped = result.skipped,
- "{CONNECTOR_NAME} ID: {} produced {} messages (V2). \
- Total: {}. Cursor: {:?}",
- self.id,
- messages.len(),
- state.processed_rows,
- state.last_timestamp.as_deref()
- );
- state.clone()
- // lock released here
- };
-
- let persisted = ConnectorState::serialize(
- &PersistedState::V2(state_snap),
- CONNECTOR_NAME,
+ state_snap.processed_rows += messages.len() as u64;
+ apply_v2_cursor_advance(
+ &mut state_snap,
+ result.max_cursor,
+ result.rows_at_max_cursor,
+ result.skipped,
+ );
+ poll_event!(
+ skipped = result.skipped,
+ "{CONNECTOR_NAME} ID: {} produced {} messages (V2). \
+ Total: {}. Cursor: {:?}",
self.id,
+ messages.len(),
+ state_snap.processed_rows,
+ state_snap.last_timestamp.as_deref()
);
+ let candidate = PersistedState::V2(state_snap);
+ let persisted =
+ ConnectorState::serialize(&candidate, CONNECTOR_NAME, self.id);
+ *self.pending_state.lock().await = Some(candidate);
match &persisted {
Some(_) => {
// Relaxed: the SDK drives poll() from a single select! loop —
@@ -616,29 +621,18 @@ impl Source for InfluxDbSource {
let messages = result.messages;
let schema = result.schema;
let msg_count = messages.len();
- // Clone before acquiring the lock so the String allocation is
- // outside the critical section; the original is moved into
- // the mutex, the clone is used for serialization.
- let for_serialize = new.clone();
- {
- let mut state = state_mu.lock().await;
- *state = new;
- poll_event!(
- "{CONNECTOR_NAME} ID: {} produced {} messages (V3). \
- Total: {}. Cursor: {:?}",
- self.id,
- msg_count,
- state.processed_rows,
- state.last_timestamp.as_deref()
- );
- // lock released here
- }
-
- let persisted = ConnectorState::serialize(
- &PersistedState::V3(for_serialize),
- CONNECTOR_NAME,
+ poll_event!(
+ "{CONNECTOR_NAME} ID: {} produced {} messages (V3). \
+ Total: {}. Cursor: {:?}",
self.id,
+ msg_count,
+ new.processed_rows,
+ new.last_timestamp.as_deref()
);
+ let candidate = PersistedState::V3(new);
+ let persisted =
+ ConnectorState::serialize(&candidate, CONNECTOR_NAME, self.id);
+ *self.pending_state.lock().await = Some(candidate);
match &persisted {
Some(_) => {
// Relaxed: the SDK drives poll() from a single select! loop —
@@ -681,6 +675,24 @@ impl Source for InfluxDbSource {
}
}
+ async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(), Error> {
+ let pending = self.pending_state.lock().await.take();
+ if result == SourceBatchResult::Ack
+ && let Some(pending) = pending
+ {
+ match (&self.version_state, pending) {
+ (VersionState::V2(state), PersistedState::V2(candidate)) => {
+ *state.lock().await = candidate;
+ }
+ (VersionState::V3(state), PersistedState::V3(candidate)) => {
+ *state.lock().await = candidate;
+ }
+ _ => return Err(Error::InvalidState),
+ }
+ }
+ Ok(())
+ }
+
async fn close(&mut self) -> Result<(), Error> {
self.client = None;
let processed = match &self.version_state {
@@ -867,6 +879,9 @@ fn apply_v2_cursor_advance(
#[cfg(test)]
mod tests {
use super::*;
+ use axum::http::{HeaderMap, StatusCode};
+ use axum::routing::{get, post};
+ use axum::{Json, Router};
use common::{V2SourceConfig, V3SourceConfig};
use secrecy::SecretString;
@@ -922,6 +937,185 @@ mod tests {
})
}
+ #[tokio::test]
+ async fn open_authenticates_health_check() {
+ for (mut config, expected_auth) in [
+ (make_v2_config(), "Token test_token"),
+ (make_v3_config(), "Bearer test_token"),
+ ] {
+ let router = Router::new().route(
+ "/health",
+ get(move |headers: HeaderMap| async move {
+ if headers
+ .get("authorization")
+ .and_then(|value| value.to_str().ok())
+ == Some(expected_auth)
+ {
+ StatusCode::OK
+ } else {
+ StatusCode::UNAUTHORIZED
+ }
+ }),
+ );
+ let (url, server) = start_http_server(router).await;
+ match &mut config {
+ InfluxDbSourceConfig::V2(config) => {
+ config.url = url;
+ config.max_open_retries = Some(1);
+ }
+ InfluxDbSourceConfig::V3(config) => {
+ config.url = url;
+ config.max_open_retries = Some(1);
+ }
+ }
+ let mut source = InfluxDbSource::new(1, config, None);
+ let opened = source.open().await;
+ server.abort();
+ let _ = server.await;
+ assert!(
+ opened.is_ok(),
+ "health check must use the configured version's authentication: {opened:?}"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn rejected_batches_preserve_cursor_and_tied_row_position() {
+ const TIMESTAMP: &str = "2026-01-01T00:00:00Z";
+ for mut config in [make_v2_config(), make_v3_config()] {
+ let is_v3 = matches!(config, InfluxDbSourceConfig::V3(_));
+ let endpoint = if is_v3 {
+ "/api/v3/query_sql"
+ } else {
+ "/api/v2/query"
+ };
+ let router = Router::new()
+ .route("/health", get(|| async { StatusCode::OK }))
+ .route(
+ endpoint,
+ post(move |Json(body): Json<serde_json::Value>| async move {
+ if is_v3 {
+ let query = body["q"].as_str().unwrap();
+ let tokens: Vec<_> = query.split_whitespace().collect();
+ let limit = tokens
+ [tokens.iter().position(|token| *token == "LIMIT").unwrap() + 1]
+ .parse::<usize>()
+ .unwrap();
+ let offset = tokens
+ [tokens.iter().position(|token| *token == "OFFSET").unwrap() + 1]
+ .parse::<usize>()
+ .unwrap();
+ if query.contains(TIMESTAMP) {
+ String::new()
+ } else {
+ (0..2)
+ .skip(offset)
+ .take(limit)
+ .map(|row| {
+ serde_json::json!({"time": TIMESTAMP, "value": row})
+ .to_string()
+ })
+ .collect::<Vec<_>>()
+ .join("\n")
+ }
+ } else {
+ let query = body["query"].as_str().unwrap();
+ let limit = query
+ .rsplit_once("limit(n: ")
+ .unwrap()
+ .1
+ .trim_end_matches(')')
+ .parse::<usize>()
+ .unwrap();
+ let mut csv = String::from("_time,_value\n");
+ for row in (0..2).take(limit) {
+ csv.push_str(&format!("{TIMESTAMP},{row}\n"));
+ }
+ csv
+ }
+ }),
+ );
+ let (url, server) = start_http_server(router).await;
+ match &mut config {
+ InfluxDbSourceConfig::V2(config) => {
+ config.url = url;
+ config.batch_size = Some(1);
+ config.poll_interval = Some("1ms".to_string());
+ }
+ InfluxDbSourceConfig::V3(config) => {
+ config.url = url;
+ config.batch_size = Some(1);
+ config.poll_interval = Some("1ms".to_string());
+ }
+ }
+ let mut source = InfluxDbSource::new(1, config.clone(), None);
+ source.open().await.unwrap();
+ let first = source.poll().await.unwrap();
+ source
+ .on_batch_result(SourceBatchResult::Nack)
+ .await
+ .unwrap();
+ let retried = source.poll().await.unwrap();
+ source
+ .on_batch_result(SourceBatchResult::Ack)
+ .await
+ .unwrap();
+ let next = source.poll().await.unwrap();
+ source
+ .on_batch_result(SourceBatchResult::Ack)
+ .await
+ .unwrap();
+ let empty = source.poll().await.unwrap();
+ source
+ .on_batch_result(SourceBatchResult::Ack)
+ .await
+ .unwrap();
+ source.close().await.unwrap();
+
+ let has_checkpoint = empty.state.is_some();
+ let mut restored = InfluxDbSource::new(1, config, empty.state);
+ restored.open().await.unwrap();
+ let after_restart = restored.poll().await.unwrap();
+ restored.close().await.unwrap();
+ server.abort();
+ let _ = server.await;
+
+ assert_eq!(first.messages.len(), 1);
+ assert_eq!(
+ retried.messages.len(),
+ 1,
+ "Nack must make the rejected batch eligible again"
+ );
+ assert_eq!(
+ serde_json::from_slice::<serde_json::Value>(&retried.messages[0].payload).unwrap(),
+ serde_json::from_slice::<serde_json::Value>(&first.messages[0].payload).unwrap(),
+ "Nack must preserve the position within the timestamp group"
+ );
+ assert_eq!(retried.messages[0].id, first.messages[0].id);
+ assert_eq!(
+ next.messages.len(),
+ 1,
+ "Ack must advance to the second tied row"
+ );
+ assert_ne!(next.messages[0].payload, first.messages[0].payload);
+ assert!(empty.messages.is_empty());
+ assert!(has_checkpoint, "empty polls must retain the checkpoint");
+ assert!(
+ after_restart.messages.is_empty(),
+ "restoring the acknowledged state must not replay rows"
+ );
+ }
+ }
+
+ async fn start_http_server(router: Router) -> (String, tokio::task::JoinHandle<()>) {
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let url = format!("http://{}", listener.local_addr().unwrap());
+ let server = tokio::spawn(async move {
+ axum::serve(listener, router).await.unwrap();
+ });
+ (url, server)
+ }
+
#[test]
fn v2_source_new_creates_v2_state() {
let source = InfluxDbSource::new(1, make_v2_config(), None);
diff --git a/core/connectors/sources/influxdb_source/src/row.rs b/core/connectors/sources/influxdb_source/src/row.rs
index 7871ec61b..a9b79b416 100644
--- a/core/connectors/sources/influxdb_source/src/row.rs
+++ b/core/connectors/sources/influxdb_source/src/row.rs
@@ -17,10 +17,8 @@
//! Query-response parsers for InfluxDB V2 (annotated CSV) and V3 (JSONL).
//!
-//! Both parsers produce `Vec<Row>` — a list of field-name → string-value maps.
-//! The cursor-tracking and payload-building logic in the source connector
-//! operates on this common representation so it runs unchanged regardless of
-//! which InfluxDB version is in use.
+//! Both parsers produce `Vec<Row>` for the version-specific cursor and payload
+//! logic. V2 retains CSV strings; V3 preserves typed JSON values.
use ahash::AHashMap;
use csv::StringRecord;
@@ -30,11 +28,9 @@ use std::sync::Arc;
/// A single row returned by a query, field name → typed JSON value.
///
-/// V2 (annotated CSV) stores all values as `Value::String` since CSV has no
-/// type information; `parse_scalar` in `build_payload` converts them to typed
-/// values when building the message payload. V3 (JSONL) stores typed values
-/// directly — numbers, booleans, and nulls arrive pre-typed from SQL, so no
-/// string-round-trip parse is needed.
+/// V2 stores cells as `Value::String`, ignoring annotation datatype/default values.
+/// `parse_scalar` infers scalars when building whole-row payloads. V3 preserves
+/// the typed JSON values returned by SQL.
///
/// Column names are interned as `Arc<str>` so they are allocated once per
/// unique name and cloned cheaply (pointer bump) for every data row.
@@ -42,29 +38,21 @@ pub(crate) type Row = AHashMap<Arc<str>, serde_json::Value>;
// ── InfluxDB V2 — annotated CSV ───────────────────────────────────────────────
-/// Return `true` if `record` is a CSV header row.
-///
-/// Checks for any of the standard InfluxDB temporal column names:
-/// `_time`, `_start`, or `_stop`. Regular time-series queries include `_time`;
-/// Flux window-aggregate queries (`count()`, `mean()`, `distinct()`) produce
-/// result tables with `_start` and `_stop` but no `_time`. Requiring only
-/// `_time` would cause those header rows to be missed, silently dropping all
-/// subsequent data rows until the next recognised header.
-///
-/// InfluxDB annotation rows (`#group`, `#datatype`, `#default`) are already
-/// filtered out earlier in [`parse_csv_rows`] by the leading-`#` check, so
-/// they will never reach this function.
-fn is_header_record(record: &StringRecord) -> bool {
- record
- .iter()
- .any(|v| v == "_time" || v == "_start" || v == "_stop")
+/// Recognize temporal headers when annotation rows are absent.
+/// Existing column positions keep payload strings from replacing the header.
+fn is_header_record(record: &StringRecord, headers: Option<&StringRecord>) -> bool {
+ record.iter().enumerate().any(|(index, value)| {
+ matches!(value, "_time" | "_start" | "_stop")
+ && headers.is_none_or(|headers| headers.get(index) == Some(value))
+ })
}
/// Parse an InfluxDB V2 annotated-CSV response body into a list of rows.
///
/// - Annotation rows (first field starts with `#`) are skipped.
/// - Blank lines are skipped.
-/// - The first non-annotation row containing `_time`, `_start`, or `_stop` becomes the header.
+/// - The row after annotations supplies the header, including custom cursor names.
+/// - Without annotations, temporal labels identify the header.
/// - Repeated identical header rows (multi-table result format) are skipped.
/// - Each subsequent data row is mapped `header[i] → row[i]`.
pub(crate) fn parse_csv_rows(csv_text: &str) -> Result<Vec<Row>, Error> {
@@ -74,6 +62,7 @@ pub(crate) fn parse_csv_rows(csv_text: &str) -> Result<Vec<Row>, Error> {
.from_reader(csv_text.as_bytes());
let mut headers: Option<StringRecord> = None;
+ let mut expect_header = false;
// Interned column names: allocated once per unique name when a header row is
// seen, then cheap-cloned (pointer bump) for every data row below.
let mut header_keys: Vec<Arc<str>> = Vec::new();
@@ -90,12 +79,14 @@ pub(crate) fn parse_csv_rows(csv_text: &str) -> Result<Vec<Row>, Error> {
if let Some(first) = record.get(0)
&& first.starts_with('#')
{
+ expect_header = true;
continue;
}
- if is_header_record(&record) {
+ if expect_header || is_header_record(&record, headers.as_ref()) {
header_keys = record.iter().map(Arc::<str>::from).collect();
headers = Some(record.clone());
+ expect_header = false;
continue;
}
@@ -223,6 +214,73 @@ mod tests {
assert!(parse_csv_rows("").unwrap().is_empty());
}
+ #[test]
+ fn csv_temporal_labels_in_payload_are_not_headers() {
+ for annotations in ["", "#datatype,dateTime:RFC3339,string\n"] {
+ let csv = format!(
+ "{annotations},_time,_value\n\
+ ,2026-01-01T00:00:00Z,_time\n\
+ ,2026-01-01T00:00:01Z,_start\n\
+ ,2026-01-01T00:00:02Z,_stop\n"
+ );
+ let rows = parse_csv_rows(&csv).unwrap();
+ let values: Vec<_> = rows
+ .iter()
+ .map(|row| row.get("_value").and_then(|value| value.as_str()))
+ .collect();
+ assert_eq!(
+ values,
+ vec![Some("_time"), Some("_start"), Some("_stop")],
+ "valid string values must not replace the active header"
+ );
+ }
+ }
+
+ #[test]
+ fn csv_annotated_tables_can_reorder_columns() {
+ let csv = "#datatype,dateTime:RFC3339,string\n\
+ ,_time,_value\n\
+ ,2026-01-01T00:00:00Z,_stop\n\
+ #datatype,string,dateTime:RFC3339\n\
+ ,_value,_time\n\
+ ,_start,2026-01-01T00:00:01Z\n";
+ let rows = parse_csv_rows(csv).unwrap();
+ assert_eq!(rows.len(), 2);
+ assert_eq!(
+ rows[0].get("_value").and_then(|value| value.as_str()),
+ Some("_stop")
+ );
+ assert_eq!(
+ rows[1].get("_value").and_then(|value| value.as_str()),
+ Some("_start")
+ );
+ assert_eq!(
+ rows[1].get("_time").and_then(|value| value.as_str()),
+ Some("2026-01-01T00:00:01Z")
+ );
+ }
+
+ #[test]
+ fn csv_annotated_header_supports_custom_cursor_names() {
+ let csv = "#datatype,dateTime:RFC3339,string\n\
+ ,event_time,payload\n\
+ ,2026-01-01T00:00:00Z,example\n";
+ let rows = parse_csv_rows(csv).unwrap();
+ assert_eq!(
+ rows.len(),
+ 1,
+ "annotations identify a table header without reserved column names"
+ );
+ assert_eq!(
+ rows[0].get("event_time").and_then(|value| value.as_str()),
+ Some("2026-01-01T00:00:00Z")
+ );
+ assert_eq!(
+ rows[0].get("payload").and_then(|value| value.as_str()),
+ Some("example")
+ );
+ }
+
#[test]
fn csv_skips_annotation_rows() {
let csv = "#group,false\n#datatype,string\n_time,_value\n2024-01-01T00:00:00Z,42\n";
diff --git a/core/connectors/sources/influxdb_source/src/v2.rs b/core/connectors/sources/influxdb_source/src/v2.rs
index 16e7ebcc8..e8d2b5993 100644
--- a/core/connectors/sources/influxdb_source/src/v2.rs
+++ b/core/connectors/sources/influxdb_source/src/v2.rs
@@ -63,8 +63,8 @@ const MAX_RESPONSE_BODY_BYTES: usize = 256 * 1024 * 1024; // 256 MiB
/// The limit is inflated by `already_seen` (rows at the current cursor
/// timestamp that were delivered in a previous batch) so that re-fetching
/// with `>= cursor` returns enough rows to skip them and still fill a full
-/// batch. Inflation is capped at `MAX_SKIP_INFLATION_FACTOR × batch_size`
-/// to prevent excessively large queries when the cursor is stuck.
+/// batch. The additional skip allowance is capped at
+/// `MAX_SKIP_INFLATION_FACTOR * batch_size`, limiting the total to eleven batches.
fn render_query(config: &V2SourceConfig, cursor: &str, already_seen: u64) -> Result<String, Error> {
validate_cursor(cursor)?;
let batch = config.batch_size.unwrap_or(500).max(1) as u64;
@@ -765,8 +765,8 @@ pub(crate) struct RowProcessingResult {
///
/// ## Message identity
///
-/// A single random UUID is generated per call; per-message IDs are derived by
-/// adding the message's position to that base, keeping PRNG work O(1) per batch.
+/// IDs add the result position to timestamp nanoseconds, with a random batch
+/// base for missing/out-of-range timestamps. Distinct rows can collide across batches.
///
/// ## Parameters
///
@@ -793,8 +793,7 @@ pub(crate) fn process_rows(
let mut max_cursor_parsed: Option<DateTime<Utc>> = None;
let mut rows_at_max_cursor = 0u64;
let mut skipped = 0u64;
- // Generate the base UUID once per poll; derive per-message IDs by addition.
- // This is O(1) PRNG calls per batch instead of O(n), measurable at batch ≥ 100.
+ // Fallback base for rows without a timestamp representable as nanoseconds.
let id_base = Uuid::new_v4().as_u128();
// Parse once outside the loop to normalise RFC 3339 variants (Z vs +00:00)
// so skip logic compares instants, not strings.
@@ -839,9 +838,8 @@ pub(crate) fn process_rows(
}
}
- // Stable ID: cursor timestamp nanoseconds + DB-absolute row position.
- // Position = already_seen + emitted so far; encodes the row's global slot
- // in the result set so IDs are stable across re-polls with changing already_seen.
+ // The position distinguishes ties within a result, but this sum is not
+ // a unique row key across timestamps or changing query boundaries.
let global_pos = already_seen as u128 + messages.len() as u128;
let msg_id = cv_dt
.and_then(|dt| dt.timestamp_nanos_opt())
diff --git a/core/connectors/sources/influxdb_source/src/v3.rs b/core/connectors/sources/influxdb_source/src/v3.rs
index 20cc7e170..7d18efe03 100644
--- a/core/connectors/sources/influxdb_source/src/v3.rs
+++ b/core/connectors/sources/influxdb_source/src/v3.rs
@@ -17,12 +17,10 @@
//! InfluxDB V3 source — SQL queries, JSONL responses, Bearer auth.
//!
-//! V3 uses strict `> cursor` semantics. DataFusion/Parquet does not guarantee
-//! stable ordering for rows that share the same timestamp, so the V2 skip-N
-//! approach is not safe here. If a full batch is returned and all rows share
-//! the same timestamp, the cursor cannot advance — the effective batch size is
-//! doubled each poll up to `stuck_batch_cap_factor × batch_size`. If the cap
-//! is reached, the circuit breaker is tripped.
+//! V3 uses strict `> cursor` semantics and requires stable timestamp/tiebreaker
+//! ordering. A full tied batch retains the cursor, advances OFFSET and doubles
+//! the query size up to its cap. Reaching the cap records a circuit-breaker
+//! failure; mixed full batches defer their final timestamp group.
use crate::common::{
DEFAULT_V3_CURSOR_FIELD, PayloadFormat, Row, RowContext, V3SourceConfig, V3State,
@@ -334,15 +332,12 @@ pub(crate) struct RowProcessingResult {
///
/// - Any row whose cursor field value fails [`normalize_v3_timestamp`] (not a valid RFC 3339
/// timestamp even after appending `"Z"`) is an immediate error.
-/// - If the batch is non-empty but *no* row contains the cursor field, an error is
-/// returned — the cursor cannot advance and the connector would re-deliver the same
-/// rows indefinitely. Individual rows that merely lack the cursor field (while at
-/// least one other row has it) still produce messages without error.
+/// - Any row with a missing or non-string cursor field is an immediate error.
///
/// ## Message identity
///
-/// A single random UUID is generated per call; per-message IDs are derived by
-/// adding the message's position to that base, keeping PRNG work O(1) per batch.
+/// IDs add the result position to timestamp nanoseconds. Distinct rows can
+/// collide across batches; these values are not unique database-row keys.
///
/// ## Parameters
///
@@ -389,8 +384,8 @@ pub(crate) fn process_rows(
))
})?;
let (cv_owned, cv_dt) = normalize_v3_timestamp(raw_cv)?;
- // Stable ID using i128 arithmetic to avoid the i64 nanosecond overflow
- // that occurs for timestamps outside ~1678-2262 CE.
+ // i128 avoids the i64 nanosecond range limit; adding the result position
+ // still permits collisions across timestamps or query boundaries.
let nanos_i128 =
cv_dt.timestamp() as i128 * 1_000_000_000 + cv_dt.timestamp_subsec_nanos() as i128;
let this_row_id = (nanos_i128 as u128).wrapping_add(db_pos);
@@ -533,7 +528,6 @@ pub(crate) async fn poll(
if cap_factor > 0
&& full_batch
&& !all_same_timestamp
- && result.rows_at_max_cursor > 1
&& let Some(penultimate) = result.penultimate_cursor
{
let safe_count = result.safe_message_count;
@@ -622,11 +616,10 @@ pub(crate) async fn poll(
None => {
warn!(
"InfluxDB V3 source — stuck-timestamp cap reached at batch size {effective_batch}; \
- tripping circuit breaker to prevent an infinite loop"
+ recording a circuit-breaker failure"
);
- // Reset effective_batch_size to base so the next poll after the
- // circuit-breaker cool-down restarts from the configured batch size
- // rather than re-entering at cap and immediately re-tripping.
+ // Restart the next query at the base size instead of returning
+ // immediately to the cap. The caller applies the failure threshold.
// Do NOT update last_timestamp_row_offset on the trip path — no
// messages were emitted, so the offset tiebreaker is unchanged.
Ok(PollResult {
@@ -709,15 +702,14 @@ pub(crate) async fn poll(
// order (the typical time-series pattern). If out-of-order or backdated
// writes are possible in your workload, a concurrent writer could insert
// rows at stuck_cursor AFTER this empty-batch poll, and those rows would
- // be silently skipped because the cursor advances past them here. In that
- // case, use stuck_batch_cap_factor = 0 to disable stuck detection and
- // accept potential re-delivery instead.
+ // be silently skipped because the cursor advances past them here.
+ // Disabling stuck detection does not make backdated writes visible.
_ if state.last_timestamp_row_offset > 0 && state.stuck_cursor.is_some() => {
warn!(
"Advancing cursor past stuck_cursor={:?} on empty follow-up batch. \
Any backdated writes at this timestamp inserted after this poll \
- will be silently skipped. Set stuck_batch_cap_factor=0 to disable \
- stuck detection if your workload has out-of-order ingestion.",
+ will be skipped. Disabling stuck detection does not recover \
+ backdated writes.",
state.stuck_cursor.as_deref()
);
state.stuck_cursor.clone()
@@ -1249,10 +1241,10 @@ mod tests {
#[cfg(test)]
mod http_tests {
use super::*;
- use axum::Router;
use axum::extract::Request;
use axum::http::{HeaderMap, StatusCode};
use axum::routing::post;
+ use axum::{Json, Router};
use secrecy::SecretString;
use std::sync::Arc;
use std::time::Duration;
@@ -1586,6 +1578,78 @@ mod http_tests {
);
}
+ #[tokio::test]
+ async fn poll_preserves_unseen_ties_after_single_maximum_row() {
+ const FIRST: &str = "2026-01-01T00:00:00Z";
+ const SECOND: &str = "2026-01-01T00:00:01Z";
+ const POLL_LIMIT: usize = 4;
+ let app = Router::new().route(
+ "/api/v3/query_sql",
+ post(|Json(body): Json<serde_json::Value>| async move {
+ let query = body["q"].as_str().unwrap();
+ let cursor = query
+ .split_once("time > '")
+ .unwrap()
+ .1
+ .split('\'')
+ .next()
+ .unwrap();
+ let tokens: Vec<_> = query.split_whitespace().collect();
+ let limit = tokens[tokens.iter().position(|token| *token == "LIMIT").unwrap() + 1]
+ .parse::<usize>()
+ .unwrap();
+ let offset = tokens
+ [tokens.iter().position(|token| *token == "OFFSET").unwrap() + 1]
+ .parse::<usize>()
+ .unwrap();
+ [(FIRST, 0), (SECOND, 1), (SECOND, 2)]
+ .into_iter()
+ .filter(|(timestamp, _)| *timestamp > cursor)
+ .skip(offset)
+ .take(limit)
+ .map(|(timestamp, value)| {
+ json!({"time": timestamp, "value": value}).to_string()
+ })
+ .collect::<Vec<_>>()
+ .join("\n")
+ }),
+ );
+ let base = start_server(app).await;
+ let config = V3SourceConfig {
+ batch_size: Some(2),
+ query: "SELECT * FROM t WHERE time > '$cursor' ORDER BY time, value LIMIT $limit OFFSET $offset".to_string(),
+ ..make_config(&base)
+ };
+ let client = make_client();
+ let mut state = V3State::default();
+ let mut values = Vec::new();
+ for _ in 0..POLL_LIMIT {
+ let result = poll(
+ &client,
+ &config,
+ "Bearer tok",
+ &state,
+ PayloadFormat::Json,
+ true,
+ )
+ .await
+ .unwrap();
+ for message in result.messages {
+ let payload: serde_json::Value = serde_json::from_slice(&message.payload).unwrap();
+ values.push(payload["value"].as_u64().unwrap());
+ }
+ state = result.new_state;
+ }
+ assert_eq!(
+ values,
+ vec![0, 1, 2],
+ "a batch boundary must not skip unseen rows at its maximum timestamp"
+ );
+ assert_eq!(state.processed_rows, 3);
+ assert_eq!(state.last_timestamp.as_deref(), Some(SECOND));
+ assert_eq!(state.last_timestamp_row_offset, 0);
+ }
+
#[tokio::test]
async fn poll_detects_stuck_batch_and_doubles_batch_size() {
// All batch_size rows share the same timestamp as the cursor → stuck.
diff --git a/core/connectors/sources/postgres_source/README.md b/core/connectors/sources/postgres_source/README.md
index d3f2feb3c..d5ea0d241 100644
--- a/core/connectors/sources/postgres_source/README.md
+++ b/core/connectors/sources/postgres_source/README.md
@@ -8,17 +8,45 @@ The PostgreSQL source connector fetches data from PostgreSQL databases and strea
- **Change Data Capture**: Monitor database changes using PostgreSQL logical replication
- **Flexible Payload Extraction**: Extract BYTEA, TEXT, or JSONB columns directly as payload
- **Custom Queries**: Use custom SQL queries with parameter substitution
-- **Delete After Read**: Automatically delete rows after processing
-- **Mark as Processed**: Mark rows as processed using a boolean column
-- **Multiple Tables**: Monitor multiple tables simultaneously
+- **Delete After Read**: Delete rows after delivery and checkpoint acknowledgement
+- **Mark as Processed**: Mark rows after acknowledgement using a boolean column
+- **Multiple Tables**: Poll multiple tables sequentially
- **Batch Processing**: Fetch data in configurable batch sizes
-- **Offset Tracking**: Resume incremental polling from the last acknowledged offset
+- **Offset Tracking**: Resume from acknowledged per-table offsets; retries can duplicate rows
## Configuration
+Use the broker credentials and main runtime configuration from the
+[source guide](https://iggy.apache.org/docs/connectors/sources/source/#configuration).
+Build from the matching checkout root:
+
+```bash
+cargo build --release -p iggy_connector_postgres_source
+```
+
+The examples use the `iggy` database and `iggy`/`iggy` credentials. Create the
+`users` and `orders` tables using the SQL on the
+[Postgres source page](https://iggy.apache.org/docs/connectors/sources/postgres/).
+Save one connector entry in the runtime's connector directory and run the runtime
+from the checkout root. Create its destination stream and topic before starting it.
+
```toml
+type = "source"
+key = "postgres"
+enabled = true
+version = 0
+name = "Postgres source"
+path = "target/release/libiggy_connector_postgres_source"
+plugin_config_format = "toml"
+
+[[streams]]
+stream = "user_events"
+topic = "users"
+schema = "json"
+batch_length = 100
+
[plugin_config]
-connection_string = "postgresql://user:pass@localhost:5432/database"
+connection_string = "postgresql://iggy:iggy@localhost:5432/iggy"
mode = "polling"
tables = ["users", "orders"]
poll_interval = "1s"
@@ -29,17 +57,17 @@ max_connections = 10
snake_case_columns = false
include_metadata = true
-# Payload extraction (optional)
-payload_column = "payload"
-payload_format = "bytea"
+# Payload extraction requires a matching column and stream schema.
+# payload_column = "payload"
+# payload_format = "bytea"
# Delete/mark processed (optional)
delete_after_read = false
-processed_column = "is_processed"
+# processed_column = "is_processed"
primary_key_column = "id"
# Custom query (optional)
-custom_query = "SELECT * FROM $table WHERE id > $offset ORDER BY id LIMIT $limit"
+# custom_query = "SELECT * FROM $table WHERE id > $offset ORDER BY id LIMIT $limit"
# CDC options (only used when mode = "cdc")
replication_slot = "iggy_slot"
@@ -53,25 +81,25 @@ 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 | `10s` | How often to poll (e.g., `1s`, `5m`) |
-| `batch_size` | u32 | `1000` | Max rows per poll |
+| `tables` | array | required | Polling tables; empty polls none. Empty captures all tables in CDC |
+| `poll_interval` | string | `10s` | Delay before each cycle; invalid values fall back to `10s` |
+| `batch_size` | u32 | `1000` | Limit per table in polling mode; CDC limits are checked at transaction boundaries |
| `tracking_column` | string | `id` | Unique, non-null column for incremental updates |
-| `initial_offset` | string | none | Starting value for tracking column |
+| `initial_offset` | string | none | Exclusive starting value when a saved table offset is absent |
| `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 |
+| `include_metadata` | bool | `true` | Polling layout; false currently adds `data.data` while retaining the envelope |
| `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; takes precedence over `processed_column` |
-| `processed_column` | string | none | Boolean column to mark as processed when `delete_after_read` is false |
+| `payload_format` | string | `json` | Selected-column format; see the schema pairings below |
+| `delete_after_read` | bool | `false` | Delete selected rows after Ack |
+| `processed_column` | string | none | Boolean column to filter on FALSE and mark after Ack |
| `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` |
+| `cdc_backend` | string | `builtin` | Only `builtin` is implemented |
| `verbose_logging` | bool | `false` | Log at info level instead of debug |
-| `max_retries` | u32 | `3` | Max attempts for transient errors; `0` and `1` both perform one attempt |
+| `max_retries` | u32 | `3` | Total attempts for transient errors, including the first; zero still makes one attempt |
| `retry_delay` | string | `1s` | Base delay between retries (e.g., `500ms`, `2s`) |
## Delivery Failures
@@ -97,7 +125,7 @@ state before restarting.
### JSON Mode (Default)
-When `payload_column` is not set, each row is wrapped in a `DatabaseRecord` JSON structure:
+When `payload_column` is not set, polling wraps each row in a `DatabaseRecord` JSON structure:
```json
{
@@ -113,20 +141,35 @@ When `payload_column` is not set, each row is wrapped in a `DatabaseRecord` JSON
}
```
-The stream config should use `schema = "json"`.
+The stream config should use `schema = "json"`. With `include_metadata = false`,
+the envelope currently remains and the row moves to `data.data`. The timestamp
+is generated during polling. The runtime does not transfer the plugin's timestamp
+fields into broker message metadata.
### Payload Column Extraction
-When `payload_column` is set, the connector extracts that column directly as the Iggy message payload. The `payload_format` option determines how the column is read:
+In polling mode, an existing `payload_column` bypasses the envelope. The `payload_format` option determines how the column is read:
| Format | Column Type | Schema | Description |
| ------ | ----------- | ------ | ----------- |
| `bytea` / `raw` | `BYTEA` | `raw` | Raw bytes passthrough |
| `text` | `TEXT` | `text` | UTF-8 text |
-| `json_direct` / `jsonb` | `JSONB` | `json` | JSON object serialized to bytes |
+| `json_direct` / `jsonb` / `jsonb_direct` | `JSON`, `JSONB` | `json` | The JSON value serialized to bytes |
+| `json` (default) | `BYTEA` | `json` | Bytes must already contain valid JSON |
+
+Null BYTEA and TEXT payloads become empty bytes. Iggy rejects empty payloads,
+so the entire batch receives Nack and five consecutive failures stop polling.
+Use non-null, non-empty BYTEA/text payloads. Null JSON/JSONB becomes JSON
+`null` and can be delivered. Missing selected columns fall back to the envelope while retaining the
+selected format's schema. Type mismatches reject the poll. Without a selected
+column, every payload-format option emits whole-row JSON.
## Payload Format Examples
+Keep the connector-entry header from the configuration above, and replace its
+`[[streams]]` and `[plugin_config]` sections with one variant. Run the matching
+SQL first; insert rows with increasing unique IDs to produce messages.
+
### BYTEA (Raw Bytes)
Extract raw bytes from a BYTEA column:
@@ -146,6 +189,8 @@ schema = "raw"
batch_length = 100
[plugin_config]
+connection_string = "postgresql://iggy:iggy@localhost:5432/iggy"
+mode = "polling"
tables = ["message_queue"]
tracking_column = "id"
payload_column = "payload"
@@ -171,6 +216,8 @@ schema = "text"
batch_length = 100
[plugin_config]
+connection_string = "postgresql://iggy:iggy@localhost:5432/iggy"
+mode = "polling"
tables = ["logs"]
tracking_column = "id"
payload_column = "message"
@@ -196,6 +243,8 @@ schema = "json"
batch_length = 100
[plugin_config]
+connection_string = "postgresql://iggy:iggy@localhost:5432/iggy"
+mode = "polling"
tables = ["events"]
tracking_column = "id"
payload_column = "data"
@@ -204,17 +253,23 @@ payload_format = "json_direct"
## Custom Query Parameters
-When using `custom_query`, these placeholders are available:
+A `custom_query` replaces the entire default query, including its ordering,
+limit and processed-row filter. These placeholders use textual substitution.
+The connector quotes and escapes `$offset` as a SQL value, including templates
+that already wrap it in single quotes. Other substitutions are inserted directly:
| Placeholder | Replaced With |
| ----------- | ------------- |
| `$table` | Current table name |
-| `$offset` | Last processed offset (or `initial_offset`) |
+| `$offset` | Last acknowledged offset, `initial_offset`, or an empty string if neither exists |
| `$limit` | `batch_size` value |
| `$now` | Current UTC timestamp (RFC3339) |
| `$now_unix` | Current Unix timestamp (seconds) |
-Example:
+This template requires a non-null, unique, increasing `created_at` column, an
+optional `scheduled_at` timestamp, and an RFC3339 `initial_offset` before the
+first row. Set `tracking_column = "created_at"` and save it as the
+`custom_query` string:
```sql
SELECT * FROM $table
@@ -239,8 +294,7 @@ column lacks a valid single-column unique index or permits null values.
## Delete After Read / Mark as Processed
-Both options may be present for compatibility. When `delete_after_read` is
-`true`, rows are deleted and `processed_column` is ignored.
+Apply these options inside the existing `[plugin_config]` section.
The resolved cleanup key is `primary_key_column`, or `tracking_column` when the
former is unset. For every configured table, it must be a non-null column with
@@ -252,7 +306,7 @@ configuration instead of applying old work to a different column or action.
### Delete After Read
-Deletes rows from the source table only after Iggy acknowledges the batch:
+Deletes selected rows after the runtime forwards the batch, saves its checkpoint and sends Ack:
```toml
[plugin_config]
@@ -262,7 +316,7 @@ primary_key_column = "id"
### Mark as Processed
-Updates a boolean column after Iggy acknowledges the batch instead of deleting:
+Updates a boolean column after Ack instead of deleting:
```toml
[plugin_config]
@@ -270,25 +324,27 @@ processed_column = "is_processed"
primary_key_column = "id"
```
-Your table needs the boolean column:
+Each configured table needs the boolean column:
```sql
ALTER TABLE users ADD COLUMN is_processed BOOLEAN DEFAULT false;
+ALTER TABLE orders ADD COLUMN is_processed BOOLEAN DEFAULT false;
```
-When `processed_column` is set, the connector automatically adds a `WHERE is_processed = FALSE` filter to the polling query, so only unprocessed rows are fetched. This improves polling efficiency as the table grows.
+When `processed_column` is set, the default polling query adds an
+`is_processed = FALSE` condition. A custom query must add its own filter.
+`delete_after_read = true` takes precedence over marking. Cleanup keys are staged
+during polling and discarded on Nack or a failed poll, leaving rows available
+for replay. Cleanup runs after delivery and checkpointing. Unfinished cleanup
+is retried before polling more rows and restored from the checkpoint after a
+restart. Three consecutive cleanup failures stop the source. Cleanup across
+tables is not atomic.
With the generated polling query, a row whose tracking value moves past the
batch boundary between poll and acknowledgement is left unchanged and returns
-in a later poll. Custom queries do not apply this boundary because their result
-order is not guaranteed. Cleanup also matches the row version captured by the
-poll, so replay cannot delete or mark a replacement row that reused the same
-key.
-
-For generated polling queries, the connector persists the acknowledged offset
-before deleting or marking rows. If it stops in between, the rows have been
-delivered but may remain unchanged in PostgreSQL. The persisted offset prevents
-those rows from being selected again.
+in a later poll. Custom queries do not apply this boundary. Cleanup also matches
+the row version captured by the poll, so replay cannot delete or mark a
+replacement row that reused the same key.
## Supported Column Types
@@ -299,13 +355,21 @@ The connector handles these PostgreSQL types in JSON mode:
| `BOOL` | boolean |
| `INT2`, `INT4`, `INT8` | number |
| `FLOAT4`, `FLOAT8` | number |
-| `NUMERIC` | string (exact decimal representation) |
-| `VARCHAR`, `TEXT`, `CHAR` | string |
-| `TIMESTAMP`, `TIMESTAMPTZ` | string (RFC3339) |
+| `NUMERIC` | exact decimal string; non-finite values become `"NaN"`, `"Infinity"`, or `"-Infinity"` |
+| `VARCHAR`, `TEXT`, `BPCHAR` (`CHAR(n)`), `NAME` | string |
+| `TIMESTAMP` | string without a timezone, such as `2024-01-15 10:30:00` |
+| `TIMESTAMPTZ` | RFC3339 string |
+| `DATE`, `TIME`, `TIMETZ`, `INTERVAL` | formatted string |
| `UUID` | string |
-| `JSON`, `JSONB` | object |
+| `JSON`, `JSONB` | original JSON value, including scalars, arrays and null |
| `BYTEA` | base64 string |
-| Other | string (fallback) |
+| Supported arrays | JSON array, preserving null elements |
+| Other | UTF-8 driver bytes if valid, otherwise base64; not a general semantic conversion |
+
+Finite NUMERIC values are decoded through BigDecimal without floating-point
+conversion, preserving exact tracking boundaries. SQL NULL remains JSON null.
+The array mappings cover boolean, integer/OID, floating-point, text, UUID,
+JSON/JSONB, date/time/timestamp and interval arrays.
## CDC Mode
@@ -313,7 +377,11 @@ CDC requires PostgreSQL 11 or newer and logical replication setup:
1. Set `wal_level = logical` in `postgresql.conf`
2. Restart PostgreSQL
-3. Use a direct connection (no pooler)
+3. Allow the login to use logical-decoding SQL functions and ensure `test_decoding` is installed and allowed
+
+The builtin backend uses ordinary SQL connections. A proxy must support its SQL
+and replication-slot operations; the connector does not open a replication-protocol
+connection. It does not create or require a publication.
```toml
[plugin_config]
@@ -322,6 +390,19 @@ tables = ["users", "orders"]
capture_operations = ["INSERT", "UPDATE", "DELETE"]
```
+The `pg_replicate` backend is not implemented. Without `cdc_pg_replicate`, startup
+rejects it; with the feature, polling returns an unimplemented-backend error.
+`capture_operations` accepts uppercase INSERT/UPDATE/DELETE, and an empty array
+emits none. An empty `tables` array captures all tables. Filters apply after the
+slot read, so acknowledged slot advances also pass excluded changes.
+
+CDC emits a JSON envelope rather than applying polling payload extraction.
+Quoted `test_decoding` values, including JSONB and arrays, remain strings.
+Deletes carry replica-identity columns; updates can include `old_data` from
+an old-key tuple. Unchanged TOAST values become null, which does not establish
+that the stored database value is null. The generated timestamp is polling time,
+not a transaction commit timestamp.
+
The connector peeks at logical changes and advances the replication slot only
after Iggy acknowledges the batch. A failed delivery leaves the slot unchanged
so the next poll can read the same changes again.
@@ -331,8 +412,6 @@ each acknowledged batch is decoded twice. Poll and decode errors do not change
the connector's runtime status. Monitor `confirmed_flush_lsn`, retained WAL, and
replication slot lag in PostgreSQL to detect a stuck CDC poller.
-The `pg_replicate` backend requires the `cdc_pg_replicate` feature flag at build time.
-
### Slot Naming
Each CDC connector must use a unique `replication_slot`. Setup accepts any
@@ -367,11 +446,11 @@ schema = "json"
batch_length = 100
[plugin_config]
-connection_string = "postgresql://user:pass@localhost:5432/mydb"
+connection_string = "postgresql://iggy:iggy@localhost:5432/iggy"
mode = "polling"
tables = ["users"]
poll_interval = "1s"
-tracking_column = "updated_at"
+tracking_column = "id"
```
### Raw Payload Passthrough
@@ -384,7 +463,7 @@ schema = "raw"
batch_length = 100
[plugin_config]
-connection_string = "postgresql://user:pass@localhost:5432/mydb"
+connection_string = "postgresql://iggy:iggy@localhost:5432/iggy"
mode = "polling"
tables = ["message_queue"]
poll_interval = "100ms"
@@ -404,7 +483,7 @@ schema = "json"
batch_length = 100
[plugin_config]
-connection_string = "postgresql://user:pass@localhost:5432/mydb"
+connection_string = "postgresql://iggy:iggy@localhost:5432/iggy"
mode = "polling"
tables = ["events"]
poll_interval = "1s"
@@ -423,7 +502,7 @@ schema = "json"
batch_length = 100
[plugin_config]
-connection_string = "postgresql://user:pass@localhost:5432/mydb"
+connection_string = "postgresql://iggy:iggy@localhost:5432/iggy"
mode = "cdc"
tables = ["users", "orders"]
capture_operations = ["INSERT", "UPDATE"]
@@ -443,11 +522,43 @@ The connector stops after three consecutive row-cleanup or replication-slot adva
### SQL Injection Protection
-All table names, column names, and identifiers are properly quoted to prevent SQL injection attacks. User-provided values in tracking offsets are also safely escaped.
+The generated polling and cleanup queries quote table/column identifiers and
+escape value literals. PostgreSQL infers value types from the compared columns,
+so numeric-looking TEXT keys keep their exact values. Custom queries use raw
+text substitution except for the quoted and escaped `$offset` value. Keep
+templates and configured names trusted and quote other substitutions as required
+by their SQL context.
+
+### Cursor and Checkpoint Limits
+
+The default polling query uses `tracking_column > last_offset`, ascending order
+and a per-table limit. Tracking values must be non-null, unique and increasing
+in the database's sort order, with a valid single-column unique index checked
+at startup. Transactions must become visible in tracking-column order; late
+inserts or updates at or below the watermark can be missed. Polling does not
+capture deletions.
+
+Polling stages its per-table offsets, last-poll time and processed-row
+count until the runtime forwards the batch, saves its MessagePack checkpoint
+and sends Ack. Nack discards that candidate. Delivery followed by a failed
+checkpoint can duplicate messages; each poll assigns fresh random UUIDs.
+The default file checkpoint is `local_state/source_postgres.state`; the runtime
+also supports HTTP state storage. Missing or undecodable plugin state starts
+fresh, while storage access errors prevent startup. Stop the connector before
+resetting its checkpoint. Five consecutive Nacks stop the SDK polling loop.
+Delete/mark operations and CDC slot advances run after Ack, as described above.
## Usage with Sink Connector
-The source and sink connectors can work together for pass-through scenarios:
+The source and sink connectors can work together for pass-through scenarios.
+For incremental reads of a sink-created table, use one input stream/topic/partition,
+keep `include_metadata = true` on the sink, and set
+`tracking_column = "iggy_offset"` with `initial_offset = "-1"` on the source,
+so the first message at offset zero is included. Message IDs are not an increasing
+cursor, and offsets from different partitions are not globally unique.
+The source requires `iggy_offset` to be non-null and backed by a single-column
+unique index; enforce both constraints on the sink-created table before starting
+the source.
### Raw Bytes Pass-through
@@ -463,10 +574,10 @@ The source and sink connectors can work together for pass-through scenarios:
In the default JSON mode (no `payload_column`), the Postgres source wraps each row in a
`DatabaseRecord` envelope containing `table_name`, `operation_type`, `timestamp`, `data`, and
-`old_data`. Sinks like Iceberg and Delta expect flat JSON matching the target table schema, so
-the envelope must be unwrapped before the data reaches the sink.
+`old_data`. Sinks like Iceberg and Delta expect JSON matching the target table schema.
+For a table whose columns match the source row, unwrap the envelope before delivery.
-**Option A — use the `unwrap_envelope` transform** on the sink side to extract the `data` field:
+**Option A - use the `unwrap_envelope` transform** on the sink side to extract the `data` field:
```toml
[transforms.unwrap_envelope]
@@ -474,5 +585,5 @@ enabled = true
field = "data"
```
-**Option B — bypass the envelope** by configuring the source with `payload_column` and
+**Option B - bypass the envelope** by configuring the source with `payload_column` and
`payload_format = "json_direct"` to emit raw JSONB directly (see Payload Column Extraction above).
diff --git a/core/connectors/sources/postgres_source/config.toml b/core/connectors/sources/postgres_source/config.toml
index 1ea3757f2..0af5f7f63 100644
--- a/core/connectors/sources/postgres_source/config.toml
+++ b/core/connectors/sources/postgres_source/config.toml
@@ -20,7 +20,7 @@ key = "postgres"
enabled = true
version = 0
name = "Postgres source"
-path = "../../target/release/libiggy_connector_postgres_source"
+path = "target/release/libiggy_connector_postgres_source"
verbose = false
[[streams]]
@@ -30,7 +30,7 @@ schema = "json"
batch_length = 100
[plugin_config]
-connection_string = "postgresql://user:pass@localhost:5432/database"
+connection_string = "postgresql://iggy:iggy@localhost:5432/iggy"
mode = "polling"
tables = ["users", "orders"]
poll_interval = "1s"
diff --git a/core/connectors/sources/postgres_source/src/lib.rs b/core/connectors/sources/postgres_source/src/lib.rs
index a337fb0d7..b835d09b5 100644
--- a/core/connectors/sources/postgres_source/src/lib.rs
+++ b/core/connectors/sources/postgres_source/src/lib.rs
@@ -1446,7 +1446,9 @@ impl PostgresSource {
if !config.payload_col.is_empty() && column.name() == config.payload_col {
extracted_payload =
Some(self.extract_payload_column(row, i, config.payload_format)?);
- continue;
+ if column.name() != config.tracking_column && column.name() != config.pk_column {
+ continue;
+ }
}
let value = extract_column_value(row, i)?;
@@ -3410,16 +3412,20 @@ mod tests {
fn given_custom_query_with_time_params_should_substitute_correctly() {
let src = PostgresSource::new(1, test_config(), None);
- let query = "SELECT * FROM $table WHERE created_at < '$now' AND epoch < $now_unix";
+ let query = "SELECT '$now', $now_unix FROM $table";
let result = src.substitute_query_params(query, "logs", &None, 100);
-
- assert!(result.contains("FROM logs"));
- assert!(!result.contains("$now"));
- let unix_value = result
- .rsplit_once("epoch < ")
- .map(|(_, value)| value)
- .unwrap();
- assert!(unix_value.parse::<i64>().is_ok());
+ let values = result
+ .strip_prefix("SELECT '")
+ .and_then(|value| value.strip_suffix(" FROM logs"))
+ .expect("query structure and table placeholder must be preserved");
+ let (timestamp, unix_seconds) = values.split_once("', ").unwrap();
+ let timestamp = DateTime::parse_from_rfc3339(timestamp)
+ .expect("$now must expand to an RFC3339 timestamp");
+ let unix_seconds: i64 = unix_seconds
+ .parse()
+ .expect("$now_unix must expand to integer seconds");
+
+ assert_eq!(unix_seconds, timestamp.timestamp());
}
#[test]
diff --git a/core/integration/tests/connectors/elasticsearch/elasticsearch_sink.rs b/core/integration/tests/connectors/elasticsearch/elasticsearch_sink.rs
index cef336cc2..d80a92add 100644
--- a/core/integration/tests/connectors/elasticsearch/elasticsearch_sink.rs
+++ b/core/integration/tests/connectors/elasticsearch/elasticsearch_sink.rs
@@ -226,7 +226,7 @@ async fn elasticsearch_sink_preserves_json_structure(
server(connectors_runtime(config_path = "tests/connectors/elasticsearch/sink.toml")),
seed = seeds::connector_stream
)]
-async fn given_rejected_document_when_indexing_should_report_runtime_error(
+async fn rejected_document_reports_runtime_error(
harness: &TestHarness,
fixture: ElasticsearchSinkFixture,
) {
diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs
index e212f1399..05c7c824e 100644
--- a/core/integration/tests/connectors/fixtures/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -80,12 +80,13 @@ pub use mongodb::{
MongoDbSinkFixture, MongoDbSinkJsonFixture, MongoDbSinkWriteConcernFixture,
};
pub use postgres::{
- PostgresOps, PostgresSinkByteaFixture, PostgresSinkFixture, PostgresSinkJsonFixture,
- PostgresSourceByteaFixture, PostgresSourceCdcFixture, PostgresSourceCdcSlowPollFixture,
- PostgresSourceDeleteFixture, PostgresSourceDeleteSlowPollFixture, PostgresSourceJsonFixture,
- PostgresSourceJsonbFixture, PostgresSourceMarkFixture, PostgresSourceNonUniqueCleanupFixture,
+ POSTGRES_LARGE_BATCH_SIZE, PostgresOps, PostgresSinkByteaFixture, PostgresSinkFixture,
+ PostgresSinkJsonFixture, PostgresSinkLargeBatchFixture, PostgresSourceByteaFixture,
+ PostgresSourceCdcFixture, PostgresSourceCdcSlowPollFixture, PostgresSourceDeleteFixture,
+ PostgresSourceDeleteSlowPollFixture, PostgresSourceJsonFixture, PostgresSourceJsonbFixture,
+ PostgresSourceMarkFixture, PostgresSourceNonUniqueCleanupFixture,
PostgresSourceNonUniqueTrackingFixture, PostgresSourceNumericTrackingFixture,
- PostgresSourceOps,
+ PostgresSourceOps, PostgresSourceTextKeyFixture,
};
pub use quickwit::{
QuickwitFixture, QuickwitOps, QuickwitPreCreatedFixture, QuickwitRawFixture,
diff --git a/core/integration/tests/connectors/fixtures/postgres/mod.rs b/core/integration/tests/connectors/fixtures/postgres/mod.rs
index 87c0c22fb..7f186d406 100644
--- a/core/integration/tests/connectors/fixtures/postgres/mod.rs
+++ b/core/integration/tests/connectors/fixtures/postgres/mod.rs
@@ -22,10 +22,13 @@ mod source;
pub use cdc::{PostgresSourceCdcFixture, PostgresSourceCdcSlowPollFixture};
pub use container::{PostgresOps, PostgresSourceOps};
-pub use sink::{PostgresSinkByteaFixture, PostgresSinkFixture, PostgresSinkJsonFixture};
+pub use sink::{
+ POSTGRES_LARGE_BATCH_SIZE, PostgresSinkByteaFixture, PostgresSinkFixture,
+ PostgresSinkJsonFixture, PostgresSinkLargeBatchFixture,
+};
pub use source::{
PostgresSourceByteaFixture, PostgresSourceDeleteFixture, PostgresSourceDeleteSlowPollFixture,
PostgresSourceJsonFixture, PostgresSourceJsonbFixture, PostgresSourceMarkFixture,
PostgresSourceNonUniqueCleanupFixture, PostgresSourceNonUniqueTrackingFixture,
- PostgresSourceNumericTrackingFixture,
+ PostgresSourceNumericTrackingFixture, PostgresSourceTextKeyFixture,
};
diff --git a/core/integration/tests/connectors/fixtures/postgres/sink.rs b/core/integration/tests/connectors/fixtures/postgres/sink.rs
index 00b891fb2..14bde9941 100644
--- a/core/integration/tests/connectors/fixtures/postgres/sink.rs
+++ b/core/integration/tests/connectors/fixtures/postgres/sink.rs
@@ -29,6 +29,10 @@ use std::collections::HashMap;
use std::time::Duration;
use tokio::time::sleep;
+pub const POSTGRES_LARGE_BATCH_SIZE: usize = 10_000;
+const ENV_SINK_BATCH_SIZE: &str = "IGGY_CONNECTORS_SINK_POSTGRES_PLUGIN_CONFIG_BATCH_SIZE";
+const ENV_SINK_BATCH_LENGTH: &str = "IGGY_CONNECTORS_SINK_POSTGRES_STREAMS_0_BATCH_LENGTH";
+
/// PostgreSQL sink connector fixture.
///
/// Starts a PostgreSQL container and provides environment variables
@@ -215,3 +219,35 @@ impl TestFixture for PostgresSinkJsonFixture {
self.inner.connectors_runtime_envs()
}
}
+
+pub struct PostgresSinkLargeBatchFixture {
+ inner: PostgresSinkFixture,
+}
+
+impl std::ops::Deref for PostgresSinkLargeBatchFixture {
+ type Target = PostgresSinkFixture;
+
+ fn deref(&self) -> &Self::Target {
+ &self.inner
+ }
+}
+
+#[async_trait]
+impl TestFixture for PostgresSinkLargeBatchFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let mut inner = PostgresSinkFixture::setup().await?;
+ inner.payload_format = SinkPayloadFormat::Json;
+ inner.schema = SinkSchema::Raw;
+ Ok(Self { inner })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ let mut envs = self.inner.connectors_runtime_envs();
+ envs.insert(ENV_SINK_BATCH_SIZE.to_string(), u32::MAX.to_string());
+ envs.insert(
+ ENV_SINK_BATCH_LENGTH.to_string(),
+ POSTGRES_LARGE_BATCH_SIZE.to_string(),
+ );
+ envs
+ }
+}
diff --git a/core/integration/tests/connectors/fixtures/postgres/source.rs b/core/integration/tests/connectors/fixtures/postgres/source.rs
index f37c59bd4..744ca85bf 100644
--- a/core/integration/tests/connectors/fixtures/postgres/source.rs
+++ b/core/integration/tests/connectors/fixtures/postgres/source.rs
@@ -802,3 +802,51 @@ impl TestFixture for PostgresSourceNonUniqueTrackingFixture {
envs
}
}
+
+pub struct PostgresSourceTextKeyFixture {
+ source: PostgresSourceJsonFixture,
+}
+
+impl PostgresOps for PostgresSourceTextKeyFixture {
+ fn container(&self) -> &PostgresContainer {
+ self.source.container()
+ }
+}
+
+impl PostgresSourceOps for PostgresSourceTextKeyFixture {
+ fn table_name(&self) -> &str {
+ self.source.table_name()
+ }
+}
+
+#[async_trait]
+impl TestFixture for PostgresSourceTextKeyFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let source = PostgresSourceJsonFixture {
+ container: PostgresContainer::start().await?,
+ };
+ let pool = source.create_pool().await?;
+ let query = format!(
+ "CREATE TABLE {} (id TEXT PRIMARY KEY, processed BOOLEAN NOT NULL DEFAULT FALSE)",
+ source.table_name()
+ );
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .execute(&pool)
+ .await
+ .expect("create validated tracking and cleanup key before source startup");
+ pool.close().await;
+ Ok(Self { source })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ let mut envs = self.source.connectors_runtime_envs();
+ envs.insert(ENV_SOURCE_PAYLOAD_COLUMN.to_string(), "id".to_string());
+ envs.insert(ENV_SOURCE_PAYLOAD_FORMAT.to_string(), "text".to_string());
+ envs.insert(
+ ENV_SOURCE_PROCESSED_COLUMN.to_string(),
+ "processed".to_string(),
+ );
+ envs.insert(ENV_SOURCE_STREAMS_0_SCHEMA.to_string(), "text".to_string());
+ envs
+ }
+}
diff --git a/core/integration/tests/connectors/mongodb/mongodb_sink.rs b/core/integration/tests/connectors/mongodb/mongodb_sink.rs
index 4aa4e500f..f10d3dd7f 100644
--- a/core/integration/tests/connectors/mongodb/mongodb_sink.rs
+++ b/core/integration/tests/connectors/mongodb/mongodb_sink.rs
@@ -618,7 +618,7 @@ async fn unordered_duplicate_partial_insert_has_exact_accounting(
server(connectors_runtime(config_path = "tests/connectors/mongodb/sink.toml")),
seed = seeds::connector_stream
)]
-async fn schema_validation_mid_batch_surfaces_hard_error_and_partial_prefix(
+async fn schema_validation_mid_batch_surfaces_hard_error_and_counts_valid_documents(
harness: &TestHarness,
fixture: MongoDbSinkFixture,
) {
@@ -705,7 +705,7 @@ async fn schema_validation_mid_batch_surfaces_hard_error_and_partial_prefix(
assert!(
id11_inserted,
- "Expected prefix message (_id=11) to be inserted before schema validation failure"
+ "Expected valid message (_id=11) to be inserted despite schema validation failures"
);
assert!(
!id12_inserted,
@@ -713,7 +713,7 @@ async fn schema_validation_mid_batch_surfaces_hard_error_and_partial_prefix(
);
assert!(
!id13_inserted,
- "Expected suffix message (_id=13) to fail the collection validator"
+ "Expected invalid message (_id=13) to fail the collection validator"
);
let total_docs = collection
@@ -722,7 +722,7 @@ async fn schema_validation_mid_batch_surfaces_hard_error_and_partial_prefix(
.expect("Failed to count documents after schema validation test");
assert_eq!(
total_docs, 1,
- "Expected exact prefix-only write accounting under schema validation failure"
+ "Expected only the valid document to count as written under schema validation failure"
);
}
diff --git a/core/integration/tests/connectors/postgres/postgres_sink.rs b/core/integration/tests/connectors/postgres/postgres_sink.rs
index f0e1fb869..fd6fd3858 100644
--- a/core/integration/tests/connectors/postgres/postgres_sink.rs
+++ b/core/integration/tests/connectors/postgres/postgres_sink.rs
@@ -17,21 +17,153 @@
use super::TEST_MESSAGE_COUNT;
use crate::connectors::fixtures::{
- PostgresOps, PostgresSinkByteaFixture, PostgresSinkFixture, PostgresSinkJsonFixture,
+ POSTGRES_LARGE_BATCH_SIZE, PostgresOps, PostgresSinkByteaFixture, PostgresSinkFixture,
+ PostgresSinkJsonFixture, PostgresSinkLargeBatchFixture,
};
use crate::connectors::{TestMessage, create_test_messages};
use bytes::Bytes;
-use iggy::prelude::{IggyMessage, Partitioning};
+use iggy::prelude::{IggyClient, IggyMessage, Partitioning};
use iggy_common::Identifier;
use iggy_common::MessageClient;
+use iggy_connector_sdk::api::ConnectorRuntimeStats;
use integration::harness::seeds;
use integration::iggy_harness;
+use std::time::Duration;
+use tokio::time::{sleep, timeout};
const SINK_TABLE: &str = "iggy_messages";
+const DEFAULT_INSERT_PARAMETERS: usize = 9;
+const MAX_DEFAULT_INSERT_ROWS: usize = u16::MAX as usize / DEFAULT_INSERT_PARAMETERS;
+const STATS_WAIT_TIMEOUT: Duration = Duration::from_secs(10);
+const STATS_POLL_INTERVAL: Duration = Duration::from_millis(100);
type SinkRow = (i64, String, String, Vec<u8>);
type SinkJsonRow = (i64, serde_json::Value);
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/sink.toml")),
+ seed = large_postgres_batch
+)]
+async fn oversized_callback_stores_all_rows(fixture: PostgresSinkLargeBatchFixture) {
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ let rows: Vec<SinkJsonRow> = fixture
+ .fetch_rows_as(
+ &pool,
+ "SELECT iggy_offset, payload FROM iggy_messages ORDER BY iggy_offset",
+ POSTGRES_LARGE_BATCH_SIZE,
+ )
+ .await
+ .expect("Every row must survive bind-limit chunking");
+ assert_eq!(rows.len(), POSTGRES_LARGE_BATCH_SIZE);
+ for (sequence, (offset, payload)) in rows.into_iter().enumerate() {
+ assert_eq!(offset, sequence as i64);
+ assert_eq!(payload, serde_json::json!({"sequence": sequence}));
+ }
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/sink.toml")),
+ seed = large_postgres_batch_with_invalid_record
+)]
+async fn failed_chunk_reports_runtime_error_and_preserves_later_chunks(
+ harness: &TestHarness,
+ fixture: PostgresSinkLargeBatchFixture,
+) {
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ let expected_rows = POSTGRES_LARGE_BATCH_SIZE - MAX_DEFAULT_INSERT_ROWS;
+ let rows: Vec<SinkJsonRow> = fixture
+ .fetch_rows_as(
+ &pool,
+ "SELECT iggy_offset, payload FROM iggy_messages ORDER BY iggy_offset",
+ expected_rows,
+ )
+ .await
+ .expect("The chunk after the rejected chunk must still be inserted");
+ assert_eq!(rows.len(), expected_rows);
+ for (position, (offset, payload)) in rows.into_iter().enumerate() {
+ let sequence = MAX_DEFAULT_INSERT_ROWS + position;
+ assert_eq!(offset, sequence as i64);
+ assert_eq!(payload, serde_json::json!({"sequence": sequence}));
+ }
+
+ let runtime = harness.connectors_runtime().expect("connectors runtime");
+ let stats_url = format!("{}/stats", runtime.http_url());
+ let http_client = reqwest::Client::new();
+ let stats = timeout(STATS_WAIT_TIMEOUT, async {
+ loop {
+ let snapshot: ConnectorRuntimeStats = http_client
+ .get(&stats_url)
+ .send()
+ .await
+ .expect("stats request")
+ .error_for_status()
+ .expect("stats status")
+ .json()
+ .await
+ .expect("stats response");
+ let sink = snapshot
+ .connectors
+ .into_iter()
+ .find(|sink| sink.key == "postgres")
+ .expect("PostgreSQL sink must be reported");
+ if sink.errors > 0 {
+ break sink;
+ }
+ sleep(STATS_POLL_INTERVAL).await;
+ }
+ })
+ .await
+ .expect("The failed chunk must reach runtime error statistics");
+ assert_eq!(stats.errors, 1);
+ assert_eq!(
+ stats.messages_consumed,
+ Some(POSTGRES_LARGE_BATCH_SIZE as u64)
+ );
+ assert_eq!(stats.messages_processed, Some(0));
+}
+
+async fn large_postgres_batch(client: &IggyClient) -> Result<(), seeds::SeedError> {
+ seed_large_postgres_batch(client, false).await
+}
+
+async fn large_postgres_batch_with_invalid_record(
+ client: &IggyClient,
+) -> Result<(), seeds::SeedError> {
+ seed_large_postgres_batch(client, true).await
+}
+
+async fn seed_large_postgres_batch(
+ client: &IggyClient,
+ invalid_first_record: bool,
+) -> Result<(), seeds::SeedError> {
+ seeds::connector_stream(client).await?;
+ let stream_id: Identifier = seeds::names::STREAM.try_into()?;
+ let topic_id: Identifier = seeds::names::TOPIC.try_into()?;
+ let mut messages = Vec::with_capacity(POSTGRES_LARGE_BATCH_SIZE);
+ for sequence in 0..POSTGRES_LARGE_BATCH_SIZE {
+ let payload = if invalid_first_record && sequence == 0 {
+ b"invalid JSON".to_vec()
+ } else {
+ serde_json::to_vec(&serde_json::json!({"sequence": sequence}))?
+ };
+ messages.push(
+ IggyMessage::builder()
+ .id(sequence as u128 + 1)
+ .payload(Bytes::from(payload))
+ .build()?,
+ );
+ }
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await?;
+ Ok(())
+}
+
#[iggy_harness(
server(connectors_runtime(config_path = "tests/connectors/postgres/sink.toml")),
seed = seeds::connector_stream
diff --git a/core/integration/tests/connectors/postgres/postgres_source.rs b/core/integration/tests/connectors/postgres/postgres_source.rs
index 375cb785e..91c3ebb90 100644
--- a/core/integration/tests/connectors/postgres/postgres_source.rs
+++ b/core/integration/tests/connectors/postgres/postgres_source.rs
@@ -15,12 +15,14 @@
// specific language governing permissions and limitations
// under the License.
+use std::fs;
use std::time::Duration;
use iggy_common::MessageClient;
use iggy_common::{Consumer, Identifier, PollingStrategy};
+use iggy_connector_sdk::ConnectorState;
use iggy_connector_sdk::api::ConnectorStatus;
-use integration::harness::seeds;
+use integration::harness::{TestHarness, seeds};
use integration::iggy_harness;
use reqwest::Client;
use tokio::time::sleep;
@@ -35,7 +37,7 @@ use crate::connectors::fixtures::{
PostgresSourceDeleteSlowPollFixture, PostgresSourceJsonFixture, PostgresSourceJsonbFixture,
PostgresSourceMarkFixture, PostgresSourceNonUniqueCleanupFixture,
PostgresSourceNonUniqueTrackingFixture, PostgresSourceNumericTrackingFixture,
- PostgresSourceOps,
+ PostgresSourceOps, PostgresSourceTextKeyFixture,
};
#[iggy_harness(
@@ -939,3 +941,346 @@ async fn state_persists_across_connector_restart(
pool.close().await;
}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn checkpoint_failure_source_replays_rows_until_acknowledged(
+ harness: &TestHarness,
+ fixture: PostgresSourceByteaFixture,
+) {
+ let runtime = harness.connectors_runtime().expect("connector runtime");
+ let checkpoint = runtime.state_path().join("source_postgres.state");
+ fs::create_dir(&checkpoint).expect("force checkpoint rename failure");
+
+ let pool = fixture.create_pool().await.expect("PostgreSQL pool");
+ let payload = b"replay after checkpoint failure";
+ fixture.insert_payload(&pool, 1, payload).await;
+ pool.close().await;
+
+ let checkpoint_failed = wait_for_failed_checkpoint(harness).await;
+ fs::remove_dir(&checkpoint).expect("restore checkpoint destination");
+ assert!(checkpoint_failed, "the produced row must receive a Nack");
+
+ let received = poll_payloads(harness, "checkpoint_failure", 2).await;
+
+ for _ in 0..POLL_ATTEMPTS {
+ if checkpoint.is_file() {
+ break;
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ assert!(
+ checkpoint.is_file(),
+ "checkpoint must recover after the failure"
+ );
+ assert!(
+ received.len() >= 2,
+ "a Nacked row must be replayed; received {} copies",
+ received.len()
+ );
+ for message in received {
+ assert_eq!(message, payload);
+ }
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn text_keys_source_tracks_and_deletes_numeric_and_quoted_values(
+ harness: &TestHarness,
+ fixture: PostgresSourceDeleteFixture,
+) {
+ let pool = fixture.create_pool().await.expect("PostgreSQL pool");
+ let query = format!(
+ "ALTER TABLE {} ALTER COLUMN id DROP DEFAULT, ALTER COLUMN id TYPE TEXT",
+ fixture.table_name()
+ );
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .execute(&pool)
+ .await
+ .expect("use TEXT keys in the empty validated table");
+ let query = format!(
+ "INSERT INTO {} (id, name, value) SELECT id, id, 0
+ FROM (VALUES ('01'), ('02'), ('O''Brien')) rows(id)",
+ fixture.table_name()
+ );
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .execute(&pool)
+ .await
+ .expect("insert TEXT keys and rows atomically");
+
+ let expected = ["01", "02", "O'Brien"];
+ let received: Vec<_> = poll_payloads(harness, "text_keys", expected.len())
+ .await
+ .into_iter()
+ .map(|payload| {
+ let row: serde_json::Value =
+ serde_json::from_slice(&payload).expect("source JSON envelope");
+ row["data"]["id"].as_str().expect("TEXT key").to_string()
+ })
+ .collect();
+ assert_eq!(
+ received, expected,
+ "TEXT keys must retain their exact values"
+ );
+ let mut remaining_rows = fixture.count_rows(&pool).await;
+ for _ in 0..POLL_ATTEMPTS {
+ if remaining_rows == 0 {
+ break;
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ remaining_rows = fixture.count_rows(&pool).await;
+ }
+ assert_eq!(remaining_rows, 0, "all selected keys must be deleted");
+ pool.close().await;
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn numeric_rows_source_preserves_decimal_strings_and_nulls(
+ harness: &TestHarness,
+ fixture: PostgresSourceJsonFixture,
+) {
+ let pool = fixture.create_pool().await.expect("PostgreSQL pool");
+ let query = format!(
+ "ALTER TABLE {} ALTER COLUMN amount TYPE NUMERIC, ALTER COLUMN amount DROP NOT NULL",
+ fixture.table_name()
+ );
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .execute(&pool)
+ .await
+ .expect("use nullable NUMERIC values in the empty validated table");
+ let query = format!(
+ "INSERT INTO {} (id, name, count, amount, active, timestamp, tag)
+ SELECT id, 'numeric', 0, amount, TRUE, 0, 'numeric' FROM (VALUES
+ (1, 12.50::NUMERIC), (2, -98765.125), (3, 0),
+ (4, 123456789.875), (5, NULL)) rows(id, amount)",
+ fixture.table_name()
+ );
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .execute(&pool)
+ .await
+ .expect("insert NUMERIC rows atomically");
+ pool.close().await;
+
+ let expected = [
+ Some("12.5"),
+ Some("-98765.125"),
+ Some("0"),
+ Some("123456789.875"),
+ None,
+ ];
+ let received = poll_payloads(harness, "numeric_rows", expected.len()).await;
+ assert_eq!(
+ received.len(),
+ expected.len(),
+ "every NUMERIC row must be delivered"
+ );
+ for (payload, expected) in received.iter().zip(expected) {
+ let row: serde_json::Value = serde_json::from_slice(payload).expect("source JSON envelope");
+ assert_eq!(
+ row["data"]["amount"].as_str(),
+ expected,
+ "NUMERIC value in {row}"
+ );
+ assert_eq!(row["data"]["amount"].is_null(), expected.is_none());
+ }
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn selected_payload_key_source_tracks_and_marks_rows(
+ harness: &TestHarness,
+ fixture: PostgresSourceTextKeyFixture,
+) {
+ let pool = fixture.create_pool().await.expect("PostgreSQL pool");
+ let query = format!(
+ "INSERT INTO {} (id) VALUES ('01'), ('02'), ('03')",
+ fixture.table_name()
+ );
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .execute(&pool)
+ .await
+ .expect("insert rows whose payload is also the tracking and primary key");
+
+ let expected = [b"01".to_vec(), b"02".to_vec(), b"03".to_vec()];
+ let received = poll_payloads(harness, "payload_key", expected.len()).await;
+ assert_eq!(received, expected, "selected keys must be delivered once");
+
+ let query = format!(
+ "SELECT COUNT(*) FROM {} WHERE processed",
+ fixture.table_name()
+ );
+ let checkpoint = harness
+ .connectors_runtime()
+ .expect("connector runtime")
+ .state_path()
+ .join("source_postgres.state");
+ let mut processed = 0i64;
+ for _ in 0..POLL_ATTEMPTS {
+ processed = sqlx::query_scalar(sqlx::AssertSqlSafe(query.as_str()))
+ .fetch_one(&pool)
+ .await
+ .expect("count marked rows");
+ if processed == expected.len() as i64 && checkpoint.is_file() {
+ break;
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ assert_eq!(
+ processed,
+ expected.len() as i64,
+ "selected payload keys must still identify rows to mark"
+ );
+
+ let state = ConnectorState(fs::read(checkpoint).expect("acknowledged checkpoint"))
+ .deserialize::<serde_json::Value>("PostgreSQL test", 1)
+ .expect("existing MessagePack state");
+ assert_eq!(
+ state[1][fixture.table_name()],
+ "03",
+ "selected payload key must still advance the cursor"
+ );
+ assert_eq!(state[2], 3, "each row must be acknowledged once");
+ pool.close().await;
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn delete_source_retains_rows_until_acknowledged(
+ harness: &TestHarness,
+ fixture: PostgresSourceDeleteFixture,
+) {
+ let pool = fixture.create_pool().await.expect("PostgreSQL pool");
+ assert_cleanup_after_ack(harness, &pool, fixture.table_name(), "TRUE").await;
+ pool.close().await;
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn mark_source_retains_unprocessed_rows_until_acknowledged(
+ harness: &TestHarness,
+ fixture: PostgresSourceMarkFixture,
+) {
+ let pool = fixture.create_pool().await.expect("PostgreSQL pool");
+ assert_cleanup_after_ack(harness, &pool, fixture.table_name(), "NOT is_processed").await;
+ pool.close().await;
+}
+
+async fn assert_cleanup_after_ack(
+ harness: &TestHarness,
+ pool: &sqlx::PgPool,
+ table: &str,
+ pending_condition: &str,
+) {
+ let checkpoint = harness
+ .connectors_runtime()
+ .expect("connector runtime")
+ .state_path()
+ .join("source_postgres.state");
+ fs::create_dir(&checkpoint).expect("force checkpoint rename failure");
+ let query = format!("INSERT INTO {table} (id, name, value) VALUES (1, 'pending', 0)");
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .expect("insert the row before checkpoint acknowledgement");
+ let failed = wait_for_failed_checkpoint(harness).await;
+ let query = format!("SELECT COUNT(*) FROM {table} WHERE {pending_condition}");
+ let pending: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(query.as_str()))
+ .fetch_one(pool)
+ .await
+ .expect("count rows before Ack");
+ fs::remove_dir(&checkpoint).expect("restore checkpoint destination");
+ assert!(failed, "the produced batch must receive Nack");
+ assert_eq!(pending, 1, "Nack must leave the row available for replay");
+
+ let received = poll_payloads(harness, "cleanup_ack", 2).await;
+ assert!(received.len() >= 2, "the rejected row must be replayed");
+ for payload in received {
+ let record: serde_json::Value = serde_json::from_slice(&payload).expect("JSON envelope");
+ assert_eq!(record["data"]["id"], 1);
+ }
+ let mut pending = 1i64;
+ for _ in 0..POLL_ATTEMPTS {
+ pending = sqlx::query_scalar(sqlx::AssertSqlSafe(query.as_str()))
+ .fetch_one(pool)
+ .await
+ .expect("count rows after Ack");
+ if pending == 0 {
+ break;
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ assert_eq!(pending, 0, "Ack must apply the requested row cleanup");
+ let state = ConnectorState(fs::read(checkpoint).expect("acknowledged checkpoint"))
+ .deserialize::<serde_json::Value>("PostgreSQL test", 1)
+ .expect("MessagePack state");
+ assert_eq!(
+ state[2], 1,
+ "a replay must not double-count the acknowledged row"
+ );
+}
+
+async fn wait_for_failed_checkpoint(harness: &TestHarness) -> bool {
+ let runtime = harness.connectors_runtime().expect("connector runtime");
+ for _ in 0..POLL_ATTEMPTS {
+ if runtime
+ .collect_logs()
+ .0
+ .contains("Failed to save state for source connector")
+ {
+ return true;
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ false
+}
+
+async fn poll_payloads(
+ harness: &TestHarness,
+ consumer_name: &str,
+ expected: usize,
+) -> Vec<Vec<u8>> {
+ let client = harness.root_client().await.expect("root client");
+ let stream: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer = Consumer::new(consumer_name.try_into().unwrap());
+ let mut received = Vec::new();
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(polled) = client
+ .poll_messages(
+ &stream,
+ &topic,
+ None,
+ &consumer,
+ &PollingStrategy::next(),
+ 10,
+ true,
+ )
+ .await
+ {
+ received.extend(
+ polled
+ .messages
+ .into_iter()
+ .map(|message| message.payload.to_vec()),
+ );
+ if received.len() >= expected {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ received
+}
diff --git a/core/integration/tests/connectors/s3/s3_sink.rs b/core/integration/tests/connectors/s3/s3_sink.rs
index 22258aba2..a3352d06e 100644
--- a/core/integration/tests/connectors/s3/s3_sink.rs
+++ b/core/integration/tests/connectors/s3/s3_sink.rs
@@ -21,7 +21,7 @@ use bytes::Bytes;
use iggy::prelude::{IggyMessage, Partitioning};
use iggy_common::Identifier;
use iggy_common::MessageClient;
-use iggy_connector_sdk::api::SinkInfoResponse;
+use iggy_connector_sdk::api::{ConnectorStatus, SinkInfoResponse};
use integration::harness::seeds;
use integration::iggy_harness;
use reqwest::Client;
@@ -53,6 +53,22 @@ async fn s3_sink_initializes_and_runs(harness: &TestHarness, fixture: S3SinkFixt
assert_eq!(sinks.len(), 1);
assert_eq!(sinks[0].key, S3_SINK_KEY);
assert!(sinks[0].enabled);
+ assert_eq!(sinks[0].status, ConnectorStatus::Running);
+
+ let keys = fixture.list_objects("").await.expect("List data objects");
+ assert!(
+ keys.is_empty(),
+ "Startup must not publish probe objects: {keys:?}"
+ );
+ let uploads = fixture
+ .bucket()
+ .list_multiparts_uploads(None, None)
+ .await
+ .expect("List multipart uploads");
+ assert!(
+ uploads.iter().all(|page| page.uploads.is_empty()),
+ "Startup must abort its multipart probe: {uploads:?}"
+ );
drop(fixture);
}
diff --git a/core/integration/tests/connectors/surrealdb/surrealdb_sink.rs b/core/integration/tests/connectors/surrealdb/surrealdb_sink.rs
index 840810377..3d1d21a28 100644
--- a/core/integration/tests/connectors/surrealdb/surrealdb_sink.rs
+++ b/core/integration/tests/connectors/surrealdb/surrealdb_sink.rs
@@ -64,7 +64,7 @@ async fn json_messages_sink_to_surrealdb(harness: &TestHarness, fixture: Surreal
server(connectors_runtime(config_path = "tests/connectors/surrealdb/sink.toml")),
seed = seeds::connector_stream
)]
-async fn given_namespace_auth_should_insert_messages(
+async fn namespace_auth_inserts_messages(
harness: &TestHarness,
fixture: SurrealDbSinkNamespaceFixture,
) {
@@ -75,7 +75,7 @@ async fn given_namespace_auth_should_insert_messages(
server(connectors_runtime(config_path = "tests/connectors/surrealdb/sink.toml")),
seed = seeds::connector_stream
)]
-async fn given_database_auth_should_insert_messages(
+async fn database_auth_inserts_messages(
harness: &TestHarness,
fixture: SurrealDbSinkDatabaseFixture,
) {