hubcio commented on code in PR #4708: URL: https://github.com/apache/iggy/pull/4708#discussion_r4470352119
########## core/connectors/sinks/scopedb_sink/src/encode.rs: ########## @@ -0,0 +1,191 @@ +// 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 std::borrow::Cow; + +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64; +use chrono::{DateTime, SecondsFormat, Utc}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Payload, TopicMetadata, owned_value_into_serde_json, +}; +use scopedb_client::DataType; +use serde_json::value::{RawValue, to_raw_value}; +use serde_json::{Map, Value, json}; + +use crate::schema::Layout; +use crate::{Settings, WriteMode}; + +pub(crate) fn row( + message: ConsumedMessage, + topic: &TopicMetadata, + metadata: &MessagesMetadata, + layout: &Layout, + settings: &Settings, +) -> Result<Box<RawValue>, Error> { + let message_offset = message.offset; + let message_timestamp = message.timestamp; + let message_id = message.id; + let message_headers = message.headers; + let mut row = match settings.mode { + WriteMode::Mapped => { + let value = json_payload(message.payload)?; + let Value::Object(mut object) = value else { + return Err(Error::InvalidRecordValue( + "mapped payload must be a JSON object".into(), + )); + }; + object.retain(|name, _| layout.writable.contains_key(name)); + object + } + WriteMode::Raw => { + let value = match layout.raw_type { + Some(DataType::Binary) => { + Value::String(BASE64.encode(payload_bytes(&message.payload)?.as_ref())) + } + Some(DataType::String) => Value::String(payload_string(message.payload)?), + Some(DataType::Object) => { + let value = raw_json(message.payload)?; + if !value.is_object() { + return Err(Error::InvalidRecordValue( + "object payload_column requires a JSON object".into(), + )); + } + value + } + Some(DataType::Any) => raw_json(message.payload)?, + _ => { + return Err(Error::SchemaMismatch( + "raw layout is not initialized".into(), + )); + } + }; + let mut object = Map::new(); + object.insert(settings.payload_column.clone(), value); + object + } + }; + if settings.include_metadata { + let offset = i64::try_from(message_offset).map_err(|_| { + Error::InvalidRecordValue("iggy_offset exceeds the NDJSON signed integer limit".into()) + })?; + let micros = i64::try_from(message_timestamp) + .map_err(|_| Error::InvalidRecordValue("iggy_timestamp is out of range".into()))?; + let timestamp = DateTime::<Utc>::from_timestamp_micros(micros) + .ok_or_else(|| Error::InvalidRecordValue("iggy_timestamp is out of range".into()))?; + row.insert("iggy_stream".into(), Value::String(topic.stream.clone())); + row.insert("iggy_topic".into(), Value::String(topic.topic.clone())); + row.insert( + "iggy_partition_id".into(), + Value::from(metadata.partition_id), + ); + row.insert("iggy_offset".into(), Value::from(offset)); + row.insert( + "iggy_timestamp".into(), + Value::String(timestamp.to_rfc3339_opts(SecondsFormat::Micros, true)), + ); + row.insert("iggy_id".into(), Value::String(message_id.to_string())); + } + if settings.include_headers { + let mut headers = Map::new(); + if let Some(values) = &message_headers { + for (name, value) in values { + let encoded = match value.as_raw() { + Ok(raw) => json!({"data":BASE64.encode(raw), "iggy_header_encoding":"base64"}), + Err(_) => Value::String(value.to_string_value()), + }; + headers.insert(name.to_string_value(), encoded); + } + } + row.insert("iggy_headers".into(), Value::Object(headers)); + } + let row = Value::Object(row); + validate_numbers(&row)?; + let encoded = to_raw_value(&row).map_err(|error| Error::Serialization(error.to_string()))?; + if encoded.get().len() > settings.max_request_bytes { + return Err(Error::InvalidRecordValue( + "row exceeds max_request_bytes".into(), + )); + } + Ok(encoded) +} + +fn payload_bytes(payload: &Payload) -> Result<Cow<'_, [u8]>, Error> { Review Comment: simplification: `payload_bytes()` duplicates the SDK conversion although `row()` owns the payload. pass `message.payload.try_into_vec()?` directly to `BASE64.encode` and remove the helper. ########## core/connectors/sinks/scopedb_sink/src/encode.rs: ########## @@ -0,0 +1,191 @@ +// 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 std::borrow::Cow; + +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64; +use chrono::{DateTime, SecondsFormat, Utc}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Payload, TopicMetadata, owned_value_into_serde_json, +}; +use scopedb_client::DataType; +use serde_json::value::{RawValue, to_raw_value}; +use serde_json::{Map, Value, json}; + +use crate::schema::Layout; +use crate::{Settings, WriteMode}; + +pub(crate) fn row( + message: ConsumedMessage, + topic: &TopicMetadata, + metadata: &MessagesMetadata, + layout: &Layout, + settings: &Settings, +) -> Result<Box<RawValue>, Error> { + let message_offset = message.offset; + let message_timestamp = message.timestamp; + let message_id = message.id; + let message_headers = message.headers; + let mut row = match settings.mode { + WriteMode::Mapped => { + let value = json_payload(message.payload)?; + let Value::Object(mut object) = value else { + return Err(Error::InvalidRecordValue( + "mapped payload must be a JSON object".into(), + )); + }; + object.retain(|name, _| layout.writable.contains_key(name)); + object + } + WriteMode::Raw => { + let value = match layout.raw_type { + Some(DataType::Binary) => { + Value::String(BASE64.encode(payload_bytes(&message.payload)?.as_ref())) + } + Some(DataType::String) => Value::String(payload_string(message.payload)?), + Some(DataType::Object) => { + let value = raw_json(message.payload)?; + if !value.is_object() { + return Err(Error::InvalidRecordValue( + "object payload_column requires a JSON object".into(), + )); + } + value + } + Some(DataType::Any) => raw_json(message.payload)?, + _ => { + return Err(Error::SchemaMismatch( + "raw layout is not initialized".into(), + )); + } + }; + let mut object = Map::new(); + object.insert(settings.payload_column.clone(), value); + object + } + }; + if settings.include_metadata { + let offset = i64::try_from(message_offset).map_err(|_| { + Error::InvalidRecordValue("iggy_offset exceeds the NDJSON signed integer limit".into()) + })?; + let micros = i64::try_from(message_timestamp) + .map_err(|_| Error::InvalidRecordValue("iggy_timestamp is out of range".into()))?; + let timestamp = DateTime::<Utc>::from_timestamp_micros(micros) + .ok_or_else(|| Error::InvalidRecordValue("iggy_timestamp is out of range".into()))?; + row.insert("iggy_stream".into(), Value::String(topic.stream.clone())); + row.insert("iggy_topic".into(), Value::String(topic.topic.clone())); + row.insert( + "iggy_partition_id".into(), + Value::from(metadata.partition_id), + ); + row.insert("iggy_offset".into(), Value::from(offset)); + row.insert( + "iggy_timestamp".into(), + Value::String(timestamp.to_rfc3339_opts(SecondsFormat::Micros, true)), + ); + row.insert("iggy_id".into(), Value::String(message_id.to_string())); + } + if settings.include_headers { + let mut headers = Map::new(); + if let Some(values) = &message_headers { + for (name, value) in values { + let encoded = match value.as_raw() { + Ok(raw) => json!({"data":BASE64.encode(raw), "iggy_header_encoding":"base64"}), + Err(_) => Value::String(value.to_string_value()), + }; + headers.insert(name.to_string_value(), encoded); + } + } + row.insert("iggy_headers".into(), Value::Object(headers)); + } + let row = Value::Object(row); + validate_numbers(&row)?; + let encoded = to_raw_value(&row).map_err(|error| Error::Serialization(error.to_string()))?; + if encoded.get().len() > settings.max_request_bytes { + return Err(Error::InvalidRecordValue( + "row exceeds max_request_bytes".into(), + )); + } + Ok(encoded) +} + +fn payload_bytes(payload: &Payload) -> Result<Cow<'_, [u8]>, Error> { + match payload { + Payload::Json(_) => payload.try_to_bytes().map(Cow::Owned), + Payload::Raw(bytes) | Payload::FlatBuffer(bytes) | Payload::Avro(bytes) => { + Ok(Cow::Borrowed(bytes)) + } + Payload::Text(text) | Payload::Proto(text) => Ok(Cow::Borrowed(text.as_bytes())), + } +} + +fn payload_string(payload: Payload) -> Result<String, Error> { + match payload { + Payload::Text(text) | Payload::Proto(text) => Ok(text), + Payload::Json(_) => { + String::from_utf8(payload.try_to_bytes()?).map_err(|_| Error::InvalidTextPayload) + } + Payload::Raw(bytes) | Payload::FlatBuffer(bytes) | Payload::Avro(bytes) => { + String::from_utf8(bytes).map_err(|_| Error::InvalidTextPayload) + } + } +} + +fn json_payload(payload: Payload) -> Result<Value, Error> { + match payload { + Payload::Json(value) => Ok(owned_value_into_serde_json(value)), + Payload::Text(text) | Payload::Proto(text) => { + serde_json::from_str(&text).map_err(|_| Error::InvalidJsonPayload) + } + _ => Err(Error::InvalidPayloadType), + } +} + +fn raw_json(payload: Payload) -> Result<Value, Error> { + match payload { + Payload::Raw(bytes) => { + serde_json::from_slice(&bytes).map_err(|_| Error::InvalidJsonPayload) + } + payload => json_payload(payload), + } +} + +fn validate_numbers(value: &Value) -> Result<(), Error> { + match value { + Value::Number(number) if number.as_i64().is_none() => { + let encoded = number.to_string(); Review Comment: simplification: the numeric check allocates and scans a temporary string for each floating-point value. reject numbers with `!number.is_i64() && !number.is_f64()` to preserve the same range checks without that conversion. ########## core/connectors/sinks/scopedb_sink/README.md: ########## @@ -0,0 +1,145 @@ +# ScopeDB Sink Connector + +The ScopeDB sink appends Iggy messages to an existing ScopeDB table through the ScopeDB Rust SDK. Each `consume()` call waits for its writes and retries to settle. It keeps no buffer across calls. Normal close drops the settled writer; cancellation stops further scheduling and retrying, while requests already sent may still commit. + +## Configuration + +```toml +type = "sink" +key = "scopedb" +enabled = true +version = 0 +name = "ScopeDB sink" +path = "target/release/libiggy_connector_scopedb_sink" +verbose = false + +[[streams]] +stream = "orders" +topics = ["events"] +schema = "json" +batch_length = 1000 +consumer_group = "scopedb_sink_connector" + +[plugin_config] +endpoint = "https://<workspace-endpoint>" +api_key = "<api-key>" +table = "events" +mode = "mapped" +include_metadata = true +include_headers = false +``` + +Inject credentials with `IGGY_CONNECTORS_SINK_SCOPEDB_PLUGIN_CONFIG_API_KEY`. Plugin config uses `SecretString` for the API key and redacts the endpoint in debug output. Embedded URL credentials, query parameters, and fragments are rejected. HTTPS, custom base paths, and the HTTP client's platform proxy settings are supported. + +| Option | Default | Meaning | +| ------ | ------- | ------- | +| `endpoint` | required | ScopeDB HTTP(S) endpoint | +| `api_key` | none | Bearer API key, when required by the endpoint | +| `database` | `scopedb` | Target database | +| `schema` | `public` | Target schema | +| `table` | required | Existing target table | +| `mode` | `mapped` | `mapped` or `raw` | +| `payload_column` | `payload` | Destination column in raw mode | +| `include_metadata` | `true` | Populate the six metadata columns below | +| `include_headers` | `false` | Populate `iggy_headers`, independently of `include_metadata` | +| `batch_size` | `100` | Maximum rows per HTTP request, at most 200,000 | +| `max_request_bytes` | `8388608` | Maximum uncompressed NDJSON request bytes; oversized individual rows are rejected | +| `max_concurrent_requests` | `1` | Concurrent requests per consume call; larger values permit out-of-order completion | +| `max_retries` | `3` | Total attempts per request, including the first, clamped to 1..3 | +| `retry_delay` | `1s` | Initial exponential backoff | +| `max_retry_delay` | `30s` | Maximum backoff, including `Retry-After` | +| `retry_ambiguous` | `false` | Explicitly allow replay of transient ambiguous outcomes; duplicates are possible | +| `timeout` | `30s` | Overall startup deadline and timeout per HTTP write attempt, including response body reads | +| `verbose_logging` | `false` | Log consume summaries at info rather than debug | + +Durations use humantime strings. Invalid duration strings fall back to the documented default with a warning; reversed retry delays are swapped. Zero write timeouts and invalid request limits fail startup. + +The identity needs permission to read the target table definition and append rows. The connector never creates tables, changes schemas, or submits transformation statements. Restart the connector after changing the table definition. + +## Modes and schema validation + +`open()` reads the table's complete Catalog definition. Missing metadata columns, computed metadata columns, and unsupported raw payload columns fail startup. + +In `mapped` mode, `Json`, `Text`, and `Proto` payloads must represent JSON objects. Writable table columns select top-level fields by name; unmatched fields and computed columns are omitted. Enabled Iggy metadata overrides payload fields with the same names. Missing columns follow ScopeDB's normal null/computed-column behavior. The connector does not implement BigQuery's default-value or missing-value policies. + +In `raw` mode, the configured payload column determines the representation: + +| Column type | Payloads | +| ----------- | -------- | +| `binary` | All payload variants, encoded as canonical padded base64 for NDJSON | +| `string` | UTF-8 bytes from Text, Proto, Raw, FlatBuffer, Avro, or serialized Json | +| `object` | Json objects, or Text/Proto/Raw values that parse as JSON objects | +| `any` | Json values, or Text/Proto/Raw values that parse as JSON | + +Raw mode preserves the representation delivered by Iggy's decoder/transforms. A Json value is serialized again; it does not preserve the original wire whitespace or original Avro/Protobuf bytes after decoding. + +NDJSON integers, including nested values and `uint` columns, must fit signed 64-bit integers. Oversized integers are rejected without narrowing. Store identifiers outside that range as strings. Schema/type validation errors returned by ScopeDB retain their request-relative row indexes through the SDK. + +## Metadata + +With `include_metadata = true`, the table must contain these writable columns: + +| Column | ScopeDB type | Value | +| ------ | ------------ | ----- | +| `iggy_stream` | `string` | Source stream | +| `iggy_topic` | `string` | Source topic | +| `iggy_partition_id` | `uint` or `int` | Partition ID | +| `iggy_offset` | `uint` or `int` | Each message's offset, checked against the NDJSON integer limit | +| `iggy_timestamp` | `timestamp` | Message timestamp converted from microseconds to UTC RFC 3339 | +| `iggy_id` | `string` | Full decimal u128 message ID | +| `iggy_headers` | `object` or `any` | Only required with `include_headers = true` | + +Text headers become strings. Binary headers become `{"data":"<base64>","iggy_header_encoding":"base64"}`. Missing headers become an empty object. `messages_metadata.current_offset` is a poll high-water mark and is not used as the row offset or a batch identity. + +Example table for mapped mode: + +```sql +CREATE TABLE events ( + id int, + event string, + iggy_stream string, + iggy_topic string, + iggy_partition_id uint, + iggy_offset uint, + iggy_timestamp timestamp, + iggy_id string, + iggy_headers object +); +``` + +## Failures and delivery semantics + +The SDK splits requests at the configured byte and row limits. Only validated HTTP 429 admission refusals retry by default, using bounded exponential backoff and `Retry-After`. HTTP 5xx, lost responses, and invalid success counts are ambiguous and are not replayed by default. Client-side write timeouts are never automatically replayed, including when `retry_ambiguous = true`. The connector adds no HTTP retry middleware or redirects that could replay a write behind the SDK's back. + +Modern ScopeDB errors describe a whole failed request, so the connector fails that batch without trying to parse row indexes from diagnostic text. + +When an older response supplies a complete, atomic rejection with structured row errors, those indexes identify rows known to be invalid. The connector deduplicates the indexes, logs and drops those rows, then appends the remaining rows from that request once more. Successful sibling requests are never replayed. Truncated or out-of-range row errors do not authorize this recovery. A second row-error response fails the remaining rows instead of starting an unbounded filtering loop. + +Every local rejection and terminal failed/unconfirmed row is logged with stream, topic, partition, and message offset. `stats()` and the shutdown log distinguish written, rejected, failed, and unconfirmed rows. These are logical input counts; opt-in unknown replay can create extra physical copies. If every message is rejected, `consume()` returns an error. There is no dead-letter queue. + +The Iggy connectors runtime currently commits consumer group offsets during polling, before the sink persists the messages. A failed consume call is not redelivered and stops the affected consumer task. A crash after the offset commit can lose that batch. Split requests can partially succeed. Review Comment: warning: the runtime logs failed `consume()` results and continues polling with `OnRead`, so outages can lose later batches too. document continued consumption and the resulting loss across subsequent batches. -- 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]
