You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "ryerraguntla (via GitHub)" <gi...@apache.org> on 2026/10/24 14:56:23 UTC
[PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
ryerraguntla opened a new pull request, #4636:
URL: https://github.com/apache/iggy/pull/4636
Fixes #3956
## What
Adds two ready-to-copy connector crates under `core/connectors/`:
- `core/connectors/sinks/sink_template/`
- `core/connectors/sources/source_template/`
Each implements every framework-level requirement already expected of a production connector — config parsing with unknown-field rejection,`SecretString`-wrapped credentials, connectivity validation in `open()`, retry/backoff + circuit breaker via `iggy_connector_sdk::retry::RetryPolicy`, batch processing / Ack-Nack cursor staging, and a canonical test suite — leaving only the backend-specific bits marked `TODO(ConnectorDeveloper)`:
- **sink_template**: the outbound data-push call to the destination system, plus a stable per-message dedup key for idempotent writes.
- **source_template**: the connection/client setup, the data-fetch call against the source system, and setting a natural per-record identity on `ProducedMessage.id` when one exists.
Also included:
- `core/connectors/docs/authoring-sinks-and-sources.md` — a written guide walking through the required patterns and why each one exists, meant to be cross-linked from the connector READMEs and reused as the basis for an Apache Iggy blog post.
- `core/connectors/BLOG_POST.md` — a narrative draft for the project blog announcing the templates.
- `.claude/skills/connector-pr-review/SKILL.md` — a new review-checklist skill (B1-B11 blockers, convention nits) encoding review patterns mined from real apache/iggy connector PRs, so recurring review comments (secrets typing, retry/circuit-breaker gaps, cursor staging, idempotency claims, zero-delivered-still-reports-Ok, ack-path budget blowups) get caught before human review instead of during it.
- Updated `connector-sink`, `connector-source`, `connector-testing`, and `connectors-overview` skills to cross-link the templates and the new review checklist, and to reflect the current `Source::batch_policy()` / NACK-retry-configurability SDK surface (#4269) and the current resolver/toolchain setup, replacing stale text that predated both.
- Workspace registration (`Cargo.toml`, `Cargo.lock`) and README entries (`sinks/README.md`, `sources/README.md`, `core/connectors/README.md`) for both new crates, each with `publish = false`.
- Example runtime configs
(`core/connectors/runtime/example_config/connectors/{sink,source}_template.toml`).
## Why
Per the discussion in #3956: connector PRs have repeatedly lost review
cycles to the same non-backend-specific issues (secrets typed as plain
`String`, missing retry/circuit-breaker, config knob name drift, missing
canonical state tests, zero-delivered batches reported as success, ack-path
work with no bound on its own retry budget). Rather than re-teaching the
same framework contract PR after PR, these templates make the required
pattern the path of least resistance — a new connector starts from working,
reviewed-pattern code and only needs the backend-specific logic filled in.
The paired `connector-pr-review` skill turns the same lessons into an automatic pre-flight check for future connector PRs.
## Testing
- [x] `cargo build -p iggy_connector_template_sink -p iggy_connector_template_source`
- [x] `cargo test -p iggy_connector_template_sink -p iggy_connector_template_source` (20/20)
- [x] `cargo clippy -p iggy_connector_template_sink -p iggy_connector_template_source --all-targets -- -D warnings`
- [x] `cargo fmt --all --check`
- [x] `cargo sort --check --no-format` (crate + workspace)
- [x] `cargo machete`
- [x] `cargo doc -p iggy_connector_template_sink -p iggy_connector_template_source --no-deps` with `RUSTDOCFLAGS="-D warnings"`
- [x] `markdownlint` on every new/changed doc
- [x] `./scripts/ci/license-headers.sh --check`
## AI Usage
- Tools: Claude (Claude Code)
- Scope: template crate authoring, the review-checklist skill, rebasing onto current master and reconciling against SDK changes that landed after the branch was first cut (e.g. #4269's `batch_policy()`), and mining real apache/iggy connector PR review comments for the checklist content.
- Verification: see Testing above. Full verification chain run locally against the actual rebased/current `apache/master`, not a stale base.
- I can explain every line of code and every checklist item if asked.
## Notes for reviewers / before marking ready
- This branch previously went out as #3965 (closed: corrupted git history from repeated `master` merges) and separately as a draft on `docs/connector-guidelines-template-and-blog-post` (closed/abandoned, also under a misleading `docs/` prefix despite shipping real code — renamed to this branch for that reason). This PR is the rebuilt, rebased, re-verified version of that work, squashed to a clean history.
## AI Usage
If AI tools were used, please answer:
1. Which tools? Claude
2. Scope of usage? Analysis of the previous connectors PR reiew, summary of the analysis and template generation
3. How did you verify the generated code works correctly? Reviewed Manually
4. Can you explain every line of the code if asked? Yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4492014055
##########
.claude/skills/connector-sink/TEMPLATE.md:
##########
@@ -1,129 +1,305 @@
-# Sink plugin skeleton
+# Sink plugin fill-in-the-blank kit
-Boilerplate for a new `core/connectors/sinks/<name>_sink/`. Adapt the
-`MySink` / `Client` types to the backend driver you're integrating.
+Copy this kit into `core/connectors/sinks/<name>_sink/`. The scaffolding
+covers config, secrets, retry, error classification, batching, logging,
+and unit-test shape. **You only implement the marked `TODO(ConnectorDeveloper)`
+sections:** build a client from the connection string, and push one
+batch.
+
+Also read [SKILL.md](SKILL.md) and pre-flight with
+[connector-pr-review](../connector-pr-review/SKILL.md) before `/ready`.
+
+Prefer starting from a compiling crate over copying this prose kit:
+`core/connectors/sinks/sink_template/` implements the same shape as real,
+tested code you can `cargo build`/`cargo test` immediately, with the same
+`TODO(ConnectorDeveloper)` markers. It also has the circuit breaker, which this
+kit leaves out to stay short. Use this kit instead only when copying a whole
+crate is more scaffolding than you need.
+
+## The delivery contract this kit is built on
+
+By default the server stores a sink's consumer offset when it answers the poll,
+before `consume()` runs, and the runtime never redelivers a batch after
+`consume()` returns an error. So delivery from Iggy is at-most-once, and the
+retrying the plugin does inside `consume()` is the only second chance a batch
+gets. The skeleton below therefore keeps trying a batch within one time budget
+(`max_delivery_time`), logs the offset range it gives up on, and leaves
+idempotent writes to you.
+
+The skeleton skips a chunk the backend rejects for good, so it does not
+override `enable_at_least_once()`, and a sink configured with
+`[delivery] guarantee = "at_least_once"` refuses to start. To support that
+mode, copy `deliver_messages()` and `enable_at_least_once()` from
+`sinks/sink_template`, which fail the call instead of skipping.
+
+## Files to create
+
+```text
+core/connectors/sinks/<name>_sink/
+├── Cargo.toml
+├── README.md
+├── config.toml
+└── src/lib.rs
+```
+
+Add a workspace member, a row in `sinks/README.md`, and a sample under
+`runtime/example_config/connectors/`.
+
+---
## Cargo.toml
```toml
-# Apache 2.0 header (copy verbatim from any existing sink Cargo.toml)
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
[package]
name = "iggy_connector_<name>_sink"
-version = "0.5.1-edge.1" # match the version most other sinks use
+version = "0.5.0" # match the version most other sinks use
edition = "2024"
license = "Apache-2.0"
publish = false
-# ...keywords, description, repository, homepage all identical to existing sinks
+description = "Apache Iggy <Name> sink connector"
+repository = "https://github.com/apache/iggy"
+homepage = "https://iggy.apache.org"
[package.metadata.cargo-machete]
-ignored = ["dashmap", "once_cell"] # used by sink_connector! macro
+ignored = ["dashmap", "once_cell"]
[lib]
-crate-type = ["cdylib", "lib"] # cdylib = runtime-loadable; lib = unit tests
+crate-type = ["cdylib", "lib"]
[dependencies]
-async-trait = { workspace = true }
-dashmap = { workspace = true }
+async-trait = { workspace = true }
+dashmap = { workspace = true }
+iggy_common = { workspace = true }
iggy_connector_sdk = { workspace = true }
-once_cell = { workspace = true }
-serde = { workspace = true }
-tokio = { workspace = true }
-tracing = { workspace = true }
-# + your client crate (reqwest, sqlx, mongodb, ...)
+once_cell = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+# TODO(ConnectorDeveloper): add your client crate as a workspace dependency
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["test-util"] } # for paused-time delivery tests
```
-Run `cargo sort --no-format --workspace` after edits.
+Run `cargo sort --no-format --workspace` after edits. Keep `Cargo.lock`
+churn limited to this crate's deps.
+
+---
+
+## config.toml (example)
+
+A sink config has a connector header, one or more `[[streams]]` entries that
+tell the runtime what to consume and how to batch it, and a `[plugin_config]`
+table for your own settings. `sinks/sink_template/config.toml` is the working
+reference.
+
+```toml
+type = "sink"
+key = "<name>"
+enabled = true
+version = 0
+name = "<Name> sink"
+path = "../../target/release/libiggy_connector_<name>_sink"
+plugin_config_format = "toml"
+verbose = false
+
+[[streams]]
+stream = "example_stream"
+topics = ["example_topic"]
+schema = "json"
+batch_length = 100
+# Long poll: a poll returns once `min_count` messages are ready or `max_wait`
+# expires. These replace `poll_interval`, which the runtime refuses for sinks.
+# `min_count` must not exceed `batch_length`; `max_wait` must be above zero.
+max_wait = "2s"
+min_count = 100
+consumer_group = "<name>_sink_connector"
+
+[plugin_config]
+# Never commit real secrets. Use env overrides in tests/ops.
+connection_string = "scheme://user:pass@host:port/db"
+batch_size = 100
+retry_delay = "500ms"
+retry_max_delay = "5s"
+# One budget for a whole consume() call. Keep it well under a minute: the call
+# blocks a runtime worker thread, and a stop waits for it.
+max_delivery_time = "20s"
+verbose_logging = false
+```
+
+---
## src/lib.rs
-Code reads top to bottom. Public types first, then `impl`, then private
-helpers.
+Replace `<Name>` / `<name>` and implement only the `TODO(ConnectorDeveloper)` blocks.
```rust
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements... */ // full Apache header
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
use async_trait::async_trait;
+use iggy_connector_sdk::retry::{parse_duration, retry_backoff};
use iggy_connector_sdk::{
ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
};
+use secrecy::{ExposeSecret, SecretString};
use serde::{Deserialize, Serialize};
-use std::str::FromStr;
use std::time::Duration;
use tokio::sync::Mutex;
+use tokio::time::{Instant, sleep};
use tracing::{debug, error, info, warn};
-sink_connector!(MySink); // generates FFI symbols + version export
+sink_connector!(NameSink);
-const CONNECTOR_NAME: &str = "My sink";
+const CONNECTOR_NAME: &str = "Name sink";
+const DEFAULT_BATCH_SIZE: u32 = 100;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+// Well under the runtime's close budget (the poll timeout plus 60 s): consume() blocks a
+// runtime worker thread and a stop waits for it.
+const DEFAULT_MAX_DELIVERY_TIME: &str = "20s";
+
+/// Backend client. Replace with the real driver type.
+struct BackendClient {
+ // TODO(ConnectorDeveloper): fields
+}
#[derive(Debug, Serialize, Deserialize)]
-pub struct MySinkConfig {
- pub endpoint: String,
+#[serde(deny_unknown_fields)]
+pub struct NameSinkConfig {
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
pub batch_size: Option<u32>,
- pub max_retries: Option<u32>,
- pub retry_delay: Option<String>, // humantime, e.g. "500ms"
- pub verbose_logging: Option<bool>, // mirror runtime's `verbose` flag
- // Every runtime-tunable field is Option<T>; defaults applied in new()
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_delivery_time: Option<String>,
+ pub verbose_logging: Option<bool>,
+ // TODO(ConnectorDeveloper): optional non-secret knobs (table, index, batch_mode, ...)
}
#[derive(Debug)]
-pub struct MySink {
+pub struct NameSink {
id: u32,
- config: MySinkConfig,
+ config: NameSinkConfig,
batch_size: usize,
- max_retries: u32,
retry_delay: Duration,
+ retry_max_delay: Duration,
+ max_delivery_time: Duration,
verbose: bool,
- client: Option<Client>,
+ client: Option<BackendClient>,
state: Mutex<State>,
}
-#[derive(Debug)]
+#[derive(Debug, Default)]
struct State {
messages_processed: u64,
errors: u64,
}
-impl MySink {
- pub fn new(id: u32, config: MySinkConfig) -> Self {
- let batch_size = config.batch_size.unwrap_or(100) as usize;
- let max_retries = config.max_retries.unwrap_or(3);
- let retry_delay = config
- .retry_delay
- .as_deref()
- .and_then(|raw| humantime::Duration::from_str(raw).ok().map(|d| *d))
- .unwrap_or_else(|| {
- warn!("Invalid retry_delay for {CONNECTOR_NAME} ID: {id}, defaulting to 500ms");
- Duration::from_millis(500)
- });
+impl NameSink {
+ pub fn new(id: u32, config: NameSinkConfig) -> Self {
+ let batch_size = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1) as usize;
+ // `parse_duration` falls back to the default with a `warn!` on a bad value.
Review Comment:
Fixed. The comment and the test now say it falls back to 1s, matching retry.rs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4491964757
##########
core/connectors/sinks/sink_template/src/lib.rs:
##########
@@ -0,0 +1,1069 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **sink** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, the destination identifier is
+//! validated before it's ever interpolated into a request, outbound calls
+//! go through a retry-wrapped client plus a circuit breaker, and messages
+//! are chunked by a configurable batch size instead of shipped as one
+//! unbounded request.
+//!
+//! Delivery follows the sink's `[delivery]` table. By default the server stores
+//! the offset when it answers the poll and the runtime never redelivers a
+//! batch after an error, so `consume()` keeps trying a batch for up to
+//! `max_delivery_time` before it gives up and logs the lost offset range. With
+//! `guarantee = "at_least_once"` the runtime stores the offset only after
+//! `consume()` succeeds, hands a failed batch over again, and stops the sink on
+//! a permanent error instead of skipping the batch. The circuit breaker only
+//! paces the attempts; it never discards a batch. Writes must be idempotent on
+//! `dedup_key()`, because retries, redelivery and failovers can repeat a
+//! message.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSink::push_batch()` — build the
+//! request/write that actually pushes one chunk of messages to your
+//! destination, using `self.config.connection_string` (and
+//! `self.config.target`, if your destination has a table/index/
+//! collection-shaped name). A second `TODO(ConnectorDeveloper)` on the
+//! `connection_string` field's doc comment just asks you to describe its
+//! expected shape — not code, but worth personalizing too. `grep` for the
+//! marker and you'll find both, plus one more on `build_raw_client()` that
+//! only applies if your destination isn't HTTP (see below).
+//!
+//! This template assumes an HTTP-ish destination and uses `reqwest` wrapped
+//! by the SDK's retry middleware, because that's what
+//! `iggy_connector_sdk::retry` is built for and it covers the common case.
+//! If your destination talks something else (a database, a queue, object
+//! storage), swap the client type in `build_raw_client()`/`push_batch()` for
+//! your driver of choice and lean on its own retry/pooling behavior — keep the
+//! surrounding shape (validation in `open()`, circuit breaker, batching,
+//! identifier validation) unchanged. See `core/connectors/sinks/postgres_sink`
+//! or `core/connectors/sinks/s3_sink` in this repo for non-HTTP examples of
+//! that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+ retry_backoff,
+};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::future::Future;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::time::{Instant, sleep};
+use tracing::{debug, error, info, warn};
+
+sink_connector!(TemplateSink);
+
+const CONNECTOR_NAME: &str = "Template sink";
+
+const REDELIVERY_AT_MOST_ONCE: &str =
+ "The runtime does not redeliver them; replay that range from Iggy to recover them.";
+const REDELIVERY_AT_LEAST_ONCE: &str = "The runtime hands them over again.";
+
+const DEFAULT_BATCH_SIZE: usize = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+// Three failed rounds per cool-down cycle bound the attempts a struggling destination sees:
+// the SDK breaker closes completely after its cool-down, so it takes `threshold` new failures
+// to open again. The cool-down has to be shorter than the delivery budget, or a call that
+// finds the breaker open could never reach a retry.
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 3;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "10s";
+// Kept well under the runtime's close budget (the poll timeout plus 60 s): `consume()` blocks
+// a runtime worker thread and a stop waits for it to return.
+const DEFAULT_MAX_DELIVERY_TIME: &str = "20s";
+const BREAKER_RECHECK_INTERVAL: Duration = Duration::from_millis(250);
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` and `target` is optional with a
+// sane default, and unknown keys are rejected outright so a typo in a TOML
+// file fails at load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSinkConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// The destination table/index/collection/bucket name. Kept as its own
+ /// field (rather than folded into `connection_string`) specifically so
+ /// it can be validated in `open()` before ever being interpolated into
+ /// a query, path, or URL — see `validate_identifier` below. Delete this
+ /// field if your destination has no such dynamic identifier.
+ pub target: String,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<usize>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// How long one `consume()` call may keep trying before it gives up on the
+ /// messages it still holds. Must be longer than `circuit_breaker_cool_down`.
+ pub max_delivery_time: Option<String>,
+
+ /// Upgrades the per-batch log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_sink::PostgresSinkConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+/// Rejects anything that isn't a plain alphanumeric/underscore identifier.
+/// Adjust the allowed character set to whatever your destination's naming
+/// rules actually are, but always validate *something* before a
+/// config-or-message-derived name is interpolated into a query, path, or
+/// URL — see `doris_sink::validate_identifier` / `surrealdb_sink::validate_identifier`
+/// in this repo for the same pattern applied to a real destination.
+fn validate_identifier(field: &str, value: &str) -> Result<(), Error> {
+ if value.is_empty() || !value.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+ return Err(Error::InvalidConfigValue(format!(
+ "{field} must be non-empty and contain only ASCII alphanumeric characters and \
+ underscores, got: {value:?}"
+ )));
+ }
+ Ok(())
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+#[derive(Debug, Default)]
+struct State {
+ invocations_count: u64,
+ messages_written: u64,
+ messages_failed: u64,
+}
+
+#[derive(Debug)]
+pub struct TemplateSink {
+ id: u32,
+ config: TemplateSinkConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size_limit: usize,
+ retry_delay: Duration,
+ retry_max_delay: Duration,
+ cool_down: Duration,
+ max_delivery_time: Duration,
+ at_least_once: bool,
+ verbose: bool,
+ state: Mutex<State>,
+ records_written_total: AtomicU64,
+}
+
+impl TemplateSink {
+ pub fn new(id: u32, config: TemplateSinkConfig) -> Self {
+ let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY);
+ let retry_max_delay =
+ parse_duration(config.retry_max_delay.as_deref(), DEFAULT_RETRY_MAX_DELAY);
+ let cool_down = parse_duration(
+ config.circuit_breaker_cool_down.as_deref(),
+ DEFAULT_CIRCUIT_BREAKER_COOL_DOWN,
+ );
+ let max_delivery_time = parse_duration(
+ config.max_delivery_time.as_deref(),
+ DEFAULT_MAX_DELIVERY_TIME,
+ );
+ let circuit_breaker = Arc::new(CircuitBreaker::new(
+ config
+ .circuit_breaker_threshold
+ .unwrap_or(DEFAULT_CIRCUIT_BREAKER_THRESHOLD),
+ cool_down,
+ ));
+ let batch_size_limit = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ client: None,
+ circuit_breaker,
+ batch_size_limit,
+ retry_delay,
+ retry_max_delay,
+ cool_down,
+ max_delivery_time,
+ at_least_once: false,
+ verbose,
+ state: Mutex::new(State::default()),
+ records_written_total: AtomicU64::new(0),
+ }
+ }
+
+ /// TODO(ConnectorDeveloper): build your actual client/connection here using
+ /// `self.config.connection_string` (and `self.config.auth_token`, if
+ /// your destination needs it). This template builds a plain
+ /// `reqwest::Client` to hand to `build_retry_client` — if you're not
+ /// talking HTTP, replace this with e.g. a database connection pool or
+ /// your driver's equivalent, store it on `self` (add a field, since
+ /// `client` here is HTTP-specific), and skip the
+ /// `check_connectivity_with_retry` call below in favor of whatever your
+ /// driver offers (a ping, a test query).
+ fn build_raw_client(&self) -> Result<reqwest::Client, Error> {
+ let timeout = parse_duration(self.config.timeout.as_deref(), DEFAULT_TIMEOUT);
+ reqwest::Client::builder()
+ .timeout(timeout)
+ .build()
+ .map_err(|e| Error::Connection(format!("failed to build HTTP client: {e}")))
+ }
+
+ /// TODO(ConnectorDeveloper): push one chunk of already-batched messages to your
+ /// destination using `self.config.connection_string` and
+ /// `self.config.target`, via `self.client` (already retry-wrapped).
+ ///
+ /// The error you return decides what `consume()` does next:
+ /// - `Error::HttpRequestFailed`, `Error::Connection` and
+ /// `Error::CannotStoreData` (network error, 5xx, timeout) are transient.
+ /// `consume()` retries the same chunk with backoff until
+ /// `max_delivery_time` runs out, and each failed round counts toward the
+ /// circuit breaker. At least once, the failed call then makes the
+ /// runtime hand the batch over again.
+ /// - `Error::PermanentHttpError` (bad schema, the destination rejects the
+ /// payload shape) will never succeed, and the breaker is left alone,
+ /// because a data problem says nothing about whether the destination is
+ /// up. By default the chunk is skipped and counted. At least once, the
+ /// call fails and the runtime stops the sink at that offset, since a
+ /// skipped chunk would let the stored offset pass an undelivered
+ /// message. See the doc comment on that variant in
+ /// `iggy_connector_sdk::Error`.
+ /// - Any other error is not retried. The chunk is counted as failed and
+ /// the error is returned once every chunk has been tried.
+ ///
+ /// By default the runtime never redelivers a batch after an error, because
+ /// the server stored the offset when it answered the poll. The retries
+ /// above are then the only second chance a batch gets, so from Iggy to this
+ /// plugin delivery is at-most-once. With `[delivery] guarantee =
+ /// "at_least_once"` the runtime stores the offset after `consume()`
+ /// succeeds and redelivers a failed batch. Duplicates come from the
+ /// retries (a request that failed part way through a write), from
+ /// redelivery, from replaying an offset range after a logged loss, and from
+ /// a server failover before the stored offset replicated. 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; // 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"
+ .to_string(),
+ ))
+ }
+
+ /// Pushes `messages` in chunks of `batch_size_limit` through `push_chunk`, which receives
+ /// the index range of one chunk.
+ ///
+ /// All chunks share one delivery budget. A chunk the destination rejects for good is skipped
+ /// and counted, except in at-least-once mode, where skipping would let the runtime store
+ /// an offset past an undelivered message: the call fails and the runtime stops the sink.
+ async fn deliver_messages<F, Fut>(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ push_chunk: F,
+ ) -> Result<(), Error>
+ where
+ F: Fn(std::ops::Range<usize>) -> Fut,
+ Fut: Future<Output = Result<(), Error>>,
+ {
+ // One budget for the whole call. With a dead destination every chunk would otherwise wait
+ // out a budget of its own, and the runtime is blocked on this call for all of them.
+ let deadline = Instant::now() + self.max_delivery_time;
+ let mut last_err: Option<Error> = None;
+ let mut written = 0u64;
+ let mut failed = 0u64;
+
+ for start in (0..messages.len()).step_by(self.batch_size_limit) {
+ let end = (start + self.batch_size_limit).min(messages.len());
+ let chunk = &messages[start..end];
+ let result = self
+ .deliver_chunk(deadline, || push_chunk(start..end))
+ .await;
+ match result {
+ Ok(()) => written += chunk.len() as u64,
+ Err(err @ Error::PermanentHttpError(_)) if self.at_least_once => {
Review Comment:
Fixed. Both arms match is_permanent(), with tests using SchemaMismatch for both modes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on PR #4636:
URL: https://github.com/apache/iggy/pull/4636#issuecomment-6385870024
/author
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4491957128
##########
core/connectors/sinks/sink_template/src/lib.rs:
##########
@@ -0,0 +1,1069 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **sink** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, the destination identifier is
+//! validated before it's ever interpolated into a request, outbound calls
+//! go through a retry-wrapped client plus a circuit breaker, and messages
+//! are chunked by a configurable batch size instead of shipped as one
+//! unbounded request.
+//!
+//! Delivery follows the sink's `[delivery]` table. By default the server stores
+//! the offset when it answers the poll and the runtime never redelivers a
+//! batch after an error, so `consume()` keeps trying a batch for up to
+//! `max_delivery_time` before it gives up and logs the lost offset range. With
+//! `guarantee = "at_least_once"` the runtime stores the offset only after
+//! `consume()` succeeds, hands a failed batch over again, and stops the sink on
+//! a permanent error instead of skipping the batch. The circuit breaker only
+//! paces the attempts; it never discards a batch. Writes must be idempotent on
+//! `dedup_key()`, because retries, redelivery and failovers can repeat a
+//! message.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSink::push_batch()` — build the
+//! request/write that actually pushes one chunk of messages to your
+//! destination, using `self.config.connection_string` (and
+//! `self.config.target`, if your destination has a table/index/
+//! collection-shaped name). A second `TODO(ConnectorDeveloper)` on the
+//! `connection_string` field's doc comment just asks you to describe its
+//! expected shape — not code, but worth personalizing too. `grep` for the
+//! marker and you'll find both, plus one more on `build_raw_client()` that
+//! only applies if your destination isn't HTTP (see below).
+//!
+//! This template assumes an HTTP-ish destination and uses `reqwest` wrapped
+//! by the SDK's retry middleware, because that's what
+//! `iggy_connector_sdk::retry` is built for and it covers the common case.
+//! If your destination talks something else (a database, a queue, object
+//! storage), swap the client type in `build_raw_client()`/`push_batch()` for
+//! your driver of choice and lean on its own retry/pooling behavior — keep the
+//! surrounding shape (validation in `open()`, circuit breaker, batching,
+//! identifier validation) unchanged. See `core/connectors/sinks/postgres_sink`
+//! or `core/connectors/sinks/s3_sink` in this repo for non-HTTP examples of
+//! that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+ retry_backoff,
+};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::future::Future;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::time::{Instant, sleep};
+use tracing::{debug, error, info, warn};
+
+sink_connector!(TemplateSink);
+
+const CONNECTOR_NAME: &str = "Template sink";
+
+const REDELIVERY_AT_MOST_ONCE: &str =
+ "The runtime does not redeliver them; replay that range from Iggy to recover them.";
+const REDELIVERY_AT_LEAST_ONCE: &str = "The runtime hands them over again.";
+
+const DEFAULT_BATCH_SIZE: usize = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+// Three failed rounds per cool-down cycle bound the attempts a struggling destination sees:
+// the SDK breaker closes completely after its cool-down, so it takes `threshold` new failures
+// to open again. The cool-down has to be shorter than the delivery budget, or a call that
+// finds the breaker open could never reach a retry.
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 3;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "10s";
+// Kept well under the runtime's close budget (the poll timeout plus 60 s): `consume()` blocks
+// a runtime worker thread and a stop waits for it to return.
+const DEFAULT_MAX_DELIVERY_TIME: &str = "20s";
+const BREAKER_RECHECK_INTERVAL: Duration = Duration::from_millis(250);
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` and `target` is optional with a
+// sane default, and unknown keys are rejected outright so a typo in a TOML
+// file fails at load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSinkConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// The destination table/index/collection/bucket name. Kept as its own
+ /// field (rather than folded into `connection_string`) specifically so
+ /// it can be validated in `open()` before ever being interpolated into
+ /// a query, path, or URL — see `validate_identifier` below. Delete this
+ /// field if your destination has no such dynamic identifier.
+ pub target: String,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<usize>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// How long one `consume()` call may keep trying before it gives up on the
+ /// messages it still holds. Must be longer than `circuit_breaker_cool_down`.
+ pub max_delivery_time: Option<String>,
+
+ /// Upgrades the per-batch log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_sink::PostgresSinkConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+/// Rejects anything that isn't a plain alphanumeric/underscore identifier.
+/// Adjust the allowed character set to whatever your destination's naming
+/// rules actually are, but always validate *something* before a
+/// config-or-message-derived name is interpolated into a query, path, or
+/// URL — see `doris_sink::validate_identifier` / `surrealdb_sink::validate_identifier`
+/// in this repo for the same pattern applied to a real destination.
+fn validate_identifier(field: &str, value: &str) -> Result<(), Error> {
+ if value.is_empty() || !value.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+ return Err(Error::InvalidConfigValue(format!(
+ "{field} must be non-empty and contain only ASCII alphanumeric characters and \
+ underscores, got: {value:?}"
+ )));
+ }
+ Ok(())
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+#[derive(Debug, Default)]
+struct State {
+ invocations_count: u64,
+ messages_written: u64,
+ messages_failed: u64,
+}
+
+#[derive(Debug)]
+pub struct TemplateSink {
+ id: u32,
+ config: TemplateSinkConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size_limit: usize,
+ retry_delay: Duration,
+ retry_max_delay: Duration,
+ cool_down: Duration,
+ max_delivery_time: Duration,
+ at_least_once: bool,
+ verbose: bool,
+ state: Mutex<State>,
+ records_written_total: AtomicU64,
+}
+
+impl TemplateSink {
+ pub fn new(id: u32, config: TemplateSinkConfig) -> Self {
+ let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY);
+ let retry_max_delay =
+ parse_duration(config.retry_max_delay.as_deref(), DEFAULT_RETRY_MAX_DELAY);
+ let cool_down = parse_duration(
+ config.circuit_breaker_cool_down.as_deref(),
+ DEFAULT_CIRCUIT_BREAKER_COOL_DOWN,
+ );
+ let max_delivery_time = parse_duration(
+ config.max_delivery_time.as_deref(),
+ DEFAULT_MAX_DELIVERY_TIME,
+ );
+ let circuit_breaker = Arc::new(CircuitBreaker::new(
+ config
+ .circuit_breaker_threshold
+ .unwrap_or(DEFAULT_CIRCUIT_BREAKER_THRESHOLD),
+ cool_down,
+ ));
+ let batch_size_limit = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ client: None,
+ circuit_breaker,
+ batch_size_limit,
+ retry_delay,
+ retry_max_delay,
+ cool_down,
+ max_delivery_time,
+ at_least_once: false,
+ verbose,
+ state: Mutex::new(State::default()),
+ records_written_total: AtomicU64::new(0),
+ }
+ }
+
+ /// TODO(ConnectorDeveloper): build your actual client/connection here using
+ /// `self.config.connection_string` (and `self.config.auth_token`, if
+ /// your destination needs it). This template builds a plain
+ /// `reqwest::Client` to hand to `build_retry_client` — if you're not
+ /// talking HTTP, replace this with e.g. a database connection pool or
+ /// your driver's equivalent, store it on `self` (add a field, since
+ /// `client` here is HTTP-specific), and skip the
+ /// `check_connectivity_with_retry` call below in favor of whatever your
+ /// driver offers (a ping, a test query).
+ fn build_raw_client(&self) -> Result<reqwest::Client, Error> {
+ let timeout = parse_duration(self.config.timeout.as_deref(), DEFAULT_TIMEOUT);
+ reqwest::Client::builder()
+ .timeout(timeout)
+ .build()
+ .map_err(|e| Error::Connection(format!("failed to build HTTP client: {e}")))
+ }
+
+ /// TODO(ConnectorDeveloper): push one chunk of already-batched messages to your
+ /// destination using `self.config.connection_string` and
+ /// `self.config.target`, via `self.client` (already retry-wrapped).
+ ///
+ /// The error you return decides what `consume()` does next:
+ /// - `Error::HttpRequestFailed`, `Error::Connection` and
+ /// `Error::CannotStoreData` (network error, 5xx, timeout) are transient.
+ /// `consume()` retries the same chunk with backoff until
+ /// `max_delivery_time` runs out, and each failed round counts toward the
+ /// circuit breaker. At least once, the failed call then makes the
+ /// runtime hand the batch over again.
+ /// - `Error::PermanentHttpError` (bad schema, the destination rejects the
+ /// payload shape) will never succeed, and the breaker is left alone,
+ /// because a data problem says nothing about whether the destination is
+ /// up. By default the chunk is skipped and counted. At least once, the
+ /// call fails and the runtime stops the sink at that offset, since a
+ /// skipped chunk would let the stored offset pass an undelivered
+ /// message. See the doc comment on that variant in
+ /// `iggy_connector_sdk::Error`.
+ /// - Any other error is not retried. The chunk is counted as failed and
+ /// the error is returned once every chunk has been tried.
+ ///
+ /// By default the runtime never redelivers a batch after an error, because
+ /// the server stored the offset when it answered the poll. The retries
+ /// above are then the only second chance a batch gets, so from Iggy to this
+ /// plugin delivery is at-most-once. With `[delivery] guarantee =
+ /// "at_least_once"` the runtime stores the offset after `consume()`
+ /// succeeds and redelivers a failed batch. Duplicates come from the
+ /// retries (a request that failed part way through a write), from
+ /// redelivery, from replaying an offset range after a logged loss, and from
+ /// a server failover before the stored offset replicated. 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; // 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"
+ .to_string(),
+ ))
+ }
+
+ /// Pushes `messages` in chunks of `batch_size_limit` through `push_chunk`, which receives
+ /// the index range of one chunk.
+ ///
+ /// All chunks share one delivery budget. A chunk the destination rejects for good is skipped
+ /// and counted, except in at-least-once mode, where skipping would let the runtime store
+ /// an offset past an undelivered message: the call fails and the runtime stops the sink.
+ async fn deliver_messages<F, Fut>(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ push_chunk: F,
+ ) -> Result<(), Error>
+ where
+ F: Fn(std::ops::Range<usize>) -> Fut,
+ Fut: Future<Output = Result<(), Error>>,
+ {
+ // One budget for the whole call. With a dead destination every chunk would otherwise wait
+ // out a budget of its own, and the runtime is blocked on this call for all of them.
+ let deadline = Instant::now() + self.max_delivery_time;
+ let mut last_err: Option<Error> = None;
+ let mut written = 0u64;
+ let mut failed = 0u64;
+
+ for start in (0..messages.len()).step_by(self.batch_size_limit) {
+ let end = (start + self.batch_size_limit).min(messages.len());
+ let chunk = &messages[start..end];
+ let result = self
+ .deliver_chunk(deadline, || push_chunk(start..end))
+ .await;
+ match result {
+ Ok(()) => written += chunk.len() as u64,
+ Err(err @ Error::PermanentHttpError(_)) if self.at_least_once => {
+ let undelivered = &messages[start..];
+ failed += undelivered.len() as u64;
+ error!(
+ "{CONNECTOR_NAME} connector with ID: {} stops at {} messages (permanent), stream: {}, topic: {}, partition: {}, offsets {}. The stored offset stays below them. Cause: {err}",
+ self.id,
+ undelivered.len(),
+ topic_metadata.stream,
+ topic_metadata.topic,
+ messages_metadata.partition_id,
+ offset_range(undelivered)
+ );
+ last_err = Some(err);
+ break;
+ }
+ Err(Error::PermanentHttpError(message)) => {
+ failed += chunk.len() as u64;
+ error!(
+ "{CONNECTOR_NAME} connector with ID: {} skipping {} messages \
+ (permanent), offsets {}: {message}",
+ self.id,
+ chunk.len(),
+ offset_range(chunk)
+ );
+ }
+ // The budget is spent, so the chunks after this one are not attempted either.
+ Err(err) if is_transient(&err) => {
+ let undelivered = &messages[start..];
+ failed += undelivered.len() as u64;
+ let redelivery = if self.at_least_once {
+ REDELIVERY_AT_LEAST_ONCE
+ } else {
+ REDELIVERY_AT_MOST_ONCE
+ };
+ error!(
+ "{CONNECTOR_NAME} connector with ID: {} gave up on {} messages after \
+ {:?}, stream: {}, topic: {}, partition: {}, offsets {}. {redelivery} \
+ Cause: {err}",
+ self.id,
+ undelivered.len(),
+ self.max_delivery_time,
+ topic_metadata.stream,
+ topic_metadata.topic,
+ messages_metadata.partition_id,
+ offset_range(undelivered)
+ );
+ last_err = Some(err);
+ break;
+ }
+ Err(err) => {
+ failed += chunk.len() as u64;
+ error!(
+ "{CONNECTOR_NAME} connector with ID: {} failed {} messages, offsets {}: \
+ {err}",
+ self.id,
+ chunk.len(),
+ offset_range(chunk)
+ );
+ last_err = Some(err);
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_written += written;
+ state.messages_failed += failed;
+ drop(state);
+ self.records_written_total
+ .fetch_add(written, Ordering::Relaxed);
+
+ match last_err {
+ None => Ok(()),
+ Some(err) => Err(err),
+ }
+ }
+
+ /// Runs `attempt` until it succeeds, fails for good, or `deadline` passes.
+ ///
+ /// A failed round counts once toward the circuit breaker, and the wait before the next
+ /// round is jittered exponential backoff. While the breaker is open nothing reaches the
+ /// destination, so a struggling destination sees at most `threshold` rounds per cool-down.
+ async fn deliver_chunk<F, Fut>(&self, deadline: Instant, mut attempt: F) -> Result<(), Error>
+ where
+ F: FnMut() -> Fut,
+ Fut: Future<Output = Result<(), Error>>,
+ {
+ let mut failed_rounds = 0;
+ let mut last_err: Option<Error> = None;
+ loop {
+ // Report the failure that opened the breaker, not the wait that ran out.
+ if let Err(still_open) = self.wait_for_closed_breaker(deadline).await {
+ return Err(last_err.unwrap_or(still_open));
+ }
+ match attempt().await {
+ Ok(()) => {
+ self.circuit_breaker.record_success();
+ return Ok(());
+ }
+ Err(err) if is_transient(&err) => {
+ self.circuit_breaker.record_failure().await;
+ failed_rounds += 1;
+ let pause =
+ retry_backoff(self.retry_delay, failed_rounds, self.retry_max_delay);
+ if Instant::now() + pause >= deadline {
Review Comment:
Fixed. Each attempt now runs under timeout(remaining budget), and a cut-off attempt counts as transient. A test hangs an attempt and checks it ends within the budget.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4492099362
##########
.claude/skills/connector-source/TEMPLATE.md:
##########
@@ -136,45 +226,32 @@ impl Source for MySource {
if self.verbose {
info!(
- "{CONNECTOR_NAME} ID: {} produced {} messages, next cursor: {:?}",
- self.id, messages.len(), next_cursor
+ "{CONNECTOR_NAME} ID: {} polled {} messages, next_cursor: {:?}",
+ self.id,
+ fetched.messages.len(),
+ fetched.next_cursor
);
}
- if messages.is_empty() {
- return Ok(ProducedMessages {
- schema: Schema::Json,
- messages,
- state: None,
- });
- }
-
- candidate.messages_produced += messages.len() as u64;
- candidate.cursor = next_cursor;
- let persisted = ConnectorState::serialize(&candidate, CONNECTOR_NAME, self.id)
- .ok_or_else(|| Error::Serialization("failed to serialize source state".into()))?;
- *self.pending.lock().await = Some(candidate);
+ let persisted = { // brief write
Review Comment:
Fixed. The kit now stages the cursor in poll() and commits it in on_batch_result on Ack. The block compiles and its tests pass, including a staged-but-not-acked test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4492028253
##########
core/connectors/sources/source_template/src/lib.rs:
##########
@@ -0,0 +1,676 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **source** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, outbound calls go through a
+//! retry-wrapped client plus a circuit breaker, and the read cursor is
+//! staged in `poll()` and only committed in `on_batch_result()` on an ACK so
+//! a dropped/nacked batch can be re-polled instead of silently lost.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSource::fetch_records()` — fetch up
+//! to `batch_size` new records from your external system, starting after
+//! `cursor`. `TemplateSource::build_raw_client()` carries a second
+//! `TODO(ConnectorDeveloper)` too, but only applies if your source isn't
+//! HTTP (see below); a third, on the `connection_string` field's doc
+//! comment, just asks you to describe its expected shape — not code.
+//! `grep` for the marker and you'll find all three.
+//!
+//! This template assumes an HTTP-ish source and uses `reqwest` wrapped by
+//! the SDK's retry middleware, because that's what `iggy_connector_sdk::retry`
+//! is built for and it covers the common case. If your source talks to
+//! something else (a database, a queue, a filesystem), swap the client type
+//! in `build_raw_client()`/`fetch_records()` for your driver of choice and lean on
+//! its own retry/pooling behavior — keep the surrounding shape (validation
+//! in `open()`, circuit breaker, cursor staging, batching) unchanged. See
+//! `core/connectors/sources/postgres_source` in this repo for a real
+//! non-HTTP example of that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+};
+use iggy_connector_sdk::{
+ ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source,
+ source::SourceBatchResult, source_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tracing::{debug, error, info, warn};
+
+source_connector!(TemplateSource);
+
+const CONNECTOR_NAME: &str = "Template source";
+
+const DEFAULT_POLL_INTERVAL: &str = "1s";
+const DEFAULT_BATCH_SIZE: u32 = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 5;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "30s";
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` is optional with a sane default,
+// and unknown keys are rejected outright so a typo in a TOML file fails at
+// load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSourceConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<u32>,
+ pub poll_interval: Option<String>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// Upgrades the per-poll log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_source::PostgresSourceConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+/// Read cursor. Kept as a plain `Option<String>` so it fits whatever ordering
+/// field your source uses (a timestamp, an auto-increment ID, an opaque
+/// pagination token, ...) — stringify it however makes sense for your data.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+struct State {
+ cursor: Option<String>,
+}
+
+/// One record as fetched from the external system, before it's turned into
+/// an Iggy message. Replace the `payload` type with whatever your
+/// `fetch_records()` actually produces.
+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,
+}
+
+#[derive(Debug)]
+pub struct TemplateSource {
+ id: u32,
+ config: TemplateSourceConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size: u32,
+ poll_interval: Duration,
+ retry_delay: Duration,
+ verbose: bool,
+ state: Mutex<State>,
+ pending_state: Mutex<Option<State>>,
+ records_produced: AtomicU64,
+}
+
+impl TemplateSource {
+ pub fn new(id: u32, config: TemplateSourceConfig, state: Option<ConnectorState>) -> Self {
+ let poll_interval = *humantime::Duration::from_str_lossy(
+ config
+ .poll_interval
+ .as_deref()
+ .unwrap_or(DEFAULT_POLL_INTERVAL),
+ );
+ let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY);
+ let circuit_breaker = Arc::new(CircuitBreaker::new(
+ config
+ .circuit_breaker_threshold
+ .unwrap_or(DEFAULT_CIRCUIT_BREAKER_THRESHOLD),
+ parse_duration(
+ config.circuit_breaker_cool_down.as_deref(),
+ DEFAULT_CIRCUIT_BREAKER_COOL_DOWN,
+ ),
+ ));
+ let batch_size = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ let restored_state = state
+ .and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+ .inspect(|s| {
+ info!(
+ "Restored state for {CONNECTOR_NAME} connector with ID: {id}. Cursor: {:?}",
+ s.cursor
+ );
+ });
+
+ Self {
+ id,
+ config,
+ client: None,
+ circuit_breaker,
+ batch_size,
+ poll_interval,
+ retry_delay,
+ verbose,
+ state: Mutex::new(restored_state.unwrap_or_default()),
+ pending_state: Mutex::new(None),
+ records_produced: AtomicU64::new(0),
+ }
+ }
+
+ fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+ ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+ }
+
+ /// TODO(ConnectorDeveloper): build your actual client/connection here using
+ /// `self.config.connection_string` (and `self.config.auth_token`, if
+ /// your system needs it). This template builds a plain `reqwest::Client`
+ /// to hand to `build_retry_client` — if you're not talking HTTP, replace
+ /// this with e.g. a `sqlx::PgPool::connect(...)` or your driver's
+ /// equivalent, store it on `self` (add a field, since `client` here is
+ /// HTTP-specific), and skip the `check_connectivity_with_retry` call
+ /// below in favor of whatever your driver offers (a ping, a test query).
+ fn build_raw_client(&self) -> Result<reqwest::Client, Error> {
+ let timeout = parse_duration(self.config.timeout.as_deref(), DEFAULT_TIMEOUT);
+ reqwest::Client::builder()
+ .timeout(timeout)
+ .build()
+ .map_err(|e| Error::Connection(format!("failed to build HTTP client: {e}")))
+ }
+
+ /// TODO(ConnectorDeveloper): fetch up to `self.batch_size` new records from your
+ /// external system, ordered after `cursor` (`None` means "from the
+ /// beginning" or "from now" — whichever is right for your source).
+ /// Use `self.config.connection_string` as the base address and
+ /// `self.client` (already retry-wrapped) to make the request. Map each
+ /// 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, 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,
+ cursor: Option<&str>,
+ ) -> Result<Vec<FetchedRecord>, Error> {
+ let _ = (client, cursor); // remove once implemented
+ Err(Error::InitError(
+ "TemplateSource::fetch_records is not implemented yet — see the TODO(ConnectorDeveloper) comment \
+ in template_source/src/lib.rs"
+ .to_string(),
+ ))
+ }
+}
+
+// ── Source trait ────────────────────────────────────────────────────────────
+
+#[async_trait]
+impl Source for TemplateSource {
+ async fn open(&mut self) -> Result<(), Error> {
+ // Structural validation happens here, not in `new()`, because only
+ // `open()` can return an error — `new()` is a plain factory function
+ // with nowhere to send a "this config is invalid" result.
+ if self
+ .config
+ .connection_string
+ .expose_secret()
+ .trim()
+ .is_empty()
+ {
+ return Err(Error::InvalidConfigValue(
+ "connection_string must not be empty".to_string(),
+ ));
+ }
+
+ info!(
+ "Opening {CONNECTOR_NAME} connector with ID: {}, batch_size: {}, poll_interval: {:?}",
+ self.id, self.batch_size, self.poll_interval
+ );
+
+ let raw_client = self.build_raw_client()?;
+
+ if let Some(health_path) = &self.config.health_check_path {
+ let base = Url::parse(self.config.connection_string.expose_secret()).map_err(|e| {
+ Error::InvalidConfigValue(format!("connection_string is not a valid URL: {e}"))
+ })?;
+ let health_url = base.join(health_path).map_err(|e| {
+ Error::InvalidConfigValue(format!("invalid health_check_path: {e}"))
+ })?;
+ check_connectivity_with_retry(
+ &raw_client,
+ health_url,
+ CONNECTOR_NAME,
+ self.id,
+ RetryPolicy {
+ max_attempts: self
+ .config
+ .max_open_retries
+ .unwrap_or(DEFAULT_MAX_OPEN_RETRIES),
+ base_delay: self.retry_delay,
+ max_delay: parse_duration(
+ self.config.open_retry_max_delay.as_deref(),
+ DEFAULT_OPEN_RETRY_MAX_DELAY,
+ ),
+ },
+ )
+ .await?;
+ } else {
+ warn!(
+ "{CONNECTOR_NAME} connector with ID: {} has no health_check_path configured — \
+ skipping the startup connectivity probe. Consider adding one.",
+ self.id
+ );
+ }
+
+ self.client = Some(build_retry_client(
+ raw_client,
+ self.config
+ .max_retries
+ .unwrap_or(DEFAULT_MAX_RETRIES)
+ .max(1),
+ self.retry_delay,
+ parse_duration(
+ self.config.retry_max_delay.as_deref(),
+ DEFAULT_RETRY_MAX_DELAY,
+ ),
+ CONNECTOR_NAME,
+ ));
+
+ info!(
+ "{CONNECTOR_NAME} connector with ID: {} opened successfully",
+ self.id
+ );
+ Ok(())
+ }
+
+ async fn poll(&self) -> Result<ProducedMessages, Error> {
+ // If the breaker is open, sleep for the normal poll interval and
+ // return an empty (not an error) result. Returning `Err` here would
+ // make the runtime retry `poll()` again immediately with no delay —
+ // see `handle_messages` in the SDK's source container — so an empty
+ // ACK-free result is how a source waits out a known-bad window
+ // without busy-looping or counting against the NACK budget.
+ if self.circuit_breaker.is_open().await {
+ warn!(
+ "{CONNECTOR_NAME} connector with ID: {} — circuit breaker OPEN, skipping poll",
+ self.id
+ );
+ tokio::time::sleep(self.poll_interval).await;
+ return Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages: Vec::new(),
+ state: None,
+ });
+ }
+ tokio::time::sleep(self.poll_interval).await;
+
+ let client = self.client.as_ref().ok_or_else(|| {
+ Error::Connection("client not initialized -- was open() called?".into())
+ })?;
+ let cursor = self.state.lock().await.cursor.clone();
+
+ let records = match self.fetch_records(client, cursor.as_deref()).await {
+ Ok(records) => {
+ self.circuit_breaker.record_success();
+ records
+ }
+ Err(err) => {
+ if !matches!(err, Error::PermanentHttpError(_)) {
+ self.circuit_breaker.record_failure().await;
+ }
+ return Err(err);
+ }
+ };
+
+ if records.is_empty() {
+ return Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages: Vec::new(),
+ state: None,
+ });
+ }
+
+ let mut messages = Vec::with_capacity(records.len());
+ let mut new_cursor = cursor;
+ for record in records {
+ let Ok(payload) = serde_json::to_vec(&record.payload) else {
+ error!(
+ "Failed to serialize a record fetched by {CONNECTOR_NAME} connector with ID: {}",
+ self.id
+ );
+ continue;
+ };
+ // Only advance the candidate cursor once the record actually made it
+ // 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: natural_id,
+ headers: None,
+ checksum: None,
+ timestamp: None,
+ origin_timestamp: None,
+ payload,
+ });
+ }
+
+ if messages.is_empty() {
+ // Every fetched record failed to serialize - no progress was
+ // made, so there's nothing to stage and no reason to write
+ // state, same as the `records.is_empty()` case above.
+ return Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages: Vec::new(),
+ state: None,
+ });
+ }
+
+ let candidate_state = State { cursor: new_cursor };
+ let persisted_state = self.serialize_state(&candidate_state).ok_or_else(|| {
+ Error::Serialization(format!(
+ "failed to serialize state for {CONNECTOR_NAME} connector with ID: {}",
+ self.id
+ ))
+ })?;
+ *self.pending_state.lock().await = Some(candidate_state);
+
+ self.records_produced
+ .fetch_add(messages.len() as u64, Ordering::Relaxed);
+
+ if self.verbose {
+ info!(
+ "{CONNECTOR_NAME} connector with ID: {} produced {} messages",
+ self.id,
+ messages.len()
+ );
+ } else {
+ debug!(
+ "{CONNECTOR_NAME} connector with ID: {} produced {} messages",
+ self.id,
+ messages.len()
+ );
+ }
+
+ Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages,
+ state: Some(persisted_state),
+ })
+ }
+
+ /// The staged cursor from `poll()` is only committed here, and only on
+ /// an ACK. A NACK (delivery failed, state-save failed, runtime is
+ /// shutting down) discards the candidate so the same range is re-polled
+ /// next time — the cursor never moves past data that wasn't confirmed
+ /// 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`.
+ /// A missing runtime result does not produce a synthetic NACK — the SDK
+ /// just keeps waiting (warning past ~30s) rather than guessing.
+ ///
+ /// 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. `on_batch_result` has no SDK-enforced deadline, but the SDK will
+ /// not call `poll()` again until it returns — slow I/O here silently
+ /// stalls the whole source, it doesn't fail loudly. If you stage more
+ /// than one operation, budget for *all* of them (a per-item retry
+ /// budget multiplies with the item count), and reject
+ /// `max_retries`/`retry_delay` combinations in `open()` that would
+ /// make that stall unbounded. 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. A source whose input genuinely
+ /// cannot be replayed after restart should instead override
+ /// `batch_policy()` with `BatchPolicy::with_max_consecutive_nacks(None)`
+ /// rather than relying on the default 5-consecutive-Nack stop.
+ async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(), Error> {
+ let candidate_state = self.pending_state.lock().await.take();
+ if result == SourceBatchResult::Ack
+ && let Some(candidate_state) = candidate_state
+ {
+ *self.state.lock().await = candidate_state;
+ }
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ let state = self.state.lock().await;
+ info!(
+ "{CONNECTOR_NAME} connector with ID: {} closed. Cursor: {:?}, total records produced: {}",
+ self.id,
+ state.cursor,
+ self.records_produced.load(Ordering::Relaxed)
+ );
+ drop(state);
+ self.client = None;
+ Ok(())
+ }
+}
+
+// Small local shim so `new()` doesn't need to pull in `humantime::Duration`'s
+// `FromStr` (which returns `Result`) just to apply a default — mirrors the
+// fallback-with-warning behavior of `iggy_connector_sdk::retry::parse_duration`
+// for the one duration field (`poll_interval`) that isn't itself optional in
+// spirit (there's always a poll interval, just maybe the default one).
+trait DurationExt {
Review Comment:
Fixed. Replaced the shim with parse_duration and dropped the humantime dependency.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4492062310
##########
.claude/skills/connector-source/SKILL.md:
##########
@@ -134,18 +172,20 @@ Match `ProducedMessages.schema` to the bytes in `messages[i].payload`:
### Concurrency
- Runtime spawns ONE `poll()` task per source. No concurrent `poll()`.
+- Only one batch is ever in flight: the SDK does not call `poll()` again until the runtime has reported `Ack`/`Nack` and `on_batch_result()` has returned for the previous batch. Past 30s without a result the SDK only warns and keeps waiting - it does not synthesize a Nack or move on.
- Don't spawn your own long-running Tokio tasks: the runtime owns lifecycle. The exception is a source that listens rather than polls, which has to own its listener; `http_source` is the worked example, and it shuts its tasks down in the last `close()` rather than leaving them to outlive the connector.
### Errors
-| Scenario | Variant |
-| ------------------------------------------- | ------------------------------------------------- |
-| Bad config in `new()`/`open()` | `Error::InitError` |
-| Cannot reach external system at startup | `Error::InitError` or `Error::Connection` |
-| Transient fetch failure (retry-worthy) | `Error::Connection` or `Error::HttpRequestFailed` |
-| Permanent fetch failure (auth, schema gone) | `Error::PermanentHttpError` |
-| Row failed to serialize | `Error::Serialization(...)` |
-| State serialization failed | `Error::Serialization(...)` |
+| Scenario | Variant |
+| --------------------------------------------------- | ------------------------------------------------- |
+| Bad config in `new()`/`open()` | `Error::InitError` |
+| Cannot reach external system at startup | `Error::InitError` or `Error::Connection` |
+| Transient fetch failure (retry-worthy) | `Error::Connection` or `Error::HttpRequestFailed` |
+| Permanent fetch failure (auth, schema gone) | `Error::PermanentHttpError` |
+| Row failed to serialize | `Error::Serialization(...)` |
+| State serialization failed | log + skip (non-fatal) |
Review Comment:
Fixed. The row now says Error::Serialization(...) from poll(), nothing sent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on PR #4636:
URL: https://github.com/apache/iggy/pull/4636#issuecomment-6657612710
@hubcio @core please review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4636:
URL: https://github.com/apache/iggy/pull/4636#issuecomment-6385836214
lets wait with this PR till [deferred polling PR](https://github.com/apache/iggy/pull/4629) is merged, it changes all connectors
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "github-actions[bot] (via GitHub) via commits" <co...@iggy.apache.org>.
github-actions[bot] commented on PR #4636:
URL: https://github.com/apache/iggy/pull/4636#issuecomment-6551336081
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs.
If you need a review, please ensure CI is green and the PR is rebased on the latest master. Don't hesitate to ping the maintainers - either `@core` on Discord or by mentioning them directly here on the PR.
Thank you for your contribution!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "github-actions[bot] (via GitHub) via commits" <co...@iggy.apache.org>.
github-actions[bot] commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4482351031
##########
.claude/skills/connector-source/TEMPLATE.md:
##########
@@ -136,45 +226,32 @@ impl Source for MySource {
if self.verbose {
info!(
- "{CONNECTOR_NAME} ID: {} produced {} messages, next cursor: {:?}",
- self.id, messages.len(), next_cursor
+ "{CONNECTOR_NAME} ID: {} polled {} messages, next_cursor: {:?}",
+ self.id,
+ fetched.messages.len(),
+ fetched.next_cursor
);
}
- if messages.is_empty() {
- return Ok(ProducedMessages {
- schema: Schema::Json,
- messages,
- state: None,
- });
- }
-
- candidate.messages_produced += messages.len() as u64;
- candidate.cursor = next_cursor;
- let persisted = ConnectorState::serialize(&candidate, CONNECTOR_NAME, self.id)
- .ok_or_else(|| Error::Serialization("failed to serialize source state".into()))?;
- *self.pending.lock().await = Some(candidate);
+ let persisted = { // brief write
Review Comment:
critical: `poll()` writes `state.cursor` before the runtime acknowledges the batch, and the kit dropped its `on_batch_result` override. A Nack therefore advances the cursor and loses that batch. Restore the `pending_state` staging and the commit-on-`Ack` handler from `core/connectors/sources/source_template/src/lib.rs`.
##########
.claude/skills/connectors-overview/SKILL.md:
##########
@@ -96,24 +102,14 @@ The connectors codebase is intentionally repetitive across plugins. Cross-plugin
### Secrets
-Any credential-bearing field (connection strings, API keys, bearer tokens, AWS keys) must be `SecretString` from the `secrecy` crate. Plain `String` for a credential is a review-blocker: `SecretString` redacts on `Debug`, so it is what keeps a credential out of a log line that formats the whole config.
-
-**`serde_secret::serialize_secret` EXPOSES the secret. It does not redact.** It calls `expose_secret()` and writes the plaintext. `SecretString` deliberately has no `Serialize` impl, and that absence is the protection - so adding `serialize_with` is what *unblocks* the derive and turns a compile-time guarantee into plaintext output. Use it only where the plaintext is the point: a wire payload, a persisted config, an API response that exposes credentials by design.
-
-So the default for a plugin config struct is **derive `Deserialize`, but not `Serialize`**. `Deserialize` is required: the SDK glue deserializes the config into the plugin's own struct (`sdk/src/{sink,source}.rs` call `serde_json::from_str::<C>` under a `DeserializeOwned` bound).
-
-What never happens is the return trip. The runtime holds plugin configuration as a `serde_json::Value` - parsed from TOML, posted as JSON to the control API, or injected by env var - and hands that across the FFI, so nothing re-serializes the plugin's struct. Leaving `Serialize` off makes that compiler-enforced instead of convention-enforced (`sources/http_source/src/lib.rs::HttpSourceConfig` does this, and comments the omission so nobody adds it back).
-
-Pattern:
+Any credential-bearing field (connection strings, API keys, bearer tokens, AWS keys) must be `SecretString` from the `secrecy` crate, with the workspace serde wrapper applied so `Debug` and serialization both redact. Runtime exposes plugin configs over the `/stats` HTTP surface via serialization - plain `String` leaks the secret to anyone who can hit the endpoint. Plain `String` for a credential is a review-blocker. Pattern (from `sinks/postgres_sink/src/lib.rs::PostgresSinkConfig`):
Review Comment:
critical: `serialize_secret` writes the plaintext, not `[REDACTED]`. The redacting helpers are `serialize_redacted` and `serialize_optional_redacted`, and the runtime serves the raw `plugin_config` value, so a plugin-side annotation cannot redact the API output. Use a redacting helper or keep `Serialize` off, and restore the warning that `serialize_secret` exposes the secret. Also at line 202.
##########
core/connectors/sources/source_template/src/lib.rs:
##########
@@ -0,0 +1,676 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **source** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, outbound calls go through a
+//! retry-wrapped client plus a circuit breaker, and the read cursor is
+//! staged in `poll()` and only committed in `on_batch_result()` on an ACK so
+//! a dropped/nacked batch can be re-polled instead of silently lost.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSource::fetch_records()` — fetch up
+//! to `batch_size` new records from your external system, starting after
+//! `cursor`. `TemplateSource::build_raw_client()` carries a second
+//! `TODO(ConnectorDeveloper)` too, but only applies if your source isn't
+//! HTTP (see below); a third, on the `connection_string` field's doc
+//! comment, just asks you to describe its expected shape — not code.
+//! `grep` for the marker and you'll find all three.
+//!
+//! This template assumes an HTTP-ish source and uses `reqwest` wrapped by
+//! the SDK's retry middleware, because that's what `iggy_connector_sdk::retry`
+//! is built for and it covers the common case. If your source talks to
+//! something else (a database, a queue, a filesystem), swap the client type
+//! in `build_raw_client()`/`fetch_records()` for your driver of choice and lean on
+//! its own retry/pooling behavior — keep the surrounding shape (validation
+//! in `open()`, circuit breaker, cursor staging, batching) unchanged. See
+//! `core/connectors/sources/postgres_source` in this repo for a real
+//! non-HTTP example of that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+};
+use iggy_connector_sdk::{
+ ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source,
+ source::SourceBatchResult, source_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tracing::{debug, error, info, warn};
+
+source_connector!(TemplateSource);
+
+const CONNECTOR_NAME: &str = "Template source";
+
+const DEFAULT_POLL_INTERVAL: &str = "1s";
+const DEFAULT_BATCH_SIZE: u32 = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 5;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "30s";
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` is optional with a sane default,
+// and unknown keys are rejected outright so a typo in a TOML file fails at
+// load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSourceConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<u32>,
+ pub poll_interval: Option<String>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// Upgrades the per-poll log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_source::PostgresSourceConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+/// Read cursor. Kept as a plain `Option<String>` so it fits whatever ordering
+/// field your source uses (a timestamp, an auto-increment ID, an opaque
+/// pagination token, ...) — stringify it however makes sense for your data.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+struct State {
+ cursor: Option<String>,
+}
+
+/// One record as fetched from the external system, before it's turned into
+/// an Iggy message. Replace the `payload` type with whatever your
+/// `fetch_records()` actually produces.
+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,
+}
+
+#[derive(Debug)]
+pub struct TemplateSource {
+ id: u32,
+ config: TemplateSourceConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size: u32,
+ poll_interval: Duration,
+ retry_delay: Duration,
+ verbose: bool,
+ state: Mutex<State>,
+ pending_state: Mutex<Option<State>>,
+ records_produced: AtomicU64,
+}
+
+impl TemplateSource {
+ pub fn new(id: u32, config: TemplateSourceConfig, state: Option<ConnectorState>) -> Self {
+ let poll_interval = *humantime::Duration::from_str_lossy(
+ config
+ .poll_interval
+ .as_deref()
+ .unwrap_or(DEFAULT_POLL_INTERVAL),
+ );
+ let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY);
+ let circuit_breaker = Arc::new(CircuitBreaker::new(
+ config
+ .circuit_breaker_threshold
+ .unwrap_or(DEFAULT_CIRCUIT_BREAKER_THRESHOLD),
+ parse_duration(
+ config.circuit_breaker_cool_down.as_deref(),
+ DEFAULT_CIRCUIT_BREAKER_COOL_DOWN,
+ ),
+ ));
+ let batch_size = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ let restored_state = state
+ .and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+ .inspect(|s| {
+ info!(
+ "Restored state for {CONNECTOR_NAME} connector with ID: {id}. Cursor: {:?}",
+ s.cursor
+ );
+ });
+
+ Self {
+ id,
+ config,
+ client: None,
+ circuit_breaker,
+ batch_size,
+ poll_interval,
+ retry_delay,
+ verbose,
+ state: Mutex::new(restored_state.unwrap_or_default()),
+ pending_state: Mutex::new(None),
+ records_produced: AtomicU64::new(0),
+ }
+ }
+
+ fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+ ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+ }
+
+ /// TODO(ConnectorDeveloper): build your actual client/connection here using
+ /// `self.config.connection_string` (and `self.config.auth_token`, if
+ /// your system needs it). This template builds a plain `reqwest::Client`
+ /// to hand to `build_retry_client` — if you're not talking HTTP, replace
+ /// this with e.g. a `sqlx::PgPool::connect(...)` or your driver's
+ /// equivalent, store it on `self` (add a field, since `client` here is
+ /// HTTP-specific), and skip the `check_connectivity_with_retry` call
+ /// below in favor of whatever your driver offers (a ping, a test query).
+ fn build_raw_client(&self) -> Result<reqwest::Client, Error> {
+ let timeout = parse_duration(self.config.timeout.as_deref(), DEFAULT_TIMEOUT);
+ reqwest::Client::builder()
+ .timeout(timeout)
+ .build()
+ .map_err(|e| Error::Connection(format!("failed to build HTTP client: {e}")))
+ }
+
+ /// TODO(ConnectorDeveloper): fetch up to `self.batch_size` new records from your
+ /// external system, ordered after `cursor` (`None` means "from the
+ /// beginning" or "from now" — whichever is right for your source).
+ /// Use `self.config.connection_string` as the base address and
+ /// `self.client` (already retry-wrapped) to make the request. Map each
+ /// 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, 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,
+ cursor: Option<&str>,
+ ) -> Result<Vec<FetchedRecord>, Error> {
+ let _ = (client, cursor); // remove once implemented
+ Err(Error::InitError(
+ "TemplateSource::fetch_records is not implemented yet — see the TODO(ConnectorDeveloper) comment \
+ in template_source/src/lib.rs"
+ .to_string(),
+ ))
+ }
+}
+
+// ── Source trait ────────────────────────────────────────────────────────────
+
+#[async_trait]
+impl Source for TemplateSource {
+ async fn open(&mut self) -> Result<(), Error> {
+ // Structural validation happens here, not in `new()`, because only
+ // `open()` can return an error — `new()` is a plain factory function
+ // with nowhere to send a "this config is invalid" result.
+ if self
+ .config
+ .connection_string
+ .expose_secret()
+ .trim()
+ .is_empty()
+ {
+ return Err(Error::InvalidConfigValue(
+ "connection_string must not be empty".to_string(),
+ ));
+ }
+
+ info!(
+ "Opening {CONNECTOR_NAME} connector with ID: {}, batch_size: {}, poll_interval: {:?}",
+ self.id, self.batch_size, self.poll_interval
+ );
+
+ let raw_client = self.build_raw_client()?;
+
+ if let Some(health_path) = &self.config.health_check_path {
+ let base = Url::parse(self.config.connection_string.expose_secret()).map_err(|e| {
+ Error::InvalidConfigValue(format!("connection_string is not a valid URL: {e}"))
+ })?;
+ let health_url = base.join(health_path).map_err(|e| {
+ Error::InvalidConfigValue(format!("invalid health_check_path: {e}"))
+ })?;
+ check_connectivity_with_retry(
+ &raw_client,
+ health_url,
+ CONNECTOR_NAME,
+ self.id,
+ RetryPolicy {
+ max_attempts: self
+ .config
+ .max_open_retries
+ .unwrap_or(DEFAULT_MAX_OPEN_RETRIES),
+ base_delay: self.retry_delay,
+ max_delay: parse_duration(
+ self.config.open_retry_max_delay.as_deref(),
+ DEFAULT_OPEN_RETRY_MAX_DELAY,
+ ),
+ },
+ )
+ .await?;
+ } else {
+ warn!(
+ "{CONNECTOR_NAME} connector with ID: {} has no health_check_path configured — \
+ skipping the startup connectivity probe. Consider adding one.",
+ self.id
+ );
+ }
+
+ self.client = Some(build_retry_client(
+ raw_client,
+ self.config
+ .max_retries
+ .unwrap_or(DEFAULT_MAX_RETRIES)
+ .max(1),
+ self.retry_delay,
+ parse_duration(
+ self.config.retry_max_delay.as_deref(),
+ DEFAULT_RETRY_MAX_DELAY,
+ ),
+ CONNECTOR_NAME,
+ ));
+
+ info!(
+ "{CONNECTOR_NAME} connector with ID: {} opened successfully",
+ self.id
+ );
+ Ok(())
+ }
+
+ async fn poll(&self) -> Result<ProducedMessages, Error> {
+ // If the breaker is open, sleep for the normal poll interval and
+ // return an empty (not an error) result. Returning `Err` here would
+ // make the runtime retry `poll()` again immediately with no delay —
+ // see `handle_messages` in the SDK's source container — so an empty
+ // ACK-free result is how a source waits out a known-bad window
+ // without busy-looping or counting against the NACK budget.
+ if self.circuit_breaker.is_open().await {
+ warn!(
+ "{CONNECTOR_NAME} connector with ID: {} — circuit breaker OPEN, skipping poll",
+ self.id
+ );
+ tokio::time::sleep(self.poll_interval).await;
+ return Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages: Vec::new(),
+ state: None,
+ });
+ }
+ tokio::time::sleep(self.poll_interval).await;
+
+ let client = self.client.as_ref().ok_or_else(|| {
+ Error::Connection("client not initialized -- was open() called?".into())
+ })?;
+ let cursor = self.state.lock().await.cursor.clone();
+
+ let records = match self.fetch_records(client, cursor.as_deref()).await {
+ Ok(records) => {
+ self.circuit_breaker.record_success();
+ records
+ }
+ Err(err) => {
+ if !matches!(err, Error::PermanentHttpError(_)) {
+ self.circuit_breaker.record_failure().await;
+ }
+ return Err(err);
+ }
+ };
+
+ if records.is_empty() {
+ return Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages: Vec::new(),
+ state: None,
+ });
+ }
+
+ let mut messages = Vec::with_capacity(records.len());
+ let mut new_cursor = cursor;
+ for record in records {
+ let Ok(payload) = serde_json::to_vec(&record.payload) else {
+ error!(
+ "Failed to serialize a record fetched by {CONNECTOR_NAME} connector with ID: {}",
+ self.id
+ );
+ continue;
+ };
+ // Only advance the candidate cursor once the record actually made it
+ // 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);
Review Comment:
warning: The loop skips a record that fails to serialize, but a later record still moves `new_cursor` past it. The next poll never fetches that record again, and the comment above claims the opposite. Return `Err` and stage nothing when a record cannot be encoded, or stop at the first failed record.
##########
.claude/skills/connector-runtime/SKILL.md:
##########
@@ -217,7 +217,7 @@ Fatal errors propagate to `main` and exit. Per-connector / per-message errors ar
All families labeled by `connector_key` + `connector_type` (histogram adds `stage`):
-- **Counters**: `iggy_connector_messages_{produced,sent,consumed,processed,filtered}_total` and `iggy_connector_errors_total`. These are the *rendered* names; each is registered without the `_total`, which the OpenMetrics encoder appends.
+- **Counters**: `iggy_connector_messages_{produced,sent,consumed,processed,filtered,errors}_total`.
Review Comment:
nit: The new counters line expands to `iggy_connector_messages_errors_total`. The runtime renders the errors counter as `iggy_connector_errors_total` (`runtime/src/metrics.rs:562`). Name the errors counter separately.
##########
core/connectors/sinks/sink_template/src/lib.rs:
##########
@@ -0,0 +1,1069 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **sink** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, the destination identifier is
+//! validated before it's ever interpolated into a request, outbound calls
+//! go through a retry-wrapped client plus a circuit breaker, and messages
+//! are chunked by a configurable batch size instead of shipped as one
+//! unbounded request.
+//!
+//! Delivery follows the sink's `[delivery]` table. By default the server stores
+//! the offset when it answers the poll and the runtime never redelivers a
+//! batch after an error, so `consume()` keeps trying a batch for up to
+//! `max_delivery_time` before it gives up and logs the lost offset range. With
+//! `guarantee = "at_least_once"` the runtime stores the offset only after
+//! `consume()` succeeds, hands a failed batch over again, and stops the sink on
+//! a permanent error instead of skipping the batch. The circuit breaker only
+//! paces the attempts; it never discards a batch. Writes must be idempotent on
+//! `dedup_key()`, because retries, redelivery and failovers can repeat a
+//! message.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSink::push_batch()` — build the
+//! request/write that actually pushes one chunk of messages to your
+//! destination, using `self.config.connection_string` (and
+//! `self.config.target`, if your destination has a table/index/
+//! collection-shaped name). A second `TODO(ConnectorDeveloper)` on the
+//! `connection_string` field's doc comment just asks you to describe its
+//! expected shape — not code, but worth personalizing too. `grep` for the
+//! marker and you'll find both, plus one more on `build_raw_client()` that
+//! only applies if your destination isn't HTTP (see below).
+//!
+//! This template assumes an HTTP-ish destination and uses `reqwest` wrapped
+//! by the SDK's retry middleware, because that's what
+//! `iggy_connector_sdk::retry` is built for and it covers the common case.
+//! If your destination talks something else (a database, a queue, object
+//! storage), swap the client type in `build_raw_client()`/`push_batch()` for
+//! your driver of choice and lean on its own retry/pooling behavior — keep the
+//! surrounding shape (validation in `open()`, circuit breaker, batching,
+//! identifier validation) unchanged. See `core/connectors/sinks/postgres_sink`
+//! or `core/connectors/sinks/s3_sink` in this repo for non-HTTP examples of
+//! that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+ retry_backoff,
+};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::future::Future;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::time::{Instant, sleep};
+use tracing::{debug, error, info, warn};
+
+sink_connector!(TemplateSink);
+
+const CONNECTOR_NAME: &str = "Template sink";
+
+const REDELIVERY_AT_MOST_ONCE: &str =
+ "The runtime does not redeliver them; replay that range from Iggy to recover them.";
+const REDELIVERY_AT_LEAST_ONCE: &str = "The runtime hands them over again.";
+
+const DEFAULT_BATCH_SIZE: usize = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+// Three failed rounds per cool-down cycle bound the attempts a struggling destination sees:
+// the SDK breaker closes completely after its cool-down, so it takes `threshold` new failures
+// to open again. The cool-down has to be shorter than the delivery budget, or a call that
+// finds the breaker open could never reach a retry.
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 3;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "10s";
+// Kept well under the runtime's close budget (the poll timeout plus 60 s): `consume()` blocks
+// a runtime worker thread and a stop waits for it to return.
+const DEFAULT_MAX_DELIVERY_TIME: &str = "20s";
+const BREAKER_RECHECK_INTERVAL: Duration = Duration::from_millis(250);
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` and `target` is optional with a
+// sane default, and unknown keys are rejected outright so a typo in a TOML
+// file fails at load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSinkConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// The destination table/index/collection/bucket name. Kept as its own
+ /// field (rather than folded into `connection_string`) specifically so
+ /// it can be validated in `open()` before ever being interpolated into
+ /// a query, path, or URL — see `validate_identifier` below. Delete this
+ /// field if your destination has no such dynamic identifier.
+ pub target: String,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<usize>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// How long one `consume()` call may keep trying before it gives up on the
+ /// messages it still holds. Must be longer than `circuit_breaker_cool_down`.
+ pub max_delivery_time: Option<String>,
+
+ /// Upgrades the per-batch log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_sink::PostgresSinkConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+/// Rejects anything that isn't a plain alphanumeric/underscore identifier.
+/// Adjust the allowed character set to whatever your destination's naming
+/// rules actually are, but always validate *something* before a
+/// config-or-message-derived name is interpolated into a query, path, or
+/// URL — see `doris_sink::validate_identifier` / `surrealdb_sink::validate_identifier`
+/// in this repo for the same pattern applied to a real destination.
+fn validate_identifier(field: &str, value: &str) -> Result<(), Error> {
+ if value.is_empty() || !value.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+ return Err(Error::InvalidConfigValue(format!(
+ "{field} must be non-empty and contain only ASCII alphanumeric characters and \
+ underscores, got: {value:?}"
+ )));
+ }
+ Ok(())
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+#[derive(Debug, Default)]
+struct State {
+ invocations_count: u64,
+ messages_written: u64,
+ messages_failed: u64,
+}
+
+#[derive(Debug)]
+pub struct TemplateSink {
+ id: u32,
+ config: TemplateSinkConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size_limit: usize,
+ retry_delay: Duration,
+ retry_max_delay: Duration,
+ cool_down: Duration,
+ max_delivery_time: Duration,
+ at_least_once: bool,
+ verbose: bool,
+ state: Mutex<State>,
+ records_written_total: AtomicU64,
+}
+
+impl TemplateSink {
+ pub fn new(id: u32, config: TemplateSinkConfig) -> Self {
+ let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY);
+ let retry_max_delay =
+ parse_duration(config.retry_max_delay.as_deref(), DEFAULT_RETRY_MAX_DELAY);
+ let cool_down = parse_duration(
+ config.circuit_breaker_cool_down.as_deref(),
+ DEFAULT_CIRCUIT_BREAKER_COOL_DOWN,
+ );
+ let max_delivery_time = parse_duration(
+ config.max_delivery_time.as_deref(),
+ DEFAULT_MAX_DELIVERY_TIME,
+ );
+ let circuit_breaker = Arc::new(CircuitBreaker::new(
+ config
+ .circuit_breaker_threshold
+ .unwrap_or(DEFAULT_CIRCUIT_BREAKER_THRESHOLD),
+ cool_down,
+ ));
+ let batch_size_limit = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ client: None,
+ circuit_breaker,
+ batch_size_limit,
+ retry_delay,
+ retry_max_delay,
+ cool_down,
+ max_delivery_time,
+ at_least_once: false,
+ verbose,
+ state: Mutex::new(State::default()),
+ records_written_total: AtomicU64::new(0),
+ }
+ }
+
+ /// TODO(ConnectorDeveloper): build your actual client/connection here using
+ /// `self.config.connection_string` (and `self.config.auth_token`, if
+ /// your destination needs it). This template builds a plain
+ /// `reqwest::Client` to hand to `build_retry_client` — if you're not
+ /// talking HTTP, replace this with e.g. a database connection pool or
+ /// your driver's equivalent, store it on `self` (add a field, since
+ /// `client` here is HTTP-specific), and skip the
+ /// `check_connectivity_with_retry` call below in favor of whatever your
+ /// driver offers (a ping, a test query).
+ fn build_raw_client(&self) -> Result<reqwest::Client, Error> {
+ let timeout = parse_duration(self.config.timeout.as_deref(), DEFAULT_TIMEOUT);
+ reqwest::Client::builder()
+ .timeout(timeout)
+ .build()
+ .map_err(|e| Error::Connection(format!("failed to build HTTP client: {e}")))
+ }
+
+ /// TODO(ConnectorDeveloper): push one chunk of already-batched messages to your
+ /// destination using `self.config.connection_string` and
+ /// `self.config.target`, via `self.client` (already retry-wrapped).
+ ///
+ /// The error you return decides what `consume()` does next:
+ /// - `Error::HttpRequestFailed`, `Error::Connection` and
+ /// `Error::CannotStoreData` (network error, 5xx, timeout) are transient.
+ /// `consume()` retries the same chunk with backoff until
+ /// `max_delivery_time` runs out, and each failed round counts toward the
+ /// circuit breaker. At least once, the failed call then makes the
+ /// runtime hand the batch over again.
+ /// - `Error::PermanentHttpError` (bad schema, the destination rejects the
+ /// payload shape) will never succeed, and the breaker is left alone,
+ /// because a data problem says nothing about whether the destination is
+ /// up. By default the chunk is skipped and counted. At least once, the
+ /// call fails and the runtime stops the sink at that offset, since a
+ /// skipped chunk would let the stored offset pass an undelivered
+ /// message. See the doc comment on that variant in
+ /// `iggy_connector_sdk::Error`.
+ /// - Any other error is not retried. The chunk is counted as failed and
+ /// the error is returned once every chunk has been tried.
+ ///
+ /// By default the runtime never redelivers a batch after an error, because
+ /// the server stored the offset when it answered the poll. The retries
+ /// above are then the only second chance a batch gets, so from Iggy to this
+ /// plugin delivery is at-most-once. With `[delivery] guarantee =
+ /// "at_least_once"` the runtime stores the offset after `consume()`
+ /// succeeds and redelivers a failed batch. Duplicates come from the
+ /// retries (a request that failed part way through a write), from
+ /// redelivery, from replaying an offset range after a logged loss, and from
+ /// a server failover before the stored offset replicated. 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; // 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"
+ .to_string(),
+ ))
+ }
+
+ /// Pushes `messages` in chunks of `batch_size_limit` through `push_chunk`, which receives
+ /// the index range of one chunk.
+ ///
+ /// All chunks share one delivery budget. A chunk the destination rejects for good is skipped
+ /// and counted, except in at-least-once mode, where skipping would let the runtime store
+ /// an offset past an undelivered message: the call fails and the runtime stops the sink.
+ async fn deliver_messages<F, Fut>(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ push_chunk: F,
+ ) -> Result<(), Error>
+ where
+ F: Fn(std::ops::Range<usize>) -> Fut,
+ Fut: Future<Output = Result<(), Error>>,
+ {
+ // One budget for the whole call. With a dead destination every chunk would otherwise wait
+ // out a budget of its own, and the runtime is blocked on this call for all of them.
+ let deadline = Instant::now() + self.max_delivery_time;
+ let mut last_err: Option<Error> = None;
+ let mut written = 0u64;
+ let mut failed = 0u64;
+
+ for start in (0..messages.len()).step_by(self.batch_size_limit) {
+ let end = (start + self.batch_size_limit).min(messages.len());
+ let chunk = &messages[start..end];
+ let result = self
+ .deliver_chunk(deadline, || push_chunk(start..end))
+ .await;
+ match result {
+ Ok(()) => written += chunk.len() as u64,
+ Err(err @ Error::PermanentHttpError(_)) if self.at_least_once => {
+ let undelivered = &messages[start..];
+ failed += undelivered.len() as u64;
+ error!(
+ "{CONNECTOR_NAME} connector with ID: {} stops at {} messages (permanent), stream: {}, topic: {}, partition: {}, offsets {}. The stored offset stays below them. Cause: {err}",
+ self.id,
+ undelivered.len(),
+ topic_metadata.stream,
+ topic_metadata.topic,
+ messages_metadata.partition_id,
+ offset_range(undelivered)
+ );
+ last_err = Some(err);
+ break;
+ }
+ Err(Error::PermanentHttpError(message)) => {
+ failed += chunk.len() as u64;
+ error!(
+ "{CONNECTOR_NAME} connector with ID: {} skipping {} messages \
+ (permanent), offsets {}: {message}",
+ self.id,
+ chunk.len(),
+ offset_range(chunk)
+ );
+ }
+ // The budget is spent, so the chunks after this one are not attempted either.
+ Err(err) if is_transient(&err) => {
+ let undelivered = &messages[start..];
+ failed += undelivered.len() as u64;
+ let redelivery = if self.at_least_once {
+ REDELIVERY_AT_LEAST_ONCE
+ } else {
+ REDELIVERY_AT_MOST_ONCE
+ };
+ error!(
+ "{CONNECTOR_NAME} connector with ID: {} gave up on {} messages after \
+ {:?}, stream: {}, topic: {}, partition: {}, offsets {}. {redelivery} \
+ Cause: {err}",
+ self.id,
+ undelivered.len(),
+ self.max_delivery_time,
+ topic_metadata.stream,
+ topic_metadata.topic,
+ messages_metadata.partition_id,
+ offset_range(undelivered)
+ );
+ last_err = Some(err);
+ break;
+ }
+ Err(err) => {
+ failed += chunk.len() as u64;
+ error!(
+ "{CONNECTOR_NAME} connector with ID: {} failed {} messages, offsets {}: \
+ {err}",
+ self.id,
+ chunk.len(),
+ offset_range(chunk)
+ );
+ last_err = Some(err);
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_written += written;
+ state.messages_failed += failed;
+ drop(state);
+ self.records_written_total
+ .fetch_add(written, Ordering::Relaxed);
+
+ match last_err {
+ None => Ok(()),
+ Some(err) => Err(err),
+ }
+ }
+
+ /// Runs `attempt` until it succeeds, fails for good, or `deadline` passes.
+ ///
+ /// A failed round counts once toward the circuit breaker, and the wait before the next
+ /// round is jittered exponential backoff. While the breaker is open nothing reaches the
+ /// destination, so a struggling destination sees at most `threshold` rounds per cool-down.
+ async fn deliver_chunk<F, Fut>(&self, deadline: Instant, mut attempt: F) -> Result<(), Error>
+ where
+ F: FnMut() -> Fut,
+ Fut: Future<Output = Result<(), Error>>,
+ {
+ let mut failed_rounds = 0;
+ let mut last_err: Option<Error> = None;
+ loop {
+ // Report the failure that opened the breaker, not the wait that ran out.
+ if let Err(still_open) = self.wait_for_closed_breaker(deadline).await {
+ return Err(last_err.unwrap_or(still_open));
+ }
+ match attempt().await {
+ Ok(()) => {
+ self.circuit_breaker.record_success();
+ return Ok(());
+ }
+ Err(err) if is_transient(&err) => {
+ self.circuit_breaker.record_failure().await;
+ failed_rounds += 1;
+ let pause =
+ retry_backoff(self.retry_delay, failed_rounds, self.retry_max_delay);
+ if Instant::now() + pause >= deadline {
Review Comment:
warning: `deliver_chunk` checks the deadline only after `attempt()` returns. The retry middleware's timed attempts and `Retry-After` sleeps can therefore run past `max_delivery_time` inside one round. Bound each attempt with the remaining budget, or build the client with one attempt and reject a `timeout` above the budget in `open()`.
##########
.claude/skills/connector-source/SKILL.md:
##########
@@ -134,18 +172,20 @@ Match `ProducedMessages.schema` to the bytes in `messages[i].payload`:
### Concurrency
- Runtime spawns ONE `poll()` task per source. No concurrent `poll()`.
+- Only one batch is ever in flight: the SDK does not call `poll()` again until the runtime has reported `Ack`/`Nack` and `on_batch_result()` has returned for the previous batch. Past 30s without a result the SDK only warns and keeps waiting - it does not synthesize a Nack or move on.
- Don't spawn your own long-running Tokio tasks: the runtime owns lifecycle. The exception is a source that listens rather than polls, which has to own its listener; `http_source` is the worked example, and it shuts its tasks down in the last `close()` rather than leaving them to outlive the connector.
### Errors
-| Scenario | Variant |
-| ------------------------------------------- | ------------------------------------------------- |
-| Bad config in `new()`/`open()` | `Error::InitError` |
-| Cannot reach external system at startup | `Error::InitError` or `Error::Connection` |
-| Transient fetch failure (retry-worthy) | `Error::Connection` or `Error::HttpRequestFailed` |
-| Permanent fetch failure (auth, schema gone) | `Error::PermanentHttpError` |
-| Row failed to serialize | `Error::Serialization(...)` |
-| State serialization failed | `Error::Serialization(...)` |
+| Scenario | Variant |
+| --------------------------------------------------- | ------------------------------------------------- |
+| Bad config in `new()`/`open()` | `Error::InitError` |
+| Cannot reach external system at startup | `Error::InitError` or `Error::Connection` |
+| Transient fetch failure (retry-worthy) | `Error::Connection` or `Error::HttpRequestFailed` |
+| Permanent fetch failure (auth, schema gone) | `Error::PermanentHttpError` |
+| Row failed to serialize | `Error::Serialization(...)` |
+| State serialization failed | log + skip (non-fatal) |
Review Comment:
nit: The table says a state serialization failure is logged and skipped. `source_template` and `random_source` return `Error::Serialization` from `poll()` and send nothing. Change the row to `Error::Serialization(...)`.
##########
core/connectors/sinks/sink_template/src/lib.rs:
##########
@@ -0,0 +1,1069 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **sink** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, the destination identifier is
+//! validated before it's ever interpolated into a request, outbound calls
+//! go through a retry-wrapped client plus a circuit breaker, and messages
+//! are chunked by a configurable batch size instead of shipped as one
+//! unbounded request.
+//!
+//! Delivery follows the sink's `[delivery]` table. By default the server stores
+//! the offset when it answers the poll and the runtime never redelivers a
+//! batch after an error, so `consume()` keeps trying a batch for up to
+//! `max_delivery_time` before it gives up and logs the lost offset range. With
+//! `guarantee = "at_least_once"` the runtime stores the offset only after
+//! `consume()` succeeds, hands a failed batch over again, and stops the sink on
+//! a permanent error instead of skipping the batch. The circuit breaker only
+//! paces the attempts; it never discards a batch. Writes must be idempotent on
+//! `dedup_key()`, because retries, redelivery and failovers can repeat a
+//! message.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSink::push_batch()` — build the
+//! request/write that actually pushes one chunk of messages to your
+//! destination, using `self.config.connection_string` (and
+//! `self.config.target`, if your destination has a table/index/
+//! collection-shaped name). A second `TODO(ConnectorDeveloper)` on the
+//! `connection_string` field's doc comment just asks you to describe its
+//! expected shape — not code, but worth personalizing too. `grep` for the
+//! marker and you'll find both, plus one more on `build_raw_client()` that
+//! only applies if your destination isn't HTTP (see below).
+//!
+//! This template assumes an HTTP-ish destination and uses `reqwest` wrapped
+//! by the SDK's retry middleware, because that's what
+//! `iggy_connector_sdk::retry` is built for and it covers the common case.
+//! If your destination talks something else (a database, a queue, object
+//! storage), swap the client type in `build_raw_client()`/`push_batch()` for
+//! your driver of choice and lean on its own retry/pooling behavior — keep the
+//! surrounding shape (validation in `open()`, circuit breaker, batching,
+//! identifier validation) unchanged. See `core/connectors/sinks/postgres_sink`
+//! or `core/connectors/sinks/s3_sink` in this repo for non-HTTP examples of
+//! that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+ retry_backoff,
+};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::future::Future;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::time::{Instant, sleep};
+use tracing::{debug, error, info, warn};
+
+sink_connector!(TemplateSink);
+
+const CONNECTOR_NAME: &str = "Template sink";
+
+const REDELIVERY_AT_MOST_ONCE: &str =
+ "The runtime does not redeliver them; replay that range from Iggy to recover them.";
+const REDELIVERY_AT_LEAST_ONCE: &str = "The runtime hands them over again.";
+
+const DEFAULT_BATCH_SIZE: usize = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+// Three failed rounds per cool-down cycle bound the attempts a struggling destination sees:
+// the SDK breaker closes completely after its cool-down, so it takes `threshold` new failures
+// to open again. The cool-down has to be shorter than the delivery budget, or a call that
+// finds the breaker open could never reach a retry.
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 3;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "10s";
+// Kept well under the runtime's close budget (the poll timeout plus 60 s): `consume()` blocks
+// a runtime worker thread and a stop waits for it to return.
+const DEFAULT_MAX_DELIVERY_TIME: &str = "20s";
+const BREAKER_RECHECK_INTERVAL: Duration = Duration::from_millis(250);
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` and `target` is optional with a
+// sane default, and unknown keys are rejected outright so a typo in a TOML
+// file fails at load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSinkConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// The destination table/index/collection/bucket name. Kept as its own
+ /// field (rather than folded into `connection_string`) specifically so
+ /// it can be validated in `open()` before ever being interpolated into
+ /// a query, path, or URL — see `validate_identifier` below. Delete this
+ /// field if your destination has no such dynamic identifier.
+ pub target: String,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<usize>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// How long one `consume()` call may keep trying before it gives up on the
+ /// messages it still holds. Must be longer than `circuit_breaker_cool_down`.
+ pub max_delivery_time: Option<String>,
+
+ /// Upgrades the per-batch log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_sink::PostgresSinkConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+/// Rejects anything that isn't a plain alphanumeric/underscore identifier.
+/// Adjust the allowed character set to whatever your destination's naming
+/// rules actually are, but always validate *something* before a
+/// config-or-message-derived name is interpolated into a query, path, or
+/// URL — see `doris_sink::validate_identifier` / `surrealdb_sink::validate_identifier`
+/// in this repo for the same pattern applied to a real destination.
+fn validate_identifier(field: &str, value: &str) -> Result<(), Error> {
+ if value.is_empty() || !value.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+ return Err(Error::InvalidConfigValue(format!(
+ "{field} must be non-empty and contain only ASCII alphanumeric characters and \
+ underscores, got: {value:?}"
+ )));
+ }
+ Ok(())
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+#[derive(Debug, Default)]
+struct State {
+ invocations_count: u64,
+ messages_written: u64,
+ messages_failed: u64,
+}
+
+#[derive(Debug)]
+pub struct TemplateSink {
+ id: u32,
+ config: TemplateSinkConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size_limit: usize,
+ retry_delay: Duration,
+ retry_max_delay: Duration,
+ cool_down: Duration,
+ max_delivery_time: Duration,
+ at_least_once: bool,
+ verbose: bool,
+ state: Mutex<State>,
+ records_written_total: AtomicU64,
Review Comment:
simplification: The sink increments `records_written_total` at line 429 and never reads it, while `State::messages_written` already tracks the same count. Delete the field, its initializer and the `fetch_add`.
##########
core/connectors/sinks/sink_template/src/lib.rs:
##########
@@ -0,0 +1,1069 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **sink** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, the destination identifier is
+//! validated before it's ever interpolated into a request, outbound calls
+//! go through a retry-wrapped client plus a circuit breaker, and messages
+//! are chunked by a configurable batch size instead of shipped as one
+//! unbounded request.
+//!
+//! Delivery follows the sink's `[delivery]` table. By default the server stores
+//! the offset when it answers the poll and the runtime never redelivers a
+//! batch after an error, so `consume()` keeps trying a batch for up to
+//! `max_delivery_time` before it gives up and logs the lost offset range. With
+//! `guarantee = "at_least_once"` the runtime stores the offset only after
+//! `consume()` succeeds, hands a failed batch over again, and stops the sink on
+//! a permanent error instead of skipping the batch. The circuit breaker only
+//! paces the attempts; it never discards a batch. Writes must be idempotent on
+//! `dedup_key()`, because retries, redelivery and failovers can repeat a
+//! message.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSink::push_batch()` — build the
+//! request/write that actually pushes one chunk of messages to your
+//! destination, using `self.config.connection_string` (and
+//! `self.config.target`, if your destination has a table/index/
+//! collection-shaped name). A second `TODO(ConnectorDeveloper)` on the
+//! `connection_string` field's doc comment just asks you to describe its
+//! expected shape — not code, but worth personalizing too. `grep` for the
+//! marker and you'll find both, plus one more on `build_raw_client()` that
+//! only applies if your destination isn't HTTP (see below).
+//!
+//! This template assumes an HTTP-ish destination and uses `reqwest` wrapped
+//! by the SDK's retry middleware, because that's what
+//! `iggy_connector_sdk::retry` is built for and it covers the common case.
+//! If your destination talks something else (a database, a queue, object
+//! storage), swap the client type in `build_raw_client()`/`push_batch()` for
+//! your driver of choice and lean on its own retry/pooling behavior — keep the
+//! surrounding shape (validation in `open()`, circuit breaker, batching,
+//! identifier validation) unchanged. See `core/connectors/sinks/postgres_sink`
+//! or `core/connectors/sinks/s3_sink` in this repo for non-HTTP examples of
+//! that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+ retry_backoff,
+};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::future::Future;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::time::{Instant, sleep};
+use tracing::{debug, error, info, warn};
+
+sink_connector!(TemplateSink);
+
+const CONNECTOR_NAME: &str = "Template sink";
+
+const REDELIVERY_AT_MOST_ONCE: &str =
+ "The runtime does not redeliver them; replay that range from Iggy to recover them.";
+const REDELIVERY_AT_LEAST_ONCE: &str = "The runtime hands them over again.";
+
+const DEFAULT_BATCH_SIZE: usize = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+// Three failed rounds per cool-down cycle bound the attempts a struggling destination sees:
+// the SDK breaker closes completely after its cool-down, so it takes `threshold` new failures
+// to open again. The cool-down has to be shorter than the delivery budget, or a call that
+// finds the breaker open could never reach a retry.
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 3;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "10s";
+// Kept well under the runtime's close budget (the poll timeout plus 60 s): `consume()` blocks
+// a runtime worker thread and a stop waits for it to return.
+const DEFAULT_MAX_DELIVERY_TIME: &str = "20s";
+const BREAKER_RECHECK_INTERVAL: Duration = Duration::from_millis(250);
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` and `target` is optional with a
+// sane default, and unknown keys are rejected outright so a typo in a TOML
+// file fails at load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSinkConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// The destination table/index/collection/bucket name. Kept as its own
+ /// field (rather than folded into `connection_string`) specifically so
+ /// it can be validated in `open()` before ever being interpolated into
+ /// a query, path, or URL — see `validate_identifier` below. Delete this
+ /// field if your destination has no such dynamic identifier.
+ pub target: String,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<usize>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// How long one `consume()` call may keep trying before it gives up on the
+ /// messages it still holds. Must be longer than `circuit_breaker_cool_down`.
+ pub max_delivery_time: Option<String>,
+
+ /// Upgrades the per-batch log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_sink::PostgresSinkConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+/// Rejects anything that isn't a plain alphanumeric/underscore identifier.
+/// Adjust the allowed character set to whatever your destination's naming
+/// rules actually are, but always validate *something* before a
+/// config-or-message-derived name is interpolated into a query, path, or
+/// URL — see `doris_sink::validate_identifier` / `surrealdb_sink::validate_identifier`
+/// in this repo for the same pattern applied to a real destination.
+fn validate_identifier(field: &str, value: &str) -> Result<(), Error> {
+ if value.is_empty() || !value.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+ return Err(Error::InvalidConfigValue(format!(
+ "{field} must be non-empty and contain only ASCII alphanumeric characters and \
+ underscores, got: {value:?}"
+ )));
+ }
+ Ok(())
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+#[derive(Debug, Default)]
+struct State {
+ invocations_count: u64,
+ messages_written: u64,
+ messages_failed: u64,
+}
+
+#[derive(Debug)]
+pub struct TemplateSink {
+ id: u32,
+ config: TemplateSinkConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size_limit: usize,
+ retry_delay: Duration,
+ retry_max_delay: Duration,
+ cool_down: Duration,
+ max_delivery_time: Duration,
+ at_least_once: bool,
+ verbose: bool,
+ state: Mutex<State>,
+ records_written_total: AtomicU64,
+}
+
+impl TemplateSink {
+ pub fn new(id: u32, config: TemplateSinkConfig) -> Self {
+ let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY);
+ let retry_max_delay =
+ parse_duration(config.retry_max_delay.as_deref(), DEFAULT_RETRY_MAX_DELAY);
+ let cool_down = parse_duration(
+ config.circuit_breaker_cool_down.as_deref(),
+ DEFAULT_CIRCUIT_BREAKER_COOL_DOWN,
+ );
+ let max_delivery_time = parse_duration(
+ config.max_delivery_time.as_deref(),
+ DEFAULT_MAX_DELIVERY_TIME,
+ );
+ let circuit_breaker = Arc::new(CircuitBreaker::new(
+ config
+ .circuit_breaker_threshold
+ .unwrap_or(DEFAULT_CIRCUIT_BREAKER_THRESHOLD),
+ cool_down,
+ ));
+ let batch_size_limit = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ client: None,
+ circuit_breaker,
+ batch_size_limit,
+ retry_delay,
+ retry_max_delay,
+ cool_down,
+ max_delivery_time,
+ at_least_once: false,
+ verbose,
+ state: Mutex::new(State::default()),
+ records_written_total: AtomicU64::new(0),
+ }
+ }
+
+ /// TODO(ConnectorDeveloper): build your actual client/connection here using
+ /// `self.config.connection_string` (and `self.config.auth_token`, if
+ /// your destination needs it). This template builds a plain
+ /// `reqwest::Client` to hand to `build_retry_client` — if you're not
+ /// talking HTTP, replace this with e.g. a database connection pool or
+ /// your driver's equivalent, store it on `self` (add a field, since
+ /// `client` here is HTTP-specific), and skip the
+ /// `check_connectivity_with_retry` call below in favor of whatever your
+ /// driver offers (a ping, a test query).
+ fn build_raw_client(&self) -> Result<reqwest::Client, Error> {
+ let timeout = parse_duration(self.config.timeout.as_deref(), DEFAULT_TIMEOUT);
+ reqwest::Client::builder()
+ .timeout(timeout)
+ .build()
+ .map_err(|e| Error::Connection(format!("failed to build HTTP client: {e}")))
+ }
+
+ /// TODO(ConnectorDeveloper): push one chunk of already-batched messages to your
+ /// destination using `self.config.connection_string` and
+ /// `self.config.target`, via `self.client` (already retry-wrapped).
+ ///
+ /// The error you return decides what `consume()` does next:
+ /// - `Error::HttpRequestFailed`, `Error::Connection` and
+ /// `Error::CannotStoreData` (network error, 5xx, timeout) are transient.
+ /// `consume()` retries the same chunk with backoff until
+ /// `max_delivery_time` runs out, and each failed round counts toward the
+ /// circuit breaker. At least once, the failed call then makes the
+ /// runtime hand the batch over again.
+ /// - `Error::PermanentHttpError` (bad schema, the destination rejects the
+ /// payload shape) will never succeed, and the breaker is left alone,
+ /// because a data problem says nothing about whether the destination is
+ /// up. By default the chunk is skipped and counted. At least once, the
+ /// call fails and the runtime stops the sink at that offset, since a
+ /// skipped chunk would let the stored offset pass an undelivered
+ /// message. See the doc comment on that variant in
+ /// `iggy_connector_sdk::Error`.
+ /// - Any other error is not retried. The chunk is counted as failed and
+ /// the error is returned once every chunk has been tried.
+ ///
+ /// By default the runtime never redelivers a batch after an error, because
+ /// the server stored the offset when it answered the poll. The retries
+ /// above are then the only second chance a batch gets, so from Iggy to this
+ /// plugin delivery is at-most-once. With `[delivery] guarantee =
+ /// "at_least_once"` the runtime stores the offset after `consume()`
+ /// succeeds and redelivers a failed batch. Duplicates come from the
+ /// retries (a request that failed part way through a write), from
+ /// redelivery, from replaying an offset range after a logged loss, and from
+ /// a server failover before the stored offset replicated. 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; // 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"
+ .to_string(),
+ ))
+ }
+
+ /// Pushes `messages` in chunks of `batch_size_limit` through `push_chunk`, which receives
+ /// the index range of one chunk.
+ ///
+ /// All chunks share one delivery budget. A chunk the destination rejects for good is skipped
+ /// and counted, except in at-least-once mode, where skipping would let the runtime store
+ /// an offset past an undelivered message: the call fails and the runtime stops the sink.
+ async fn deliver_messages<F, Fut>(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ push_chunk: F,
+ ) -> Result<(), Error>
+ where
+ F: Fn(std::ops::Range<usize>) -> Fut,
+ Fut: Future<Output = Result<(), Error>>,
+ {
+ // One budget for the whole call. With a dead destination every chunk would otherwise wait
+ // out a budget of its own, and the runtime is blocked on this call for all of them.
+ let deadline = Instant::now() + self.max_delivery_time;
+ let mut last_err: Option<Error> = None;
+ let mut written = 0u64;
+ let mut failed = 0u64;
+
+ for start in (0..messages.len()).step_by(self.batch_size_limit) {
+ let end = (start + self.batch_size_limit).min(messages.len());
+ let chunk = &messages[start..end];
+ let result = self
+ .deliver_chunk(deadline, || push_chunk(start..end))
+ .await;
+ match result {
+ Ok(()) => written += chunk.len() as u64,
+ Err(err @ Error::PermanentHttpError(_)) if self.at_least_once => {
Review Comment:
nit: Both permanent branches match only `Error::PermanentHttpError`. Other permanent variants such as `SchemaMismatch` and `Serialization` fall into the unclassified branch, so later chunks are still written. Match `err.is_permanent()` in both arms.
##########
core/connectors/sources/source_template/src/lib.rs:
##########
@@ -0,0 +1,676 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **source** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, outbound calls go through a
+//! retry-wrapped client plus a circuit breaker, and the read cursor is
+//! staged in `poll()` and only committed in `on_batch_result()` on an ACK so
+//! a dropped/nacked batch can be re-polled instead of silently lost.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSource::fetch_records()` — fetch up
+//! to `batch_size` new records from your external system, starting after
+//! `cursor`. `TemplateSource::build_raw_client()` carries a second
+//! `TODO(ConnectorDeveloper)` too, but only applies if your source isn't
+//! HTTP (see below); a third, on the `connection_string` field's doc
+//! comment, just asks you to describe its expected shape — not code.
+//! `grep` for the marker and you'll find all three.
+//!
+//! This template assumes an HTTP-ish source and uses `reqwest` wrapped by
+//! the SDK's retry middleware, because that's what `iggy_connector_sdk::retry`
+//! is built for and it covers the common case. If your source talks to
+//! something else (a database, a queue, a filesystem), swap the client type
+//! in `build_raw_client()`/`fetch_records()` for your driver of choice and lean on
+//! its own retry/pooling behavior — keep the surrounding shape (validation
+//! in `open()`, circuit breaker, cursor staging, batching) unchanged. See
+//! `core/connectors/sources/postgres_source` in this repo for a real
+//! non-HTTP example of that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+};
+use iggy_connector_sdk::{
+ ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source,
+ source::SourceBatchResult, source_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tracing::{debug, error, info, warn};
+
+source_connector!(TemplateSource);
+
+const CONNECTOR_NAME: &str = "Template source";
+
+const DEFAULT_POLL_INTERVAL: &str = "1s";
+const DEFAULT_BATCH_SIZE: u32 = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 5;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "30s";
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` is optional with a sane default,
+// and unknown keys are rejected outright so a typo in a TOML file fails at
+// load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSourceConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<u32>,
+ pub poll_interval: Option<String>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// Upgrades the per-poll log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_source::PostgresSourceConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+/// Read cursor. Kept as a plain `Option<String>` so it fits whatever ordering
+/// field your source uses (a timestamp, an auto-increment ID, an opaque
+/// pagination token, ...) — stringify it however makes sense for your data.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+struct State {
+ cursor: Option<String>,
+}
+
+/// One record as fetched from the external system, before it's turned into
+/// an Iggy message. Replace the `payload` type with whatever your
+/// `fetch_records()` actually produces.
+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,
+}
+
+#[derive(Debug)]
+pub struct TemplateSource {
+ id: u32,
+ config: TemplateSourceConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size: u32,
+ poll_interval: Duration,
+ retry_delay: Duration,
+ verbose: bool,
+ state: Mutex<State>,
+ pending_state: Mutex<Option<State>>,
+ records_produced: AtomicU64,
+}
+
+impl TemplateSource {
+ pub fn new(id: u32, config: TemplateSourceConfig, state: Option<ConnectorState>) -> Self {
+ let poll_interval = *humantime::Duration::from_str_lossy(
+ config
+ .poll_interval
+ .as_deref()
+ .unwrap_or(DEFAULT_POLL_INTERVAL),
+ );
+ let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY);
+ let circuit_breaker = Arc::new(CircuitBreaker::new(
+ config
+ .circuit_breaker_threshold
+ .unwrap_or(DEFAULT_CIRCUIT_BREAKER_THRESHOLD),
+ parse_duration(
+ config.circuit_breaker_cool_down.as_deref(),
+ DEFAULT_CIRCUIT_BREAKER_COOL_DOWN,
+ ),
+ ));
+ let batch_size = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ let restored_state = state
+ .and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+ .inspect(|s| {
+ info!(
+ "Restored state for {CONNECTOR_NAME} connector with ID: {id}. Cursor: {:?}",
+ s.cursor
+ );
+ });
+
+ Self {
+ id,
+ config,
+ client: None,
+ circuit_breaker,
+ batch_size,
+ poll_interval,
+ retry_delay,
+ verbose,
+ state: Mutex::new(restored_state.unwrap_or_default()),
+ pending_state: Mutex::new(None),
+ records_produced: AtomicU64::new(0),
+ }
+ }
+
+ fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+ ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+ }
+
+ /// TODO(ConnectorDeveloper): build your actual client/connection here using
+ /// `self.config.connection_string` (and `self.config.auth_token`, if
+ /// your system needs it). This template builds a plain `reqwest::Client`
+ /// to hand to `build_retry_client` — if you're not talking HTTP, replace
+ /// this with e.g. a `sqlx::PgPool::connect(...)` or your driver's
+ /// equivalent, store it on `self` (add a field, since `client` here is
+ /// HTTP-specific), and skip the `check_connectivity_with_retry` call
+ /// below in favor of whatever your driver offers (a ping, a test query).
+ fn build_raw_client(&self) -> Result<reqwest::Client, Error> {
+ let timeout = parse_duration(self.config.timeout.as_deref(), DEFAULT_TIMEOUT);
+ reqwest::Client::builder()
+ .timeout(timeout)
+ .build()
+ .map_err(|e| Error::Connection(format!("failed to build HTTP client: {e}")))
+ }
+
+ /// TODO(ConnectorDeveloper): fetch up to `self.batch_size` new records from your
+ /// external system, ordered after `cursor` (`None` means "from the
+ /// beginning" or "from now" — whichever is right for your source).
+ /// Use `self.config.connection_string` as the base address and
+ /// `self.client` (already retry-wrapped) to make the request. Map each
+ /// 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, 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,
+ cursor: Option<&str>,
+ ) -> Result<Vec<FetchedRecord>, Error> {
+ let _ = (client, cursor); // remove once implemented
+ Err(Error::InitError(
+ "TemplateSource::fetch_records is not implemented yet — see the TODO(ConnectorDeveloper) comment \
+ in template_source/src/lib.rs"
+ .to_string(),
+ ))
+ }
+}
+
+// ── Source trait ────────────────────────────────────────────────────────────
+
+#[async_trait]
+impl Source for TemplateSource {
+ async fn open(&mut self) -> Result<(), Error> {
+ // Structural validation happens here, not in `new()`, because only
+ // `open()` can return an error — `new()` is a plain factory function
+ // with nowhere to send a "this config is invalid" result.
+ if self
+ .config
+ .connection_string
+ .expose_secret()
+ .trim()
+ .is_empty()
+ {
+ return Err(Error::InvalidConfigValue(
+ "connection_string must not be empty".to_string(),
+ ));
+ }
+
+ info!(
+ "Opening {CONNECTOR_NAME} connector with ID: {}, batch_size: {}, poll_interval: {:?}",
+ self.id, self.batch_size, self.poll_interval
+ );
+
+ let raw_client = self.build_raw_client()?;
+
+ if let Some(health_path) = &self.config.health_check_path {
+ let base = Url::parse(self.config.connection_string.expose_secret()).map_err(|e| {
+ Error::InvalidConfigValue(format!("connection_string is not a valid URL: {e}"))
+ })?;
+ let health_url = base.join(health_path).map_err(|e| {
+ Error::InvalidConfigValue(format!("invalid health_check_path: {e}"))
+ })?;
+ check_connectivity_with_retry(
+ &raw_client,
+ health_url,
+ CONNECTOR_NAME,
+ self.id,
+ RetryPolicy {
+ max_attempts: self
+ .config
+ .max_open_retries
+ .unwrap_or(DEFAULT_MAX_OPEN_RETRIES),
+ base_delay: self.retry_delay,
+ max_delay: parse_duration(
+ self.config.open_retry_max_delay.as_deref(),
+ DEFAULT_OPEN_RETRY_MAX_DELAY,
+ ),
+ },
+ )
+ .await?;
+ } else {
+ warn!(
+ "{CONNECTOR_NAME} connector with ID: {} has no health_check_path configured — \
+ skipping the startup connectivity probe. Consider adding one.",
+ self.id
+ );
+ }
+
+ self.client = Some(build_retry_client(
+ raw_client,
+ self.config
+ .max_retries
+ .unwrap_or(DEFAULT_MAX_RETRIES)
+ .max(1),
+ self.retry_delay,
+ parse_duration(
+ self.config.retry_max_delay.as_deref(),
+ DEFAULT_RETRY_MAX_DELAY,
+ ),
+ CONNECTOR_NAME,
+ ));
+
+ info!(
+ "{CONNECTOR_NAME} connector with ID: {} opened successfully",
+ self.id
+ );
+ Ok(())
+ }
+
+ async fn poll(&self) -> Result<ProducedMessages, Error> {
+ // If the breaker is open, sleep for the normal poll interval and
+ // return an empty (not an error) result. Returning `Err` here would
+ // make the runtime retry `poll()` again immediately with no delay —
+ // see `handle_messages` in the SDK's source container — so an empty
+ // ACK-free result is how a source waits out a known-bad window
+ // without busy-looping or counting against the NACK budget.
+ if self.circuit_breaker.is_open().await {
+ warn!(
+ "{CONNECTOR_NAME} connector with ID: {} — circuit breaker OPEN, skipping poll",
+ self.id
+ );
+ tokio::time::sleep(self.poll_interval).await;
+ return Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages: Vec::new(),
+ state: None,
+ });
+ }
+ tokio::time::sleep(self.poll_interval).await;
+
+ let client = self.client.as_ref().ok_or_else(|| {
+ Error::Connection("client not initialized -- was open() called?".into())
+ })?;
+ let cursor = self.state.lock().await.cursor.clone();
+
+ let records = match self.fetch_records(client, cursor.as_deref()).await {
+ Ok(records) => {
+ self.circuit_breaker.record_success();
+ records
+ }
+ Err(err) => {
+ if !matches!(err, Error::PermanentHttpError(_)) {
+ self.circuit_breaker.record_failure().await;
+ }
+ return Err(err);
+ }
+ };
+
+ if records.is_empty() {
+ return Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages: Vec::new(),
+ state: None,
+ });
+ }
+
+ let mut messages = Vec::with_capacity(records.len());
+ let mut new_cursor = cursor;
+ for record in records {
+ let Ok(payload) = serde_json::to_vec(&record.payload) else {
+ error!(
+ "Failed to serialize a record fetched by {CONNECTOR_NAME} connector with ID: {}",
+ self.id
+ );
+ continue;
+ };
+ // Only advance the candidate cursor once the record actually made it
+ // 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: natural_id,
+ headers: None,
+ checksum: None,
+ timestamp: None,
+ origin_timestamp: None,
+ payload,
+ });
+ }
+
+ if messages.is_empty() {
+ // Every fetched record failed to serialize - no progress was
+ // made, so there's nothing to stage and no reason to write
+ // state, same as the `records.is_empty()` case above.
+ return Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages: Vec::new(),
+ state: None,
+ });
+ }
+
+ let candidate_state = State { cursor: new_cursor };
+ let persisted_state = self.serialize_state(&candidate_state).ok_or_else(|| {
+ Error::Serialization(format!(
+ "failed to serialize state for {CONNECTOR_NAME} connector with ID: {}",
+ self.id
+ ))
+ })?;
+ *self.pending_state.lock().await = Some(candidate_state);
+
+ self.records_produced
+ .fetch_add(messages.len() as u64, Ordering::Relaxed);
+
+ if self.verbose {
+ info!(
+ "{CONNECTOR_NAME} connector with ID: {} produced {} messages",
+ self.id,
+ messages.len()
+ );
+ } else {
+ debug!(
+ "{CONNECTOR_NAME} connector with ID: {} produced {} messages",
+ self.id,
+ messages.len()
+ );
+ }
+
+ Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages,
+ state: Some(persisted_state),
+ })
+ }
+
+ /// The staged cursor from `poll()` is only committed here, and only on
+ /// an ACK. A NACK (delivery failed, state-save failed, runtime is
+ /// shutting down) discards the candidate so the same range is re-polled
+ /// next time — the cursor never moves past data that wasn't confirmed
+ /// 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`.
+ /// A missing runtime result does not produce a synthetic NACK — the SDK
+ /// just keeps waiting (warning past ~30s) rather than guessing.
+ ///
+ /// 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. `on_batch_result` has no SDK-enforced deadline, but the SDK will
+ /// not call `poll()` again until it returns — slow I/O here silently
+ /// stalls the whole source, it doesn't fail loudly. If you stage more
+ /// than one operation, budget for *all* of them (a per-item retry
+ /// budget multiplies with the item count), and reject
+ /// `max_retries`/`retry_delay` combinations in `open()` that would
+ /// make that stall unbounded. 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. A source whose input genuinely
+ /// cannot be replayed after restart should instead override
+ /// `batch_policy()` with `BatchPolicy::with_max_consecutive_nacks(None)`
+ /// rather than relying on the default 5-consecutive-Nack stop.
+ async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(), Error> {
+ let candidate_state = self.pending_state.lock().await.take();
+ if result == SourceBatchResult::Ack
+ && let Some(candidate_state) = candidate_state
+ {
+ *self.state.lock().await = candidate_state;
+ }
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ let state = self.state.lock().await;
+ info!(
+ "{CONNECTOR_NAME} connector with ID: {} closed. Cursor: {:?}, total records produced: {}",
+ self.id,
+ state.cursor,
+ self.records_produced.load(Ordering::Relaxed)
+ );
+ drop(state);
+ self.client = None;
+ Ok(())
+ }
+}
+
+// Small local shim so `new()` doesn't need to pull in `humantime::Duration`'s
+// `FromStr` (which returns `Result`) just to apply a default — mirrors the
+// fallback-with-warning behavior of `iggy_connector_sdk::retry::parse_duration`
+// for the one duration field (`poll_interval`) that isn't itself optional in
+// spirit (there's always a poll interval, just maybe the default one).
+trait DurationExt {
Review Comment:
simplification: `DurationExt` wraps one `humantime` parse for a single call site. The crate already imports `parse_duration`. Replace the shim with `parse_duration(config.poll_interval.as_deref(), DEFAULT_POLL_INTERVAL)` and drop the `humantime` dependency.
##########
core/connectors/sinks/sink_template/src/lib.rs:
##########
@@ -0,0 +1,1069 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **sink** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, the destination identifier is
+//! validated before it's ever interpolated into a request, outbound calls
+//! go through a retry-wrapped client plus a circuit breaker, and messages
+//! are chunked by a configurable batch size instead of shipped as one
+//! unbounded request.
+//!
+//! Delivery follows the sink's `[delivery]` table. By default the server stores
+//! the offset when it answers the poll and the runtime never redelivers a
+//! batch after an error, so `consume()` keeps trying a batch for up to
+//! `max_delivery_time` before it gives up and logs the lost offset range. With
+//! `guarantee = "at_least_once"` the runtime stores the offset only after
+//! `consume()` succeeds, hands a failed batch over again, and stops the sink on
+//! a permanent error instead of skipping the batch. The circuit breaker only
+//! paces the attempts; it never discards a batch. Writes must be idempotent on
+//! `dedup_key()`, because retries, redelivery and failovers can repeat a
+//! message.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSink::push_batch()` — build the
+//! request/write that actually pushes one chunk of messages to your
+//! destination, using `self.config.connection_string` (and
+//! `self.config.target`, if your destination has a table/index/
+//! collection-shaped name). A second `TODO(ConnectorDeveloper)` on the
+//! `connection_string` field's doc comment just asks you to describe its
+//! expected shape — not code, but worth personalizing too. `grep` for the
+//! marker and you'll find both, plus one more on `build_raw_client()` that
+//! only applies if your destination isn't HTTP (see below).
+//!
+//! This template assumes an HTTP-ish destination and uses `reqwest` wrapped
+//! by the SDK's retry middleware, because that's what
+//! `iggy_connector_sdk::retry` is built for and it covers the common case.
+//! If your destination talks something else (a database, a queue, object
+//! storage), swap the client type in `build_raw_client()`/`push_batch()` for
+//! your driver of choice and lean on its own retry/pooling behavior — keep the
+//! surrounding shape (validation in `open()`, circuit breaker, batching,
+//! identifier validation) unchanged. See `core/connectors/sinks/postgres_sink`
+//! or `core/connectors/sinks/s3_sink` in this repo for non-HTTP examples of
+//! that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+ retry_backoff,
+};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::future::Future;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::time::{Instant, sleep};
+use tracing::{debug, error, info, warn};
+
+sink_connector!(TemplateSink);
+
+const CONNECTOR_NAME: &str = "Template sink";
+
+const REDELIVERY_AT_MOST_ONCE: &str =
+ "The runtime does not redeliver them; replay that range from Iggy to recover them.";
+const REDELIVERY_AT_LEAST_ONCE: &str = "The runtime hands them over again.";
+
+const DEFAULT_BATCH_SIZE: usize = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+// Three failed rounds per cool-down cycle bound the attempts a struggling destination sees:
+// the SDK breaker closes completely after its cool-down, so it takes `threshold` new failures
+// to open again. The cool-down has to be shorter than the delivery budget, or a call that
+// finds the breaker open could never reach a retry.
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 3;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "10s";
+// Kept well under the runtime's close budget (the poll timeout plus 60 s): `consume()` blocks
+// a runtime worker thread and a stop waits for it to return.
+const DEFAULT_MAX_DELIVERY_TIME: &str = "20s";
+const BREAKER_RECHECK_INTERVAL: Duration = Duration::from_millis(250);
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` and `target` is optional with a
+// sane default, and unknown keys are rejected outright so a typo in a TOML
+// file fails at load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSinkConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// The destination table/index/collection/bucket name. Kept as its own
+ /// field (rather than folded into `connection_string`) specifically so
+ /// it can be validated in `open()` before ever being interpolated into
+ /// a query, path, or URL — see `validate_identifier` below. Delete this
+ /// field if your destination has no such dynamic identifier.
+ pub target: String,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<usize>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// How long one `consume()` call may keep trying before it gives up on the
+ /// messages it still holds. Must be longer than `circuit_breaker_cool_down`.
+ pub max_delivery_time: Option<String>,
+
+ /// Upgrades the per-batch log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_sink::PostgresSinkConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+/// Rejects anything that isn't a plain alphanumeric/underscore identifier.
+/// Adjust the allowed character set to whatever your destination's naming
+/// rules actually are, but always validate *something* before a
+/// config-or-message-derived name is interpolated into a query, path, or
+/// URL — see `doris_sink::validate_identifier` / `surrealdb_sink::validate_identifier`
+/// in this repo for the same pattern applied to a real destination.
+fn validate_identifier(field: &str, value: &str) -> Result<(), Error> {
+ if value.is_empty() || !value.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+ return Err(Error::InvalidConfigValue(format!(
+ "{field} must be non-empty and contain only ASCII alphanumeric characters and \
+ underscores, got: {value:?}"
+ )));
+ }
+ Ok(())
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+#[derive(Debug, Default)]
+struct State {
+ invocations_count: u64,
+ messages_written: u64,
+ messages_failed: u64,
+}
+
+#[derive(Debug)]
+pub struct TemplateSink {
+ id: u32,
+ config: TemplateSinkConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size_limit: usize,
+ retry_delay: Duration,
+ retry_max_delay: Duration,
+ cool_down: Duration,
+ max_delivery_time: Duration,
+ at_least_once: bool,
+ verbose: bool,
+ state: Mutex<State>,
+ records_written_total: AtomicU64,
+}
+
+impl TemplateSink {
+ pub fn new(id: u32, config: TemplateSinkConfig) -> Self {
+ let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY);
+ let retry_max_delay =
+ parse_duration(config.retry_max_delay.as_deref(), DEFAULT_RETRY_MAX_DELAY);
+ let cool_down = parse_duration(
+ config.circuit_breaker_cool_down.as_deref(),
+ DEFAULT_CIRCUIT_BREAKER_COOL_DOWN,
+ );
+ let max_delivery_time = parse_duration(
+ config.max_delivery_time.as_deref(),
+ DEFAULT_MAX_DELIVERY_TIME,
+ );
+ let circuit_breaker = Arc::new(CircuitBreaker::new(
+ config
+ .circuit_breaker_threshold
+ .unwrap_or(DEFAULT_CIRCUIT_BREAKER_THRESHOLD),
+ cool_down,
+ ));
+ let batch_size_limit = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ client: None,
+ circuit_breaker,
+ batch_size_limit,
+ retry_delay,
+ retry_max_delay,
+ cool_down,
+ max_delivery_time,
+ at_least_once: false,
+ verbose,
+ state: Mutex::new(State::default()),
+ records_written_total: AtomicU64::new(0),
+ }
+ }
+
+ /// TODO(ConnectorDeveloper): build your actual client/connection here using
+ /// `self.config.connection_string` (and `self.config.auth_token`, if
+ /// your destination needs it). This template builds a plain
+ /// `reqwest::Client` to hand to `build_retry_client` — if you're not
+ /// talking HTTP, replace this with e.g. a database connection pool or
+ /// your driver's equivalent, store it on `self` (add a field, since
+ /// `client` here is HTTP-specific), and skip the
+ /// `check_connectivity_with_retry` call below in favor of whatever your
+ /// driver offers (a ping, a test query).
+ fn build_raw_client(&self) -> Result<reqwest::Client, Error> {
+ let timeout = parse_duration(self.config.timeout.as_deref(), DEFAULT_TIMEOUT);
+ reqwest::Client::builder()
+ .timeout(timeout)
+ .build()
+ .map_err(|e| Error::Connection(format!("failed to build HTTP client: {e}")))
+ }
+
+ /// TODO(ConnectorDeveloper): push one chunk of already-batched messages to your
+ /// destination using `self.config.connection_string` and
+ /// `self.config.target`, via `self.client` (already retry-wrapped).
+ ///
+ /// The error you return decides what `consume()` does next:
+ /// - `Error::HttpRequestFailed`, `Error::Connection` and
+ /// `Error::CannotStoreData` (network error, 5xx, timeout) are transient.
+ /// `consume()` retries the same chunk with backoff until
+ /// `max_delivery_time` runs out, and each failed round counts toward the
+ /// circuit breaker. At least once, the failed call then makes the
+ /// runtime hand the batch over again.
+ /// - `Error::PermanentHttpError` (bad schema, the destination rejects the
+ /// payload shape) will never succeed, and the breaker is left alone,
+ /// because a data problem says nothing about whether the destination is
+ /// up. By default the chunk is skipped and counted. At least once, the
+ /// call fails and the runtime stops the sink at that offset, since a
+ /// skipped chunk would let the stored offset pass an undelivered
+ /// message. See the doc comment on that variant in
+ /// `iggy_connector_sdk::Error`.
+ /// - Any other error is not retried. The chunk is counted as failed and
+ /// the error is returned once every chunk has been tried.
+ ///
+ /// By default the runtime never redelivers a batch after an error, because
+ /// the server stored the offset when it answered the poll. The retries
+ /// above are then the only second chance a batch gets, so from Iggy to this
+ /// plugin delivery is at-most-once. With `[delivery] guarantee =
+ /// "at_least_once"` the runtime stores the offset after `consume()`
+ /// succeeds and redelivers a failed batch. Duplicates come from the
+ /// retries (a request that failed part way through a write), from
+ /// redelivery, from replaying an offset range after a logged loss, and from
+ /// a server failover before the stored offset replicated. 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
Review Comment:
warning: The `push_batch()` guidance fails only when zero rows of a non-empty chunk are accepted. `deliver_messages` then counts a partially rejected chunk as fully written, so at-least-once stores the offset past the rejected rows. Require a failed call whenever a non-empty chunk loses any row in at-least-once mode.
##########
.claude/skills/connector-source/TEMPLATE.md:
##########
@@ -1,64 +1,134 @@
-# Source plugin skeleton
+# Source plugin fill-in-the-blank kit
-Boilerplate for a new `core/connectors/sources/<name>_source/`. Adapt
-the `MySource` / `Client` / row types to the backend driver.
+Copy this kit into `core/connectors/sources/<name>_source/`. The
+scaffolding covers config, secrets, sleep-first poll, lock discipline,
+staged/committed state serde (ACK/NACK batch acknowledgment, #3855),
+retry classification, logging, and the six canonical state tests.
+**You only implement the marked `TODO(ConnectorDeveloper)` sections:** build a
+client from the connection string, and fetch the next batch (advancing
+a cursor).
-`Cargo.toml` is identical to a sink's (see
-[connector-sink/TEMPLATE.md](../connector-sink/TEMPLATE.md)) - only
-the crate name suffix changes (`iggy_connector_<name>_source`) and the
-upstream client dep.
+Also read [SKILL.md](SKILL.md) and pre-flight with
+[connector-pr-review](../connector-pr-review/SKILL.md) before `/ready`.
+
+Prefer starting from a compiling crate over copying this prose kit:
+`core/connectors/sources/source_template/` implements the same shape as
+real, tested code you can `cargo build`/`cargo test` immediately, with the
+same `TODO(ConnectorDeveloper)` markers. Use this kit instead only when
+copying a whole crate is more scaffolding than you need.
+
+## Files to create
+
+```text
+core/connectors/sources/<name>_source/
+├── Cargo.toml
+├── README.md
+├── config.toml
+└── src/lib.rs
+```
+
+Add a workspace member, a row in `sources/README.md`, and a sample under
+`runtime/example_config/connectors/`.
+
+---
+
+## Cargo.toml
+
+Same shape as the sink kit (`cdylib` + `lib`, workspace deps, Apache
+header). Only the package name suffix changes:
+
+```toml
+name = "iggy_connector_<name>_source"
+# ... identical metadata / machete ignored / crate-type ...
+# TODO(ConnectorDeveloper): add your client crate as a workspace dependency
+```
+
+---
+
+## config.toml (example)
+
+```toml
+path = "../../target/release/libiggy_connector_<name>_source"
+
+[[sources]]
Review Comment:
warning: The sample uses `[[sources]]` and `[sources.<name>.plugin_config]`, but the runtime loads one `type = "source"` connector per file with a flat `[plugin_config]` table. The sample does not parse. Replace it with the format in `core/connectors/runtime/example_config/connectors/source_template.toml`.
##########
.claude/skills/connector-sink/TEMPLATE.md:
##########
@@ -1,129 +1,305 @@
-# Sink plugin skeleton
+# Sink plugin fill-in-the-blank kit
-Boilerplate for a new `core/connectors/sinks/<name>_sink/`. Adapt the
-`MySink` / `Client` types to the backend driver you're integrating.
+Copy this kit into `core/connectors/sinks/<name>_sink/`. The scaffolding
+covers config, secrets, retry, error classification, batching, logging,
+and unit-test shape. **You only implement the marked `TODO(ConnectorDeveloper)`
+sections:** build a client from the connection string, and push one
+batch.
+
+Also read [SKILL.md](SKILL.md) and pre-flight with
+[connector-pr-review](../connector-pr-review/SKILL.md) before `/ready`.
+
+Prefer starting from a compiling crate over copying this prose kit:
+`core/connectors/sinks/sink_template/` implements the same shape as real,
+tested code you can `cargo build`/`cargo test` immediately, with the same
+`TODO(ConnectorDeveloper)` markers. It also has the circuit breaker, which this
+kit leaves out to stay short. Use this kit instead only when copying a whole
+crate is more scaffolding than you need.
+
+## The delivery contract this kit is built on
+
+By default the server stores a sink's consumer offset when it answers the poll,
+before `consume()` runs, and the runtime never redelivers a batch after
+`consume()` returns an error. So delivery from Iggy is at-most-once, and the
+retrying the plugin does inside `consume()` is the only second chance a batch
+gets. The skeleton below therefore keeps trying a batch within one time budget
+(`max_delivery_time`), logs the offset range it gives up on, and leaves
+idempotent writes to you.
+
+The skeleton skips a chunk the backend rejects for good, so it does not
+override `enable_at_least_once()`, and a sink configured with
+`[delivery] guarantee = "at_least_once"` refuses to start. To support that
+mode, copy `deliver_messages()` and `enable_at_least_once()` from
+`sinks/sink_template`, which fail the call instead of skipping.
+
+## Files to create
+
+```text
+core/connectors/sinks/<name>_sink/
+├── Cargo.toml
+├── README.md
+├── config.toml
+└── src/lib.rs
+```
+
+Add a workspace member, a row in `sinks/README.md`, and a sample under
+`runtime/example_config/connectors/`.
+
+---
## Cargo.toml
```toml
-# Apache 2.0 header (copy verbatim from any existing sink Cargo.toml)
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
[package]
name = "iggy_connector_<name>_sink"
-version = "0.5.1-edge.1" # match the version most other sinks use
+version = "0.5.0" # match the version most other sinks use
edition = "2024"
license = "Apache-2.0"
publish = false
-# ...keywords, description, repository, homepage all identical to existing sinks
+description = "Apache Iggy <Name> sink connector"
+repository = "https://github.com/apache/iggy"
+homepage = "https://iggy.apache.org"
[package.metadata.cargo-machete]
-ignored = ["dashmap", "once_cell"] # used by sink_connector! macro
+ignored = ["dashmap", "once_cell"]
[lib]
-crate-type = ["cdylib", "lib"] # cdylib = runtime-loadable; lib = unit tests
+crate-type = ["cdylib", "lib"]
[dependencies]
-async-trait = { workspace = true }
-dashmap = { workspace = true }
+async-trait = { workspace = true }
+dashmap = { workspace = true }
+iggy_common = { workspace = true }
iggy_connector_sdk = { workspace = true }
-once_cell = { workspace = true }
-serde = { workspace = true }
-tokio = { workspace = true }
-tracing = { workspace = true }
-# + your client crate (reqwest, sqlx, mongodb, ...)
+once_cell = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+# TODO(ConnectorDeveloper): add your client crate as a workspace dependency
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["test-util"] } # for paused-time delivery tests
```
-Run `cargo sort --no-format --workspace` after edits.
+Run `cargo sort --no-format --workspace` after edits. Keep `Cargo.lock`
+churn limited to this crate's deps.
+
+---
+
+## config.toml (example)
+
+A sink config has a connector header, one or more `[[streams]]` entries that
+tell the runtime what to consume and how to batch it, and a `[plugin_config]`
+table for your own settings. `sinks/sink_template/config.toml` is the working
+reference.
+
+```toml
+type = "sink"
+key = "<name>"
+enabled = true
+version = 0
+name = "<Name> sink"
+path = "../../target/release/libiggy_connector_<name>_sink"
+plugin_config_format = "toml"
+verbose = false
+
+[[streams]]
+stream = "example_stream"
+topics = ["example_topic"]
+schema = "json"
+batch_length = 100
+# Long poll: a poll returns once `min_count` messages are ready or `max_wait`
+# expires. These replace `poll_interval`, which the runtime refuses for sinks.
+# `min_count` must not exceed `batch_length`; `max_wait` must be above zero.
+max_wait = "2s"
+min_count = 100
+consumer_group = "<name>_sink_connector"
+
+[plugin_config]
+# Never commit real secrets. Use env overrides in tests/ops.
+connection_string = "scheme://user:pass@host:port/db"
+batch_size = 100
+retry_delay = "500ms"
+retry_max_delay = "5s"
+# One budget for a whole consume() call. Keep it well under a minute: the call
+# blocks a runtime worker thread, and a stop waits for it.
+max_delivery_time = "20s"
+verbose_logging = false
+```
+
+---
## src/lib.rs
-Code reads top to bottom. Public types first, then `impl`, then private
-helpers.
+Replace `<Name>` / `<name>` and implement only the `TODO(ConnectorDeveloper)` blocks.
```rust
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements... */ // full Apache header
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
use async_trait::async_trait;
+use iggy_connector_sdk::retry::{parse_duration, retry_backoff};
use iggy_connector_sdk::{
ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
};
+use secrecy::{ExposeSecret, SecretString};
use serde::{Deserialize, Serialize};
-use std::str::FromStr;
use std::time::Duration;
use tokio::sync::Mutex;
+use tokio::time::{Instant, sleep};
use tracing::{debug, error, info, warn};
-sink_connector!(MySink); // generates FFI symbols + version export
+sink_connector!(NameSink);
-const CONNECTOR_NAME: &str = "My sink";
+const CONNECTOR_NAME: &str = "Name sink";
+const DEFAULT_BATCH_SIZE: u32 = 100;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+// Well under the runtime's close budget (the poll timeout plus 60 s): consume() blocks a
+// runtime worker thread and a stop waits for it.
+const DEFAULT_MAX_DELIVERY_TIME: &str = "20s";
+
+/// Backend client. Replace with the real driver type.
+struct BackendClient {
+ // TODO(ConnectorDeveloper): fields
+}
#[derive(Debug, Serialize, Deserialize)]
-pub struct MySinkConfig {
- pub endpoint: String,
+#[serde(deny_unknown_fields)]
+pub struct NameSinkConfig {
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
pub batch_size: Option<u32>,
- pub max_retries: Option<u32>,
- pub retry_delay: Option<String>, // humantime, e.g. "500ms"
- pub verbose_logging: Option<bool>, // mirror runtime's `verbose` flag
- // Every runtime-tunable field is Option<T>; defaults applied in new()
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_delivery_time: Option<String>,
+ pub verbose_logging: Option<bool>,
+ // TODO(ConnectorDeveloper): optional non-secret knobs (table, index, batch_mode, ...)
}
#[derive(Debug)]
-pub struct MySink {
+pub struct NameSink {
id: u32,
- config: MySinkConfig,
+ config: NameSinkConfig,
batch_size: usize,
- max_retries: u32,
retry_delay: Duration,
+ retry_max_delay: Duration,
+ max_delivery_time: Duration,
verbose: bool,
- client: Option<Client>,
+ client: Option<BackendClient>,
state: Mutex<State>,
}
-#[derive(Debug)]
+#[derive(Debug, Default)]
struct State {
messages_processed: u64,
errors: u64,
}
-impl MySink {
- pub fn new(id: u32, config: MySinkConfig) -> Self {
- let batch_size = config.batch_size.unwrap_or(100) as usize;
- let max_retries = config.max_retries.unwrap_or(3);
- let retry_delay = config
- .retry_delay
- .as_deref()
- .and_then(|raw| humantime::Duration::from_str(raw).ok().map(|d| *d))
- .unwrap_or_else(|| {
- warn!("Invalid retry_delay for {CONNECTOR_NAME} ID: {id}, defaulting to 500ms");
- Duration::from_millis(500)
- });
+impl NameSink {
+ pub fn new(id: u32, config: NameSinkConfig) -> Self {
+ let batch_size = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1) as usize;
+ // `parse_duration` falls back to the default with a `warn!` on a bad value.
Review Comment:
warning: `parse_duration()` falls back to one second on a parse error, not to the default it was given. The comment is wrong, and `given_invalid_retry_delay_should_fall_back_to_default` at line 490 expects `500ms` and fails. Correct the comment and the expected value, or parse with `humantime` and fall back to `DEFAULT_RETRY_DELAY`.
##########
.claude/skills/connector-source/TEMPLATE.md:
##########
@@ -1,64 +1,134 @@
-# Source plugin skeleton
+# Source plugin fill-in-the-blank kit
-Boilerplate for a new `core/connectors/sources/<name>_source/`. Adapt
-the `MySource` / `Client` / row types to the backend driver.
+Copy this kit into `core/connectors/sources/<name>_source/`. The
+scaffolding covers config, secrets, sleep-first poll, lock discipline,
+staged/committed state serde (ACK/NACK batch acknowledgment, #3855),
+retry classification, logging, and the six canonical state tests.
+**You only implement the marked `TODO(ConnectorDeveloper)` sections:** build a
+client from the connection string, and fetch the next batch (advancing
+a cursor).
-`Cargo.toml` is identical to a sink's (see
-[connector-sink/TEMPLATE.md](../connector-sink/TEMPLATE.md)) - only
-the crate name suffix changes (`iggy_connector_<name>_source`) and the
-upstream client dep.
+Also read [SKILL.md](SKILL.md) and pre-flight with
+[connector-pr-review](../connector-pr-review/SKILL.md) before `/ready`.
+
+Prefer starting from a compiling crate over copying this prose kit:
+`core/connectors/sources/source_template/` implements the same shape as
+real, tested code you can `cargo build`/`cargo test` immediately, with the
+same `TODO(ConnectorDeveloper)` markers. Use this kit instead only when
+copying a whole crate is more scaffolding than you need.
+
+## Files to create
+
+```text
+core/connectors/sources/<name>_source/
+├── Cargo.toml
+├── README.md
+├── config.toml
+└── src/lib.rs
+```
+
+Add a workspace member, a row in `sources/README.md`, and a sample under
+`runtime/example_config/connectors/`.
+
+---
+
+## Cargo.toml
+
+Same shape as the sink kit (`cdylib` + `lib`, workspace deps, Apache
+header). Only the package name suffix changes:
+
+```toml
+name = "iggy_connector_<name>_source"
+# ... identical metadata / machete ignored / crate-type ...
+# TODO(ConnectorDeveloper): add your client crate as a workspace dependency
+```
+
+---
+
+## config.toml (example)
+
+```toml
+path = "../../target/release/libiggy_connector_<name>_source"
+
+[[sources]]
+key = "<name>"
+enabled = true
+
+[sources.<name>.plugin_config]
+connection_string = "scheme://user:pass@host:port/db"
+poll_interval = "5s"
+batch_size = 100
+max_retries = 3
+retry_delay = "500ms"
+verbose_logging = false
+```
+
+Defaults in this file must match `DEFAULT_*` consts in code.
+
+---
## src/lib.rs
-Code reads top to bottom. Public types first, `impl Source` after,
-helpers below.
+Replace `<Name>` / `<name>` and implement only the `TODO(ConnectorDeveloper)` blocks.
```rust
/* Apache 2.0 header */
-use std::str::FromStr;
-use std::time::Duration;
-
use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
use iggy_connector_sdk::{
ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source,
source::SourceBatchResult, source_connector,
};
+use secrecy::{ExposeSecret, SecretString};
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;
use tokio::time::sleep;
use tracing::{debug, error, info, warn};
-source_connector!(MySource);
+source_connector!(NameSource);
-const CONNECTOR_NAME: &str = "My source";
+const CONNECTOR_NAME: &str = "Name source";
+const DEFAULT_POLL_INTERVAL: &str = "5s";
+const DEFAULT_BATCH_SIZE: u32 = 100;
+const DEFAULT_MAX_RETRIES: u32 = 3; // total attempts
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+
+struct BackendClient {
+ // TODO(ConnectorDeveloper): fields
+}
#[derive(Debug, Serialize, Deserialize)]
-pub struct MySourceConfig {
- pub endpoint: String,
- pub poll_interval: Option<String>, // humantime, "10s" default
+#[serde(deny_unknown_fields)]
+pub struct NameSourceConfig {
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+ pub poll_interval: Option<String>,
pub batch_size: Option<u32>,
pub max_retries: Option<u32>,
pub retry_delay: Option<String>,
pub verbose_logging: Option<bool>,
+ // TODO(ConnectorDeveloper): optional non-secret knobs (query, table, index, ...)
}
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize)]
struct State {
- cursor: Option<String>, // WAL LSN, scroll id, timestamp, ...
- last_offset: u64,
+ /// Opaque backend cursor (LSN, scroll id, timestamp, PK, ...). Keep O(1).
+ cursor: Option<String>,
messages_produced: u64,
}
#[derive(Debug)]
-pub struct MySource {
+pub struct NameSource {
Review Comment:
warning: The struct is now `NameSource`, but the inherent impl below still targets `MySource`, so a copied crate does not compile. Rename the impl, and restore the `Duration` and `FromStr` imports and the `Clone` derive that the kit dropped.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #4636:
URL: https://github.com/apache/iggy/pull/4636#issuecomment-6385880961
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/4636?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 84.89%. Comparing base ([`717719c`](https://app.codecov.io/gh/apache/iggy/commit/717719c8e60fedfa3334fd243d646077d91f6298?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`e349759`](https://app.codecov.io/gh/apache/iggy/commit/e3497595e816991f4b1c97420cee9e4e9a2b21b8?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 25 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4636 +/- ##
============================================
- Coverage 89.78% 84.89% -4.89%
Complexity 2146 2146
============================================
Files 1343 1422 +79
Lines 291806 329766 +37960
Branches 248040 278932 +30892
============================================
+ Hits 261989 279953 +17964
- Misses 24096 43636 +19540
- Partials 5721 6177 +456
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/4636/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/4636/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.01% <ø> (-6.03%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/4636/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.18% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/4636/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.54% <ø> (+2.22%)` | :arrow_up: |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/4636/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.71% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/4636/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.59% <ø> (+2.72%)` | :arrow_up: |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/4636/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `96.45% <ø> (+1.22%)` | :arrow_up: |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/4636/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `69.96% <ø> (-1.87%)` | :arrow_down: |
| [C++ SDK](https://app.codecov.io/gh/apache/iggy/pull/4636/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.69% <ø> (ø)` | |
[see 362 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/4636/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on PR #4636:
URL: https://github.com/apache/iggy/pull/4636#issuecomment-6657590595
/review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4492066898
##########
.claude/skills/connector-runtime/SKILL.md:
##########
@@ -217,7 +217,7 @@ Fatal errors propagate to `main` and exit. Per-connector / per-message errors ar
All families labeled by `connector_key` + `connector_type` (histogram adds `stage`):
-- **Counters**: `iggy_connector_messages_{produced,sent,consumed,processed,filtered}_total` and `iggy_connector_errors_total`. These are the *rendered* names; each is registered without the `_total`, which the OpenMetrics encoder appends.
+- **Counters**: `iggy_connector_messages_{produced,sent,consumed,processed,filtered,errors}_total`.
Review Comment:
Fixed. The errors counter is named separately as iggy_connector_errors_total.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "hubcio (via GitHub) via commits" <co...@iggy.apache.org>.
hubcio commented on PR #4636:
URL: https://github.com/apache/iggy/pull/4636#issuecomment-6661329853
/skill team-review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on PR #4636:
URL: https://github.com/apache/iggy/pull/4636#issuecomment-6700105028
/review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4491960866
##########
core/connectors/sinks/sink_template/src/lib.rs:
##########
@@ -0,0 +1,1069 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **sink** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, the destination identifier is
+//! validated before it's ever interpolated into a request, outbound calls
+//! go through a retry-wrapped client plus a circuit breaker, and messages
+//! are chunked by a configurable batch size instead of shipped as one
+//! unbounded request.
+//!
+//! Delivery follows the sink's `[delivery]` table. By default the server stores
+//! the offset when it answers the poll and the runtime never redelivers a
+//! batch after an error, so `consume()` keeps trying a batch for up to
+//! `max_delivery_time` before it gives up and logs the lost offset range. With
+//! `guarantee = "at_least_once"` the runtime stores the offset only after
+//! `consume()` succeeds, hands a failed batch over again, and stops the sink on
+//! a permanent error instead of skipping the batch. The circuit breaker only
+//! paces the attempts; it never discards a batch. Writes must be idempotent on
+//! `dedup_key()`, because retries, redelivery and failovers can repeat a
+//! message.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSink::push_batch()` — build the
+//! request/write that actually pushes one chunk of messages to your
+//! destination, using `self.config.connection_string` (and
+//! `self.config.target`, if your destination has a table/index/
+//! collection-shaped name). A second `TODO(ConnectorDeveloper)` on the
+//! `connection_string` field's doc comment just asks you to describe its
+//! expected shape — not code, but worth personalizing too. `grep` for the
+//! marker and you'll find both, plus one more on `build_raw_client()` that
+//! only applies if your destination isn't HTTP (see below).
+//!
+//! This template assumes an HTTP-ish destination and uses `reqwest` wrapped
+//! by the SDK's retry middleware, because that's what
+//! `iggy_connector_sdk::retry` is built for and it covers the common case.
+//! If your destination talks something else (a database, a queue, object
+//! storage), swap the client type in `build_raw_client()`/`push_batch()` for
+//! your driver of choice and lean on its own retry/pooling behavior — keep the
+//! surrounding shape (validation in `open()`, circuit breaker, batching,
+//! identifier validation) unchanged. See `core/connectors/sinks/postgres_sink`
+//! or `core/connectors/sinks/s3_sink` in this repo for non-HTTP examples of
+//! that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+ retry_backoff,
+};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::future::Future;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::time::{Instant, sleep};
+use tracing::{debug, error, info, warn};
+
+sink_connector!(TemplateSink);
+
+const CONNECTOR_NAME: &str = "Template sink";
+
+const REDELIVERY_AT_MOST_ONCE: &str =
+ "The runtime does not redeliver them; replay that range from Iggy to recover them.";
+const REDELIVERY_AT_LEAST_ONCE: &str = "The runtime hands them over again.";
+
+const DEFAULT_BATCH_SIZE: usize = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+// Three failed rounds per cool-down cycle bound the attempts a struggling destination sees:
+// the SDK breaker closes completely after its cool-down, so it takes `threshold` new failures
+// to open again. The cool-down has to be shorter than the delivery budget, or a call that
+// finds the breaker open could never reach a retry.
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 3;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "10s";
+// Kept well under the runtime's close budget (the poll timeout plus 60 s): `consume()` blocks
+// a runtime worker thread and a stop waits for it to return.
+const DEFAULT_MAX_DELIVERY_TIME: &str = "20s";
+const BREAKER_RECHECK_INTERVAL: Duration = Duration::from_millis(250);
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` and `target` is optional with a
+// sane default, and unknown keys are rejected outright so a typo in a TOML
+// file fails at load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSinkConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// The destination table/index/collection/bucket name. Kept as its own
+ /// field (rather than folded into `connection_string`) specifically so
+ /// it can be validated in `open()` before ever being interpolated into
+ /// a query, path, or URL — see `validate_identifier` below. Delete this
+ /// field if your destination has no such dynamic identifier.
+ pub target: String,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<usize>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// How long one `consume()` call may keep trying before it gives up on the
+ /// messages it still holds. Must be longer than `circuit_breaker_cool_down`.
+ pub max_delivery_time: Option<String>,
+
+ /// Upgrades the per-batch log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_sink::PostgresSinkConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+/// Rejects anything that isn't a plain alphanumeric/underscore identifier.
+/// Adjust the allowed character set to whatever your destination's naming
+/// rules actually are, but always validate *something* before a
+/// config-or-message-derived name is interpolated into a query, path, or
+/// URL — see `doris_sink::validate_identifier` / `surrealdb_sink::validate_identifier`
+/// in this repo for the same pattern applied to a real destination.
+fn validate_identifier(field: &str, value: &str) -> Result<(), Error> {
+ if value.is_empty() || !value.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+ return Err(Error::InvalidConfigValue(format!(
+ "{field} must be non-empty and contain only ASCII alphanumeric characters and \
+ underscores, got: {value:?}"
+ )));
+ }
+ Ok(())
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+#[derive(Debug, Default)]
+struct State {
+ invocations_count: u64,
+ messages_written: u64,
+ messages_failed: u64,
+}
+
+#[derive(Debug)]
+pub struct TemplateSink {
+ id: u32,
+ config: TemplateSinkConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size_limit: usize,
+ retry_delay: Duration,
+ retry_max_delay: Duration,
+ cool_down: Duration,
+ max_delivery_time: Duration,
+ at_least_once: bool,
+ verbose: bool,
+ state: Mutex<State>,
+ records_written_total: AtomicU64,
+}
+
+impl TemplateSink {
+ pub fn new(id: u32, config: TemplateSinkConfig) -> Self {
+ let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY);
+ let retry_max_delay =
+ parse_duration(config.retry_max_delay.as_deref(), DEFAULT_RETRY_MAX_DELAY);
+ let cool_down = parse_duration(
+ config.circuit_breaker_cool_down.as_deref(),
+ DEFAULT_CIRCUIT_BREAKER_COOL_DOWN,
+ );
+ let max_delivery_time = parse_duration(
+ config.max_delivery_time.as_deref(),
+ DEFAULT_MAX_DELIVERY_TIME,
+ );
+ let circuit_breaker = Arc::new(CircuitBreaker::new(
+ config
+ .circuit_breaker_threshold
+ .unwrap_or(DEFAULT_CIRCUIT_BREAKER_THRESHOLD),
+ cool_down,
+ ));
+ let batch_size_limit = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ client: None,
+ circuit_breaker,
+ batch_size_limit,
+ retry_delay,
+ retry_max_delay,
+ cool_down,
+ max_delivery_time,
+ at_least_once: false,
+ verbose,
+ state: Mutex::new(State::default()),
+ records_written_total: AtomicU64::new(0),
+ }
+ }
+
+ /// TODO(ConnectorDeveloper): build your actual client/connection here using
+ /// `self.config.connection_string` (and `self.config.auth_token`, if
+ /// your destination needs it). This template builds a plain
+ /// `reqwest::Client` to hand to `build_retry_client` — if you're not
+ /// talking HTTP, replace this with e.g. a database connection pool or
+ /// your driver's equivalent, store it on `self` (add a field, since
+ /// `client` here is HTTP-specific), and skip the
+ /// `check_connectivity_with_retry` call below in favor of whatever your
+ /// driver offers (a ping, a test query).
+ fn build_raw_client(&self) -> Result<reqwest::Client, Error> {
+ let timeout = parse_duration(self.config.timeout.as_deref(), DEFAULT_TIMEOUT);
+ reqwest::Client::builder()
+ .timeout(timeout)
+ .build()
+ .map_err(|e| Error::Connection(format!("failed to build HTTP client: {e}")))
+ }
+
+ /// TODO(ConnectorDeveloper): push one chunk of already-batched messages to your
+ /// destination using `self.config.connection_string` and
+ /// `self.config.target`, via `self.client` (already retry-wrapped).
+ ///
+ /// The error you return decides what `consume()` does next:
+ /// - `Error::HttpRequestFailed`, `Error::Connection` and
+ /// `Error::CannotStoreData` (network error, 5xx, timeout) are transient.
+ /// `consume()` retries the same chunk with backoff until
+ /// `max_delivery_time` runs out, and each failed round counts toward the
+ /// circuit breaker. At least once, the failed call then makes the
+ /// runtime hand the batch over again.
+ /// - `Error::PermanentHttpError` (bad schema, the destination rejects the
+ /// payload shape) will never succeed, and the breaker is left alone,
+ /// because a data problem says nothing about whether the destination is
+ /// up. By default the chunk is skipped and counted. At least once, the
+ /// call fails and the runtime stops the sink at that offset, since a
+ /// skipped chunk would let the stored offset pass an undelivered
+ /// message. See the doc comment on that variant in
+ /// `iggy_connector_sdk::Error`.
+ /// - Any other error is not retried. The chunk is counted as failed and
+ /// the error is returned once every chunk has been tried.
+ ///
+ /// By default the runtime never redelivers a batch after an error, because
+ /// the server stored the offset when it answered the poll. The retries
+ /// above are then the only second chance a batch gets, so from Iggy to this
+ /// plugin delivery is at-most-once. With `[delivery] guarantee =
+ /// "at_least_once"` the runtime stores the offset after `consume()`
+ /// succeeds and redelivers a failed batch. Duplicates come from the
+ /// retries (a request that failed part way through a write), from
+ /// redelivery, from replaying an offset range after a logged loss, and from
+ /// a server failover before the stored offset replicated. 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
Review Comment:
Fixed. The guidance now requires Err unless every row of a non-empty chunk was accepted, and says why at-least-once needs it. The README and B9 match.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4492024165
##########
core/connectors/sinks/sink_template/src/lib.rs:
##########
@@ -0,0 +1,1069 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **sink** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, the destination identifier is
+//! validated before it's ever interpolated into a request, outbound calls
+//! go through a retry-wrapped client plus a circuit breaker, and messages
+//! are chunked by a configurable batch size instead of shipped as one
+//! unbounded request.
+//!
+//! Delivery follows the sink's `[delivery]` table. By default the server stores
+//! the offset when it answers the poll and the runtime never redelivers a
+//! batch after an error, so `consume()` keeps trying a batch for up to
+//! `max_delivery_time` before it gives up and logs the lost offset range. With
+//! `guarantee = "at_least_once"` the runtime stores the offset only after
+//! `consume()` succeeds, hands a failed batch over again, and stops the sink on
+//! a permanent error instead of skipping the batch. The circuit breaker only
+//! paces the attempts; it never discards a batch. Writes must be idempotent on
+//! `dedup_key()`, because retries, redelivery and failovers can repeat a
+//! message.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSink::push_batch()` — build the
+//! request/write that actually pushes one chunk of messages to your
+//! destination, using `self.config.connection_string` (and
+//! `self.config.target`, if your destination has a table/index/
+//! collection-shaped name). A second `TODO(ConnectorDeveloper)` on the
+//! `connection_string` field's doc comment just asks you to describe its
+//! expected shape — not code, but worth personalizing too. `grep` for the
+//! marker and you'll find both, plus one more on `build_raw_client()` that
+//! only applies if your destination isn't HTTP (see below).
+//!
+//! This template assumes an HTTP-ish destination and uses `reqwest` wrapped
+//! by the SDK's retry middleware, because that's what
+//! `iggy_connector_sdk::retry` is built for and it covers the common case.
+//! If your destination talks something else (a database, a queue, object
+//! storage), swap the client type in `build_raw_client()`/`push_batch()` for
+//! your driver of choice and lean on its own retry/pooling behavior — keep the
+//! surrounding shape (validation in `open()`, circuit breaker, batching,
+//! identifier validation) unchanged. See `core/connectors/sinks/postgres_sink`
+//! or `core/connectors/sinks/s3_sink` in this repo for non-HTTP examples of
+//! that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+ retry_backoff,
+};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::future::Future;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::time::{Instant, sleep};
+use tracing::{debug, error, info, warn};
+
+sink_connector!(TemplateSink);
+
+const CONNECTOR_NAME: &str = "Template sink";
+
+const REDELIVERY_AT_MOST_ONCE: &str =
+ "The runtime does not redeliver them; replay that range from Iggy to recover them.";
+const REDELIVERY_AT_LEAST_ONCE: &str = "The runtime hands them over again.";
+
+const DEFAULT_BATCH_SIZE: usize = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+// Three failed rounds per cool-down cycle bound the attempts a struggling destination sees:
+// the SDK breaker closes completely after its cool-down, so it takes `threshold` new failures
+// to open again. The cool-down has to be shorter than the delivery budget, or a call that
+// finds the breaker open could never reach a retry.
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 3;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "10s";
+// Kept well under the runtime's close budget (the poll timeout plus 60 s): `consume()` blocks
+// a runtime worker thread and a stop waits for it to return.
+const DEFAULT_MAX_DELIVERY_TIME: &str = "20s";
+const BREAKER_RECHECK_INTERVAL: Duration = Duration::from_millis(250);
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` and `target` is optional with a
+// sane default, and unknown keys are rejected outright so a typo in a TOML
+// file fails at load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSinkConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// The destination table/index/collection/bucket name. Kept as its own
+ /// field (rather than folded into `connection_string`) specifically so
+ /// it can be validated in `open()` before ever being interpolated into
+ /// a query, path, or URL — see `validate_identifier` below. Delete this
+ /// field if your destination has no such dynamic identifier.
+ pub target: String,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<usize>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// How long one `consume()` call may keep trying before it gives up on the
+ /// messages it still holds. Must be longer than `circuit_breaker_cool_down`.
+ pub max_delivery_time: Option<String>,
+
+ /// Upgrades the per-batch log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_sink::PostgresSinkConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+/// Rejects anything that isn't a plain alphanumeric/underscore identifier.
+/// Adjust the allowed character set to whatever your destination's naming
+/// rules actually are, but always validate *something* before a
+/// config-or-message-derived name is interpolated into a query, path, or
+/// URL — see `doris_sink::validate_identifier` / `surrealdb_sink::validate_identifier`
+/// in this repo for the same pattern applied to a real destination.
+fn validate_identifier(field: &str, value: &str) -> Result<(), Error> {
+ if value.is_empty() || !value.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+ return Err(Error::InvalidConfigValue(format!(
+ "{field} must be non-empty and contain only ASCII alphanumeric characters and \
+ underscores, got: {value:?}"
+ )));
+ }
+ Ok(())
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+#[derive(Debug, Default)]
+struct State {
+ invocations_count: u64,
+ messages_written: u64,
+ messages_failed: u64,
+}
+
+#[derive(Debug)]
+pub struct TemplateSink {
+ id: u32,
+ config: TemplateSinkConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size_limit: usize,
+ retry_delay: Duration,
+ retry_max_delay: Duration,
+ cool_down: Duration,
+ max_delivery_time: Duration,
+ at_least_once: bool,
+ verbose: bool,
+ state: Mutex<State>,
+ records_written_total: AtomicU64,
Review Comment:
Fixed. Removed the unread field, its initializer and the fetch_add.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4492107313
##########
core/connectors/sources/source_template/src/lib.rs:
##########
@@ -0,0 +1,676 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Template Apache Iggy **source** connector.
+//!
+//! Everything in this file is already wired up and follows the patterns the
+//! connector-review checklist expects: config validation happens in
+//! `open()`, secrets are `SecretString`, outbound calls go through a
+//! retry-wrapped client plus a circuit breaker, and the read cursor is
+//! staged in `poll()` and only committed in `on_batch_result()` on an ACK so
+//! a dropped/nacked batch can be re-polled instead of silently lost.
+//!
+//! There is exactly **one** place you need to write code, marked
+//! `TODO(ConnectorDeveloper)`: `TemplateSource::fetch_records()` — fetch up
+//! to `batch_size` new records from your external system, starting after
+//! `cursor`. `TemplateSource::build_raw_client()` carries a second
+//! `TODO(ConnectorDeveloper)` too, but only applies if your source isn't
+//! HTTP (see below); a third, on the `connection_string` field's doc
+//! comment, just asks you to describe its expected shape — not code.
+//! `grep` for the marker and you'll find all three.
+//!
+//! This template assumes an HTTP-ish source and uses `reqwest` wrapped by
+//! the SDK's retry middleware, because that's what `iggy_connector_sdk::retry`
+//! is built for and it covers the common case. If your source talks to
+//! something else (a database, a queue, a filesystem), swap the client type
+//! in `build_raw_client()`/`fetch_records()` for your driver of choice and lean on
+//! its own retry/pooling behavior — keep the surrounding shape (validation
+//! in `open()`, circuit breaker, cursor staging, batching) unchanged. See
+//! `core/connectors/sources/postgres_source` in this repo for a real
+//! non-HTTP example of that same shape.
+
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{
+ CircuitBreaker, RetryPolicy, build_retry_client, check_connectivity_with_retry, parse_duration,
+};
+use iggy_connector_sdk::{
+ ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source,
+ source::SourceBatchResult, source_connector,
+};
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tracing::{debug, error, info, warn};
+
+source_connector!(TemplateSource);
+
+const CONNECTOR_NAME: &str = "Template source";
+
+const DEFAULT_POLL_INTERVAL: &str = "1s";
+const DEFAULT_BATCH_SIZE: u32 = 100;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 10;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s";
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 5;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "30s";
+
+// ── Configuration ───────────────────────────────────────────────────────────
+//
+// Every tunable except `connection_string` is optional with a sane default,
+// and unknown keys are rejected outright so a typo in a TOML file fails at
+// load time instead of silently doing nothing.
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TemplateSourceConfig {
+ /// TODO(ConnectorDeveloper): document the exact shape this connector expects, e.g.
+ /// "<https://api.example.com>" or "postgres://user:pass@host:5432/db".
+ /// `SecretString` because DSNs commonly embed credentials — never plain
+ /// `String` for this field, see `PostgresSinkConfig::connection_string`
+ /// in `sinks/postgres_sink` for the same pattern.
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+
+ /// Example of a secret-shaped setting. `SecretString` keeps it out of
+ /// `Debug`/log output; delete this field if `connection_string` already
+ /// carries all required auth. Read it with `.expose_secret()` (from the
+ /// `secrecy::ExposeSecret` trait) at the one place you actually need the
+ /// plaintext — e.g. when building an auth header in `build_raw_client()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
+
+ /// Optional path (e.g. "/health") probed with retry during `open()`
+ /// before the connector is considered ready. Leave unset if your target
+ /// has no health endpoint — the probe is skipped, not failed, in that case.
+ pub health_check_path: Option<String>,
+
+ pub batch_size: Option<u32>,
+ pub poll_interval: Option<String>,
+ pub timeout: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub retry_max_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+ pub open_retry_max_delay: Option<String>,
+ pub circuit_breaker_threshold: Option<u32>,
+ pub circuit_breaker_cool_down: Option<String>,
+
+ /// Upgrades the per-poll log line from `debug!` to `info!`. Mirrors the
+ /// runtime's own `verbose` flag; keep the field name `verbose_logging`
+ /// (see `postgres_source::PostgresSourceConfig::verbose_logging`).
+ pub verbose_logging: Option<bool>,
+}
+
+// ── Internal state ──────────────────────────────────────────────────────────
+
+/// Read cursor. Kept as a plain `Option<String>` so it fits whatever ordering
+/// field your source uses (a timestamp, an auto-increment ID, an opaque
+/// pagination token, ...) — stringify it however makes sense for your data.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+struct State {
+ cursor: Option<String>,
+}
+
+/// One record as fetched from the external system, before it's turned into
+/// an Iggy message. Replace the `payload` type with whatever your
+/// `fetch_records()` actually produces.
+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,
+}
+
+#[derive(Debug)]
+pub struct TemplateSource {
+ id: u32,
+ config: TemplateSourceConfig,
+ client: Option<ClientWithMiddleware>,
+ circuit_breaker: Arc<CircuitBreaker>,
+ batch_size: u32,
+ poll_interval: Duration,
+ retry_delay: Duration,
+ verbose: bool,
+ state: Mutex<State>,
+ pending_state: Mutex<Option<State>>,
+ records_produced: AtomicU64,
+}
+
+impl TemplateSource {
+ pub fn new(id: u32, config: TemplateSourceConfig, state: Option<ConnectorState>) -> Self {
+ let poll_interval = *humantime::Duration::from_str_lossy(
+ config
+ .poll_interval
+ .as_deref()
+ .unwrap_or(DEFAULT_POLL_INTERVAL),
+ );
+ let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY);
+ let circuit_breaker = Arc::new(CircuitBreaker::new(
+ config
+ .circuit_breaker_threshold
+ .unwrap_or(DEFAULT_CIRCUIT_BREAKER_THRESHOLD),
+ parse_duration(
+ config.circuit_breaker_cool_down.as_deref(),
+ DEFAULT_CIRCUIT_BREAKER_COOL_DOWN,
+ ),
+ ));
+ let batch_size = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ let restored_state = state
+ .and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id))
+ .inspect(|s| {
+ info!(
+ "Restored state for {CONNECTOR_NAME} connector with ID: {id}. Cursor: {:?}",
+ s.cursor
+ );
+ });
+
+ Self {
+ id,
+ config,
+ client: None,
+ circuit_breaker,
+ batch_size,
+ poll_interval,
+ retry_delay,
+ verbose,
+ state: Mutex::new(restored_state.unwrap_or_default()),
+ pending_state: Mutex::new(None),
+ records_produced: AtomicU64::new(0),
+ }
+ }
+
+ fn serialize_state(&self, state: &State) -> Option<ConnectorState> {
+ ConnectorState::serialize(state, CONNECTOR_NAME, self.id)
+ }
+
+ /// TODO(ConnectorDeveloper): build your actual client/connection here using
+ /// `self.config.connection_string` (and `self.config.auth_token`, if
+ /// your system needs it). This template builds a plain `reqwest::Client`
+ /// to hand to `build_retry_client` — if you're not talking HTTP, replace
+ /// this with e.g. a `sqlx::PgPool::connect(...)` or your driver's
+ /// equivalent, store it on `self` (add a field, since `client` here is
+ /// HTTP-specific), and skip the `check_connectivity_with_retry` call
+ /// below in favor of whatever your driver offers (a ping, a test query).
+ fn build_raw_client(&self) -> Result<reqwest::Client, Error> {
+ let timeout = parse_duration(self.config.timeout.as_deref(), DEFAULT_TIMEOUT);
+ reqwest::Client::builder()
+ .timeout(timeout)
+ .build()
+ .map_err(|e| Error::Connection(format!("failed to build HTTP client: {e}")))
+ }
+
+ /// TODO(ConnectorDeveloper): fetch up to `self.batch_size` new records from your
+ /// external system, ordered after `cursor` (`None` means "from the
+ /// beginning" or "from now" — whichever is right for your source).
+ /// Use `self.config.connection_string` as the base address and
+ /// `self.client` (already retry-wrapped) to make the request. Map each
+ /// 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, 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,
+ cursor: Option<&str>,
+ ) -> Result<Vec<FetchedRecord>, Error> {
+ let _ = (client, cursor); // remove once implemented
+ Err(Error::InitError(
+ "TemplateSource::fetch_records is not implemented yet — see the TODO(ConnectorDeveloper) comment \
+ in template_source/src/lib.rs"
+ .to_string(),
+ ))
+ }
+}
+
+// ── Source trait ────────────────────────────────────────────────────────────
+
+#[async_trait]
+impl Source for TemplateSource {
+ async fn open(&mut self) -> Result<(), Error> {
+ // Structural validation happens here, not in `new()`, because only
+ // `open()` can return an error — `new()` is a plain factory function
+ // with nowhere to send a "this config is invalid" result.
+ if self
+ .config
+ .connection_string
+ .expose_secret()
+ .trim()
+ .is_empty()
+ {
+ return Err(Error::InvalidConfigValue(
+ "connection_string must not be empty".to_string(),
+ ));
+ }
+
+ info!(
+ "Opening {CONNECTOR_NAME} connector with ID: {}, batch_size: {}, poll_interval: {:?}",
+ self.id, self.batch_size, self.poll_interval
+ );
+
+ let raw_client = self.build_raw_client()?;
+
+ if let Some(health_path) = &self.config.health_check_path {
+ let base = Url::parse(self.config.connection_string.expose_secret()).map_err(|e| {
+ Error::InvalidConfigValue(format!("connection_string is not a valid URL: {e}"))
+ })?;
+ let health_url = base.join(health_path).map_err(|e| {
+ Error::InvalidConfigValue(format!("invalid health_check_path: {e}"))
+ })?;
+ check_connectivity_with_retry(
+ &raw_client,
+ health_url,
+ CONNECTOR_NAME,
+ self.id,
+ RetryPolicy {
+ max_attempts: self
+ .config
+ .max_open_retries
+ .unwrap_or(DEFAULT_MAX_OPEN_RETRIES),
+ base_delay: self.retry_delay,
+ max_delay: parse_duration(
+ self.config.open_retry_max_delay.as_deref(),
+ DEFAULT_OPEN_RETRY_MAX_DELAY,
+ ),
+ },
+ )
+ .await?;
+ } else {
+ warn!(
+ "{CONNECTOR_NAME} connector with ID: {} has no health_check_path configured — \
+ skipping the startup connectivity probe. Consider adding one.",
+ self.id
+ );
+ }
+
+ self.client = Some(build_retry_client(
+ raw_client,
+ self.config
+ .max_retries
+ .unwrap_or(DEFAULT_MAX_RETRIES)
+ .max(1),
+ self.retry_delay,
+ parse_duration(
+ self.config.retry_max_delay.as_deref(),
+ DEFAULT_RETRY_MAX_DELAY,
+ ),
+ CONNECTOR_NAME,
+ ));
+
+ info!(
+ "{CONNECTOR_NAME} connector with ID: {} opened successfully",
+ self.id
+ );
+ Ok(())
+ }
+
+ async fn poll(&self) -> Result<ProducedMessages, Error> {
+ // If the breaker is open, sleep for the normal poll interval and
+ // return an empty (not an error) result. Returning `Err` here would
+ // make the runtime retry `poll()` again immediately with no delay —
+ // see `handle_messages` in the SDK's source container — so an empty
+ // ACK-free result is how a source waits out a known-bad window
+ // without busy-looping or counting against the NACK budget.
+ if self.circuit_breaker.is_open().await {
+ warn!(
+ "{CONNECTOR_NAME} connector with ID: {} — circuit breaker OPEN, skipping poll",
+ self.id
+ );
+ tokio::time::sleep(self.poll_interval).await;
+ return Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages: Vec::new(),
+ state: None,
+ });
+ }
+ tokio::time::sleep(self.poll_interval).await;
+
+ let client = self.client.as_ref().ok_or_else(|| {
+ Error::Connection("client not initialized -- was open() called?".into())
+ })?;
+ let cursor = self.state.lock().await.cursor.clone();
+
+ let records = match self.fetch_records(client, cursor.as_deref()).await {
+ Ok(records) => {
+ self.circuit_breaker.record_success();
+ records
+ }
+ Err(err) => {
+ if !matches!(err, Error::PermanentHttpError(_)) {
+ self.circuit_breaker.record_failure().await;
+ }
+ return Err(err);
+ }
+ };
+
+ if records.is_empty() {
+ return Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages: Vec::new(),
+ state: None,
+ });
+ }
+
+ let mut messages = Vec::with_capacity(records.len());
+ let mut new_cursor = cursor;
+ for record in records {
+ let Ok(payload) = serde_json::to_vec(&record.payload) else {
+ error!(
+ "Failed to serialize a record fetched by {CONNECTOR_NAME} connector with ID: {}",
+ self.id
+ );
+ continue;
+ };
+ // Only advance the candidate cursor once the record actually made it
+ // 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);
Review Comment:
Fixed. The payload is now a Result, and stage_batch() stops at the first record that cannot be converted. The cursor stays at the last good record. If the first record is bad, the poll fails. Tests: a bad record in the middle, and a bad first record.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4492114316
##########
.claude/skills/connector-source/TEMPLATE.md:
##########
@@ -1,64 +1,134 @@
-# Source plugin skeleton
+# Source plugin fill-in-the-blank kit
-Boilerplate for a new `core/connectors/sources/<name>_source/`. Adapt
-the `MySource` / `Client` / row types to the backend driver.
+Copy this kit into `core/connectors/sources/<name>_source/`. The
+scaffolding covers config, secrets, sleep-first poll, lock discipline,
+staged/committed state serde (ACK/NACK batch acknowledgment, #3855),
+retry classification, logging, and the six canonical state tests.
+**You only implement the marked `TODO(ConnectorDeveloper)` sections:** build a
+client from the connection string, and fetch the next batch (advancing
+a cursor).
-`Cargo.toml` is identical to a sink's (see
-[connector-sink/TEMPLATE.md](../connector-sink/TEMPLATE.md)) - only
-the crate name suffix changes (`iggy_connector_<name>_source`) and the
-upstream client dep.
+Also read [SKILL.md](SKILL.md) and pre-flight with
+[connector-pr-review](../connector-pr-review/SKILL.md) before `/ready`.
+
+Prefer starting from a compiling crate over copying this prose kit:
+`core/connectors/sources/source_template/` implements the same shape as
+real, tested code you can `cargo build`/`cargo test` immediately, with the
+same `TODO(ConnectorDeveloper)` markers. Use this kit instead only when
+copying a whole crate is more scaffolding than you need.
+
+## Files to create
+
+```text
+core/connectors/sources/<name>_source/
+├── Cargo.toml
+├── README.md
+├── config.toml
+└── src/lib.rs
+```
+
+Add a workspace member, a row in `sources/README.md`, and a sample under
+`runtime/example_config/connectors/`.
+
+---
+
+## Cargo.toml
+
+Same shape as the sink kit (`cdylib` + `lib`, workspace deps, Apache
+header). Only the package name suffix changes:
+
+```toml
+name = "iggy_connector_<name>_source"
+# ... identical metadata / machete ignored / crate-type ...
+# TODO(ConnectorDeveloper): add your client crate as a workspace dependency
+```
+
+---
+
+## config.toml (example)
+
+```toml
+path = "../../target/release/libiggy_connector_<name>_source"
+
+[[sources]]
+key = "<name>"
+enabled = true
+
+[sources.<name>.plugin_config]
+connection_string = "scheme://user:pass@host:port/db"
+poll_interval = "5s"
+batch_size = 100
+max_retries = 3
+retry_delay = "500ms"
+verbose_logging = false
+```
+
+Defaults in this file must match `DEFAULT_*` consts in code.
+
+---
## src/lib.rs
-Code reads top to bottom. Public types first, `impl Source` after,
-helpers below.
+Replace `<Name>` / `<name>` and implement only the `TODO(ConnectorDeveloper)` blocks.
```rust
/* Apache 2.0 header */
-use std::str::FromStr;
-use std::time::Duration;
-
use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
use iggy_connector_sdk::{
ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source,
source::SourceBatchResult, source_connector,
};
+use secrecy::{ExposeSecret, SecretString};
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;
use tokio::time::sleep;
use tracing::{debug, error, info, warn};
-source_connector!(MySource);
+source_connector!(NameSource);
-const CONNECTOR_NAME: &str = "My source";
+const CONNECTOR_NAME: &str = "Name source";
+const DEFAULT_POLL_INTERVAL: &str = "5s";
+const DEFAULT_BATCH_SIZE: u32 = 100;
+const DEFAULT_MAX_RETRIES: u32 = 3; // total attempts
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+
+struct BackendClient {
+ // TODO(ConnectorDeveloper): fields
+}
#[derive(Debug, Serialize, Deserialize)]
-pub struct MySourceConfig {
- pub endpoint: String,
- pub poll_interval: Option<String>, // humantime, "10s" default
+#[serde(deny_unknown_fields)]
+pub struct NameSourceConfig {
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+ pub poll_interval: Option<String>,
pub batch_size: Option<u32>,
pub max_retries: Option<u32>,
pub retry_delay: Option<String>,
pub verbose_logging: Option<bool>,
+ // TODO(ConnectorDeveloper): optional non-secret knobs (query, table, index, ...)
}
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize)]
struct State {
- cursor: Option<String>, // WAL LSN, scroll id, timestamp, ...
- last_offset: u64,
+ /// Opaque backend cursor (LSN, scroll id, timestamp, PK, ...). Keep O(1).
+ cursor: Option<String>,
messages_produced: u64,
}
#[derive(Debug)]
-pub struct MySource {
+pub struct NameSource {
Review Comment:
Fixed. The impl matches the struct and the imports and derives are restored. The whole block compiles and passes clippy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add sink/source connector templates, authoring guide, and blog post [iggy]
Posted by "ryerraguntla (via GitHub) via commits" <co...@iggy.apache.org>.
ryerraguntla commented on code in PR #4636:
URL: https://github.com/apache/iggy/pull/4636#discussion_r4492103454
##########
.claude/skills/connectors-overview/SKILL.md:
##########
@@ -96,24 +102,14 @@ The connectors codebase is intentionally repetitive across plugins. Cross-plugin
### Secrets
-Any credential-bearing field (connection strings, API keys, bearer tokens, AWS keys) must be `SecretString` from the `secrecy` crate. Plain `String` for a credential is a review-blocker: `SecretString` redacts on `Debug`, so it is what keeps a credential out of a log line that formats the whole config.
-
-**`serde_secret::serialize_secret` EXPOSES the secret. It does not redact.** It calls `expose_secret()` and writes the plaintext. `SecretString` deliberately has no `Serialize` impl, and that absence is the protection - so adding `serialize_with` is what *unblocks* the derive and turns a compile-time guarantee into plaintext output. Use it only where the plaintext is the point: a wire payload, a persisted config, an API response that exposes credentials by design.
-
-So the default for a plugin config struct is **derive `Deserialize`, but not `Serialize`**. `Deserialize` is required: the SDK glue deserializes the config into the plugin's own struct (`sdk/src/{sink,source}.rs` call `serde_json::from_str::<C>` under a `DeserializeOwned` bound).
-
-What never happens is the return trip. The runtime holds plugin configuration as a `serde_json::Value` - parsed from TOML, posted as JSON to the control API, or injected by env var - and hands that across the FFI, so nothing re-serializes the plugin's struct. Leaving `Serialize` off makes that compiler-enforced instead of convention-enforced (`sources/http_source/src/lib.rs::HttpSourceConfig` does this, and comments the omission so nobody adds it back).
-
-Pattern:
+Any credential-bearing field (connection strings, API keys, bearer tokens, AWS keys) must be `SecretString` from the `secrecy` crate, with the workspace serde wrapper applied so `Debug` and serialization both redact. Runtime exposes plugin configs over the `/stats` HTTP surface via serialization - plain `String` leaks the secret to anyone who can hit the endpoint. Plain `String` for a credential is a review-blocker. Pattern (from `sinks/postgres_sink/src/lib.rs::PostgresSinkConfig`):
Review Comment:
Fixed. The docs now say serialize_secret writes plaintext and SecretString redacts only Debug and Display. They name serialize_redacted for redaction and explain that the runtime serves the raw plugin_config value. I kept serialize_secret in the templates because it is the convention in the bundled plugins and the checklist requires it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org