Repository navigation
feat(connectors): add sink/source connector templates, authoring guide, and blog post - #4636
ryerraguntla wants to merge 11 commits into
Conversation
|
lets wait with this PR till deferred polling PR is merged, it changes all connectors |
|
/author |
github-actions
Bot
added
S-waiting-on-author
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #4636 +/- ##
=============================================
- Coverage 91.06% 81.05% -10.02%
+ Complexity 3878 3875 -3
=============================================
Files 1562 1564 +2
Lines 405733 406409 +676
Branches 339602 340281 +679
=============================================
- Hits 369496 329395 -40101
- Misses 28844 69554 +40710
- Partials 7393 7460 +67
🚀 New features to boost your workflow:
|
|
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 Thank you for your contribution! |
Squashed for a clean rebase onto current master.
# Conflicts: # .claude/skills/connector-pr-review/SKILL.md # core/connectors/sinks/sink_template/README.md # core/connectors/sinks/sink_template/src/lib.rs # core/connectors/sources/source_template/README.md # core/connectors/sources/source_template/src/lib.rs
e349759 to
edac134
Compare
|
/ready |
github-actions
Bot
added
S-waiting-on-review
|
/skill team-review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Summary: The review found a data-loss defect in the source fill-in-the-blank kit: poll() commits the cursor before the runtime acknowledges the batch. It also found that the sink template's delivery budget does not bound a hung request, that the secrets guidance claims serialize_secret redacts a credential when it writes the plaintext, and several smaller template and documentation defects.
Counts: critical 2, warning 6, nit 3, simplification 2
This review was generated by Claude Code 2.1.310 on deepseek-flash[1m]. Review the output before you act on it.
| 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 |
There was a problem hiding this 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.
| 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`): |
There was a problem hiding this 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.
| failed_rounds += 1; | ||
| let pause = | ||
| retry_backoff(self.retry_delay, failed_rounds, self.retry_max_delay); | ||
| if Instant::now() + pause >= deadline { |
There was a problem hiding this 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().
| // 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); |
There was a problem hiding this 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.
| /// 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 |
There was a problem hiding this 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.
| 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`. |
There was a problem hiding this 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.
| .await; | ||
| match result { | ||
| Ok(()) => written += chunk.len() as u64, | ||
| Err(err @ Error::PermanentHttpError(_)) if self.at_least_once => { |
There was a problem hiding this 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.
| | 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) | |
There was a problem hiding this 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(...).
| at_least_once: bool, | ||
| verbose: bool, | ||
| state: Mutex<State>, | ||
| records_written_total: AtomicU64, |
There was a problem hiding this 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.
| // 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 { |
There was a problem hiding this 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.
github-actions
Bot
added
S-waiting-on-author
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 inopen(), retry/backoff + circuit breaker viaiggy_connector_sdk::retry::RetryPolicy, batch processing / Ack-Nack cursor staging, and a canonical test suite — leaving only the backend-specific bits markedTODO(ConnectorDeveloper):ProducedMessage.idwhen 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.connector-sink,connector-source,connector-testing, andconnectors-overviewskills to cross-link the templates and the new review checklist, and to reflect the currentSource::batch_policy()/ NACK-retry-configurability SDK surface (fix(connectors): make source NACK retries configurable and observable #4269) and the current resolver/toolchain setup, replacing stale text that predated both.Cargo.toml,Cargo.lock) and README entries (sinks/README.md,sources/README.md,core/connectors/README.md) for both new crates, each withpublish = false.(
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, missingcanonical 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-reviewskill turns the same lessons into an automatic pre-flight check for future connector PRs.Testing
cargo build -p iggy_connector_template_sink -p iggy_connector_template_sourcecargo test -p iggy_connector_template_sink -p iggy_connector_template_source(20/20)cargo clippy -p iggy_connector_template_sink -p iggy_connector_template_source --all-targets -- -D warningscargo fmt --all --checkcargo sort --check --no-format(crate + workspace)cargo machetecargo doc -p iggy_connector_template_sink -p iggy_connector_template_source --no-depswithRUSTDOCFLAGS="-D warnings"markdownlinton every new/changed doc./scripts/ci/license-headers.sh --checkAI Usage
batch_policy()), and mining real apache/iggy connector PR review comments for the checklist content.apache/master, not a stale base.Notes for reviewers / before marking ready
mastermerges) and separately as a draft ondocs/connector-guidelines-template-and-blog-post(closed/abandoned, also under a misleadingdocs/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: