mattp5657 opened a new pull request, #3873: URL: https://github.com/apache/iggy/pull/3873
## Which issue does this PR address? Closes #3504 ## Rationale Iggy connectors ship sinks for several external systems but not for OpenSearch, a widely used search/analytics backend. This adds one, modeled on the existing `elasticsearch_sink` shape but closing a retry gap that sink still has (see Known trade-offs). ## What changed? Adds `core/connectors/sinks/opensearch_sink/`, following the sink lifecycle end to end: - **`open()`:** validates config (URL shape, credential pairing, `document_id_field` constraints), then `retry_on_open`-wraps a cluster health check, an index-exists check, and, when `create_index_if_not_exists` (default `true`), index creation with an optional custom mapping. Capped at `max_open_retries` with exponential backoff and jitter (shared `retry` helpers). - **`consume()`:** batches incoming messages into `batch_size` chunks, builds an `iggy_*`-enriched document per message (hashed or field-derived `_id`), and hands each chunk to `index_chunk`. - **`index_chunk()`:** POSTs a `_bulk` request and loops up to `max_retries` with the same backoff. `_bulk` answers `200` even when individual documents fail, so each response is parsed per item rather than trusting the top-level status: permanent failures (4xx, e.g. a mapping conflict) are recorded immediately, while transient ones (429/5xx) shrink the pending set to just the rejected documents and get resent, so a partial rejection under load doesn't re-index or lose the rest of the chunk. Counts from earlier attempts are merged into the outcome so a later failure doesn't erase already-indexed documents from the tally. - **`close()`:** drops the client, no special teardown. **Integration tests** (`core/integration/tests/connectors/opensearch/`) run against a real container (`testcontainers-modules`, reused across tests via `ReuseDirective::Always`, per-test-unique index names for isolation), covering the happy path plus a static mapping conflict, a missing index with index-creation disabled, and confirming a failing chunk doesn't block chunks queued behind it. **Credentials:** HTTP Basic auth only (`username`/`password`, both-or-neither validated at config time); `password` is a `SecretString`, never logged or serialized. AWS SigV4 (AWS-managed OpenSearch / Serverless) is not supported. **Known trade-offs**, deliberately out of scope here, verified against current `master`: - `elasticsearch_sink` has the same gap this PR fixes for OpenSearch: `bulk_index_documents` (`elasticsearch_sink/src/lib.rs:205-219`) tallies per-item `_bulk` failures into `errors_count` but never retries the transient subset (e.g. 429 `es_rejected_execution_exception`). Worth a follow-up issue rather than folding into this PR. - The connectors runtime discards a sink's `consume()` return value: `core/connectors/runtime/src/sink.rs:740-748` invokes the FFI `consume` callback as a bare statement, never binding its `i32` result, so `process_messages` always returns `Ok`. Combined with offsets auto-committing at poll time (`sink.rs:522`), a plugin-level failure never reaches connector status, `last_error`, or `/stats`, and the batch is never redelivered. Pre-existing, repo-wide, affects every sink. - `meilisearch_sink` (`lib.rs:451-454`) and `elasticsearch_sink` (`lib.rs:319-328`) both silently drop `iggy_headers`/`_iggy_headers`: `BTreeMap<HeaderKey, HeaderValue>` can't serialize as a JSON object (`serde_json` requires string keys), and both sinks swallow that error via `if let Ok(...)` instead of surfacing it. `core/common` even ships a `serialize_headers` workaround for this exact case that neither sink uses. - Single-node transport: like `elasticsearch_source`, the client is built on `SingleNodeConnectionPool` (`lib.rs:208`) with no cluster sniffing or multi-node failover. A dead configured node fails every request rather than routing around it. ## Local Execution - Passed - Pre-commit hooks ran ## AI Usage 1. **Tools:** Claude Code (Sonnet 5). 2. Used for scaffolding and iteration. 3. **Verification:** Full local compilation, extensive integration testing against a live container. 4. Reviewed and can explain every line if asked. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
