mattp5657 commented on code in PR #3873: URL: https://github.com/apache/iggy/pull/3873#discussion_r4265184353
########## core/connectors/sinks/opensearch_sink/src/lib.rs: ########## @@ -0,0 +1,3565 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use async_trait::async_trait; +use base64::{Engine as _, engine::general_purpose}; +use bytes::{BufMut, Bytes, BytesMut}; +use iggy_common::{HeaderKey, HeaderValue, IggyTimestamp, calculate_256}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, + convert::owned_value_to_serde_json, + retry::{is_transient_status, parse_duration, parse_retry_after, retry_backoff}, + sink_connector, +}; +use opensearch::{ + BulkParts, OpenSearch, + auth::Credentials, + cluster::ClusterHealthParts, + http::{ + StatusCode, + headers::RETRY_AFTER, + transport::{SingleNodeConnectionPool, TransportBuilder}, + }, + indices::{IndicesCreateParts, IndicesExistsParts}, + params::Refresh, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; +use std::{ + collections::BTreeMap, + future::Future, + net::IpAddr, + sync::atomic::{AtomicBool, AtomicU64, Ordering}, + time::Duration, +}; +use tokio::time::sleep; +use tracing::{debug, error, info, warn}; +use url::Url; + +sink_connector!(OpenSearchSink); + +const DEFAULT_CREATE_INDEX_IF_NOT_EXISTS: bool = true; +const DEFAULT_INCLUDE_METADATA: bool = true; +const DEFAULT_BATCH_SIZE: usize = 1000; +const DEFAULT_TIMEOUT: &str = "30s"; +const DEFAULT_RETRY_DELAY: &str = "500ms"; +const DEFAULT_MAX_RETRY_DELAY: &str = "5s"; +const DEFAULT_MAX_RETRIES: u32 = 3; +const DEFAULT_MAX_OPEN_RETRIES: u32 = 5; +const ENCODING_BASE64: &str = "base64"; +const ENCODING_UTF8: &str = "utf8"; +const GENERATED_ID_PREFIX: &str = "iggy_"; +const INDEX_ALREADY_EXISTS_ERROR: &str = "resource_already_exists_exception"; + +/// OpenSearch rejects `_id` values longer than 512 bytes. Payload-supplied IDs +/// are checked before the batch is built because that rejection fails the whole +/// `_bulk` call with an `action_request_validation_exception` rather than the +/// one item: a single oversized ID would cost every document in its chunk. +const MAX_DOCUMENT_ID_BYTES: usize = 512; + +// Connector-owned rather than exposing `opensearch::params::Refresh` directly: +// keeps this config's wire format under this connector's own control instead +// of an internal detail of the opensearch crate. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] +pub enum RefreshPolicy { + #[serde(rename = "true")] + True, + #[serde(rename = "false")] + False, + #[serde(rename = "wait_for")] + WaitFor, +} + +impl From<RefreshPolicy> for Refresh { + fn from(policy: RefreshPolicy) -> Self { + match policy { + RefreshPolicy::True => Refresh::True, + RefreshPolicy::False => Refresh::False, + RefreshPolicy::WaitFor => Refresh::WaitFor, + } + } +} + +// No `Serialize`: nothing serializes this type, and `serde_secret::serialize_secret` +// writes a `SecretString` credential in plaintext. +#[derive(Debug, Default, Deserialize)] +pub struct OpenSearchSinkConfig { + pub url: String, + pub index: String, + pub username: Option<String>, + pub password: Option<SecretString>, + pub document_id_field: Option<String>, + pub create_index_if_not_exists: Option<bool>, + pub index_mapping: Option<Value>, + pub include_metadata: Option<bool>, + pub batch_size: Option<usize>, + pub timeout: Option<String>, + pub refresh: Option<RefreshPolicy>, + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, + pub max_retry_delay: Option<String>, + pub max_open_retries: Option<u32>, + pub verbose_logging: Option<bool>, +} + +pub struct OpenSearchSink { + id: u32, + config: ResolvedOpenSearchSinkConfig, + client: Option<OpenSearch>, + invocations_count: AtomicU64, + documents_indexed: AtomicU64, + errors_count: AtomicU64, + // Warns at most once per connector instance instead of once per record, + // so a persistently misconfigured document_id_field can't flood logs. + missing_document_id_field_warned: AtomicBool, + // A stream with an unsupported schema would otherwise warn at the message rate. + invalid_record_warned: AtomicBool, +} + +// `OpenSearch` derives `Debug` down through its `Transport`, and +// `opensearch::auth::Credentials::Basic` derives `Debug` on its raw +// `(String, String)` without redaction, so a derived `Debug` on this struct +// would print the Basic-auth password in plaintext once `client` is set. +impl std::fmt::Debug for OpenSearchSink { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("OpenSearchSink") + .field("id", &self.id) + .field("config", &self.config) + .field("client", &self.client.is_some()) + .field("invocations_count", &self.invocations_count) + .field("documents_indexed", &self.documents_indexed) + .field("errors_count", &self.errors_count) + .field( + "missing_document_id_field_warned", + &self.missing_document_id_field_warned, + ) + .field("invalid_record_warned", &self.invalid_record_warned) + .finish() + } +} + +struct ResolvedOpenSearchSinkConfig { + url: String, + index: String, + username: Option<String>, + password: Option<SecretString>, + document_id_field: Option<String>, + create_index_if_not_exists: bool, + index_mapping: Option<Value>, + include_metadata: bool, + batch_size: usize, + timeout: Duration, + refresh: Option<Refresh>, + max_retries: u32, + retry_delay: Duration, + max_retry_delay: Duration, + max_open_retries: u32, + verbose_logging: bool, +} + +// `derive(Debug)` would print `url` verbatim; `open()` isn't what redacts it. +impl std::fmt::Debug for ResolvedOpenSearchSinkConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ResolvedOpenSearchSinkConfig") + .field("url", &redact_url_credentials(&self.url)) + .field("index", &self.index) + .field("username", &self.username) + .field("password", &self.password) + .field("document_id_field", &self.document_id_field) + .field( + "create_index_if_not_exists", + &self.create_index_if_not_exists, + ) + .field("index_mapping", &self.index_mapping) + .field("include_metadata", &self.include_metadata) + .field("batch_size", &self.batch_size) + .field("timeout", &self.timeout) + .field("refresh", &self.refresh) + .field("max_retries", &self.max_retries) + .field("retry_delay", &self.retry_delay) + .field("max_retry_delay", &self.max_retry_delay) + .field("max_open_retries", &self.max_open_retries) + .field("verbose_logging", &self.verbose_logging) + .finish() + } +} + +impl ResolvedOpenSearchSinkConfig { + // Not a `From` impl: the swap warning below needs the connector ID, + // which only the caller (`OpenSearchSink::new`) has in scope. + fn resolve(id: u32, config: OpenSearchSinkConfig) -> Self { + let mut retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + let mut max_retry_delay = + parse_duration(config.max_retry_delay.as_deref(), DEFAULT_MAX_RETRY_DELAY); + if retry_delay > max_retry_delay { + warn!( + "OpenSearch sink connector ID: {id}: retry_delay ({:?}) exceeds max_retry_delay ({:?}). Swapping values.", + retry_delay, max_retry_delay + ); + std::mem::swap(&mut retry_delay, &mut max_retry_delay); + } + + Self { + url: config.url, + index: config.index.trim().to_string(), + username: trimmed_non_empty(config.username), + password: config + .password + .filter(|password| !is_blank_secret(password)), + document_id_field: trimmed_non_empty(config.document_id_field), + create_index_if_not_exists: config + .create_index_if_not_exists + .unwrap_or(DEFAULT_CREATE_INDEX_IF_NOT_EXISTS), + index_mapping: config.index_mapping, + include_metadata: config.include_metadata.unwrap_or(DEFAULT_INCLUDE_METADATA), + // Clamped to 1: `[T]::chunks(0)` panics, and `documents.chunks(batch_size)` would hit it on `batch_size = 0`. + batch_size: config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1), + // Clamped to 1s: a zero timeout expires every request at once, so open() never connects. + timeout: parse_duration(config.timeout.as_deref(), DEFAULT_TIMEOUT) + .max(Duration::from_secs(1)), + refresh: config.refresh.map(Into::into), + max_retries: config.max_retries.unwrap_or(DEFAULT_MAX_RETRIES), + retry_delay, + max_retry_delay, + max_open_retries: config.max_open_retries.unwrap_or(DEFAULT_MAX_OPEN_RETRIES), + verbose_logging: config.verbose_logging.unwrap_or(false), + } + } +} + +impl OpenSearchSink { + pub fn new(id: u32, config: OpenSearchSinkConfig) -> Self { + Self { + id, + config: ResolvedOpenSearchSinkConfig::resolve(id, config), + client: None, + invocations_count: AtomicU64::new(0), + documents_indexed: AtomicU64::new(0), + errors_count: AtomicU64::new(0), + missing_document_id_field_warned: AtomicBool::new(false), + invalid_record_warned: AtomicBool::new(false), + } + } + + fn validate_config(&self) -> Result<(), Error> { + if self.config.index.is_empty() { + return Err(Error::InvalidConfigValue( + "OpenSearch index cannot be empty".to_string(), + )); + } + + match (&self.config.username, &self.config.password) { + (Some(_), None) => Err(Error::InvalidConfigValue( + "OpenSearch username is set without a password".to_string(), + )), + (None, Some(_)) => Err(Error::InvalidConfigValue( + "OpenSearch password is set without a username".to_string(), + )), + _ => Ok(()), + } + } + + /// Takes the normalized URL rather than normalizing again, so + /// `normalize_url`'s warnings are emitted once per `open()`. + fn create_client(&self, normalized_url: &str) -> Result<OpenSearch, Error> { + warn_if_credentials_use_insecure_http( + self.id, + &self.config.url, + normalized_url, + self.config.password.is_some(), + ); + + let url = Url::parse(normalized_url) + .map_err(|error| Error::Connection(format!("Invalid OpenSearch URL: {error}")))?; + // The transport defaults to no timeout at all, and unlike the per-call + // `tokio::time::timeout` guards this also covers reading response bodies. + let mut builder = + TransportBuilder::new(SingleNodeConnectionPool::new(url)).timeout(self.config.timeout); + if let (Some(username), Some(password)) = (&self.config.username, &self.config.password) { + builder = builder.auth(Credentials::Basic( + username.to_owned(), + password.expose_secret().to_owned(), + )); + } + + let transport = builder.build().map_err(|error| { + Error::Connection(format!("Failed to build OpenSearch transport: {error}")) + })?; + Ok(OpenSearch::new(transport)) + } + + /// Retries a transiently failing `open()`-time call, so one blip does not + /// park the connector in `Error` until an operator restarts it. + async fn retry_on_open<T, F, Fut>(&self, operation: &str, call: F) -> Result<T, Error> + where + F: Fn() -> Fut, + Fut: Future<Output = Result<T, Error>>, + { + let mut retries = 0u32; + + loop { + let failure = match call().await { + Ok(value) => return Ok(value), + Err(error) if is_transient_error(&error) => error.to_string(), + Err(error) => return Err(error), + }; + + if retries >= self.config.max_open_retries { + return Err(Error::InitError(format!( + "OpenSearch {operation} failed after {} retries ({failure})", + self.config.max_open_retries + ))); + } + + retries += 1; + self.sleep_before_retry( + operation, + retries, + self.config.max_open_retries, + &failure, + None, + ) + .await; + } + } + + async fn check_connectivity(&self, client: &OpenSearch) -> Result<(), Error> { + self.retry_on_open("health check", || self.cluster_health(client)) + .await + } + + async fn cluster_health(&self, client: &OpenSearch) -> Result<(), Error> { + let response = tokio::time::timeout( + self.config.timeout, + client.cluster().health(ClusterHealthParts::None).send(), + ) + .await + .map_err(|_| { + Error::HttpRequestFailed(format!( + "OpenSearch health check timed out after {:?}", + self.config.timeout + )) + })? + .map_err(|error| map_client_error("health check", error))?; + + let status = response.status_code(); + if status.is_success() { + return Ok(()); + } + + // A 403 only means cluster:monitor/health is missing, not that the cluster is down. + if status == StatusCode::FORBIDDEN { + warn!( + "OpenSearch sink connector ID: {}: health check returned 403: the configured user lacks the cluster-scoped cluster:monitor/health privilege. Treating the cluster as reachable; grant that privilege to restore the check.", + self.id + ); + return Ok(()); + } + + let body = response.text().await.unwrap_or_default(); + Err(map_status_error("health check", status, &body)) + } + + async fn ensure_index_exists(&self, client: &OpenSearch) -> Result<(), Error> { + if self + .retry_on_open("index existence check", || self.index_exists(client)) + .await? + { + info!( + "OpenSearch sink connector ID: {}: index '{}' already exists", + self.id, self.config.index + ); + return Ok(()); + } + + if !self.config.create_index_if_not_exists { + return Err(Error::InitError(format!( + "OpenSearch index '{}' does not exist and create_index_if_not_exists=false", + self.config.index + ))); + } + + self.retry_on_open("index creation", || self.create_index(client)) + .await + } + + async fn index_exists(&self, client: &OpenSearch) -> Result<bool, Error> { + let response = client + .indices() + .exists(IndicesExistsParts::Index(&[&self.config.index])) + .send() + .await + .map_err(|error| map_client_error("index existence check", error))?; + + let status = response.status_code(); + if status.is_success() { + return Ok(true); + } + + // A missing index comes back as a 404 response, not a transport error, + // so anything else is a genuine failure worth surfacing. + if status == StatusCode::NOT_FOUND { + return Ok(false); + } + + let body = response.text().await.unwrap_or_default(); + Err(map_status_error("index existence check", status, &body)) + } + + async fn create_index(&self, client: &OpenSearch) -> Result<(), Error> { + info!( + "OpenSearch sink connector ID: {}: creating index '{}'", + self.id, self.config.index + ); + + let indices = client.indices(); + let request = indices.create(IndicesCreateParts::Index(&self.config.index)); + let response = if let Some(mapping) = &self.config.index_mapping { + request.body(mapping.clone()).send().await + } else { + request.send().await + } + .map_err(|error| map_client_error("index creation", error))?; + + let status = response.status_code(); + if status.is_success() { + info!( + "OpenSearch sink connector ID: {}: created index '{}'", + self.id, self.config.index + ); + return Ok(()); + } + + let body = response.text().await.unwrap_or_default(); + // Another runtime instance winning the create race is not an error. + if is_index_already_exists_error(&body) { + info!( + "OpenSearch sink connector ID: {}: index '{}' was created concurrently", + self.id, self.config.index + ); + return Ok(()); + } + + Err(map_status_error("index creation", status, &body)) + } + + fn prepare_document( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + mut message: ConsumedMessage, + ) -> Result<PreparedDocument, Error> { + let payload = + std::mem::replace(&mut message.payload, Payload::Raw(Vec::new())).into_json_document(); + + let mut document = match payload { + Payload::Json(value) => document_from_json(owned_value_to_serde_json(&value)), + Payload::Raw(bytes) => document_from_raw(bytes), + Payload::Text(text) | Payload::Proto(text) => Map::from_iter([ + ("text".to_string(), Value::String(text)), + ("data_type".to_string(), Value::String("text".to_string())), + ]), + _ => { + return Err(Error::InvalidRecordValue(format!( + "Unsupported payload format for OpenSearch sink: {}", + messages_metadata.schema + ))); + } + }; + + let id = match self.document_id_from_field(&document)? { + Some(id) => id, + None => generated_document_id( + topic_metadata, + messages_metadata, + message.offset, + message.id, + ), + }; + + if self.config.include_metadata { + inject_metadata( + self.id, + &mut document, + topic_metadata, + messages_metadata, + &message, + ); + } + + Ok(PreparedDocument { + id, + document: Value::Object(document), + }) + } + + fn log_invalid_record(&self, reason: &str) { + if self.invalid_record_warned.swap(true, Ordering::Relaxed) { + debug!( + "Dropping invalid OpenSearch sink record for connector ID: {}, reason: {reason}", + self.id + ); + return; + } + warn!( + "Dropping invalid OpenSearch sink record for connector ID: {}, reason: {reason}. Further drops are counted and logged at debug level.", + self.id + ); + } + + fn document_id_from_field( + &self, + document: &Map<String, Value>, + ) -> Result<Option<String>, Error> { + let Some(field) = self.config.document_id_field.as_deref() else { + return Ok(None); + }; + let Some(value) = document.get(field) else { + if !self + .missing_document_id_field_warned + .swap(true, Ordering::Relaxed) + { + warn!( + "OpenSearch sink connector ID: {}: document_id_field '{field}' missing from a message; falling back to a generated ID. Further occurrences will not be logged individually.", + self.id + ); + } + return Ok(None); + }; + + let id = match value { + Value::String(text) => text.clone(), + Value::Number(number) => number.to_string(), + Value::Bool(flag) => flag.to_string(), + Value::Null | Value::Array(_) | Value::Object(_) => { + return Err(Error::InvalidRecordValue(format!( + "OpenSearch document_id_field '{field}' must be a string, number, or boolean" + ))); + } + }; + + // Checked via `.trim()`, but `id` itself stays untrimmed - whitespace can be meaningful. + if id.trim().is_empty() { + return Err(Error::InvalidRecordValue(format!( + "OpenSearch document_id_field '{field}' is empty" + ))); + } + if id.len() > MAX_DOCUMENT_ID_BYTES { + return Err(Error::InvalidRecordValue(format!( + "OpenSearch document_id_field '{field}' exceeds the {MAX_DOCUMENT_ID_BYTES} byte limit" + ))); + } + + Ok(Some(id)) + } + + async fn index_documents( + &self, + client: &OpenSearch, + documents: Vec<PreparedDocument>, + ) -> Result<usize, PartialIndexError> { + let total = documents.len(); + let mut indexed = 0usize; + // The runtime commits the offset regardless of this error, so returning + // early would drop the remaining chunks for good. The first failure is + // kept, not the last, matching `BulkOutcome::merge`: it's the one an + // operator can act on if later chunks fail differently (e.g. a + // permanent mapping conflict followed by a transient timeout). + let mut first_error: Option<Error> = None; + + for chunk in documents.chunks(self.config.batch_size) { + match self.index_chunk(client, chunk).await { + Ok(outcome) => { + indexed += outcome.indexed; + if let Some(error) = outcome.into_error(&self.config.index) { + first_error = first_error.take().or(Some(error)); + } + } + Err(error) => first_error = first_error.take().or(Some(error)), + } + debug!( + "OpenSearch sink with ID: {} indexed {}/{} documents", + self.id, indexed, total + ); + } + + match first_error { + Some(error) => Err(PartialIndexError { + indexed, + failed: total - indexed, + error, + }), + None => Ok(indexed), + } + } + + async fn index_chunk( + &self, + client: &OpenSearch, + documents: &[PreparedDocument], + ) -> Result<BulkOutcome, Error> { + // Only the documents OpenSearch has not yet accepted. A per-item + // transient rejection (429 under load, most commonly) shrinks this to + // just those documents rather than resending the whole chunk. + let mut pending: Vec<&PreparedDocument> = documents.iter().collect(); + // Rebuilt only when `pending` shrinks; `Bytes` clones are a refcount + // bump, so a whole-request retry resends without re-serializing. + let mut body = build_bulk_body(&self.config.index, &pending)?; + let mut outcome = BulkOutcome::default(); + let mut retries = 0u32; + + loop { + let request = client.bulk(BulkParts::None).body(vec![body.clone()]); + let request = match self.config.refresh { + Some(refresh) => request.refresh(refresh), + None => request, + }; + + let mut retry_after = None; + let failure = match tokio::time::timeout(self.config.timeout, request.send()).await { + Ok(Ok(response)) => { + let status = response.status_code(); + if !status.is_success() { + // Read before the body is consumed below: `text()`/`json()` + // take `response` by value, so the header is gone once + // either is called. + if status == StatusCode::TOO_MANY_REQUESTS { + retry_after = response + .headers() + .get(RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(parse_retry_after); + } + let response_body = response.text().await.unwrap_or_default(); + let error = map_status_error("bulk request", status, &response_body); + if !is_transient_status(status) { + // Documents an earlier attempt in this same retry loop + // already indexed must still count; only `pending` (not + // yet accepted) is failed by this non-transient status. + outcome.merge(BulkOutcome { + indexed: 0, + failed: pending.len(), + transient: false, + first_failure: Some(error.to_string()), + }); + return Ok(outcome); + } + error.to_string() + } else { + // A bulk call answers 200 even when individual documents + // fail, so the per-item results decide the outcome. + match response.json::<BulkResponseBody>().await { + Ok(payload) => match parse_bulk_response(&payload, pending.len()) { + Ok(attempt) => { + let retry_set = documents_at(&pending, &attempt.retryable); + + if retry_set.is_empty() || retries >= self.config.max_retries { + outcome.merge(attempt.into_outcome()); + return Ok(outcome); + } + + // Only the permanent half is final; the retryable half + // is settled by a later attempt. + outcome.merge(BulkOutcome { + indexed: attempt.indexed, + failed: attempt.permanent_failed, + transient: false, + first_failure: attempt.first_permanent_failure, + }); + + let rejected = retry_set.len(); + pending = retry_set; + body = build_bulk_body(&self.config.index, &pending)?; + format!( + "{rejected} document(s) rejected with a transient status" + ) + } + // A 200 whose `items` array can't be trusted to cover + // every pending document leaves the true per-item + // outcome unknown. Retrying is safe because indexing + // is idempotent by `_id`, so this is treated as + // transient rather than silently counting the chunk + // as handled. + Err(error) => { + format!("failed to parse bulk response items: {error}") + } + }, + // A 200 with an unparsable body leaves the true + // per-item outcome unknown. Retrying is safe + // because indexing is idempotent by `_id`, so this + // is treated as transient rather than hard-failing + // a chunk OpenSearch may already have accepted. + Err(error) => { + format!("failed to parse bulk response body: {error}") + } + } + } + } + Ok(Err(error)) => { + if !is_transient_client_error(&error) { + let error = map_client_error("bulk request", error); + outcome.merge(BulkOutcome { + indexed: 0, + failed: pending.len(), + transient: false, + first_failure: Some(error.to_string()), + }); + return Ok(outcome); + } + error.to_string() + } + Err(_) => format!("timed out after {:?}", self.config.timeout), + }; + + // Reported through the outcome rather than `Err` so documents + // accepted by earlier attempts still count as indexed. + if retries >= self.config.max_retries { + outcome.merge(BulkOutcome { + indexed: 0, + failed: pending.len(), + transient: true, + first_failure: Some(format!( + "OpenSearch bulk request failed after {} retries ({failure})", + self.config.max_retries + )), + }); + return Ok(outcome); + } + + retries += 1; + self.sleep_before_retry( + "bulk request", + retries, + self.config.max_retries, + &failure, + retry_after, + ) + .await; + } + } + + /// Waits out the backoff for an already-incremented `retries`. The first + /// retry (`retries == 1`) uses attempt `0` so it sleeps `retry_delay` + /// itself rather than `retry_delay * 2`. Jitter can push the raw backoff + /// up to 20% above `max_retry_delay`, so the jittered result is clamped + /// back down to it. `retry_after` overrides the computed backoff but is + /// still clamped to `max_retry_delay` - a hostile or misconfigured + /// server naming an hours-long wait can't force an unbounded sleep. + async fn sleep_before_retry( + &self, + operation: &str, + retries: u32, + max_retries: u32, + failure: &str, + retry_after: Option<Duration>, + ) { + let delay = retry_after + .map(|wait| wait.min(self.config.max_retry_delay)) + .unwrap_or_else(|| { + retry_backoff( + self.config.retry_delay, + retries, + self.config.max_retry_delay, + ) + }); + warn!( + "OpenSearch sink connector ID: {}: {} failed (retry {}/{}): {}. Retrying in {:?}...", + self.id, operation, retries, max_retries, failure, delay + ); + sleep(delay).await; + } +} + +#[async_trait] +impl Sink for OpenSearchSink { + async fn open(&mut self) -> Result<(), Error> { + self.validate_config()?; + let normalized_url = normalize_url(self.id, &self.config.url)?; + info!( + "Opening OpenSearch sink connector with ID: {} for URL: {}, index: {}", + self.id, normalized_url, self.config.index + ); + + let client = self.create_client(&normalized_url)?; + self.check_connectivity(&client).await?; + self.ensure_index_exists(&client).await?; + self.client = Some(client); + + info!( + "Successfully opened OpenSearch sink connector with ID: {}", + self.id + ); + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec<ConsumedMessage>, + ) -> Result<(), Error> { + let invocation = self.invocations_count.fetch_add(1, Ordering::Relaxed) + 1; + + let verbose = self.config.verbose_logging; + macro_rules! log_progress { + ($($tt:tt)+) => { + if verbose { + info!($($tt)+) + } else { + debug!($($tt)+) + } + }; + } + + log_progress!( + "OpenSearch sink with ID: {} received: {} messages, schema: {}, stream: {}, topic: {}, partition: {}, offset: {}, invocation: {}", + self.id, + messages.len(), + messages_metadata.schema, + topic_metadata.stream, + topic_metadata.topic, + messages_metadata.partition_id, + messages_metadata.current_offset, + invocation + ); + + let client = self + .client + .as_ref() + .ok_or_else(|| Error::Connection("OpenSearch client not initialized".to_string()))?; + + let messages_count = messages.len(); + let mut documents = Vec::with_capacity(messages_count); + let mut invalid_records = 0usize; + let mut preparation_errors = 0usize; + for message in messages { + match self.prepare_document(topic_metadata, &messages_metadata, message) { + Ok(document) => documents.push(document), + Err(Error::InvalidRecordValue(reason)) => { + invalid_records += 1; + self.log_invalid_record(&reason); + } + // A single message's preparation failing must not discard the + // documents already built from earlier messages in this batch. + Err(error) => { + preparation_errors += 1; + error!( + "Failed to prepare OpenSearch sink document for connector ID: {}, error: {}", + self.id, error + ); + } + } + } + if invalid_records > 0 || preparation_errors > 0 { + self.errors_count.fetch_add( + (invalid_records + preparation_errors) as u64, + Ordering::Relaxed, + ); + } + + if documents.is_empty() { Review Comment: Should we hold off on this until the runtime-level discussion in #2927 and #2928 settles? Dropping a bad record, counting it and continuing is the runtime's own convention for records that fail decoding or transforms, and the README documents it for this sink. Failing the batch wouldn't recover the records anyway, since offsets are committed at poll time. -- 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]
