This is an automated email from the ASF dual-hosted git repository.
ryerraguntla pushed a commit to branch
docs/connector-guidelines-template-and-blog-post
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to
refs/heads/docs/connector-guidelines-template-and-blog-post by this push:
new 0c338c5a7 Fixed couple of trivial issues
0c338c5a7 is described below
commit 0c338c5a7126ebf5d7c8c25a85adf97754084a53
Author: ryerraguntla <[email protected]>
AuthorDate: Sat Oct 10 20:51:25 2026 -0400
Fixed couple of trivial issues
---
.claude/skills/connector-pr-review/SKILL.md | 130 ++++++++++++++++++++-
.claude/skills/connector-testing/SKILL.md | 19 +--
core/connectors/sinks/sink_template/README.md | 25 ++++
core/connectors/sinks/sink_template/src/lib.rs | 72 ++++++++++--
core/connectors/sources/source_template/README.md | 27 +++++
core/connectors/sources/source_template/src/lib.rs | 40 ++++++-
6 files changed, 289 insertions(+), 24 deletions(-)
diff --git a/.claude/skills/connector-pr-review/SKILL.md
b/.claude/skills/connector-pr-review/SKILL.md
index 4725fcd6e..0ae07f3cd 100644
--- a/.claude/skills/connector-pr-review/SKILL.md
+++ b/.claude/skills/connector-pr-review/SKILL.md
@@ -136,6 +136,75 @@ that state on success, then calls `on_batch_result(Ack |
Nack)`. Only
re-emit an entire index without calling that out in README.
- [ ] Config flag combos that no-op should `warn!` or `Err`, not silently
ignore.
+### B9. Zero delivered is a failure, even if the call "succeeded"
+
+A bulk/batch API call returning 200 does not mean every row in it landed.
+Recurring, independently-discovered bug across four different sinks:
+
+- [ ] If your destination's write API can partially or fully reject a batch
+ while the call itself succeeds (HTTP 200 with per-item errors, a bulk
+ index response, a stream-load response), count accepted items and
+ return `Err` when the accepted count is **zero** for a non-empty input —
+ never `Ok(())`. BigQuery (#4394): "when every row of a non-empty batch
+ is rejected, `consume` still returns `Ok`." Elasticsearch (#4154): bulk
+ index knows the rejected count but still returns `Ok`. Doris (#3215):
+ "if every message in the batch is non-JSON... `consume()` returns
+ `Ok(())`" with the batch silently gone. S3 (#3103): a formatter failure
+ produced an empty placeholder that was still counted as uploaded.
+- [ ] A per-record encode/serialize failure must not silently substitute an
+ empty/null/default value that then gets counted as delivered — skip
+ and count it as lost (visibly), or fail the whole chunk.
+
+### B10. A wrapping timeout does not cancel in-flight work
+
+`tokio::time::timeout(dur, fut)` dropping `fut` on expiry does **not** mean
+the operation didn't happen — for a blocking-executor call, an FFI call, or
+anything already past the "no going back" point in the wire protocol, the
+work keeps running after the timeout fires and can still complete
+server-side. Confirmed independently on two unrelated backends:
+
+- [ ] Treat an expired wrapping timeout as **unknown outcome**, not
+ **didn't happen** — do not assume it's safe to retry the identical
+ operation without a dedup/idempotency key. RabbitMQ (#3973): "Wrapping
+ `basic_publish` in `timeout()` does not make it abortable... usually
+ already delivered." Postgres (#3957): "`timeout` drops the client
+ future, it does not abort the statement — a DELETE... can still commit
+ server-side after this logs a failure."
+- [ ] Prefer a protocol/transport-level timeout over (or in addition to) a
+ wrapping `tokio::time::timeout`: `reqwest::ClientBuilder::timeout`,
+ a connection-string timeout parameter (`?connection_timeout=` for AMQP),
+ or a driver's own `statement_timeout`. A bare `TcpStream::connect` run
+ on a blocking executor is not cancelled by an outer async timeout and
+ leaks a pool thread until the kernel's own SYN timeout (RabbitMQ #3973).
+- [ ] This is why `build_raw_client()` in the sink/source templates sets
+ `.timeout()` on the `reqwest::Client` itself rather than wrapping calls
+ — copy that pattern, don't wrap instead of it.
+
+### B11. Ack-path work must fit inside the SDK's fixed batch-result window
+
+If `on_batch_result` (or any post-ack cleanup) does real I/O — not just an
+in-memory state swap — that I/O runs inside the SDK's ~30s batch-result
+timeout, and a NACK from that timeout is often **fatal** (stops the source
+for good after repeated consecutive NACKs), not a retry signal.
+
+- [ ] Any retry/timeout budget inside `on_batch_result` must be sized well
+ under the SDK's window, and **multiplied by how many operations run
+ inside it** — not assumed to be one shared budget. Postgres (#3957,
+ critical): "10s is per operation, not per batch... N tables burn
+ N\*10s inside the SDK's 30s window."
+- [ ] User-configurable `max_retries`/`retry_delay` can produce a wait time
+ that exceeds a hardcoded internal deadline — validate the combination
+ in `open()` and reject configs that cannot fit, rather than letting
+ the ack path blow its budget at runtime. Same PR: "`max_retries=3`
+ with `retry_delay=\"2s\"`, both straight from the readme, needs 12s
+ and fails startup."
+- [ ] A failed destructive/cleanup op inside `on_batch_result` must not be
+ silently logged and ignored while the batch is still ACKed — count
+ consecutive failures and return `Err` past a threshold so the SDK
+ stops the source, rather than resending the same broken operation
+ forever. Same PR: "a failed slot advance only logs. the batch is
+ already ACKed, so nothing stops it."
+
## High-frequency convention nits
These are "cheap" but burn full review rounds when missed.
@@ -164,25 +233,66 @@ These are "cheap" but burn full review rounds when missed.
- [ ] New fields: `Option<T>` + `#[serde(default)]` where needed.
- [ ] Prefer `#[serde(deny_unknown_fields)]` on plugin config so typo’d knobs
fail loud.
+- [ ] Reject degenerate numeric values (`0`, empty string) for a field where
+ zero makes a check always-true or always-false, rather than letting it
+ silently misbehave — `batch_size`/`max_messages_per_file` zero
+ triggered S3 (#3103) to rotate on every single message. If a sibling
+ connector already clamps a field like this (`.max(1)`), clamp
+ consistently rather than leaving one sibling to reject what another
+ silently accepts.
+- [ ] Never derive a value that must stay identical across a retry of the
+ *same* logical batch (a destination key, filename, or dedup id) from
+ wall-clock time (`Utc::now()`) — derive it from the batch's own fixed
+ fields (an offset, a message id, `origin_timestamp`). S3 (#3103): a
+ `{timestamp}` path placeholder rendered at upload time broke the
+ README's own idempotent-retry claim.
### C3. Crate / path / docs checklist
- [ ] `[lib] crate-type = ["cdylib", "lib"]`.
+- [ ] `publish = false` in `[package]` — every in-tree plugin needs it, and
+ it's easy to forget on a new crate (missed on S3 #3103 and Doris
+ #3215, both caught in review).
- [ ] Example TOML plugin path uses `../../target/release/lib…` like siblings.
- [ ] Row added to `sinks/README.md` or `sources/README.md`.
- [ ] Sample under `runtime/example_config/connectors/`.
- [ ] README defaults **byte-equal** to consts in code (diff them).
- [ ] No links to non-existent docs.
+- [ ] Helper modules stay crate-private (no `pub mod`) unless genuinely part
+ of the plugin's public API — S3 (#3103) exposed `buffer`/`client`/etc.
+ as public on a cdylib plugin where every sibling keeps them private.
### C4. Hot path
- [ ] No `payload.clone().try_to_bytes()` — use `try_to_bytes(&self)`.
- [ ] `Vec::with_capacity(n)` for per-batch buffers.
- [ ] No `std::sync::Mutex` across `.await`; use `tokio::sync::Mutex`.
+- [ ] Even a `tokio::sync::Mutex` must not be **held across** a slow network
+ call or a retry sleep — lock, clone/extract what you need, drop the
+ guard, *then* do I/O. This isn't just a deadlock rule: holding the
+ lock across the call serializes every concurrent caller sharing that
+ plugin instance onto one upload/write at a time, killing the
+ parallelism the runtime's per-partition tasks were meant to give you.
+ S3 (#3103): one `buffers.lock().await` held through `put_object()`
+ *and* the retry sleep capped the whole sink's throughput to one
+ upload in flight regardless of partition count.
- [ ] `&self` on `consume` / `poll` (interior mutability only).
- [ ] No `tokio::spawn` inside plugin code.
- [ ] No `unwrap()`/`expect()` on external I/O outside tests.
- [ ] No eager `format!` around tracing args.
+- [ ] Never `let _ = fallible_call()` on a side effect that matters (a
+ cleanup op, a quoted identifier, a staged delete) — log or propagate
+ the error. Postgres (#3957): `let _` on a staged-operation error made
+ an always-failing delete invisible.
+- [ ] If retrying raw HTTP status codes (not just SDK `Error` variants),
+ classify by family — 5xx, 408, 429 retry; other 4xx fail fast. S3
+ (#3103) retried `AccessDenied`/`NoSuchBucket`/`InvalidBucketName`
+ three times each, burning the whole retry budget on failures no retry
+ could fix.
+- [ ] Never log a substring, prefix, or fragment of a credential "for
+ diagnostics" — it still identifies the credential type/shape for zero
+ diagnostic value. S3 (#3103) logged the first 3 characters of an AWS
+ access key (`AKIA`/`ASIA`/`AROA` prefix).
### C5. Containers / fixtures
@@ -238,10 +348,22 @@ Paste into the PR description (or run mentally before
`/ready`):
Recurring comments mined from connector PRs including (non-exhaustive):
SurrealDB sink (#3453), Meilisearch sink/source (#3497/#3498), OpenSearch
source (#3515), Quickwit convention (#3523), MySQL source (#3568), JDBC
-source (#3588), Doris retry (#3574), Redshift sink (#3654), Airflow trigger
-(#3716), Fluss sink (#3782). Highest-density themes: delivery/offset
-semantics, idempotency IDs, transient/permanent mapping, config-name drift,
-secrets, README/code drift, false-green tests, CI/lockfile hygiene.
+source (#3588), Doris retry (#3574) and sink review (#3215), Redshift sink
+(#3654), Airflow trigger (#3716), Fluss sink (#3782), S3 sink (#3103),
+Google BigQuery sink (#4394), Delta sink production hardening (#4219),
+RabbitMQ sink (#3973), Elasticsearch sink batch semantics (#4154),
+Postgres source ACK/NACK + CDC hardening (#3957), the shared
+`retry_async`/`RetryPolicy` SDK helper review (#4104). Highest-density
+themes: delivery/offset semantics, idempotency IDs, transient/permanent
+mapping, config-name drift, secrets, README/code drift, false-green tests,
+CI/lockfile hygiene, zero-delivered-still-reports-Ok, a wrapping timeout
+that doesn't cancel in-flight work, ack-path work blowing its fixed budget,
+and holding a lock across slow I/O.
+
+Re-derive or extend this list with `gh pr list --repo apache/iggy --search
+"connectors in:title" --state all --limit 100` plus `gh api
+repos/apache/iggy/pulls/<N>/comments --paginate` for PRs under
+`core/connectors/`.
---
diff --git a/.claude/skills/connector-testing/SKILL.md
b/.claude/skills/connector-testing/SKILL.md
index 9e6669e05..fcbbbab75 100644
--- a/.claude/skills/connector-testing/SKILL.md
+++ b/.claude/skills/connector-testing/SKILL.md
@@ -100,18 +100,23 @@ fn test_config() -> RandomSourceConfig {
### Async testing pattern
-The trait methods are `async`. For unit tests, build a runtime locally - don't
add `#[tokio::test]` in `src/lib.rs` (it pulls test macros into prod build):
+The trait methods are `async`. Prefer `#[tokio::test]` directly on the test fn
- `#[cfg(test)]`
+gates the whole module out of a plain `cargo build`, and resolver v2 (default
since edition 2021,
+which this workspace uses) keeps `tokio`'s `test-util`/`macros` dev-dependency
features out of that
+build too, so there's no prod-build leak to avoid. This is also the dominant
pattern in-tree
+(`postgres_source::tests` uses `#[tokio::test]` throughout, no manual runtime):
```rust
-#[test]
-fn given_X_should_Y() {
- let runtime = tokio::runtime::Runtime::new().unwrap();
- runtime.block_on(async {
- // await calls here
- });
+#[tokio::test]
+async fn given_x_should_y() {
+ // await calls directly, no manual Runtime::new()
}
```
+A manual `tokio::runtime::Runtime::new().unwrap().block_on(...)` inside a
plain `#[test]` still
+works and you'll see it in some older plugin tests (`random_source`), but
don't ask for it in
+review - it's equivalent, not a correctness requirement.
+
### Mandatory: four canonical source state tests
Every source must have these. Copy from
`sources/random_source/src/lib.rs::tests`:
diff --git a/core/connectors/sinks/sink_template/README.md
b/core/connectors/sinks/sink_template/README.md
index 86da056d8..5924b3422 100644
--- a/core/connectors/sinks/sink_template/README.md
+++ b/core/connectors/sinks/sink_template/README.md
@@ -7,6 +7,31 @@ doc comment at the top of `src/lib.rs` for the full rationale,
and the
`iggy-connector-review` skill / the "Building Connectors That Pass Review"
blog post for the checklist this template is built against.
+## Delivery Semantics
+
+The framework around `push_batch()` gives you retry (`max_retries` total
attempts,
+`retry_delay`/`retry_max_delay` backoff) and a circuit breaker that opens after
+`circuit_breaker_threshold` consecutive failures. Everything below that line
+depends on what you write in `push_batch()` — fill in the three blanks with
your
+real behavior before shipping:
+
+1. **Transient failure**: retries up to `max_retries` total attempts, then
+ `consume()` returns the error. The runtime's handling of that error is fatal
+ to the whole connector — see `connector-pr-review` B2.
+2. **Permanent failure** (bad schema, destination rejects the payload shape):
+ return `Error::PermanentHttpError` from `push_batch()` — `consume()` drops
+ and counts that one batch instead of taking the whole connector down.
+3. **Duplication window**: at-least-once. A batch can be retried after a
+ partial write (e.g. your destination accepted some rows then the connection
+ dropped before the response), so a retry can re-send already-written rows.
+4. **Dedup key**: none by default — `push_batch()` doesn't build one yet.
+ Before shipping, derive a stable key from `message.id` plus
+ `topic_metadata`/`messages_metadata` (see
`mongodb_sink::build_composite_document_id`
+ for `stream:topic:partition:message_id`) and make your destination's write
+ idempotent on it (upsert / `ON CONFLICT` / unique index) — otherwise the
+ at-least-once retry above will create duplicates. See `connector-pr-review`
+ B5.
+
## What's already done for you
- Config parsing with `#[serde(deny_unknown_fields)]` so a typo in a TOML
diff --git a/core/connectors/sinks/sink_template/src/lib.rs
b/core/connectors/sinks/sink_template/src/lib.rs
index 0a86bc733..80a8c97ab 100644
--- a/core/connectors/sinks/sink_template/src/lib.rs
+++ b/core/connectors/sinks/sink_template/src/lib.rs
@@ -230,12 +230,36 @@ impl TemplateSink {
/// returned, which the runtime treats as fatal for the whole connector —
/// so a permanent classification is what keeps one bad message from
/// taking every future batch down with it.
+ ///
+ /// Delivery is at-least-once: a retried batch can re-send rows your
+ /// destination already wrote from an earlier attempt that failed after
+ /// partially succeeding. Build a stable key with `dedup_key()` below for
+ /// every message you write, and make the write idempotent on it (upsert /
+ /// `ON CONFLICT` / unique index) — see `connector-pr-review` B5.
+ ///
+ /// If your destination's write call can succeed at the transport level
+ /// while rejecting some or all rows individually (a bulk/batch API
+ /// returning 200 with per-item errors, a stream-load response, a bulk
+ /// index response), count accepted rows yourself and return `Err` when
+ /// that count is zero for a non-empty `batch` — never `Ok(())`. Four
+ /// unrelated sinks shipped exactly this bug in review (BigQuery,
+ /// Elasticsearch, Doris, S3): the call "succeeded," `consume()` reported
+ /// success, and the batch was gone with no signal anywhere. See
+ /// `connector-pr-review` B9.
async fn push_batch(
&self,
client: &ClientWithMiddleware,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
batch: &[ConsumedMessage],
) -> Result<(), Error> {
- let _ = (client, batch); // remove once implemented
+ let _ = client; // remove once implemented
+ for message in batch {
+ // TODO(ConnectorDeveloper): use `key` as the identity your
destination's
+ // write is idempotent on (upsert key, `ON CONFLICT` target,
document _id).
+ let key = dedup_key(topic_metadata, messages_metadata, message);
+ let _ = key;
+ }
Err(Error::InitError(
"TemplateSink::push_batch is not implemented yet — see the
TODO(ConnectorDeveloper) comment in \
template_sink/src/lib.rs"
@@ -244,6 +268,23 @@ impl TemplateSink {
}
}
+/// Stable per-message dedup key: `stream:topic:partition:message_id`. Mirrors
+/// `mongodb_sink::build_composite_document_id` — copy this, not a random UUID,
+/// for whatever identity column/key your destination's upsert needs. Offset
+/// is deliberately excluded: it only identifies a message within one
+/// partition's log, not across the retries and redeliveries this key has to
+/// survive.
+fn dedup_key(
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ message: &ConsumedMessage,
+) -> String {
+ format!(
+ "{}:{}:{}:{}",
+ topic_metadata.stream, topic_metadata.topic,
messages_metadata.partition_id, message.id
+ )
+}
+
// ── Sink trait
────────────────────────────────────────────────────────────────
#[async_trait]
@@ -395,7 +436,10 @@ impl Sink for TemplateSink {
let mut failed = 0u64;
for batch in messages.chunks(self.batch_size_limit) {
- match self.push_batch(client, batch).await {
+ match self
+ .push_batch(client, topic_metadata, &messages_metadata, batch)
+ .await
+ {
Ok(()) => written += batch.len() as u64,
Err(Error::PermanentHttpError(message)) => {
failed += batch.len() as u64;
@@ -498,7 +542,7 @@ mod tests {
}
#[tokio::test]
- async fn open_rejects_empty_connection_string() {
+ async fn given_empty_connection_string_should_reject_open() {
let mut config = test_config();
config.connection_string = SecretString::from(" ");
let mut sink = TemplateSink::new(1, config);
@@ -509,7 +553,7 @@ mod tests {
}
#[tokio::test]
- async fn open_rejects_invalid_target_identifier() {
+ async fn given_invalid_target_identifier_should_reject_open() {
let mut config = test_config();
config.target = "events; DROP TABLE users;--".to_string();
let mut sink = TemplateSink::new(1, config);
@@ -520,17 +564,17 @@ mod tests {
}
#[test]
- fn validate_identifier_accepts_plain_names() {
+ fn validate_identifier_should_accept_plain_names() {
assert!(validate_identifier("target", "events_v2").is_ok());
}
#[test]
- fn validate_identifier_rejects_empty() {
+ fn validate_identifier_should_reject_empty() {
assert!(validate_identifier("target", "").is_err());
}
#[test]
- fn validate_identifier_rejects_path_and_query_characters() {
+ fn validate_identifier_should_reject_path_and_query_characters() {
for bad in [
"../etc/passwd",
"events?x=1",
@@ -544,8 +588,18 @@ mod tests {
}
}
+ #[test]
+ fn dedup_key_should_join_stream_topic_partition_and_message_id() {
+ let (topic_metadata, messages_metadata) =
topic_and_messages_metadata();
+ let message = consumed_message();
+ assert_eq!(
+ dedup_key(&topic_metadata, &messages_metadata, &message),
+ "s:t:1:1"
+ );
+ }
+
#[tokio::test]
- async fn consume_short_circuits_when_breaker_is_open() {
+ async fn given_open_circuit_breaker_should_short_circuit_consume() {
let sink = TemplateSink::new(1, test_config());
sink.circuit_breaker.record_failure().await;
sink.circuit_breaker.record_failure().await;
@@ -592,7 +646,7 @@ mod tests {
// of the chunking loop end to end, including the failed-count and
// circuit-breaker bookkeeping.
#[tokio::test]
- async fn
consume_with_unimplemented_push_batch_returns_err_and_counts_failure() {
+ async fn
given_unimplemented_push_batch_should_return_err_and_count_failure() {
let mut sink = TemplateSink::new(1, test_config());
sink.open().await.expect("open should succeed");
let (topic_metadata, messages_metadata) =
topic_and_messages_metadata();
diff --git a/core/connectors/sources/source_template/README.md
b/core/connectors/sources/source_template/README.md
index 0ae77ac2c..234f10d21 100644
--- a/core/connectors/sources/source_template/README.md
+++ b/core/connectors/sources/source_template/README.md
@@ -7,6 +7,33 @@ doc comment at the top of `src/lib.rs` for the full rationale,
and the
`iggy-connector-review` skill / the "Building Connectors That Pass Review"
blog post for the checklist this template is built against.
+## Delivery Semantics
+
+The framework around `fetch_records()` gives you retry (`max_retries` total
+attempts), a circuit breaker, and the ACK/NACK cursor handshake described
+below. Everything past "what fetches" depends on what you write in
+`fetch_records()` — fill in the blanks with your real behavior before
+shipping:
+
+1. **Transient failure**: retries up to `max_retries` total attempts, then
+ `poll()` returns the error; the circuit breaker counts it. The runtime
+ retries `poll()` again next interval.
+2. **Permanent failure** (auth rejected, schema gone): return
+ `Error::PermanentHttpError` from `fetch_records()` — `poll()` does not
+ trip the circuit breaker on it, since a bad config/auth state isn't a
+ connectivity signal the breaker should react to.
+3. **Duplication window**: at-least-once. `poll()` stages the cursor as
+ *candidate* state; the runtime persists it only after the batch is sent,
+ then reports `Ack`/`Nack`. A NACK (or a crash between send and the
+ persisted-state write) redelivers the same range on the next poll.
+4. **Dedup key**: set `FetchedRecord::natural_id` from a true per-record
+ identity (row PK, document `_id`, event UUID) whenever your source has
+ one — it becomes `ProducedMessage.id`, which a downstream sink can key an
+ idempotent upsert on (see `mongodb_sink`) so the redelivery above doesn't
+ produce a duplicate at the destination. Leave it `None` only when records
+ genuinely have no natural identity (a pure generator). See
+ `connector-pr-review` B5.
+
## What's already done for you
- Config parsing with `#[serde(deny_unknown_fields)]` so a typo in a TOML
diff --git a/core/connectors/sources/source_template/src/lib.rs
b/core/connectors/sources/source_template/src/lib.rs
index 430e7e13a..7dafc097f 100644
--- a/core/connectors/sources/source_template/src/lib.rs
+++ b/core/connectors/sources/source_template/src/lib.rs
@@ -143,6 +143,19 @@ struct FetchedRecord {
/// The value `cursor` should advance to once this record's batch is
/// acknowledged — typically this record's timestamp/ID/token.
cursor_value: String,
+ /// TODO(ConnectorDeveloper): set this from a true per-record identity —
+ /// a numeric row PK, a hash of a document `_id`, an event UUID the
+ /// upstream system itself assigned — so `ProducedMessage.id` below is
+ /// stable across a NACK-triggered redelivery of this same record. A
+ /// *fresh* random id per emit defeats a downstream sink's dedup-by-id
+ /// (see `connector-pr-review` B5): a redelivered row would arrive with a
+ /// different id than its first delivery, so a sink keying its upsert on
+ /// `message.id` would store it twice instead of overwriting. Leave this
+ /// `None` only when records genuinely have no natural per-record
+ /// identity (a pure generator, e.g. `random_source`) — `cursor_value`
+ /// alone usually isn't it either, since a timestamp-based cursor is
+ /// commonly shared by more than one record.
+ natural_id: Option<u128>,
payload: serde_json::Value,
}
@@ -234,7 +247,8 @@ impl TemplateSource {
/// result row/document/event to a `FetchedRecord`, using something that
/// monotonically increases (a timestamp, an ID, a page token) as
/// `cursor_value` so the state-staging logic in `poll()` below can
- /// advance the cursor correctly.
+ /// advance the cursor correctly, and set `natural_id` from a true
+ /// per-record identity if one exists (see the field's own doc comment).
async fn fetch_records(
&self,
client: &ClientWithMiddleware,
@@ -391,9 +405,10 @@ impl Source for TemplateSource {
// into `messages` - advancing on a dropped (unserializable) record
// would stage a cursor past data that was never produced, and a
// later Ack would commit past it permanently.
+ let natural_id = record.natural_id;
new_cursor = Some(record.cursor_value);
messages.push(ProducedMessage {
- id: None,
+ id: natural_id,
headers: None,
checksum: None,
timestamp: None,
@@ -453,6 +468,23 @@ impl Source for TemplateSource {
/// delivered. If `fetch_records()` ever needs to perform a destructive
/// read against the source (delete-after-read, mark-as-processed),
/// stage that side effect the same way and only apply it here on `Ack`.
+ ///
+ /// If you add real I/O here (an external ack call, a delete, a mark-
+ /// processed update), two things from a real connector's review history
+ /// apply directly:
+ ///
+ /// 1. This runs inside the SDK's fixed ~30s batch-result window, and a
+ /// size-your-own-timeout mistake here is fatal, not just slow — a
+ /// NACK from that window stops the source for good after repeated
+ /// consecutive NACKs. If you stage more than one operation, budget
+ /// for *all* of them inside the window, not one op's worth, and
+ /// reject `max_retries`/`retry_delay` combinations in `open()` that
+ /// could blow it. See `connector-pr-review` B11.
+ /// 2. A failed op must not be silently logged while the batch is still
+ /// ACKed — the cursor has already committed past it by the time you
+ /// find out. Count consecutive failures and return `Err` past a
+ /// threshold so the SDK stops the source, instead of resending the
+ /// same broken operation forever.
async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(),
Error> {
let candidate_state = self.pending_state.lock().await.take();
if result == SourceBatchResult::Ack
@@ -534,7 +566,7 @@ mod tests {
}
#[tokio::test]
- async fn open_rejects_empty_connection_string() {
+ async fn given_empty_connection_string_should_reject_open() {
let mut config = test_config();
config.connection_string = SecretString::from(" ");
let mut source = TemplateSource::new(1, config, None);
@@ -622,7 +654,7 @@ mod tests {
}
#[tokio::test(start_paused = true)]
- async fn poll_returns_empty_without_error_when_circuit_is_open() {
+ async fn given_open_circuit_breaker_should_return_empty_without_error() {
let source = TemplateSource::new(1, test_config(), None);
source.circuit_breaker.record_failure().await;
source.circuit_breaker.record_failure().await;
