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/08/25 15:46:24 UTC
[PR] feat(connectors): add sink_template and source_template starting points [iggy]
ryerraguntla opened a new pull request, #3965:
URL: https://github.com/apache/iggy/pull/3965
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`, 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.
- **source_template**: the connection string/client setup and the data-fetch call against the source system.
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.
- Updated `.claude/skills/connector-*` skill definitions, including a new `connector-pr-review` skill intended to catch the same recurring review issues (missing retry/circuit-breaker, plaintext secrets, unvalidated config, missing cursor staging) automatically on future connector PRs.
- Workspace registration (`Cargo.toml`, `Cargo.lock`) and README entries (`core/connectors/sinks/README.md`, `core/connectors/sources/README.md`,
`core/connectors/README.md`) for both new crates.
- Example runtime configs (`core/connectors/runtime/example_config/connectors/{sink,source}_template.toml`).
### Why
Per the discussion in #3956: connector PRs have been hitting the same review comments repeatedly (improperly typed secrets, missing state-staging, inconsistent error handling, thin test coverage). Rather than repeating that feedback PR after PR, these templates make the required pattern the path of least resistance — a new connector starts from working, reviewed code and only needs the backend-specific logic filled in.
Local Execution
Passed
Pre-commit hooks ran
### Testing
- [ ] `cargo fmt --all`
- [ ] `cargo clippy --all-targets --all-features -- -D warnings`
- [ ] `cargo build -p iggy_connector_template_sink -p iggy_connector_template_source`
- [ ] `cargo test -p iggy_connector_template_sink -p iggy_connector_template_source`
- [ ] `cargo machete`
- [ ] `cargo sort --workspace`
- [ ] `typos`
- [ ] `./scripts/ci/license-headers.sh --check`
AI Usage
Claude
Implementation was AI-generated with human review
Each line of the code is reviewed and can be explained by developer
--
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_template and source_template starting points [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on PR #3965:
URL: https://github.com/apache/iggy/pull/3965#issuecomment-5668023619
Sure @justinmclean . Currently I am heads down with Kafka gateway , after the next PR on gateway I will jump onto this topic followed by CVE discussions PRs. Thanks for reminding and following up
--
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_template and source_template starting points [iggy]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3965:
URL: https://github.com/apache/iggy/pull/3965#issuecomment-5555874274
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_template and source_template starting points [iggy]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3965:
URL: https://github.com/apache/iggy/pull/3965#discussion_r3854867827
##########
core/connectors/sources/source_template/src/lib.rs:
##########
@@ -0,0 +1,586 @@
+// 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 are exactly **two** places you need to touch, each marked
+//! `TODO(ConnectorDeveloper)`:
+//! 1. `TemplateSource::connect()` — build your actual client/connection
+//! from `config.connection_string` (and `config.auth_token`, if used).
+//! 2. `TemplateSource::fetch_records()` — fetch up to `batch_size` new
+//! records from your external system, starting after `cursor`.
+//!
+//! 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 `connect()`/`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, ConnectivityConfig, 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::{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)]
Review Comment:
Deriving `Serialize` for the config while also using `serialize_secret` / `serialize_optional_secret` on secret fields risks accidental plaintext secret exposure anywhere the config is serialized (e.g., logs, debug endpoints, metrics labels, etc.). If plugin config structs don’t need to be serialized, remove the `Serialize` derive (and the `serialize_with` attrs); if serialization is required, use a redacting serializer (e.g., `serialize_redacted`/`serialize_optional_redacted`, if available in `iggy_common`) so secrets cannot be emitted as plaintext.
##########
core/connectors/sources/source_template/src/lib.rs:
##########
@@ -0,0 +1,586 @@
+// 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 are exactly **two** places you need to touch, each marked
+//! `TODO(ConnectorDeveloper)`:
+//! 1. `TemplateSource::connect()` — build your actual client/connection
+//! from `config.connection_string` (and `config.auth_token`, if used).
+//! 2. `TemplateSource::fetch_records()` — fetch up to `batch_size` new
+//! records from your external system, starting after `cursor`.
+//!
+//! 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 `connect()`/`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, ConnectivityConfig, 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::{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,
Review Comment:
Deriving `Serialize` for the config while also using `serialize_secret` / `serialize_optional_secret` on secret fields risks accidental plaintext secret exposure anywhere the config is serialized (e.g., logs, debug endpoints, metrics labels, etc.). If plugin config structs don’t need to be serialized, remove the `Serialize` derive (and the `serialize_with` attrs); if serialization is required, use a redacting serializer (e.g., `serialize_redacted`/`serialize_optional_redacted`, if available in `iggy_common`) so secrets cannot be emitted as plaintext.
##########
core/connectors/sources/source_template/src/lib.rs:
##########
@@ -0,0 +1,586 @@
+// 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 are exactly **two** places you need to touch, each marked
+//! `TODO(ConnectorDeveloper)`:
+//! 1. `TemplateSource::connect()` — build your actual client/connection
+//! from `config.connection_string` (and `config.auth_token`, if used).
+//! 2. `TemplateSource::fetch_records()` — fetch up to `batch_size` new
+//! records from your external system, starting after `cursor`.
+//!
+//! 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 `connect()`/`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, ConnectivityConfig, 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::{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 `connect()`.
+ #[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>,
+}
+
+// ── 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,
+ 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,
+ 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 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,
+ 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.
+ 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,
+ &ConnectivityConfig {
+ max_open_retries: self
+ .config
+ .max_open_retries
+ .unwrap_or(DEFAULT_MAX_OPEN_RETRIES),
+ open_retry_max_delay: parse_duration(
+ self.config.open_retry_max_delay.as_deref(),
+ DEFAULT_OPEN_RETRY_MAX_DELAY,
+ ),
+ retry_delay: self.retry_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 {
+ new_cursor = Some(record.cursor_value);
+ 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;
+ };
Review Comment:
`new_cursor` is advanced before serialization, but serialization failures `continue` (dropping the record). This will still stage/commit a cursor that can skip past an un-emitted record (silent data loss). Fix by only advancing `new_cursor` after the message is successfully built/pushed (or alternatively, treat serialization failure as a hard error and return `Err(Error::Serialization(...))` so the cursor cannot advance past the bad record).
##########
core/connectors/sinks/sink_template/src/lib.rs:
##########
@@ -0,0 +1,504 @@
+// 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.
+//!
+//! There is exactly **one** place you need to touch, 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).
+//!
+//! 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 `connect()`/`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, ConnectivityConfig, build_retry_client, check_connectivity_with_retry,
+ parse_duration,
+};
+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::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tracing::{error, info, warn};
+
+sink_connector!(TemplateSink);
+
+const CONNECTOR_NAME: &str = "Template sink";
+
+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";
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 5;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "30s";
+
+// ── 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 `connect()`.
+ #[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>,
+}
+
+/// 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,
+ 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 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_limit = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+
+ Self {
+ id,
+ config,
+ client: None,
+ circuit_breaker,
+ batch_size_limit,
+ retry_delay,
+ 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).
+ /// Distinguish permanent failures (bad schema, destination rejects the
+ /// payload shape — will not succeed on retry) from transient ones
+ /// (network error, 5xx, timeout — should retry) by returning
+ /// `Error::PermanentHttpError` for the former; see the doc comment on
+ /// that variant in `iggy_connector_sdk::Error` for why the distinction
+ /// matters to the circuit breaker.
+ async fn push_batch(
+ &self,
+ client: &ClientWithMiddleware,
+ batch: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let _ = (client, batch); // remove once implemented
+ Err(Error::InitError(
+ "TemplateSink::push_batch is not implemented yet — see the TODO(ConnectorDeveloper) comment in \
+ template_sink/src/lib.rs"
+ .to_string(),
+ ))
+ }
+}
+
+// ── Sink trait ────────────────────────────────────────────────────────────────
+
+#[async_trait]
+impl Sink for TemplateSink {
+ 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(),
+ ));
+ }
+ validate_identifier("target", &self.config.target)?;
+
+ info!(
+ "Opening {CONNECTOR_NAME} connector with ID: {}, target: {}, batch_size: {}",
+ self.id, self.config.target, self.batch_size_limit
+ );
+
+ 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,
+ &ConnectivityConfig {
+ max_open_retries: self
+ .config
+ .max_open_retries
+ .unwrap_or(DEFAULT_MAX_OPEN_RETRIES),
+ open_retry_max_delay: parse_duration(
+ self.config.open_retry_max_delay.as_deref(),
+ DEFAULT_OPEN_RETRY_MAX_DELAY,
+ ),
+ retry_delay: self.retry_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,
Review Comment:
The config knob name `retry_max_delay` conflicts with the canonical naming documented in the newly-added `connector-pr-review` skill (`max_retry_delay`). Since this PR’s purpose is to standardize patterns, consider renaming the field/consts to `max_retry_delay` and updating all related TOMLs/READMEs/templates to match, to avoid future connector implementations drifting on naming again.
##########
core/connectors/sources/source_template/src/lib.rs:
##########
@@ -0,0 +1,586 @@
+// 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 are exactly **two** places you need to touch, each marked
+//! `TODO(ConnectorDeveloper)`:
+//! 1. `TemplateSource::connect()` — build your actual client/connection
+//! from `config.connection_string` (and `config.auth_token`, if used).
+//! 2. `TemplateSource::fetch_records()` — fetch up to `batch_size` new
+//! records from your external system, starting after `cursor`.
+//!
+//! 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 `connect()`/`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, ConnectivityConfig, 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::{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 `connect()`.
+ #[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>,
+}
+
+// ── 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,
+ 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,
+ 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 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,
+ 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.
+ 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,
+ &ConnectivityConfig {
+ max_open_retries: self
+ .config
+ .max_open_retries
+ .unwrap_or(DEFAULT_MAX_OPEN_RETRIES),
+ open_retry_max_delay: parse_duration(
+ self.config.open_retry_max_delay.as_deref(),
+ DEFAULT_OPEN_RETRY_MAX_DELAY,
+ ),
+ retry_delay: self.retry_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 {
+ new_cursor = Some(record.cursor_value);
+ 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;
+ };
+ messages.push(ProducedMessage {
+ id: None,
+ headers: None,
+ checksum: None,
+ timestamp: None,
+ origin_timestamp: None,
+ payload,
+ });
+ }
+
+ 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);
+
+ 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, batch timed out, 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`.
+ 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 {
+ fn from_str_lossy(s: &str) -> humantime::Duration;
+}
+impl DurationExt for humantime::Duration {
+ fn from_str_lossy(s: &str) -> humantime::Duration {
+ use std::str::FromStr;
+ humantime::Duration::from_str(s).unwrap_or_else(|_| {
+ warn!("Invalid poll_interval {s:?}, falling back to {DEFAULT_POLL_INTERVAL}");
+ humantime::Duration::from_str(DEFAULT_POLL_INTERVAL)
+ .expect("DEFAULT_POLL_INTERVAL must itself be a valid duration literal")
+ })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn test_config() -> TemplateSourceConfig {
+ TemplateSourceConfig {
+ connection_string: SecretString::from("https://api.example.com"),
+ auth_token: None,
+ health_check_path: None,
+ batch_size: Some(50),
+ poll_interval: Some("50ms".to_string()),
+ timeout: Some("5s".to_string()),
+ max_retries: Some(2),
+ retry_delay: Some("10ms".to_string()),
+ retry_max_delay: Some("100ms".to_string()),
+ max_open_retries: Some(2),
+ open_retry_max_delay: Some("100ms".to_string()),
+ circuit_breaker_threshold: Some(3),
+ circuit_breaker_cool_down: Some("50ms".to_string()),
+ }
+ }
+
+ #[tokio::test]
+ async fn open_rejects_empty_connection_string() {
+ let mut config = test_config();
+ config.connection_string = SecretString::from(" ");
+ let mut source = TemplateSource::new(1, config, None);
+ let result = source.open().await;
+ assert!(matches!(result, Err(Error::InvalidConfigValue(_))));
+ }
+
+ #[tokio::test]
+ async fn given_no_state_should_start_with_no_cursor() {
+ let source = TemplateSource::new(1, test_config(), None);
+ assert_eq!(source.state.lock().await.cursor, None);
+ }
+
+ #[tokio::test]
+ async fn given_persisted_state_should_restore_cursor() {
+ let state = State {
+ cursor: Some("2024-01-01T00:00:00Z".to_string()),
+ };
+ let serialized = rmp_serde::to_vec(&state).expect("failed to serialize state");
+ let source = TemplateSource::new(1, test_config(), Some(ConnectorState(serialized)));
+ assert_eq!(
+ source.state.lock().await.cursor,
+ Some("2024-01-01T00:00:00Z".to_string())
+ );
+ }
+
+ #[tokio::test]
+ async fn given_invalid_persisted_state_should_start_fresh() {
+ let source = TemplateSource::new(
+ 1,
+ test_config(),
+ Some(ConnectorState(b"not valid msgpack".to_vec())),
+ );
+ assert_eq!(source.state.lock().await.cursor, None);
+ }
+
+ #[test]
+ fn state_should_be_serializable_and_deserializable() {
+ let original = State {
+ cursor: Some("2024-01-01T00:00:00Z".to_string()),
+ };
+
+ let serialized = rmp_serde::to_vec(&original).expect("failed to serialize state");
+ let deserialized: State =
+ rmp_serde::from_slice(&serialized).expect("failed to deserialize state");
+
+ assert_eq!(original.cursor, deserialized.cursor);
+ }
+
+ #[tokio::test]
+ async fn given_ack_should_commit_staged_cursor() {
+ let source = TemplateSource::new(1, test_config(), None);
+ *source.pending_state.lock().await = Some(State {
+ cursor: Some("next-cursor".to_string()),
+ });
+
+ source
+ .on_batch_result(SourceBatchResult::Ack)
+ .await
+ .expect("ACK should be applied");
+
+ assert_eq!(
+ source.state.lock().await.cursor,
+ Some("next-cursor".to_string())
+ );
+ assert!(source.pending_state.lock().await.is_none());
+ }
+
+ #[tokio::test]
+ async fn given_nack_should_discard_staged_cursor() {
+ let source = TemplateSource::new(1, test_config(), None);
+ *source.pending_state.lock().await = Some(State {
+ cursor: Some("next-cursor".to_string()),
+ });
+
+ source
+ .on_batch_result(SourceBatchResult::Nack)
+ .await
+ .expect("NACK should be applied");
+
+ // The committed cursor is unchanged (still None) — the candidate is
+ // simply discarded so the same range is polled again.
+ assert_eq!(source.state.lock().await.cursor, None);
+ assert!(source.pending_state.lock().await.is_none());
+ }
+
+ #[tokio::test]
+ async fn poll_returns_empty_without_error_when_circuit_is_open() {
+ let source = TemplateSource::new(1, test_config(), None);
+ source.circuit_breaker.record_failure().await;
+ source.circuit_breaker.record_failure().await;
+ source.circuit_breaker.record_failure().await;
+ assert!(source.circuit_breaker.is_open().await);
+
+ let result = source
+ .poll()
+ .await
Review Comment:
This test will incur a real `tokio::time::sleep(self.poll_interval)` during `poll()` (50ms in `test_config()`), slowing the suite and potentially introducing timing flakiness. Prefer using Tokio’s time control (e.g., `tokio::time::pause()` + `advance()`) so the test is deterministic and near-instant.
##########
core/connectors/sinks/sink_template/src/lib.rs:
##########
@@ -0,0 +1,504 @@
+// 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.
+//!
+//! There is exactly **one** place you need to touch, 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).
+//!
+//! 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 `connect()`/`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, ConnectivityConfig, build_retry_client, check_connectivity_with_retry,
+ parse_duration,
+};
+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::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tracing::{error, info, warn};
+
+sink_connector!(TemplateSink);
+
+const CONNECTOR_NAME: &str = "Template sink";
+
+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";
+const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 5;
+const DEFAULT_CIRCUIT_BREAKER_COOL_DOWN: &str = "30s";
+
+// ── 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 `connect()`.
+ #[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>,
Review Comment:
The config knob name `retry_max_delay` conflicts with the canonical naming documented in the newly-added `connector-pr-review` skill (`max_retry_delay`). Since this PR’s purpose is to standardize patterns, consider renaming the field/consts to `max_retry_delay` and updating all related TOMLs/READMEs/templates to match, to avoid future connector implementations drifting on naming again.
##########
core/connectors/sources/source_template/src/lib.rs:
##########
@@ -0,0 +1,586 @@
+// 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 are exactly **two** places you need to touch, each marked
+//! `TODO(ConnectorDeveloper)`:
+//! 1. `TemplateSource::connect()` — build your actual client/connection
+//! from `config.connection_string` (and `config.auth_token`, if used).
+//! 2. `TemplateSource::fetch_records()` — fetch up to `batch_size` new
+//! records from your external system, starting after `cursor`.
+//!
+//! 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 `connect()`/`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, ConnectivityConfig, 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::{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 `connect()`.
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub auth_token: Option<SecretString>,
Review Comment:
Deriving `Serialize` for the config while also using `serialize_secret` / `serialize_optional_secret` on secret fields risks accidental plaintext secret exposure anywhere the config is serialized (e.g., logs, debug endpoints, metrics labels, etc.). If plugin config structs don’t need to be serialized, remove the `Serialize` derive (and the `serialize_with` attrs); if serialization is required, use a redacting serializer (e.g., `serialize_redacted`/`serialize_optional_redacted`, if available in `iggy_common`) so secrets cannot be emitted as plaintext.
--
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_template and source_template starting points [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3965:
URL: https://github.com/apache/iggy/pull/3965#issuecomment-5413293397
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3965?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 83.65%. Comparing base ([`2bd9d0b`](https://app.codecov.io/gh/apache/iggy/commit/2bd9d0bce11fc401dc2bec38c126bf9f0ffe8b48?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`8f097c8`](https://app.codecov.io/gh/apache/iggy/commit/8f097c8525038ca01003e82ba3e62ecedb5cb18c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 7 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3965 +/- ##
============================================
- Coverage 84.11% 83.65% -0.47%
Complexity 1358 1358
============================================
Files 1214 1212 -2
Lines 170054 169806 -248
Branches 137611 137570 -41
============================================
- Hits 143045 142051 -994
- Misses 23305 24047 +742
- Partials 3704 3708 +4
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/3965/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/3965/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.94% <ø> (ø)` | |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3965/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `66.67% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3965/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.52% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3965/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.13% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3965/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.48% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3965/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `88.98% <ø> (-6.93%)` | :arrow_down: |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3965/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `68.29% <ø> (ø)` | |
[see 54 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3965/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_template and source_template starting points [iggy]
Posted by "justinmclean (via GitHub)" <gi...@apache.org>.
justinmclean commented on PR #3965:
URL: https://github.com/apache/iggy/pull/3965#issuecomment-5656595866
@ryerraguntla this has conflicts with master. Worth rebasing before it gets closed as stale?
--
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_template and source_template starting points [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #3965:
URL: https://github.com/apache/iggy/pull/3965#issuecomment-5757899743
moving this to draft to not trigger stale bot
--
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_template and source_template starting points [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #3965:
URL: https://github.com/apache/iggy/pull/3965#issuecomment-5879932145
history broken in this PR, please reopen a new one as fresh
<img width="165" height="49" alt="image" src="https://github.com/user-attachments/assets/5bd96530-fa5d-48b8-b72d-4b1a026c2cae" />
--
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_template and source_template starting points [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #3965:
URL: https://github.com/apache/iggy/pull/3965#issuecomment-5792181050
@ryerraguntla looks like merge is broken in this PR
<img width="286" height="87" alt="image" src="https://github.com/user-attachments/assets/d6abf6ad-9d61-48a1-a956-a3d2b6f65eed" />
--
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_template and source_template starting points [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio closed pull request #3965: feat(connectors): add sink_template and source_template starting points
URL: https://github.com/apache/iggy/pull/3965
--
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