chore: add connector-review skill - #4121
Open
slbotbm wants to merge 2 commits into
Open
Conversation
Contributor
Author
|
I don't have claude, so could someone please test this on one of the connectors PRs? |
Contributor
|
I can test it on my OpenSearch one later today. |
Contributor
Author
|
@mattp5657 thanks! |
Contributor
|
Testing the connector-review skill on PR #3873 (OpenSearch sink) Ran the skill end to end to validate it:
Demonstrates the skill's core value: independent multi-angle review, adversarial self-checking that catches its own citation errors, and a sweep pass that guards against groupthink, all before a human reads a line. Full review below: |
Contributor
Connector Review - opensearch_sink (f3d9688)Target: Review: add OpenSearch sink connectorConfirmed (expert + clean-room validator)
ContestedNone. Retracted (validator REMOVE)None. Pre-existing (origin pre-*, not blocking)None as findings against this PR. Two repo-wide gaps were independently checked by the runtime and plugin experts and confirmed correctly disclosed in the PR body rather than hidden: Simplification opportunities (non-blocking)
Verdict: APPROVEFour independent experts (plugin, runtime, sdk, testing) plus clean-room validation and an adversarial sweep found zero critical or warning findings. Every confirmed item is a nit-level naming/logging/comment-accuracy issue or a latent (not live) gap; two are already self-documented by the author as intentional tradeoffs. Counts: critical 0, warning 0, nit 7, simplify 3 Raw findings per expertplugin# Plugin review: PR 3873 (opensearch_sink) — role: plugin
Exemplars compared: `meilisearch_sink` (closest — same batch_size=1000 default,
same `Payload` dispatch shape, same `InvalidRecordValue` on unsupported
payload), `elasticsearch_sink` (closest sibling search-engine sink, same
`owned_value_to_serde_json` usage, same `close()` shape).
## Findings
[nit] opensearch_sink::index_documents (core/connectors/sinks/opensearch_sink/src/lib.rs:494) - keeps the *first* chunk error across a multi-chunk consume(), not the last. Every other multi-batch sink (`http_sink:595`, `mongodb_sink:241`, `surrealdb_sink:596`) uses a last-err accumulator. Deliberate and commented ("kept, not the last, matching BulkOutcome::merge... the one an operator can act on"), and every chunk is still processed and counted regardless of which error surfaces — counts (`indexed`/`failed`) are correct either way, only the reported message text differs. Fix: none required; flagging for consistency awareness only. (intro, conf:H)
[nit] OpenSearchSink::retry_on_open (core/connectors/sinks/opensearch_sink/src/lib.rs:263-290, call sites :280,:338) - a non-transient open()-time failure returns the raw `PermanentHttpError`/`Connection` error directly instead of wrapping it in `Error::InitError`, unlike every other sink's connectivity-failure convention (`postgres_sink:171`, `mongodb_sink:179-214`, `http_sink:347`, `doris_sink:253`, `redshift_sink:146`, `surrealdb_sink:362`, `clickhouse_sink:176` all wrap in `InitError`). Confirmed intentional and unit-tested (`given_permanent_open_failure_should_not_retry`, lib.rs:2259-2274, asserts `PermanentHttpError` survives untouched). Behaviorally inert: `manager/sink.rs` flips `ConnectorStatus::Error` on any `Err` from `open()` regardless of variant (sink.rs:106), so this is pure error-message classification, not a functional gap. (intro, conf:M)
[simplify] sanitize_url_for_log (core/connectors/sinks/opensearch_sink/src/lib.rs:1296-1308) - the username/password-stripping branches are dead in practice: its only call site (`open()`, lib.rs:685) always passes `normalized_url`, and `normalize_url` (lib.rs:1221) already rejects any URL with embedded credentials before it can reach here. Already documented as intentional defense-in-depth via the comment above the fn. Simpler: could be removed or reduced to a `to_string()` alias since the branch never fires today, but doing so would remove the safety net for any future caller that bypasses `normalize_url`. Not worth changing. Saves: ~6 lines, but conf:L this is worth doing. (intro, conf:L)
## Verified correct (no finding, checked because charter mandates it)
- `SecretString` on `password`; no `Serialize` on `OpenSearchSinkConfig` (comment at lib.rs:74 explains why — matches convention, e.g. no sink derives `Serialize` on its config type).
- Custom `Debug` impls on both `OpenSearchSink` and `ResolvedOpenSearchSinkConfig` redact the password and URL credentials (lib.rs:109-165) — needed because `opensearch::auth::Credentials::Basic` derives an unredacted `Debug`; verified via `given_debug_formatted_*_should_redact_password` tests (lib.rs:2988-3040).
- `consume(&self, ...)` — no interior `&mut self`, no locks, no `tokio::spawn`/`block_on`.
- `BTreeMap<HeaderKey, HeaderValue>` headers handled via `headers_to_json` (lib.rs:1113), and unlike the pre-existing gap the PR documents in `elasticsearch_sink`/`meilisearch_sink` (headers silently dropped because `serde_json` can't serialize a `HeaderKey`-keyed map), this sink actually converts and indexes them — a real fix, not a regression.
- `try_to_bytes`/no-clone JSON path: `prepare_document` uses `mem::replace` to take `message.payload` by value (lib.rs:408), then `owned_value_to_serde_json(&value)` (a reference-taking conversion helper already used identically by `elasticsearch_sink:195`) — matches SDK convention, not a clone-then-replace anti-pattern.
- `Vec::with_capacity(messages_count)` at lib.rs:741.
- `[lib] crate-type = ["cdylib", "lib"]` present in Cargo.toml.
- `verbose_logging` mirrors `debug!`→`info!` correctly in both `consume()` receive-log and indexed-count log (lib.rs:709-733, 780-790).
- 3-attempt cap (`DEFAULT_MAX_RETRIES = 3`), exponential backoff + jitter shared via SDK `retry` module, matches sibling sinks.
- Idempotency: dedup-on-write via deterministic `_id` (hash of stream/topic/partition/offset/message-id, or user `document_id_field`), verified end-to-end by `connectors::opensearch::opensearch_sink` (resends under existing `order_id`, asserts count unchanged) — correctly distinguished from the ES-auto-`_id` non-idempotent case the PR body calls out.
- No `unwrap()`/`expect()` on external-I/O `Result`s in production code (verified via grep over lib.rs:1-1350; all `unwrap()`/`expect()` usage is in `#[cfg(test)] mod tests`).
- Config forward-compat: every `OpenSearchSinkConfig` field is `Option<T>` with defaults resolved in `From<OpenSearchSinkConfig> for ResolvedOpenSearchSinkConfig` (lib.rs:167-203), conflict (retry_delay > max_retry_delay) fixed with `warn!` + swap in `new()`-time conversion, never in `consume()`.
- `Permanent*` vs transient classification: `is_transient_error`/`is_transient_client_error`/`map_status_error`/`map_client_error` all correctly route through `Error::HttpRequestFailed` (transient) vs `Error::PermanentHttpError` (permanent), including the per-item `_bulk` response classification (429/5xx retryable, everything else including mapping errors permanent) — unit-tested extensively (lib.rs:1971-2194).
- Drop accounting: `invalid_records`/`preparation_errors` both roll into `errors_count` (lib.rs:765-770); a single bad message in a batch does not lose the rest (comment at lib.rs:754 states this explicitly and `given_permanently_failing_chunk_should_not_abandon_later_chunks` proves the chunk-boundary case at the unit level, integration test `opensearch_sink_failures.rs` proves it against a live server).
- FFI/runtime consume()-swallow gap and `elasticsearch_sink`/`meilisearch_sink` header-drop gap: both correctly classified in the PR body as `pre-untouched` (verified against current `master` citations, e.g. `sink.rs:740-748`, `elasticsearch_sink/src/lib.rs:205-219`) — out of scope for this PR, not introduced by it.
- No `#[repr(C)]`, `Schema` variant, consumer-group, or `state.rs` changes — no STOP-tripwire scope.
- `.config/nextest.toml` OpenSearch test-group addition mirrors the existing Elasticsearch group exactly (serialized, shared reusable container by fixed name) — consistent pattern, not scope creep.
## Simplifications: none (beyond the low-priority dead-branch note above, not worth acting on)
## Verdict: APPROVEruntime# PR 3873 review — runtime lane (FFI host, `runtime/src/`)
Scope check first: `git diff --stat` / `files.txt` touch zero files under
`core/connectors/runtime/src/`. Only `runtime/example_config/connectors/opensearch_sink.toml`
(new example, no code). This PR is a new sink plugin
(`core/connectors/sinks/opensearch_sink/`) plus its integration tests, README/registry
entries, and workspace/nextest/bump-version plumbing. Nothing in my ownership
(FFI dispatch, plugin lifecycle, `plugin_id`, `LogCallback`, container `Arc`,
`DashMap`, state atomic-rename, metrics/config plumbing) is modified.
## Verified against runtime source (not changed by this PR)
- PR body claims: `core/connectors/runtime/src/sink.rs:740-748` invokes the FFI
`consume` callback as a bare statement, discarding its `i32`, so a plugin
failure never reaches `ConnectorStatus`/`last_error`/`/stats`, and combined
with `AutoCommit::When(PollingMessages)` (`sink.rs:522`) the offset is
already committed by the time `consume` runs. Read `sink.rs:480-756`
directly: confirmed accurate on both counts — `process_messages` calls
`(consume)(plugin_id, ...)` at `sink.rs:740` with no binding, and
`setup_sink_consumers` builds the consumer with
`.auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages))` at
`sink.rs:522`. Pre-existing, untouched by this diff, correctly scoped out
of the PR rather than silently relied on. (pre-untouched, conf:H)
- `opensearch_sink/README.md`'s "Delivery Semantics" section states the same
gap and additionally claims it verified the failure mode against a live
server (mapping-conflict batch logged as `PermanentHttpError`, connector
stayed `Running`). Consistent with the runtime code read above — a
plugin-level `Err` from `consume()` has no path back into the host's status
machinery, so "stays Running" is the only possible outcome given current
`sink.rs`. No fix expected or requested here; this is the same repo-wide
limitation the PR states affects every sink.
## Connector-mandatory checks, in scope for this plugin
- `SecretString` on `password` (`lib.rs:591`), no `Serialize` derive on
`OpenSearchSinkConfig`/`ResolvedOpenSearchSinkConfig` — both have hand
comments explaining why (`lib.rs:584-585`, `lib.rs:651`). `Debug` impls are
hand-written to redact the URL and never print `password` in plaintext via
a derived path (`lib.rs:619-630`, `652-675`). Matches `postgres_sink`/
`http_sink` convention.
- `consume(&self, ...)` (`lib.rs:1211`) — immutable receiver, matches the
`Sink` trait and every other sink; no interior mutability via `Mutex`, only
`AtomicU64` counters, so no lock-across-`.await` risk.
- No `tokio::spawn` / `block_on` anywhere in the plugin (grepped the whole
file) — retries use `tokio::time::sleep`/`tokio::time::timeout` inline in
the `&self` async fns, which is how the runtime expects a plugin to behave
under its own executor.
- `[lib] crate-type = ["cdylib", "lib"]` present (`Cargo.toml:221-222`).
- `BTreeMap<HeaderKey, HeaderValue>` used for headers (`lib.rs:1623`,
matches `ConsumedMessage.headers` shape), converted explicitly to JSON
since `HeaderKey`/`HeaderValue` aren't `Serialize` as map keys — same
workaround shape the PR body says `elasticsearch_sink`/`meilisearch_sink`
skip (verified true against master's `serialize_headers` remark; this sink
does the conversion by hand rather than reusing that helper, but does not
drop headers, which is the actual bug in the other two).
- `Vec::with_capacity(messages_count)` at `lib.rs:1251`.
- FFI panic-safety: `documents_at` (`lib.rs:1438-1446`) explicitly drops
out-of-range positions from a server-echoed `items` array instead of
indexing/panicking, with a comment citing exactly the right reason — a
panic across the `extern "C"` plugin boundary aborts the whole connectors
runtime process. Backed by a dedicated regression test
(`given_out_of_range_positions_should_drop_them_instead_of_panicking`,
`lib.rs:2731-2741`). This is the one place in the plugin that actually
touches a runtime-host invariant, and it's handled correctly.
- Logging: `info!`/`debug!` pair gated on `verbose_logging`
(`lib.rs:1219-1243`, `1290-1300`) duplicates the exact same format string
under both branches. Checked whether this is a plugin-specific tell:
`mongodb_sink/src/lib.rs:263-273` and `redshift_sink/src/lib.rs:417-427` do
the identical `if self.verbose { info!(...) } else { debug!(...) }`
duplication already on master. Established repo-wide pattern, not
introduced here — not flagging as a new simplify target for this PR alone.
(pre-untouched convention, closest exemplar: `mongodb_sink`)
- Example config path has no platform extension
(`path = "target/release/libiggy_connector_opensearch_sink"`,
`example_config/connectors/opensearch_sink.toml:9`) — matches
`delta_sink.toml`/`clickhouse_sink.toml`/`iceberg_sink.toml`, the runtime
resolves the `.so`/`.dylib`/`.dll` suffix itself. Correct.
## STOP tripwires
None triggered: no `#[repr(C)]` touched, no `Schema` variant change, no
transient↔`Permanent*` promotion in runtime code (the plugin's own
classification in `is_transient_error`/`map_client_error`/`map_status_error`
is plugin-local, not a runtime reclassification), no consumer-group default
rename (`opensearch_sink_connector` is this plugin's own new default, not a
rename of an existing one), no plugin path resolution change, no
`iggy_connector_sdk` trait change.
## Findings
None in the runtime lane — no `runtime/src/` files changed, and everything
this plugin does that touches a runtime-owned invariant (FFI boundary panic
safety, crate-type, `&self` receiver, no blocking calls, config
forward-compat via `Option<T>`) is handled correctly.
Simplifications: none (no runtime-owned code changed; the one duplication
noted above — verbose/info vs debug logging — is a pre-existing repo-wide
pattern, not something this PR should be asked to fix unilaterally).
Verdict: APPROVE - no runtime/src/ changes; the plugin's own runtime-facing
behavior (FFI panic safety, crate-type, blocking-call discipline, config
forward-compat) is correct, and its two disclosed pre-existing runtime gaps
(consume() return value discarded, offset auto-committed before consume())
were independently verified against current `sink.rs` and are accurately
scoped out of this PR rather than papered over.sdk# SDK role review — PR #3873 (opensearch_sink) @ f3d9688d
Scope note: no files under `core/connectors/sdk/src/` are touched by this diff.
Review covers whether the new plugin correctly *uses* the existing SDK
contract (`Sink` trait, `Error` enum, `retry.rs`, `convert.rs`, `sink_connector!`
macro) rather than a contract change. Closest exemplar per PR body:
`elasticsearch_sink` (also compared against `http_sink`, `meilisearch_sink` for
config-field conventions).
## Findings
[nit] opensearch_sink::OpenSearchSinkConfig (core/connectors/sinks/opensearch_sink/src/lib.rs:74-75) - comment "the only in-tree helper for a `SecretString` field writes the credential in plaintext" is factually wrong: `iggy_common::serde_secret::serialize_optional_redacted` (core/common/src/utils/serde_secret.rs:85-93) exists precisely for a struct that must serialize but must not expose the secret, and predates this PR (added before f3d9688d; the module's doc header even warns against exactly this kind of inverted claim after a prior fix, commit f900e7abe "fix the inverted serde_secret redaction claim", #3803). Omitting `Serialize` entirely is a defensible choice on its own merits (elasticsearch_sink instead derives `Serialize` with `serialize_optional_secret`, i.e. plaintext-on-purpose for the control API; clickhouse_sink/delta_sink/doris_sink/redshift_sink also omit `Serialize` like this PR does — split convention, not a deviation), but the justification comment misstates the available tools and could steer a future maintainer wrong. Fix: reword to something like "No `Serialize`: nothing in-tree serializes this type, and `serialize_optional_redacted` is unused here because dropping the impl entirely is simpler than remembering to route password through it." (intro, conf:H)
Simplifications: none. The bulk-retry state machine (`BulkOutcome`/`BulkAttempt`/`index_chunk`) looks large but every branch is exercised by a named test (`given_*`) and each piece (partial-chunk accounting, transient vs permanent classification, first-failure retention) maps to a real OpenSearch `_bulk` behavior called out in the PR body; collapsing it would drop correctness, not just lines. `AtomicU64` counters instead of `elasticsearch_sink`'s `Mutex<State>` is already a simplification (no lock, no lock-across-await concern).
## SDK-contract compliance checklist
- `Sink` trait (core/connectors/sdk/src/lib.rs:133-147): `open(&mut self)`, `consume(&self, ...)`, `close(&mut self)` signatures match exactly; `consume` takes `&self` as required (no interior `Mutex` needed, uses `AtomicU64`).
- `Send + Sync` on `Sink`: satisfied automatically — every field (`AtomicU64`, `Option<OpenSearch>`, `ResolvedOpenSearchSinkConfig` of `String`/`bool`/`Duration`/`Option<Value>`/`Option<Refresh>`/`SecretString`) is `Send + Sync`. No manual impl needed or attempted.
- `sink_connector!(OpenSearchSink)` (lib.rs:53) — plain macro invocation, matches the macro's expansion in core/connectors/sdk/src/sink.rs:239-318 (duplicate-ID guard, `0`/`-1`/`1` FFI codes, `INSTANCES` global, `version()` static) — all handled by the macro, plugin does nothing FFI-level itself. `crate-type = ["cdylib", "lib"]` set correctly in Cargo.toml:32-33.
- `Error` variants used (`InvalidConfigValue`, `Connection`, `InitError`, `HttpRequestFailed`, `InvalidRecordValue`, `PermanentHttpError`, `Serialization`) — all pre-existing variants in core/connectors/sdk/src/lib.rs:389+, used per their documented retry semantics: `HttpRequestFailed` = transient (checked by `is_transient_error`/`retry_on_open`), `PermanentHttpError` = non-retryable data/schema issue (matches the variant's own doc comment about circuit breakers not tripping on bad data). No new `Error` variant added — correct, none of these needed distinct handling.
- `retry.rs` helpers (`exponential_backoff`, `jitter`, `parse_duration`, `is_transient_status`) used with correct signatures. `is_transient_status(status: reqwest::StatusCode)` called with `opensearch::http::StatusCode` — verified same type: `opensearch::http::mod.rs:41` is `pub use reqwest::StatusCode;`, and `reqwest::StatusCode` is itself `pub use http::{StatusCode, ...}` (reqwest-0.12.28/src/lib.rs:279) — no type mismatch, compiles.
- `max_retries` semantics ("N retries after the initial attempt", not "N total attempts") differs from `retry.rs::HttpRetryMiddleware`'s self-documented "total attempt count" convention, but matches the established sink-level convention: http_sink's own doc comment (`batch_length * (max_retries + 1) * ...`, http_sink/src/lib.rs:1191) uses the same "+1 for initial attempt" semantics. Not a deviation — two different sub-conventions coexist pre-PR (middleware-level vs bespoke-retry-loop sinks), and this plugin correctly follows the bespoke-loop one it actually implements.
- `convert::owned_value_to_serde_json` (core/connectors/sdk/src/convert.rs:27) called with `&simd_json::OwnedValue` from a destructured `Payload::Json` — signature matches, same usage pattern as elasticsearch_sink.
- `Payload` match in `prepare_document` (lib.rs:410-423) handles `Json`/`Raw`/`Text` explicitly, `Proto`/`FlatBuffer`/`Avro` fall to an explicit `Error::InvalidRecordValue` (not a silent drop) — acceptable scope limit, no SDK contract violation.
- `SecretString` on `password` (config field, lib.rs:81) with `ExposeSecret` used only inside `create_client` (Basic auth header) — never logged, never in a `Display`/`Debug` path (manual `Debug` impls on both `OpenSearchSink` and `ResolvedOpenSearchSinkConfig` explicitly avoid deriving through the client/password, with an accurate comment explaining why on lib.rs:105-108 and lib.rs:141).
- No `tokio::spawn` / `block_on` / `std::sync::Mutex` anywhere in the plugin (verified by grep across lib.rs); all `.unwrap()`/`.expect()` calls are confined to `#[cfg(test)] mod tests` (first non-test line is 1351, all unwrap/expect hits are ≥1405).
- BDD test naming: all unit tests are `#[test] fn given_..._should_...` — consistent 3-part naming, matches CLAUDE.md convention.
- Config forward-compat: every `OpenSearchSinkConfig` field is `Option<T>` except the two required ones (`url`, `index`), consistent with `#[serde(default)]`-equivalent forward compatibility (no `#[serde(default)]` needed since there's no `Default` derive requirement beyond what's already `Option`-wrapped; struct does derive `Default`).
- `Refresh` config field: `Option<Refresh>` where `opensearch::params::Refresh` derives `Deserialize` with `#[serde(rename = "true"/"false"/"wait_for")]` (opensearch-2.4.0/src/params.rs:152-159) — `refresh = "false"` in config.toml/README/example_config all deserialize correctly.
- Workspace `Cargo.toml` diff (lines 89-101 of diff.patch) only adds the new workspace member and the `opensearch = "2.4.0"` dependency — no version bump to `iggy_connector_sdk` or any STOP-tripwire crate in this PR's actual diff (the version-bump lines visible via a stale `git diff HEAD~1` are from the unrelated `Merge branch 'master'` commit at HEAD, not from this PR's content — confirmed against diff.patch directly).
## Verdict: APPROVE - clean SDK-contract usage; one nit-level comment inaccuracy (not a correctness or security issue — the credential itself is still never leaked, since `Serialize` is correctly omitted either way)testing# Testing review - PR #3873 (opensearch_sink)
[nit] core/integration/tests/connectors/opensearch/opensearch_sink.rs::given_json_messages_when_sink_consumes_should_index_documents_and_upsert_by_natural_key (opensearch_sink.rs:33) - integration test uses unit-test BDD `given_when_should` naming instead of `connector-testing` SKILL.md's mandated declarative `<subject>_<action>_<observation>` for integration tests ("never given_should there"). Same pattern in opensearch_sink_failures.rs:132 (`given_missing_index_and_mapping_conflict_should_isolate_failures_from_healthy_sibling`). Fix: rename to e.g. `json_messages_sink_indexes_and_upserts_by_natural_key` / `missing_index_and_mapping_conflict_isolate_from_healthy_sibling`, matching `postgres_sink.rs::json_messages_sink_stores_as_bytea` or `elasticsearch_sink.rs::elasticsearch_sink_stores_json_messages`. Low priority: `meilisearch_sink.rs::given_json_messages_when_sink_consumes_should_index_documents` (near-identical name/shape to this PR's test) and `doris_sink.rs`, `quickwit_sink.rs`, `runtime/benchmark.rs`, `runtime/http_state.rs` all already use the same `given_should` style in integration tests, so this is repo-wide drift the PR inherited rather than introduced. (origin: pre-surfaced, conf: H)
Simplifications: none. Unit test file (`core/connectors/sinks/opensearch_sink/src/lib.rs:1351-3085`, ~1730 lines) is large but every test targets a distinct branch with a comment justifying why (e.g. the four separate `given_debug_formatted_*_should_redact_password` tests each catch a different leak surface: config Debug, sink Debug with no client, sink Debug with a real opened `OpenSearch` client - which has its own un-redacted `Debug` derive down through `Credentials::Basic` - and URL-embedded credentials). No duplicate/mocked-should-be-real-infra tests, no fixed sleeps outside legitimate poll loops or backoff-under-test.
## What was verified
- **BDD naming, unit tests**: consistent `given_X_should_Y` / `given_X_when_Y_should_Z` throughout `opensearch_sink/src/lib.rs::tests` (lib.rs:1408-3084). No `test_foo`/`does_bar` names.
- **Config helper**: `base_config()` (lib.rs:1386) matches the naming convention of the closest exemplar `meilisearch_sink::tests::base_config` (not `test_config`, which is fine - `meilisearch_sink`, `mongodb_sink` (`given_default_config`), and `http_sink` (`given_default_config`) all name this helper differently; no single fixed name is enforced repo-wide).
- **`#[tokio::test]` in `src/lib.rs`**: used for all async unit tests (lib.rs:2241 onward) rather than local `Runtime::new()`. This looks like a skill-guideline deviation at first read, but the closest exemplars for wiremock-backed sink tests, `http_sink/src/lib.rs` and `mongodb_sink/src/lib.rs`, both already use `#[tokio::test]` directly in `src/lib.rs`. The `Runtime::new()` guidance in the skill is written for source state round-trip tests (`random_source`-style); sinks with real async HTTP mocking already establish `#[tokio::test]` as the norm. Not a defect. (origin: pre-surfaced, conf: H)
- **Sink pure-logic coverage** (mandatory checklist item): config defaults + fallback/clamping (lib.rs:1409-1442), all three `Payload` variants incl. non-object JSON wrapping and raw-non-JSON base64 fallback (lib.rs:1494-1715), header encoding incl. the dynamic-mapping-collision regression test (lib.rs:1648-1684), `_bulk` response/query building and parsing incl. malformed/short/missing-items edge cases (lib.rs:1923-2231), and transient-vs-permanent classification for both client errors and per-item bulk statuses (lib.rs:2845-2856, 1170-1195, plus the wiremock-driven retry tests at lib.rs:2394-2653). This is the most thorough sink test suite of the set I compared against (`elasticsearch_sink`, `meilisearch_sink`, `mongodb_sink`, `http_sink`).
- **No four canonical source-state tests required**: this is a sink, not a source - correctly out of scope.
- **Integration layout**: `core/integration/tests/connectors/opensearch/{mod.rs,opensearch_sink.rs,opensearch_sink_failures.rs,sink.toml,failure_states.toml,failure_states/*.toml}` plus `fixtures/opensearch/{mod.rs,container.rs,sink.rs,failure.rs}` mirrors `postgres`/`elasticsearch` exactly: `TestFixture` impl, `ConfigEnv` env-injection (all three documented forms: plugin-config leaf, indexed `streams_N_*`, top-level `path`), `iggy-test-opensearch` fixed name + `ReuseDirective::Always` matching the `elasticsearch`/`doris` reuse pattern (skill's "as of now only elasticsearch and doris share a container" note is now stale text, not a PR defect), polling helpers (`wait_for_document_count`, `wait_for_status`) instead of fixed sleeps, `iggy-test-` container prefix honored.
- **`.config/nextest.toml`**: new `[test-groups.opensearch]` + `max-threads = 1` override correctly serializes the reused-container test group, matching the `elasticsearch`/`doris` blocks immediately above it (nextest.toml:56-69).
- **Two integration tests cover exactly what's claimed**: happy path (natural-key upsert idempotency proven end-to-end by resending `order_id: A-1` at a new offset, headers round-trip, both `Payload::Raw` branches via a second stream/topic on the `raw` schema) and a real failure-isolation test (missing-index `open()` failure vs. a live `mapper_parsing_exception` mid-`consume()`, three-sink isolation, chunk-boundary survival cross-checked against the unit-level `given_permanently_failing_chunk_should_not_abandon_later_chunks`). The module doc in `opensearch_sink_failures.rs:18-65` explicitly traces the runtime code path (`sink.rs::process_messages` discarding the FFI `consume` return) backing its "not visible anywhere" claim rather than asserting it blind.
- **Doc/config sync**: `opensearch_sink/README.md` config table, `opensearch_sink/config.toml`, and `runtime/example_config/connectors/opensearch_sink.toml` all list the identical `[plugin_config]` field set (defaults match `OpenSearchSinkConfig`'s `unwrap_or` values in lib.rs). `core/connectors/README.md` and `core/connectors/sinks/README.md` both gained the new sink row in alphabetical position. No `max_connections` field applies here (sink has no connection pool); retry knobs (`max_retries`, `retry_delay`, `max_retry_delay`, `max_open_retries`) and `verbose_logging` are all present in both TOMLs and documented.
- **Investigated and ruled out**: an apparent "PR deletes `rabbitmq_sink`" signal from a raw `git diff master -- ...` was a false positive from the local `master` ref having advanced past this PR's actual merge-base (post-merge commits landed on master after this branch's base). The actual `diff.patch` supplied for review contains zero references to `rabbitmq_sink`; confirmed via direct grep. Not a real finding.
Verdict: APPROVEValidation record
|
Contributor
|
@slbotbm Please see above. |
spetz
requested changes
Sep 10, 2026
| - `<TOPIC>`: `<TARGET>` lowercased, chars outside `[a-z0-9-]` replaced by `-`, repeats collapsed, trimmed, max 40 chars (`PR3123` -> `pr3123`, `origin/master..HEAD` -> `origin-master-head`). Empty -> `date +%s`. | ||
| - `<DIR>` = `<session scratchpad dir from your system prompt>/review-<TOPIC>`. `mkdir -p` it. | ||
| - PR: `gh pr view <PR> --json title,body,headRefOid > <DIR>/pr.json`, `gh pr diff <PR> > <DIR>/diff.patch`, `gh pr diff <PR> --name-only > <DIR>/files.txt`. `<SHORTCOMMIT>` = first 8 of `headRefOid`. | ||
| - Ref range or bare branch: `git diff $(git merge-base origin/master HEAD)..HEAD > <DIR>/diff.patch`, same with `--name-only`, `<SHORTCOMMIT>` = `git rev-parse --short=8 HEAD`. No `pr.json` on this path. |
Contributor
There was a problem hiding this comment.
This command ignores <TARGET> and always compares the current checkout against origin/master. Supplying another branch or revision range therefore reviews the wrong changes. Construct the diff from the supplied target and verify the checkout against that target’s head.
| Simplify: dead payload arms, skip branches that never fire, single-variant enums, near-duplicate path/config helpers. | ||
| - **runtime**: Connectors runtime + FFI host engineer. Owns `runtime/src/`. | ||
| Focus: FFI pointer lifetimes (call-duration only), `plugin_id` monotonic never reused, `LogCallback` static, container `Arc` outlives tasks (no unload mid-call), `DashMap` keyed by TOML key + status counters on transitions. | ||
| Loops: sink (autocommit timing, `consumer.next()` Err auto-committed drop, decode/transform drop-and-continue, postcard encode, nonzero-return, flush on offset gap, failed batch still emits); source (flume handoff, close-then-`cleanup_sender`-then-drain, state save after Iggy send, transform `Err` never flips status); state atomic-rename protocol; `restart_guard.try_lock()`. |
Contributor
There was a problem hiding this comment.
“Transform Err never flips status” contradicts the runtime and line 67. source_forwarding_loop rejects batches with processing errors, calls context.sources.set_error, and returns NACK. Update this instruction to match that behavior.
| Sink focus: lifecycle (`open` connectivity fail-fast `InitError`, `close` flush + `.take()` + final stats), config `Option` defaults + conflict fix in `new()` + `warn!` (never error, never validate in `consume()`), durations `Option<String>` humantime (no `humantime_serde`). | ||
| Sink payloads/errors: dispatch Json/Raw/Text minimum (`InvalidPayloadType` or base64 fallback), `try_to_bytes` + `mem::replace` + `with_capacity`, header `is_empty` check + binary-base64 vs text split in `BTreeMap`, `InvalidRecordValue` skip+log, `WriteFailure` caller-decides, `CatalogCommitError` txn consumed and not idempotent. | ||
| Sink retry/idempotency: 3-attempt cap, `Retry-After` honored, per-status custom strategy, SQLSTATE-style transient map; dedup-on-write via message `id` (mongo composite `_id` reference; ES auto-`_id` is NOT idempotent); `last_err` pattern (process all batches, return last); batch `chunks` default 100, no cross-`consume` buffering, no config mutation in `consume()`. | ||
| Source focus: `poll()` `Err` only logs, loop continues, status never flips (runtime sets `Error` on transform/encode/send/save failure only); always return state incl. empty polls; poll-return-to-save crash window means at-least-once; sleep-first or idle spins CPU; single `poll()` task. |
Contributor
There was a problem hiding this comment.
Returning state on every poll does not guarantee at-least-once delivery. Advancing a cursor before delivery can skip messages after a failed batch. Add checks for Source::on_batch_result: stage cursor changes and destructive operations during polling, apply them only after ACK, and discard staged changes on NACK.
github-actions
Bot
added
S-waiting-on-author
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds a
connector-reviewskill that uses team-review's structure with the concerns outlined in the.claude/connector-*skills.