From 43d08eaf6b1359524343cca5478a30ebcdef177d Mon Sep 17 00:00:00 2001
From: ryankert01 <ryankert01@gmail.com>
Date: Wed, 6 May 2026 01:50:34 +0800
Subject: [PATCH 1/6] feat(connectors): add Apache Doris sink connector (#3112)

Sink connector that writes Iggy messages to Apache Doris via the HTTP
Stream Load API. v1 scope: JSON payloads only, HTTP Basic auth,
pre-created tables only (no DDL).

Behaviour:
- Manual 307/308 redirect following (capped at 5) so the Authorization
  header survives the FE -> BE hop, which reqwest strips by default.
- Deterministic per-batch label
  ({prefix}-{stream}-{topic}-{partition}-{first_offset}-{last_offset})
  so replays are deduplicated by Doris within label_keep_max_second.
- Response body Status field drives error classification: Success and
  "Label Already Exists" -> Ok; Publish Timeout -> CannotStoreData
  (transient); Fail or any unknown status -> PermanentHttpError so the
  runtime DLQs the batch instead of looping.
- Optional columns / where / max_filter_ratio / batch_size / timeout
  forwarded as Stream Load headers.
- Password held as secrecy::SecretString; auth header wrapped in
  SecretString so Debug derivation never leaks the base64 credential.
- Client built in open() with InitError on failure; fe_url validated
  there too so a bad config fails at startup rather than first batch.

Tests: 6 integration tests under core/integration/tests/connectors/doris
backed by an apache/doris all-in-one testcontainer (FE HTTP + FE MySQL).
Coverage includes happy path, 1k-row bulk, max_filter_ratio skip path,
label-replay dedupe, missing-target-table (proves no auto-create), and
the columns derived-expression header. The container must bind host:8040
1:1 because the FE 307-redirects to 127.0.0.1:8040; tests are serialized
via a 'doris' nextest test-group (max-threads = 1) so concurrent test
processes don't race for that port.
---
 .config/nextest.toml                          |  12 +
 Cargo.lock                                    |  15 +
 Cargo.toml                                    |   2 +
 core/connectors/README.md                     |   1 +
 .../example_config/connectors/doris_sink.toml |  43 ++
 core/connectors/sinks/README.md               |   1 +
 core/connectors/sinks/doris_sink/Cargo.toml   |  43 ++
 core/connectors/sinks/doris_sink/README.md    |  74 +++
 core/connectors/sinks/doris_sink/config.toml  |  43 ++
 core/connectors/sinks/doris_sink/src/lib.rs   | 565 ++++++++++++++++
 .../tests/connectors/doris/doris_sink.rs      | 498 ++++++++++++++
 .../integration/tests/connectors/doris/mod.rs |  20 +
 .../tests/connectors/doris/sink.toml          |  20 +
 .../connectors/fixtures/doris/container.rs    | 611 ++++++++++++++++++
 .../tests/connectors/fixtures/doris/mod.rs    |  25 +
 .../tests/connectors/fixtures/mod.rs          |   5 +
 core/integration/tests/connectors/mod.rs      |   1 +
 17 files changed, 1979 insertions(+)
 create mode 100644 core/connectors/runtime/example_config/connectors/doris_sink.toml
 create mode 100644 core/connectors/sinks/doris_sink/Cargo.toml
 create mode 100644 core/connectors/sinks/doris_sink/README.md
 create mode 100644 core/connectors/sinks/doris_sink/config.toml
 create mode 100644 core/connectors/sinks/doris_sink/src/lib.rs
 create mode 100644 core/integration/tests/connectors/doris/doris_sink.rs
 create mode 100644 core/integration/tests/connectors/doris/mod.rs
 create mode 100644 core/integration/tests/connectors/doris/sink.toml
 create mode 100644 core/integration/tests/connectors/fixtures/doris/container.rs
 create mode 100644 core/integration/tests/connectors/fixtures/doris/mod.rs

diff --git a/.config/nextest.toml b/.config/nextest.toml
index fb488b6948..47f6b5723d 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -21,6 +21,18 @@
 filter = 'package(integration) and test(cli::system::test_cli_session_scenario::should_be_successful)'
 threads-required = "num-cpus"
 
+# Doris testcontainer must bind host:8040 1:1 (FE→BE 307 redirects to
+# 127.0.0.1:8040; BE's priority_networks pins the advertise address) and the
+# all-in-one image is heavy enough that running it alongside other
+# container-backed tests (e.g. elasticsearch) starves their startup timers.
+# threads-required = "num-cpus" forces each doris test to run alone, matching
+# the existing pattern used elsewhere in this file. The longer slow-timeout
+# accounts for cold container boot (~60-90s on Linux CI) plus the test body.
+[[profile.default.overrides]]
+filter = 'package(integration) and test(/connectors::doris::/)'
+threads-required = "num-cpus"
+slow-timeout = { period = "60s", terminate-after = 8 }
+
 [profile.default]
 slow-timeout = { period = "30s", terminate-after = 4 }
 
diff --git a/Cargo.lock b/Cargo.lock
index 3a48443372..f5080e3de0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6563,6 +6563,21 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "iggy_connector_doris_sink"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "base64 0.22.1",
+ "iggy_connector_sdk",
+ "reqwest 0.13.3",
+ "secrecy",
+ "serde",
+ "serde_json",
+ "simd-json",
+ "tracing",
+]
+
 [[package]]
 name = "iggy_connector_elasticsearch_sink"
 version = "0.4.0"
diff --git a/Cargo.toml b/Cargo.toml
index 949a498f2a..fe2413af0a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -34,6 +34,7 @@ members = [
     "core/connectors/runtime",
     "core/connectors/sdk",
     "core/connectors/sinks/delta_sink",
+    "core/connectors/sinks/doris_sink",
     "core/connectors/sinks/elasticsearch_sink",
     "core/connectors/sinks/http_sink",
     "core/connectors/sinks/iceberg_sink",
@@ -268,6 +269,7 @@ socket2 = "0.6.3"
 sqlx = { version = "0.8.6", features = [
     "runtime-tokio-rustls",
     "postgres",
+    "mysql",
     "chrono",
     "uuid",
     "json",
diff --git a/core/connectors/README.md b/core/connectors/README.md
index cdef67789a..c36624b1a8 100644
--- a/core/connectors/README.md
+++ b/core/connectors/README.md
@@ -78,6 +78,7 @@ Each sink should have its own, custom configuration, which is passed along with
 
 ### Available Sinks
 
+- **Doris Sink** - loads JSON messages into Apache Doris tables via the Stream Load HTTP API
 - **Elasticsearch Sink** - sends messages to Elasticsearch indices
 - **Iceberg Sink** - writes data to Apache Iceberg tables via REST catalog
 - **PostgreSQL Sink** - stores messages in PostgreSQL database tables
diff --git a/core/connectors/runtime/example_config/connectors/doris_sink.toml b/core/connectors/runtime/example_config/connectors/doris_sink.toml
new file mode 100644
index 0000000000..bbc167397d
--- /dev/null
+++ b/core/connectors/runtime/example_config/connectors/doris_sink.toml
@@ -0,0 +1,43 @@
+# 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.
+
+type = "sink"
+key = "doris"
+enabled = true
+version = 0
+name = "Doris sink"
+path = "<BASE_DIR>/target/release/libiggy_connector_doris_sink"
+plugin_config_format = "toml"
+verbose = false
+
+[[streams]]
+stream = "events"
+topics = ["doris_events"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "doris_sink"
+
+[plugin_config]
+fe_url = "http://localhost:8030"
+database = "iggy_demo"
+table = "events"
+username = "root"
+password = "replace_with_secret"
+label_prefix = "iggy"
+batch_size = 1000
+timeout_secs = 30
diff --git a/core/connectors/sinks/README.md b/core/connectors/sinks/README.md
index 55c639c336..fc0c83e1db 100644
--- a/core/connectors/sinks/README.md
+++ b/core/connectors/sinks/README.md
@@ -8,6 +8,7 @@ Sink connectors are responsible for writing data from Iggy streams to external s
 
 | Sink | Description |
 | ---- | ----------- |
+| **doris_sink** | Loads JSON messages into Apache Doris tables via the Stream Load HTTP API |
 | **elasticsearch_sink** | Sends messages to Elasticsearch indices for full-text search and analytics |
 | **iceberg_sink** | Writes data to Apache Iceberg tables via REST catalog with S3/GCS/Azure storage |
 | **postgres_sink** | Stores messages in PostgreSQL database tables with configurable schemas |
diff --git a/core/connectors/sinks/doris_sink/Cargo.toml b/core/connectors/sinks/doris_sink/Cargo.toml
new file mode 100644
index 0000000000..f2795b7ffe
--- /dev/null
+++ b/core/connectors/sinks/doris_sink/Cargo.toml
@@ -0,0 +1,43 @@
+# 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.
+
+[package]
+name = "iggy_connector_doris_sink"
+version = "0.1.0"
+description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming"]
+categories = ["command-line-utilities", "database", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+async-trait = { workspace = true }
+base64 = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+reqwest = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+simd-json = { workspace = true }
+tracing = { workspace = true }
diff --git a/core/connectors/sinks/doris_sink/README.md b/core/connectors/sinks/doris_sink/README.md
new file mode 100644
index 0000000000..d3633337f2
--- /dev/null
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -0,0 +1,74 @@
+# Apache Doris Sink
+
+The Doris sink connector consumes JSON messages from Iggy streams and writes them to a pre-created Apache Doris table via Doris's [Stream Load HTTP API](https://doris.apache.org/docs/data-operate/import/import-way/stream-load-manual).
+
+## Requirements
+
+- The target Doris **database and table must be pre-created** before enabling the sink. The connector never issues DDL.
+- Messages must arrive with `Payload::Json` (i.e. the configured stream schema is `json`). Other payload types are skipped with a warning.
+- The Iggy message JSON shape must match the target table columns. Use the optional `columns` plugin setting if the column order differs from the JSON keys.
+
+## How it works
+
+1. For each batch of messages, the connector serializes the JSON payloads into a JSON array.
+2. It computes a deterministic Stream Load `label` of the form `{label_prefix}-{stream}-{topic}-{partition}-{first_offset}-{last_offset}`. Doris dedupes loads by label inside its `label_keep_max_second` window, so a replayed batch (after restart, retry, etc.) is silently absorbed instead of producing duplicates.
+3. It `PUT`s the batch to `{fe_url}/api/{database}/{table}/_stream_load` with HTTP Basic auth and the headers `format: json`, `strip_outer_array: true`, `label: <label>`.
+4. The Doris frontend (FE) responds with a `307 Temporary Redirect` to a backend (BE). The connector follows the redirect manually so that the `Authorization` header is preserved across the hop (`reqwest`'s default policy strips it on cross-host redirects).
+5. The HTTP body is parsed as JSON and the `Status` field decides the outcome:
+   - `Success` → batch accepted.
+   - `Label Already Exists` → idempotent replay, treated as success.
+   - `Publish Timeout` or HTTP `5xx` → transient error (`Error::CannotStoreData`); the runtime can retry.
+   - `Fail` or HTTP `4xx` → permanent error (`Error::PermanentHttpError`); retrying is not useful.
+
+## Configuration
+
+| Field | Required | Default | Description |
+| --- | --- | --- | --- |
+| `fe_url` | yes | — | Doris frontend HTTP base URL, e.g. `http://localhost:8030`. |
+| `database` | yes | — | Target database. |
+| `table` | yes | — | Target table. |
+| `username` | yes | — | Doris user with `LOAD_PRIV` on the table. |
+| `password` | yes | — | Doris user password. Stored as a `secrecy::SecretString` and never logged. |
+| `label_prefix` | no | `iggy` | Prefix for the deterministic Stream Load label. |
+| `batch_size` | no | `1000` | Maximum number of messages per Stream Load request. |
+| `timeout_secs` | no | `30` | Per-request HTTP timeout. |
+| `max_filter_ratio` | no | unset | Forwarded as the `max_filter_ratio` Stream Load header. |
+| `columns` | no | unset | Forwarded as the `columns` Stream Load header. |
+| `where` | no | unset | Forwarded as the `where` Stream Load header. |
+
+### Example
+
+```toml
+type = "sink"
+key = "doris"
+enabled = true
+version = 0
+name = "Doris sink"
+path = "target/release/libiggy_connector_doris_sink"
+plugin_config_format = "toml"
+
+[[streams]]
+stream = "events"
+topics = ["doris_events"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "doris_sink"
+
+[plugin_config]
+fe_url = "http://localhost:8030"
+database = "iggy_demo"
+table = "events"
+username = "root"
+password = "replace_with_secret"
+label_prefix = "iggy"
+batch_size = 1000
+timeout_secs = 30
+```
+
+## Limitations (v1)
+
+- JSON payload only. CSV and raw-text payloads are not supported yet.
+- HTTP Basic auth only.
+- No automatic table creation.
+- No built-in retry middleware or circuit breaker — the runtime decides whether to redrive a failing batch. A hardening pass with `iggy_connector_sdk::retry::*` is planned as a follow-up.
diff --git a/core/connectors/sinks/doris_sink/config.toml b/core/connectors/sinks/doris_sink/config.toml
new file mode 100644
index 0000000000..5faa0c03da
--- /dev/null
+++ b/core/connectors/sinks/doris_sink/config.toml
@@ -0,0 +1,43 @@
+# 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.
+
+type = "sink"
+key = "doris"
+enabled = true
+version = 0
+name = "Doris sink"
+path = "../../target/release/libiggy_connector_doris_sink"
+plugin_config_format = "toml"
+verbose = false
+
+[[streams]]
+stream = ""
+topics = []
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "doris_sink"
+
+[plugin_config]
+fe_url = "http://localhost:8030"
+database = ""
+table = ""
+username = "root"
+password = ""
+label_prefix = "iggy"
+batch_size = 1000
+timeout_secs = 30
diff --git a/core/connectors/sinks/doris_sink/src/lib.rs b/core/connectors/sinks/doris_sink/src/lib.rs
new file mode 100644
index 0000000000..ff5a39ac84
--- /dev/null
+++ b/core/connectors/sinks/doris_sink/src/lib.rs
@@ -0,0 +1,565 @@
+/* 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 iggy_connector_sdk::{
+    ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector,
+};
+use reqwest::{Method, StatusCode, header};
+use secrecy::{ExposeSecret, SecretString};
+use serde::Deserialize;
+use std::time::Duration;
+use tracing::{debug, error, info, warn};
+
+sink_connector!(DorisSink);
+
+const DEFAULT_LABEL_PREFIX: &str = "iggy";
+const DEFAULT_BATCH_SIZE: u32 = 1000;
+const DEFAULT_TIMEOUT_SECS: u64 = 30;
+// Doris's FE Stream Load returns 307 to a BE; reqwest's default policy strips
+// the Authorization header on cross-host redirects, so we follow redirects
+// manually. This cap protects against a misbehaving cluster looping.
+const MAX_REDIRECTS: u8 = 5;
+
+#[derive(Debug)]
+pub struct DorisSink {
+    id: u32,
+    config: DorisSinkConfig,
+    client: Option<reqwest::Client>,
+    auth_header: SecretString,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct DorisSinkConfig {
+    pub fe_url: String,
+    pub database: String,
+    pub table: String,
+    pub username: String,
+    pub password: SecretString,
+    pub label_prefix: Option<String>,
+    pub max_filter_ratio: Option<f64>,
+    pub columns: Option<String>,
+    #[serde(rename = "where")]
+    pub where_clause: Option<String>,
+    pub timeout_secs: Option<u64>,
+    pub batch_size: Option<u32>,
+}
+
+#[derive(Debug, Deserialize)]
+struct StreamLoadResponse {
+    #[serde(rename = "Status")]
+    status: String,
+    #[serde(rename = "Message")]
+    #[serde(default)]
+    message: String,
+    #[serde(rename = "NumberLoadedRows")]
+    #[serde(default)]
+    number_loaded_rows: u64,
+    #[serde(rename = "NumberFilteredRows")]
+    #[serde(default)]
+    number_filtered_rows: u64,
+}
+
+impl DorisSink {
+    pub fn new(id: u32, config: DorisSinkConfig) -> Self {
+        let credential = format!("{}:{}", config.username, config.password.expose_secret());
+        let auth_header = SecretString::from(format!(
+            "Basic {}",
+            general_purpose::STANDARD.encode(credential)
+        ));
+
+        DorisSink {
+            id,
+            config,
+            client: None,
+            auth_header,
+        }
+    }
+
+    fn build_client(&self) -> Result<reqwest::Client, Error> {
+        let timeout = Duration::from_secs(self.config.timeout_secs.unwrap_or(DEFAULT_TIMEOUT_SECS));
+        // We follow redirects manually to keep the Authorization header alive
+        // across the FE -> BE hop. reqwest strips Authorization on cross-host
+        // 307s by design.
+        reqwest::Client::builder()
+            .redirect(reqwest::redirect::Policy::none())
+            .timeout(timeout)
+            .build()
+            .map_err(|e| Error::InitError(format!("Failed to build Doris HTTP client: {e}")))
+    }
+
+    fn stream_load_url(&self) -> String {
+        format!(
+            "{}/api/{}/{}/_stream_load",
+            self.config.fe_url.trim_end_matches('/'),
+            self.config.database,
+            self.config.table,
+        )
+    }
+
+    fn label_prefix(&self) -> &str {
+        self.config
+            .label_prefix
+            .as_deref()
+            .unwrap_or(DEFAULT_LABEL_PREFIX)
+    }
+
+    fn batch_size(&self) -> usize {
+        self.config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1) as usize
+    }
+
+    /// Build the deterministic label for one batch. Same inputs always produce
+    /// the same label, so a replayed batch (after restart, retry, etc.) gets
+    /// deduplicated by Doris within its label-keep window.
+    fn build_label(
+        &self,
+        topic_metadata: &TopicMetadata,
+        partition_id: u32,
+        first_offset: u64,
+        last_offset: u64,
+    ) -> String {
+        format!(
+            "{}-{}-{}-{}-{}-{}",
+            sanitize_label(self.label_prefix()),
+            sanitize_label(&topic_metadata.stream),
+            sanitize_label(&topic_metadata.topic),
+            partition_id,
+            first_offset,
+            last_offset,
+        )
+    }
+
+    async fn send_stream_load(
+        &self,
+        label: &str,
+        body: Vec<u8>,
+    ) -> Result<StreamLoadResponse, Error> {
+        let client = self.client.as_ref().ok_or_else(|| {
+            Error::InitError(format!(
+                "Doris sink ID {} called before open() — client not initialized",
+                self.id
+            ))
+        })?;
+        let mut url = self.stream_load_url();
+        let mut redirects = 0u8;
+
+        loop {
+            let mut request = client
+                .request(Method::PUT, &url)
+                .header(header::AUTHORIZATION, self.auth_header.expose_secret())
+                .header(header::EXPECT, "100-continue")
+                .header("format", "json")
+                .header("strip_outer_array", "true")
+                .header("label", label)
+                .body(body.clone());
+
+            if let Some(ratio) = self.config.max_filter_ratio {
+                request = request.header("max_filter_ratio", ratio.to_string());
+            }
+            if let Some(columns) = &self.config.columns {
+                request = request.header("columns", columns);
+            }
+            if let Some(where_clause) = &self.config.where_clause {
+                request = request.header("where", where_clause);
+            }
+
+            let response = request.send().await.map_err(|e| {
+                error!("Doris sink ID {} HTTP request failed: {e}", self.id);
+                Error::HttpRequestFailed(e.to_string())
+            })?;
+
+            let status = response.status();
+            if matches!(
+                status,
+                StatusCode::TEMPORARY_REDIRECT | StatusCode::PERMANENT_REDIRECT
+            ) {
+                redirects += 1;
+                if redirects > MAX_REDIRECTS {
+                    return Err(Error::HttpRequestFailed(format!(
+                        "Doris sink ID {} exceeded max redirects ({MAX_REDIRECTS})",
+                        self.id
+                    )));
+                }
+                let Some(location) = response
+                    .headers()
+                    .get(header::LOCATION)
+                    .and_then(|v| v.to_str().ok())
+                else {
+                    return Err(Error::HttpRequestFailed(format!(
+                        "Doris sink ID {} got {status} with no Location header",
+                        self.id
+                    )));
+                };
+                debug!("Doris sink ID {} following redirect to {location}", self.id);
+                url = location.to_string();
+                continue;
+            }
+
+            let response_text = response.text().await.unwrap_or_default();
+
+            if !status.is_success() {
+                let msg = format!(
+                    "Doris sink ID {} stream load returned HTTP {status}: {response_text}",
+                    self.id
+                );
+                error!("{msg}");
+                return Err(if status.is_server_error() {
+                    Error::CannotStoreData(msg)
+                } else {
+                    Error::PermanentHttpError(msg)
+                });
+            }
+
+            return parse_stream_load_response(&response_text);
+        }
+    }
+}
+
+/// Replace Doris-label-illegal characters with `_`. Doris allows
+/// [A-Za-z0-9_-]. Anything else -> `_`.
+fn sanitize_label(value: &str) -> String {
+    value
+        .chars()
+        .map(|c| {
+            if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
+                c
+            } else {
+                '_'
+            }
+        })
+        .collect()
+}
+
+fn parse_stream_load_response(body: &str) -> Result<StreamLoadResponse, Error> {
+    let parsed: StreamLoadResponse = serde_json::from_str(body).map_err(|e| {
+        Error::CannotStoreData(format!(
+            "Failed to parse Doris stream load response: {e}. Body: {body}"
+        ))
+    })?;
+    Ok(parsed)
+}
+
+fn classify_status(response: &StreamLoadResponse) -> Result<(), Error> {
+    match response.status.as_str() {
+        "Success" => Ok(()),
+        "Label Already Exists" => {
+            // Idempotent replay — the data was already loaded with this label.
+            // Treat as success so the runtime advances the consumer offset.
+            info!(
+                "Doris reported 'Label Already Exists' (loaded={}, filtered={}); treating as success.",
+                response.number_loaded_rows, response.number_filtered_rows
+            );
+            Ok(())
+        }
+        "Publish Timeout" => Err(Error::CannotStoreData(format!(
+            "Doris stream load Publish Timeout: {}",
+            response.message
+        ))),
+        "Fail" => Err(Error::PermanentHttpError(format!(
+            "Doris stream load failed: {}",
+            response.message
+        ))),
+        // Default unknown statuses to permanent: silently retrying an
+        // unrecognized failure (e.g. a future Doris error variant) is worse
+        // than surfacing it. The runtime will route the batch to its DLQ
+        // path instead of hammering the FE forever.
+        other => Err(Error::PermanentHttpError(format!(
+            "Doris stream load returned unexpected status '{other}': {}",
+            response.message
+        ))),
+    }
+}
+
+#[async_trait]
+impl Sink for DorisSink {
+    async fn open(&mut self) -> Result<(), Error> {
+        // Validate the URL up front so a bad config fails at startup rather
+        // than on the first batch.
+        reqwest::Url::parse(&self.stream_load_url()).map_err(|e| {
+            Error::InvalidConfigValue(format!(
+                "Doris sink ID {} has invalid fe_url '{}': {e}",
+                self.id, self.config.fe_url
+            ))
+        })?;
+
+        self.client = Some(self.build_client()?);
+
+        info!(
+            "Opened Doris sink ID {} for {}.{} at {}",
+            self.id, self.config.database, self.config.table, self.config.fe_url
+        );
+        Ok(())
+    }
+
+    async fn consume(
+        &self,
+        topic_metadata: &TopicMetadata,
+        messages_metadata: MessagesMetadata,
+        messages: Vec<ConsumedMessage>,
+    ) -> Result<(), Error> {
+        if messages.is_empty() {
+            return Ok(());
+        }
+
+        let total = messages.len();
+        debug!(
+            "Doris sink ID {} received {total} messages for {}.{}",
+            self.id, self.config.database, self.config.table
+        );
+
+        let batch_size = self.batch_size();
+        let mut first_error: Option<Error> = None;
+
+        for chunk in messages.chunks(batch_size) {
+            let json_values: Vec<&simd_json::OwnedValue> = chunk
+                .iter()
+                .filter_map(|m| match &m.payload {
+                    Payload::Json(value) => Some(value),
+                    _ => {
+                        warn!(
+                            "Doris sink ID {} dropping non-JSON payload (schema={})",
+                            self.id, messages_metadata.schema
+                        );
+                        None
+                    }
+                })
+                .collect();
+
+            if json_values.is_empty() {
+                continue;
+            }
+
+            let body = match simd_json::to_vec(&json_values) {
+                Ok(b) => b,
+                Err(e) => {
+                    let err =
+                        Error::CannotStoreData(format!("Failed to serialize batch for Doris: {e}"));
+                    if first_error.is_none() {
+                        first_error = Some(err);
+                    }
+                    continue;
+                }
+            };
+
+            let first_offset = chunk.first().map(|m| m.offset).unwrap_or(0);
+            let last_offset = chunk.last().map(|m| m.offset).unwrap_or(0);
+            let label = self.build_label(
+                topic_metadata,
+                messages_metadata.partition_id,
+                first_offset,
+                last_offset,
+            );
+
+            match self.send_stream_load(&label, body).await {
+                Ok(response) => match classify_status(&response) {
+                    Ok(()) => {
+                        info!(
+                            "Doris sink ID {} loaded {} rows (filtered={}) into {}.{} (label={label})",
+                            self.id,
+                            response.number_loaded_rows,
+                            response.number_filtered_rows,
+                            self.config.database,
+                            self.config.table,
+                        );
+                    }
+                    Err(e) => {
+                        error!("Doris sink ID {} batch failed: {e}", self.id);
+                        if first_error.is_none() {
+                            first_error = Some(e);
+                        }
+                    }
+                },
+                Err(e) => {
+                    error!("Doris sink ID {} HTTP failed: {e}", self.id);
+                    if first_error.is_none() {
+                        first_error = Some(e);
+                    }
+                }
+            }
+        }
+
+        if let Some(err) = first_error {
+            return Err(err);
+        }
+        Ok(())
+    }
+
+    async fn close(&mut self) -> Result<(), Error> {
+        info!("Doris sink ID {} closed.", self.id);
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn make_config() -> DorisSinkConfig {
+        DorisSinkConfig {
+            fe_url: "http://localhost:8030".into(),
+            database: "test_db".into(),
+            table: "test_tbl".into(),
+            username: "root".into(),
+            password: SecretString::from("pw"),
+            label_prefix: None,
+            max_filter_ratio: None,
+            columns: None,
+            where_clause: None,
+            timeout_secs: None,
+            batch_size: None,
+        }
+    }
+
+    #[test]
+    fn stream_load_url_is_well_formed() {
+        let sink = DorisSink::new(1, make_config());
+        assert_eq!(
+            sink.stream_load_url(),
+            "http://localhost:8030/api/test_db/test_tbl/_stream_load"
+        );
+    }
+
+    #[test]
+    fn stream_load_url_strips_trailing_slash() {
+        let mut cfg = make_config();
+        cfg.fe_url = "http://localhost:8030/".into();
+        let sink = DorisSink::new(1, cfg);
+        assert!(!sink.stream_load_url().contains("//api"));
+    }
+
+    #[test]
+    fn label_is_deterministic() {
+        let sink = DorisSink::new(1, make_config());
+        let topic = TopicMetadata {
+            stream: "events".into(),
+            topic: "orders".into(),
+        };
+        let a = sink.build_label(&topic, 7, 100, 199);
+        let b = sink.build_label(&topic, 7, 100, 199);
+        assert_eq!(a, b);
+        assert_eq!(a, "iggy-events-orders-7-100-199");
+    }
+
+    #[test]
+    fn label_sanitizes_illegal_chars() {
+        let sink = DorisSink::new(1, make_config());
+        let topic = TopicMetadata {
+            stream: "events.v1".into(),
+            topic: "orders/inbound".into(),
+        };
+        let label = sink.build_label(&topic, 0, 0, 0);
+        // dots and slashes are not allowed in Doris labels.
+        assert!(!label.contains('.'));
+        assert!(!label.contains('/'));
+        assert_eq!(label, "iggy-events_v1-orders_inbound-0-0-0");
+    }
+
+    #[test]
+    fn label_prefix_default_is_iggy() {
+        let sink = DorisSink::new(1, make_config());
+        assert_eq!(sink.label_prefix(), "iggy");
+    }
+
+    #[test]
+    fn label_prefix_uses_config_value() {
+        let mut cfg = make_config();
+        cfg.label_prefix = Some("prod".into());
+        let sink = DorisSink::new(1, cfg);
+        assert_eq!(sink.label_prefix(), "prod");
+    }
+
+    #[test]
+    fn batch_size_floors_at_one() {
+        let mut cfg = make_config();
+        cfg.batch_size = Some(0);
+        let sink = DorisSink::new(1, cfg);
+        assert_eq!(sink.batch_size(), 1);
+    }
+
+    #[test]
+    fn classify_success_returns_ok() {
+        let r = StreamLoadResponse {
+            status: "Success".into(),
+            message: String::new(),
+            number_loaded_rows: 10,
+            number_filtered_rows: 0,
+        };
+        assert!(classify_status(&r).is_ok());
+    }
+
+    #[test]
+    fn classify_label_already_exists_returns_ok() {
+        let r = StreamLoadResponse {
+            status: "Label Already Exists".into(),
+            message: String::new(),
+            number_loaded_rows: 0,
+            number_filtered_rows: 0,
+        };
+        assert!(classify_status(&r).is_ok());
+    }
+
+    #[test]
+    fn classify_publish_timeout_is_transient() {
+        let r = StreamLoadResponse {
+            status: "Publish Timeout".into(),
+            message: "be unreachable".into(),
+            number_loaded_rows: 0,
+            number_filtered_rows: 0,
+        };
+        assert!(matches!(
+            classify_status(&r).unwrap_err(),
+            Error::CannotStoreData(_)
+        ));
+    }
+
+    #[test]
+    fn classify_fail_is_permanent() {
+        let r = StreamLoadResponse {
+            status: "Fail".into(),
+            message: "schema mismatch".into(),
+            number_loaded_rows: 0,
+            number_filtered_rows: 0,
+        };
+        assert!(matches!(
+            classify_status(&r).unwrap_err(),
+            Error::PermanentHttpError(_)
+        ));
+    }
+
+    #[test]
+    fn parse_stream_load_response_handles_minimal_json() {
+        let body = r#"{"Status":"Success"}"#;
+        let r = parse_stream_load_response(body).unwrap();
+        assert_eq!(r.status, "Success");
+        assert_eq!(r.number_loaded_rows, 0);
+    }
+
+    #[test]
+    fn parse_stream_load_response_rejects_garbage() {
+        let body = "not json";
+        assert!(parse_stream_load_response(body).is_err());
+    }
+
+    #[test]
+    fn auth_header_is_basic_b64() {
+        let sink = DorisSink::new(1, make_config());
+        // base64("root:pw") = cm9vdDpwdw==
+        assert_eq!(sink.auth_header.expose_secret(), "Basic cm9vdDpwdw==");
+    }
+}
diff --git a/core/integration/tests/connectors/doris/doris_sink.rs b/core/integration/tests/connectors/doris/doris_sink.rs
new file mode 100644
index 0000000000..d5d69f2b41
--- /dev/null
+++ b/core/integration/tests/connectors/doris/doris_sink.rs
@@ -0,0 +1,498 @@
+/*
+ * 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 crate::connectors::create_test_messages;
+use crate::connectors::fixtures::{
+    DorisOps, DorisSinkColumnsMappingFixture, DorisSinkFixture, DorisSinkMaxFilterRatioFixture,
+    DorisSinkPreCreatedFixture,
+};
+use bytes::Bytes;
+use iggy::prelude::{IggyMessage, Partitioning};
+use iggy_common::Identifier;
+use iggy_common::MessageClient;
+use integration::harness::seeds;
+use integration::iggy_harness;
+use serde::{Deserialize, Serialize};
+
+const TEST_DATABASE: &str = "iggy_test_db";
+const TEST_TABLE: &str = "test_topic";
+
+#[iggy_harness(
+    server(connectors_runtime(config_path = "tests/connectors/doris/sink.toml")),
+    seed = seeds::connector_stream
+)]
+async fn given_existent_doris_table_should_store(
+    harness: &TestHarness,
+    fixture: DorisSinkPreCreatedFixture,
+) {
+    let client = harness.root_client().await.unwrap();
+    let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+    let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+    let message_count = 10;
+    let test_messages = create_test_messages(message_count);
+    let payloads: Vec<Bytes> = test_messages
+        .iter()
+        .map(|m| Bytes::from(serde_json::to_vec(m).expect("serialize")))
+        .collect();
+
+    let mut messages: Vec<IggyMessage> = payloads
+        .iter()
+        .enumerate()
+        .map(|(i, p)| {
+            IggyMessage::builder()
+                .id((i + 1) as u128)
+                .payload(p.clone())
+                .build()
+                .expect("build message")
+        })
+        .collect();
+
+    client
+        .send_messages(
+            &stream_id,
+            &topic_id,
+            &Partitioning::partition_id(0),
+            &mut messages,
+        )
+        .await
+        .expect("send messages");
+
+    let count = fixture
+        .wait_for_rows(TEST_DATABASE, TEST_TABLE, message_count as i64)
+        .await
+        .expect("rows");
+    assert_eq!(count, message_count as i64);
+}
+
+#[iggy_harness(
+    server(connectors_runtime(config_path = "tests/connectors/doris/sink.toml")),
+    seed = seeds::connector_stream
+)]
+async fn given_bulk_message_send_should_store(
+    harness: &TestHarness,
+    fixture: DorisSinkPreCreatedFixture,
+) {
+    let client = harness.root_client().await.unwrap();
+    let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+    let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+    let message_count = 1000;
+    let test_messages = create_test_messages(message_count);
+
+    let mut messages: Vec<IggyMessage> = test_messages
+        .iter()
+        .enumerate()
+        .map(|(i, m)| {
+            let payload = Bytes::from(serde_json::to_vec(m).expect("serialize"));
+            IggyMessage::builder()
+                .id((i + 1) as u128)
+                .payload(payload)
+                .build()
+                .expect("build message")
+        })
+        .collect();
+
+    client
+        .send_messages(
+            &stream_id,
+            &topic_id,
+            &Partitioning::partition_id(0),
+            &mut messages,
+        )
+        .await
+        .expect("send messages");
+
+    let count = fixture
+        .wait_for_rows(TEST_DATABASE, TEST_TABLE, message_count as i64)
+        .await
+        .expect("rows");
+    assert_eq!(count, message_count as i64);
+}
+
+#[iggy_harness(
+    server(connectors_runtime(config_path = "tests/connectors/doris/sink.toml")),
+    seed = seeds::connector_stream
+)]
+async fn given_invalid_messages_should_skip_via_max_filter_ratio(
+    harness: &TestHarness,
+    fixture: DorisSinkMaxFilterRatioFixture,
+) {
+    let client = harness.root_client().await.unwrap();
+    let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+    let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+    // Two valid messages.
+    let valid: Vec<Bytes> = create_test_messages(2)
+        .iter()
+        .map(|m| Bytes::from(serde_json::to_vec(m).expect("serialize")))
+        .collect();
+
+    // One message whose JSON does not match the table columns. Doris will
+    // reject it as a "filtered" row; max_filter_ratio = 0.5 covers up to
+    // half the batch, so the load still succeeds and the two valid rows land.
+    #[derive(Debug, Serialize, Deserialize)]
+    struct WrongShape {
+        unrelated: f64,
+    }
+    let invalid =
+        Bytes::from(serde_json::to_vec(&WrongShape { unrelated: 1.0 }).expect("serialize"));
+
+    let payloads = [valid[0].clone(), invalid, valid[1].clone()];
+    let mut messages: Vec<IggyMessage> = payloads
+        .iter()
+        .enumerate()
+        .map(|(i, p)| {
+            IggyMessage::builder()
+                .id((i + 1) as u128)
+                .payload(p.clone())
+                .build()
+                .expect("build message")
+        })
+        .collect();
+
+    client
+        .send_messages(
+            &stream_id,
+            &topic_id,
+            &Partitioning::partition_id(0),
+            &mut messages,
+        )
+        .await
+        .expect("send messages");
+
+    let count = fixture
+        .wait_for_rows(TEST_DATABASE, TEST_TABLE, 2)
+        .await
+        .expect("rows");
+    assert_eq!(count, 2, "only the two valid rows should land");
+}
+
+#[iggy_harness(
+    server(connectors_runtime(config_path = "tests/connectors/doris/sink.toml")),
+    seed = seeds::connector_stream
+)]
+async fn given_replayed_label_should_dedupe(harness: &TestHarness, fixture: DorisSinkFixture) {
+    // This fixture variant does NOT pre-create the table; the test creates
+    // it explicitly before producing messages so we can drop and recreate
+    // between rounds without disturbing the connector runtime's state.
+    fixture
+        .create_table(TEST_DATABASE, TEST_TABLE)
+        .await
+        .expect("create table");
+
+    let client = harness.root_client().await.unwrap();
+    let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+    let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+    let message_count = 5;
+    let test_messages = create_test_messages(message_count);
+
+    // Round 1: send and observe rows land.
+    let mut round1: Vec<IggyMessage> = test_messages
+        .iter()
+        .enumerate()
+        .map(|(i, m)| {
+            let payload = Bytes::from(serde_json::to_vec(m).expect("serialize"));
+            IggyMessage::builder()
+                .id((i + 1) as u128)
+                .payload(payload)
+                .build()
+                .expect("build message")
+        })
+        .collect();
+    client
+        .send_messages(
+            &stream_id,
+            &topic_id,
+            &Partitioning::partition_id(0),
+            &mut round1,
+        )
+        .await
+        .expect("send messages round 1");
+
+    let count = fixture
+        .wait_for_rows(TEST_DATABASE, TEST_TABLE, message_count as i64)
+        .await
+        .expect("rows after round 1");
+    assert_eq!(count, message_count as i64);
+
+    // Round 2: send the same payloads again with the same Iggy IDs. The
+    // connector generates a deterministic Stream Load label per (stream,
+    // topic, partition, first_offset, last_offset). Because Iggy assigns
+    // monotonic offsets, the second batch lands at different offsets and
+    // gets a different label — so duplicates WOULD land if the replay
+    // deduplication only relied on Iggy IDs. The point of this test is the
+    // converse case below.
+    //
+    // Instead we verify the label-dedupe path by issuing a manual Stream
+    // Load with the SAME label as the one the connector just used. Doris
+    // must respond with "Label Already Exists", which the connector maps
+    // to Ok — so the row count must NOT increase.
+    // Use the fixture's count helper (raw_sql under the hood) — Doris's
+    // MySQL frontend rejects prepared sqlx::query/query_scalar statements.
+    let row_before = fixture
+        .count_rows(TEST_DATABASE, TEST_TABLE)
+        .await
+        .expect("count before");
+    assert_eq!(row_before, message_count as i64);
+
+    // Replay the same offsets => same label => Doris dedupes server-side.
+    let label = format!(
+        "iggy_test-{}-{}-0-0-{}",
+        sanitize(seeds::names::STREAM),
+        sanitize(seeds::names::TOPIC),
+        message_count - 1,
+    );
+
+    let body = serde_json::to_vec(&test_messages).expect("serialize replay body");
+
+    let client_http = reqwest::Client::builder()
+        .redirect(reqwest::redirect::Policy::none())
+        .build()
+        .unwrap();
+
+    let url = format!(
+        "{}/api/{TEST_DATABASE}/{TEST_TABLE}/_stream_load",
+        fixture.container().fe_url()
+    );
+
+    // Manually follow the FE -> BE 307 once. Cap iterations so a misbehaving
+    // cluster can't hang the test process indefinitely.
+    const MAX_REDIRECTS: u8 = 5;
+    let mut current_url = url.clone();
+    let mut redirects = 0u8;
+    let response = loop {
+        let resp = client_http
+            .put(&current_url)
+            .basic_auth("root", Some(""))
+            // Doris's FE rejects Stream Load PUTs that don't carry
+            // `Expect: 100-continue` (the connector sets this; the manual
+            // probe needs the same).
+            .header(reqwest::header::EXPECT, "100-continue")
+            .header("format", "json")
+            .header("strip_outer_array", "true")
+            .header("label", &label)
+            .body(body.clone())
+            .send()
+            .await
+            .expect("stream load");
+        let status = resp.status();
+        if status == reqwest::StatusCode::TEMPORARY_REDIRECT
+            || status == reqwest::StatusCode::PERMANENT_REDIRECT
+        {
+            redirects += 1;
+            assert!(
+                redirects <= MAX_REDIRECTS,
+                "exceeded {MAX_REDIRECTS} redirects following Stream Load"
+            );
+            let loc = resp
+                .headers()
+                .get(reqwest::header::LOCATION)
+                .and_then(|v| v.to_str().ok())
+                .expect("Location header")
+                .to_string();
+            current_url = loc;
+            continue;
+        }
+        break resp;
+    };
+    let body_text = response.text().await.expect("body");
+    // Require Doris to report dedupe explicitly. A `"Success"` here would
+    // mean the label drifted from what the connector built — which doubles
+    // rows and is caught by `row_after == message_count`, but we want the
+    // dedupe path itself to be load-bearing in this test.
+    assert!(
+        body_text.contains("Label Already Exists"),
+        "expected Doris to dedupe by label, got: {body_text}"
+    );
+
+    let row_after = fixture
+        .count_rows(TEST_DATABASE, TEST_TABLE)
+        .await
+        .expect("count after");
+    assert_eq!(
+        row_after, message_count as i64,
+        "label dedupe must not produce duplicates"
+    );
+}
+
+/// Connector targets a table that does not exist. Every Stream Load PUT
+/// gets a `Status: "Fail"` (or the FE returns a 4xx) — the connector must
+/// classify that as `PermanentHttpError`, NOT `CannotStoreData`. The
+/// integration assertions here are coarse on purpose (we can't read the
+/// connector's internal Result), but they prove three things any future
+/// regression would break:
+///   1. The connector does not silently auto-create the missing table.
+///   2. The connector does not write into any *other* table by mistake.
+///   3. The Doris cluster (and our connection pool) survives the failed
+///      load attempts intact.
+#[iggy_harness(
+    server(connectors_runtime(config_path = "tests/connectors/doris/sink.toml")),
+    seed = seeds::connector_stream
+)]
+async fn given_missing_target_table_should_not_create_or_corrupt(
+    harness: &TestHarness,
+    fixture: DorisSinkFixture,
+) {
+    let client = harness.root_client().await.unwrap();
+    let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+    let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+    let test_messages = create_test_messages(5);
+    let mut messages: Vec<IggyMessage> = test_messages
+        .iter()
+        .enumerate()
+        .map(|(i, m)| {
+            let payload = Bytes::from(serde_json::to_vec(m).expect("serialize"));
+            IggyMessage::builder()
+                .id((i + 1) as u128)
+                .payload(payload)
+                .build()
+                .expect("build message")
+        })
+        .collect();
+
+    client
+        .send_messages(
+            &stream_id,
+            &topic_id,
+            &Partitioning::partition_id(0),
+            &mut messages,
+        )
+        .await
+        .expect("send messages");
+
+    // Give the connector enough time to receive the batch and attempt at
+    // least one Stream Load PUT against the missing table.
+    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+
+    // 1. The target table must NOT have been auto-created.
+    let exists = fixture
+        .table_exists(TEST_DATABASE, TEST_TABLE)
+        .await
+        .expect("information_schema query");
+    assert!(
+        !exists,
+        "Doris sink must NOT auto-create the target table on a failed load"
+    );
+
+    // 2 + 3. The cluster is still healthy and `iggy_test_db` has no rogue
+    // tables — proves we didn't write anywhere unexpected and that the
+    // connector's HTTP failures didn't take Doris down.
+    let pool = fixture.pool().await.expect("pool after failed loads");
+    use sqlx::Row;
+    let rows = sqlx::raw_sql(
+        "SELECT TABLE_NAME FROM information_schema.tables \
+         WHERE TABLE_SCHEMA = 'iggy_test_db'",
+    )
+    .fetch_all(&pool)
+    .await
+    .expect("list tables in iggy_test_db");
+    let tables: Vec<String> = rows
+        .iter()
+        .filter_map(|r| r.try_get::<String, _>("TABLE_NAME").ok())
+        .collect();
+    assert!(
+        tables.is_empty(),
+        "iggy_test_db should be empty after failed loads, found: {tables:?}"
+    );
+}
+
+/// Verifies the `columns` Stream Load config is wired through end-to-end.
+/// The pre-created table has an extra `calculated INT NOT NULL` column that
+/// is NOT in the JSON payload; the only way the load succeeds is if the
+/// connector forwards the configured `columns` header to Doris so it can
+/// derive `calculated = count + 1` server-side.
+#[iggy_harness(
+    server(connectors_runtime(config_path = "tests/connectors/doris/sink.toml")),
+    seed = seeds::connector_stream
+)]
+async fn given_columns_config_should_apply_derived_expression(
+    harness: &TestHarness,
+    fixture: DorisSinkColumnsMappingFixture,
+) {
+    let client = harness.root_client().await.unwrap();
+    let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+    let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+    let message_count = 5;
+    let test_messages = create_test_messages(message_count);
+    let mut messages: Vec<IggyMessage> = test_messages
+        .iter()
+        .enumerate()
+        .map(|(i, m)| {
+            let payload = Bytes::from(serde_json::to_vec(m).expect("serialize"));
+            IggyMessage::builder()
+                .id((i + 1) as u128)
+                .payload(payload)
+                .build()
+                .expect("build message")
+        })
+        .collect();
+
+    client
+        .send_messages(
+            &stream_id,
+            &topic_id,
+            &Partitioning::partition_id(0),
+            &mut messages,
+        )
+        .await
+        .expect("send messages");
+
+    let count = fixture
+        .wait_for_rows(TEST_DATABASE, TEST_TABLE, message_count as i64)
+        .await
+        .expect("rows");
+    assert_eq!(count, message_count as i64);
+
+    // `create_test_messages` produces count = (i - 1) * 10 for i in 1..=N,
+    // so SUM(count) = 0 + 10 + 20 + 30 + 40 = 100 and with the derived
+    // expression `calculated = count + 1` we expect SUM(calculated - count)
+    // to equal exactly the row count (one per row).
+    let pool = fixture.pool().await.expect("pool");
+    use sqlx::Row;
+    let row = sqlx::raw_sql(&format!(
+        "SELECT SUM(calculated - `count`) AS delta FROM {TEST_DATABASE}.{TEST_TABLE}"
+    ))
+    .fetch_one(&pool)
+    .await
+    .expect("sum query");
+    let delta: i64 = row
+        .try_get::<i64, _>("delta")
+        .expect("delta column decodes as i64");
+    assert_eq!(
+        delta, message_count as i64,
+        "expected calculated = count + 1 per row (delta sum = {message_count}), got {delta}"
+    );
+}
+
+fn sanitize(value: &str) -> String {
+    value
+        .chars()
+        .map(|c| {
+            if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
+                c
+            } else {
+                '_'
+            }
+        })
+        .collect()
+}
diff --git a/core/integration/tests/connectors/doris/mod.rs b/core/integration/tests/connectors/doris/mod.rs
new file mode 100644
index 0000000000..b75e42c973
--- /dev/null
+++ b/core/integration/tests/connectors/doris/mod.rs
@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+
+mod doris_sink;
diff --git a/core/integration/tests/connectors/doris/sink.toml b/core/integration/tests/connectors/doris/sink.toml
new file mode 100644
index 0000000000..59f191e0a4
--- /dev/null
+++ b/core/integration/tests/connectors/doris/sink.toml
@@ -0,0 +1,20 @@
+# 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.
+
+[connectors]
+config_type = "local"
+config_dir = "../connectors/sinks/doris_sink"
diff --git a/core/integration/tests/connectors/fixtures/doris/container.rs b/core/integration/tests/connectors/fixtures/doris/container.rs
new file mode 100644
index 0000000000..8786fc0ee8
--- /dev/null
+++ b/core/integration/tests/connectors/fixtures/doris/container.rs
@@ -0,0 +1,611 @@
+/*
+ * 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.
+ */
+
+// Doris's FE returns 307 redirects to a BE address that is the BE container's
+// internal Docker network address. On Linux Docker hosts (the iggy CI
+// environment) those addresses are routable from the host, so the connector's
+// manual-redirect-following works against an off-the-shelf testcontainer.
+// On macOS Docker Desktop they are not routable, so these tests are
+// effectively Linux-only without extra networking work — see the README for
+// the limitation note.
+
+use async_trait::async_trait;
+use integration::harness::{TestBinaryError, TestFixture};
+use once_cell::sync::Lazy;
+use sqlx::mysql::{MySqlConnectOptions, MySqlPool, MySqlPoolOptions};
+use std::collections::HashMap;
+use std::time::Duration;
+use testcontainers_modules::testcontainers::core::wait::HttpWaitStrategy;
+use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor};
+use testcontainers_modules::testcontainers::runners::AsyncRunner;
+use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage, ImageExt};
+use tokio::sync::{Mutex, OwnedMutexGuard};
+use tokio::time::sleep;
+use tracing::info;
+use uuid::Uuid;
+
+// Global serialization lock for Doris fixtures. The all-in-one image binds
+// the BE HTTP port 1:1 (host:8040 → container:8040) so the FE→BE 307
+// redirect remains reachable; only one container per test process can hold
+// that port at a time. Each fixture holds the guard for its lifetime and
+// drops it on Drop, releasing the slot for the next test.
+static DORIS_LOCK: Lazy<std::sync::Arc<Mutex<()>>> =
+    Lazy::new(|| std::sync::Arc::new(Mutex::new(())));
+
+const DORIS_IMAGE: &str = "apache/doris";
+const DORIS_TAG: &str = "doris-all-in-one-2.1.0";
+const FE_HTTP_PORT: u16 = 8030;
+const FE_MYSQL_PORT: u16 = 9030;
+const BE_HTTP_PORT: u16 = 8040;
+const FE_HEALTH_ENDPOINT: &str = "/api/health";
+
+const DEFAULT_TEST_DATABASE: &str = "iggy_test_db";
+const DEFAULT_TEST_TABLE: &str = "test_topic";
+const DEFAULT_USER: &str = "root";
+const DEFAULT_PASSWORD: &str = "";
+
+// Doris's all-in-one image typically reports `Alive: true` within ~40s of
+// container start, but a cold Docker host or constrained CI runner can push
+// that out. The wait window is sized to swallow those outliers without
+// hanging a test slot for a truly broken cluster.
+const DEFAULT_BE_REGISTRATION_ATTEMPTS: usize = 180;
+const DEFAULT_BE_REGISTRATION_INTERVAL_MS: u64 = 2000;
+
+// 90 seconds (180 × 500ms) — bulk tests with 1000 rows can take ~30 s on a
+// fresh container, and a Doris instance under memory pressure (e.g. when it
+// is the 3rd or 4th container to spin up serially) takes longer.
+pub const DEFAULT_POLL_ATTEMPTS: usize = 180;
+pub const DEFAULT_POLL_INTERVAL_MS: u64 = 500;
+
+pub const ENV_SINK_FE_URL: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_FE_URL";
+pub const ENV_SINK_DATABASE: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_DATABASE";
+pub const ENV_SINK_TABLE: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_TABLE";
+pub const ENV_SINK_USERNAME: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_USERNAME";
+pub const ENV_SINK_PASSWORD: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_PASSWORD";
+pub const ENV_SINK_LABEL_PREFIX: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_LABEL_PREFIX";
+pub const ENV_SINK_MAX_FILTER_RATIO: &str =
+    "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_MAX_FILTER_RATIO";
+pub const ENV_SINK_COLUMNS: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_COLUMNS";
+pub const ENV_SINK_BATCH_SIZE: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_BATCH_SIZE";
+pub const ENV_SINK_STREAMS_0_STREAM: &str = "IGGY_CONNECTORS_SINK_DORIS_STREAMS_0_STREAM";
+pub const ENV_SINK_STREAMS_0_TOPICS: &str = "IGGY_CONNECTORS_SINK_DORIS_STREAMS_0_TOPICS";
+pub const ENV_SINK_STREAMS_0_SCHEMA: &str = "IGGY_CONNECTORS_SINK_DORIS_STREAMS_0_SCHEMA";
+pub const ENV_SINK_STREAMS_0_CONSUMER_GROUP: &str =
+    "IGGY_CONNECTORS_SINK_DORIS_STREAMS_0_CONSUMER_GROUP";
+pub const ENV_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_DORIS_PATH";
+
+/// The schema used by `TestMessage` in the integration tests.
+///
+/// `replication_num = 1` because the all-in-one container has a single BE.
+/// `storage_medium = HDD` because that BE advertises HDD; without this
+/// override Doris defaults the table to SSD and the FE rejects CREATE with
+/// "Failed to find enough backend ... storage medium: SSD".
+const TEST_TABLE_DDL_TEMPLATE: &str = "
+CREATE TABLE IF NOT EXISTS {db}.{table} (
+    id BIGINT NOT NULL,
+    name VARCHAR(64) NOT NULL,
+    count INT NOT NULL,
+    amount DOUBLE NOT NULL,
+    active BOOLEAN NOT NULL,
+    timestamp BIGINT NOT NULL
+)
+DUPLICATE KEY(id)
+DISTRIBUTED BY HASH(id) BUCKETS 1
+PROPERTIES (
+    \"replication_num\" = \"1\",
+    \"storage_medium\" = \"HDD\"
+);
+";
+
+/// Same shape as `TEST_TABLE_DDL_TEMPLATE` plus an extra `calculated INT NOT
+/// NULL` column. The columns-mapping test populates `calculated` via a Stream
+/// Load `columns` derived expression (`calculated = count + 1`); without that
+/// header the load would fail because `calculated` has no source field in the
+/// JSON payload.
+const TEST_TABLE_WITH_CALCULATED_DDL_TEMPLATE: &str = "
+CREATE TABLE IF NOT EXISTS {db}.{table} (
+    id BIGINT NOT NULL,
+    name VARCHAR(64) NOT NULL,
+    count INT NOT NULL,
+    amount DOUBLE NOT NULL,
+    active BOOLEAN NOT NULL,
+    timestamp BIGINT NOT NULL,
+    calculated INT NOT NULL
+)
+DUPLICATE KEY(id)
+DISTRIBUTED BY HASH(id) BUCKETS 1
+PROPERTIES (
+    \"replication_num\" = \"1\",
+    \"storage_medium\" = \"HDD\"
+);
+";
+
+/// Stream Load `columns` header used by `DorisSinkColumnsMappingFixture`. The
+/// six leading names match the JSON payload's keys; `calculated = count + 1`
+/// instructs Doris to derive the seventh column from the loaded `count`.
+pub const COLUMNS_MAPPING_HEADER: &str =
+    "id, name, count, amount, active, timestamp, calculated = count + 1";
+
+pub struct DorisContainer {
+    #[allow(dead_code)]
+    container: ContainerAsync<GenericImage>,
+    fe_http_host_port: u16,
+    fe_mysql_host_port: u16,
+    // Held for the lifetime of the fixture so the next Doris fixture cannot
+    // start until this one is dropped.
+    #[allow(dead_code)]
+    serial_guard: OwnedMutexGuard<()>,
+}
+
+impl DorisContainer {
+    pub async fn start() -> Result<Self, TestBinaryError> {
+        let serial_guard = DORIS_LOCK.clone().lock_owned().await;
+        let unique_network = format!("iggy-doris-{}", Uuid::new_v4());
+
+        // The all-in-one image ships without a Dockerfile EXPOSE directive,
+        // so we have to publish every port we want via `with_mapped_port`.
+        // FE HTTP and FE MySQL get ephemeral host ports (the connector and
+        // tests connect via the resolved mapping). BE HTTP must be pinned
+        // 1:1 — the FE always returns Location: http://127.0.0.1:8040/...
+        // for the Stream Load redirect, and that's only reachable from the
+        // host if container:8040 is bound to host:8040. The static
+        // DORIS_LOCK mutex keeps two containers from racing for that port.
+        let container = GenericImage::new(DORIS_IMAGE, DORIS_TAG)
+            .with_wait_for(WaitFor::http(
+                HttpWaitStrategy::new(FE_HEALTH_ENDPOINT)
+                    .with_port(FE_HTTP_PORT.tcp())
+                    .with_expected_status_code(200u16),
+            ))
+            .with_network(unique_network)
+            .with_mapped_port(0, FE_HTTP_PORT.tcp())
+            .with_mapped_port(0, FE_MYSQL_PORT.tcp())
+            .with_mapped_port(BE_HTTP_PORT, BE_HTTP_PORT.tcp())
+            .start()
+            .await
+            .map_err(|e| TestBinaryError::FixtureSetup {
+                fixture_type: "DorisContainer".to_string(),
+                message: format!("Failed to start container: {e}"),
+            })?;
+
+        info!("Started Doris all-in-one container");
+
+        let ports = container
+            .ports()
+            .await
+            .map_err(|e| TestBinaryError::FixtureSetup {
+                fixture_type: "DorisContainer".to_string(),
+                message: format!("Failed to read mapped ports: {e}"),
+            })?;
+
+        let fe_http_host_port = ports.map_to_host_port_ipv4(FE_HTTP_PORT).ok_or_else(|| {
+            TestBinaryError::FixtureSetup {
+                fixture_type: "DorisContainer".to_string(),
+                message: "No host mapping for Doris FE HTTP port".to_string(),
+            }
+        })?;
+        let fe_mysql_host_port = ports.map_to_host_port_ipv4(FE_MYSQL_PORT).ok_or_else(|| {
+            TestBinaryError::FixtureSetup {
+                fixture_type: "DorisContainer".to_string(),
+                message: "No host mapping for Doris FE MySQL port".to_string(),
+            }
+        })?;
+
+        info!("Doris FE HTTP -> {fe_http_host_port}, FE MySQL -> {fe_mysql_host_port}");
+
+        let this = Self {
+            container,
+            fe_http_host_port,
+            fe_mysql_host_port,
+            serial_guard,
+        };
+
+        this.wait_for_be_alive().await?;
+        this.create_test_database().await?;
+
+        Ok(this)
+    }
+
+    pub fn fe_url(&self) -> String {
+        format!("http://127.0.0.1:{}", self.fe_http_host_port)
+    }
+
+    async fn create_pool(&self) -> Result<MySqlPool, TestBinaryError> {
+        // sqlx ordinarily emits a `SET sql_mode = (SELECT CONCAT(...))`
+        // session init when connecting to MySQL. Doris rejects that with
+        // "Set statement does't support non-constant expr.", which kills
+        // the connection before any query runs. Disabling these two flags
+        // suppresses both offending SET statements.
+        let opts = MySqlConnectOptions::new()
+            .host("127.0.0.1")
+            .port(self.fe_mysql_host_port)
+            .username(DEFAULT_USER)
+            .pipes_as_concat(false)
+            .no_engine_substitution(false);
+
+        MySqlPoolOptions::new()
+            .max_connections(2)
+            .acquire_timeout(Duration::from_secs(10))
+            .connect_with(opts)
+            .await
+            .map_err(|e| TestBinaryError::FixtureSetup {
+                fixture_type: "DorisContainer".to_string(),
+                message: format!("Failed to connect to Doris over MySQL: {e}"),
+            })
+    }
+
+    /// Doris reports BE alive=true via `SHOW BACKENDS` only after the BE has
+    /// successfully registered with the FE. Stream Load won't work until then,
+    /// so we block here.
+    async fn wait_for_be_alive(&self) -> Result<(), TestBinaryError> {
+        let mut last_diag = String::from("never connected");
+        for attempt in 0..DEFAULT_BE_REGISTRATION_ATTEMPTS {
+            match self.create_pool().await {
+                Ok(pool) => {
+                    // `SHOW BACKENDS` cannot be sent through `sqlx::query` —
+                    // sqlx prepares the statement and Doris's MySQL frontend
+                    // rejects PREPARE for anything other than SELECT/INSERT
+                    // ("Only support prepare SelectStmt or InsertStmt now").
+                    // `sqlx::raw_sql` skips PREPARE and dispatches the bytes
+                    // directly, so SHOW/DDL works.
+                    match sqlx::raw_sql("SHOW BACKENDS").fetch_all(&pool).await {
+                        Ok(rows) => {
+                            let any_alive = rows.iter().any(|row| {
+                                use sqlx::Row;
+                                row.try_get::<String, _>("Alive")
+                                    .map(|v| v.eq_ignore_ascii_case("true"))
+                                    .unwrap_or(false)
+                            });
+                            if any_alive {
+                                info!("Doris BE registered alive after {attempt} attempts");
+                                return Ok(());
+                            }
+                            last_diag = format!("{} backend rows, none alive", rows.len());
+                        }
+                        Err(e) => {
+                            last_diag = format!("SHOW BACKENDS failed: {e}");
+                        }
+                    }
+                }
+                Err(e) => {
+                    last_diag = format!("connect failed: {e}");
+                }
+            }
+            // Surface progress every ~30 seconds so a stuck cluster is
+            // distinguishable from one that's still bootstrapping.
+            if attempt > 0 && attempt % (30_000 / DEFAULT_BE_REGISTRATION_INTERVAL_MS as usize) == 0
+            {
+                info!("Doris BE not yet alive after {attempt} attempts ({last_diag})");
+            }
+            sleep(Duration::from_millis(DEFAULT_BE_REGISTRATION_INTERVAL_MS)).await;
+        }
+        Err(TestBinaryError::FixtureSetup {
+            fixture_type: "DorisContainer".to_string(),
+            message: format!(
+                "Doris BE did not register within {DEFAULT_BE_REGISTRATION_ATTEMPTS} attempts ({last_diag})"
+            ),
+        })
+    }
+
+    async fn create_test_database(&self) -> Result<(), TestBinaryError> {
+        let pool = self.create_pool().await?;
+        // raw_sql avoids the PREPARE path Doris doesn't accept for DDL.
+        sqlx::raw_sql(&format!(
+            "CREATE DATABASE IF NOT EXISTS {DEFAULT_TEST_DATABASE}"
+        ))
+        .execute(&pool)
+        .await
+        .map_err(|e| TestBinaryError::FixtureSetup {
+            fixture_type: "DorisContainer".to_string(),
+            message: format!("Failed to create test database: {e}"),
+        })?;
+        Ok(())
+    }
+}
+
+#[async_trait]
+pub trait DorisOps: Sync {
+    fn container(&self) -> &DorisContainer;
+
+    async fn pool(&self) -> Result<MySqlPool, TestBinaryError> {
+        self.container().create_pool().await
+    }
+
+    async fn create_table(&self, database: &str, table: &str) -> Result<(), TestBinaryError> {
+        self.create_table_with_template(database, table, TEST_TABLE_DDL_TEMPLATE)
+            .await
+    }
+
+    async fn create_table_with_template(
+        &self,
+        database: &str,
+        table: &str,
+        ddl_template: &str,
+    ) -> Result<(), TestBinaryError> {
+        let pool = self.pool().await?;
+        let ddl = ddl_template
+            .replace("{db}", database)
+            .replace("{table}", table);
+
+        // Even after `SHOW BACKENDS` reports `Alive: true`, the FE briefly
+        // doesn't know about the BE's storage paths and rejects CREATE TABLE
+        // with "Failed to find enough backend ... storage medium: ...". The
+        // BE typically catches up within a couple of seconds once it sends
+        // its first tablet/disk report. Retry through that window.
+        let mut last_err: Option<sqlx::Error> = None;
+        for _ in 0..30 {
+            match sqlx::raw_sql(&ddl).execute(&pool).await {
+                Ok(_) => return Ok(()),
+                Err(e) => {
+                    let msg = e.to_string();
+                    if msg.contains("Failed to find enough backend") {
+                        last_err = Some(e);
+                        sleep(Duration::from_secs(2)).await;
+                        continue;
+                    }
+                    return Err(TestBinaryError::FixtureSetup {
+                        fixture_type: "DorisOps".to_string(),
+                        message: format!("Failed to create table {database}.{table}: {e}"),
+                    });
+                }
+            }
+        }
+        Err(TestBinaryError::FixtureSetup {
+            fixture_type: "DorisOps".to_string(),
+            message: format!(
+                "Failed to create table {database}.{table} after 30 retries waiting for BE storage report: {}",
+                last_err.map(|e| e.to_string()).unwrap_or_default(),
+            ),
+        })
+    }
+
+    /// Returns true iff `database.table` is registered in Doris's
+    /// `information_schema.tables`. Used by the missing-target-table test to
+    /// assert the connector did NOT silently auto-create on a failed load.
+    async fn table_exists(&self, database: &str, table: &str) -> Result<bool, TestBinaryError> {
+        let pool = self.pool().await?;
+        let rows = sqlx::raw_sql(&format!(
+            "SELECT TABLE_NAME FROM information_schema.tables \
+             WHERE TABLE_SCHEMA = '{database}' AND TABLE_NAME = '{table}'"
+        ))
+        .fetch_all(&pool)
+        .await
+        .map_err(|e| TestBinaryError::InvalidState {
+            message: format!("Failed to query information_schema for {database}.{table}: {e}"),
+        })?;
+        Ok(!rows.is_empty())
+    }
+
+    async fn count_rows(&self, database: &str, table: &str) -> Result<i64, TestBinaryError> {
+        let pool = self.pool().await?;
+        // SELECT supports PREPARE in Doris but raw_sql is consistent with
+        // the rest of the fixture and avoids any future surprises.
+        use sqlx::Row;
+        let row = sqlx::raw_sql(&format!("SELECT COUNT(*) AS c FROM {database}.{table}"))
+            .fetch_one(&pool)
+            .await
+            .map_err(|e| TestBinaryError::InvalidState {
+                message: format!("Failed to count rows in {database}.{table}: {e}"),
+            })?;
+        // Doris returns COUNT(*) as BIGINT; sqlx decodes that to i64.
+        let count: i64 = row.try_get(0).map_err(|e| TestBinaryError::InvalidState {
+            message: format!("Failed to read count column: {e}"),
+        })?;
+        Ok(count)
+    }
+
+    async fn wait_for_rows(
+        &self,
+        database: &str,
+        table: &str,
+        expected: i64,
+    ) -> Result<i64, TestBinaryError> {
+        let mut last = 0i64;
+        for _ in 0..DEFAULT_POLL_ATTEMPTS {
+            // Doris is not strongly consistent for very freshly loaded data
+            // until the publish phase completes. Poll until the expected
+            // count is observed (or until we exhaust attempts).
+            if let Ok(c) = self.count_rows(database, table).await {
+                last = c;
+                if c >= expected {
+                    return Ok(c);
+                }
+            }
+            sleep(Duration::from_millis(DEFAULT_POLL_INTERVAL_MS)).await;
+        }
+        Err(TestBinaryError::InvalidState {
+            message: format!(
+                "Expected {expected} rows in {database}.{table} but observed {last} after {DEFAULT_POLL_ATTEMPTS} polls"
+            ),
+        })
+    }
+}
+
+fn build_connector_envs(fe_url: &str) -> HashMap<String, String> {
+    use integration::harness::seeds;
+
+    HashMap::from([
+        (ENV_SINK_FE_URL.to_string(), fe_url.to_string()),
+        (
+            ENV_SINK_DATABASE.to_string(),
+            DEFAULT_TEST_DATABASE.to_string(),
+        ),
+        (ENV_SINK_TABLE.to_string(), DEFAULT_TEST_TABLE.to_string()),
+        (ENV_SINK_USERNAME.to_string(), DEFAULT_USER.to_string()),
+        (ENV_SINK_PASSWORD.to_string(), DEFAULT_PASSWORD.to_string()),
+        (ENV_SINK_LABEL_PREFIX.to_string(), "iggy_test".to_string()),
+        (ENV_SINK_BATCH_SIZE.to_string(), "1000".to_string()),
+        (
+            ENV_SINK_STREAMS_0_STREAM.to_string(),
+            seeds::names::STREAM.to_string(),
+        ),
+        (
+            ENV_SINK_STREAMS_0_TOPICS.to_string(),
+            format!("[{}]", seeds::names::TOPIC),
+        ),
+        (ENV_SINK_STREAMS_0_SCHEMA.to_string(), "json".to_string()),
+        (
+            ENV_SINK_STREAMS_0_CONSUMER_GROUP.to_string(),
+            "doris_sink".to_string(),
+        ),
+        (
+            ENV_SINK_PATH.to_string(),
+            "../../target/debug/libiggy_connector_doris_sink".to_string(),
+        ),
+    ])
+}
+
+/// Doris fixture where the test is responsible for creating the table
+/// (e.g. to exercise different DDL or to verify failure when the table
+/// is absent).
+pub struct DorisSinkFixture {
+    container: DorisContainer,
+}
+
+impl DorisOps for DorisSinkFixture {
+    fn container(&self) -> &DorisContainer {
+        &self.container
+    }
+}
+
+#[async_trait]
+impl TestFixture for DorisSinkFixture {
+    async fn setup() -> Result<Self, TestBinaryError> {
+        let container = DorisContainer::start().await?;
+        Ok(Self { container })
+    }
+
+    fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+        build_connector_envs(&self.container.fe_url())
+    }
+}
+
+/// Doris fixture where the target table is pre-created during setup.
+/// Mirrors `QuickwitPreCreatedFixture` so tests can rely on the table
+/// being present from the moment the connector starts.
+pub struct DorisSinkPreCreatedFixture {
+    inner: DorisSinkFixture,
+}
+
+impl std::ops::Deref for DorisSinkPreCreatedFixture {
+    type Target = DorisSinkFixture;
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl DorisOps for DorisSinkPreCreatedFixture {
+    fn container(&self) -> &DorisContainer {
+        self.inner.container()
+    }
+}
+
+#[async_trait]
+impl TestFixture for DorisSinkPreCreatedFixture {
+    async fn setup() -> Result<Self, TestBinaryError> {
+        let inner = DorisSinkFixture::setup().await?;
+        inner
+            .create_table(DEFAULT_TEST_DATABASE, DEFAULT_TEST_TABLE)
+            .await?;
+        Ok(Self { inner })
+    }
+
+    fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+        self.inner.connectors_runtime_envs()
+    }
+}
+
+/// Pre-created table fixture that additionally tells the connector to use
+/// `max_filter_ratio = 0.5`, so a batch with up to half non-conforming rows
+/// still loads the conforming subset.
+pub struct DorisSinkMaxFilterRatioFixture {
+    inner: DorisSinkPreCreatedFixture,
+}
+
+impl std::ops::Deref for DorisSinkMaxFilterRatioFixture {
+    type Target = DorisSinkPreCreatedFixture;
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl DorisOps for DorisSinkMaxFilterRatioFixture {
+    fn container(&self) -> &DorisContainer {
+        self.inner.container()
+    }
+}
+
+#[async_trait]
+impl TestFixture for DorisSinkMaxFilterRatioFixture {
+    async fn setup() -> Result<Self, TestBinaryError> {
+        let inner = DorisSinkPreCreatedFixture::setup().await?;
+        Ok(Self { inner })
+    }
+
+    fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+        let mut envs = self.inner.connectors_runtime_envs();
+        envs.insert(ENV_SINK_MAX_FILTER_RATIO.to_string(), "0.5".to_string());
+        envs
+    }
+}
+
+/// Pre-creates a target table whose schema has an extra column (`calculated`)
+/// that does NOT exist in the JSON payload, and configures the connector with
+/// a `columns` Stream Load header that derives that column from `count`. The
+/// load fails without the `columns` config flowing through correctly, so a
+/// passing test proves the config wiring end-to-end.
+pub struct DorisSinkColumnsMappingFixture {
+    inner: DorisSinkFixture,
+}
+
+impl std::ops::Deref for DorisSinkColumnsMappingFixture {
+    type Target = DorisSinkFixture;
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl DorisOps for DorisSinkColumnsMappingFixture {
+    fn container(&self) -> &DorisContainer {
+        self.inner.container()
+    }
+}
+
+#[async_trait]
+impl TestFixture for DorisSinkColumnsMappingFixture {
+    async fn setup() -> Result<Self, TestBinaryError> {
+        let inner = DorisSinkFixture::setup().await?;
+        inner
+            .create_table_with_template(
+                DEFAULT_TEST_DATABASE,
+                DEFAULT_TEST_TABLE,
+                TEST_TABLE_WITH_CALCULATED_DDL_TEMPLATE,
+            )
+            .await?;
+        Ok(Self { inner })
+    }
+
+    fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+        let mut envs = self.inner.connectors_runtime_envs();
+        envs.insert(
+            ENV_SINK_COLUMNS.to_string(),
+            COLUMNS_MAPPING_HEADER.to_string(),
+        );
+        envs
+    }
+}
diff --git a/core/integration/tests/connectors/fixtures/doris/mod.rs b/core/integration/tests/connectors/fixtures/doris/mod.rs
new file mode 100644
index 0000000000..a295636551
--- /dev/null
+++ b/core/integration/tests/connectors/fixtures/doris/mod.rs
@@ -0,0 +1,25 @@
+/*
+ * 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.
+ */
+
+mod container;
+
+pub use container::{
+    DorisOps, DorisSinkColumnsMappingFixture, DorisSinkFixture, DorisSinkMaxFilterRatioFixture,
+    DorisSinkPreCreatedFixture,
+};
diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs
index a020fa42c7..f224c4213d 100644
--- a/core/integration/tests/connectors/fixtures/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -18,6 +18,7 @@
  */
 
 mod delta;
+mod doris;
 mod elasticsearch;
 mod http;
 mod iceberg;
@@ -28,6 +29,10 @@ mod quickwit;
 mod wiremock;
 
 pub use delta::{DeltaFixture, DeltaS3Fixture};
+pub use doris::{
+    DorisOps, DorisSinkColumnsMappingFixture, DorisSinkFixture, DorisSinkMaxFilterRatioFixture,
+    DorisSinkPreCreatedFixture,
+};
 pub use elasticsearch::{ElasticsearchSinkFixture, ElasticsearchSourcePreCreatedFixture};
 pub use http::{
     HttpSinkIndividualFixture, HttpSinkJsonArrayFixture, HttpSinkMultiTopicFixture,
diff --git a/core/integration/tests/connectors/mod.rs b/core/integration/tests/connectors/mod.rs
index 83600be271..01d1fbb5e3 100644
--- a/core/integration/tests/connectors/mod.rs
+++ b/core/integration/tests/connectors/mod.rs
@@ -19,6 +19,7 @@
 
 mod api;
 mod delta;
+mod doris;
 mod elasticsearch;
 mod fixtures;
 mod http;

From 9fd85d642f8f66bb50f6c8a953d68bd7106c1deb Mon Sep 17 00:00:00 2001
From: ryankert01 <ryankert01@gmail.com>
Date: Mon, 11 May 2026 20:45:09 +0800
Subject: [PATCH 2/6] fixup(connectors): address Doris sink review findings
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Addresses review feedback on the Doris sink connector before merge.

Correctness:
- Label format now appends an 8-hex blake3 of the *raw* stream/topic names,
  so streams that sanitize identically (e.g. `events.v1` vs `events_v1`)
  can no longer collide and silently dedupe against each other in Doris.
  Each variable-length segment is also truncated; total label is bounded
  under Doris's 128-char cap regardless of input length.
- `build_label` is now a pure `pub` free function. The integration test's
  manual label construction (used to verify server-side dedupe) now calls
  it directly, so the test cannot drift from the production format.
- `consume` tracks the *most severe* error across chunks via `record_error`:
  permanent shadows transient. The previous first-error strategy let a
  transient error from chunk N hide a permanent error from chunk M and
  caused the runtime to retry forever instead of routing to DLQ.
- HTTP 408 (Request Timeout) and 429 (Too Many Requests) classified as
  `CannotStoreData` (transient). They are 4xx but recoverable; the old
  code lumped them with all 4xx and DLQ'd retryable conditions.
- Parse failures on the response body now return `PermanentHttpError`.
  An unparseable 200-OK is almost always a Doris bug or proxy interference
  — retrying the same bytes won't help.

Security:
- `open()` rejects `database`/`table` values outside `[A-Za-z0-9_]+`.
  Doris would reject them server-side anyway, but rejecting at config-load
  also prevents path traversal in the `/api/{db}/{table}/_stream_load` URL.
- `open()` emits a `warn!` when `fe_url` is `http://` and the host is
  not loopback. README's new "Security notes" section spells out the
  trust boundary the manual-redirect-following implies (a compromised FE
  could exfiltrate credentials via a hostile `Location` header).
- Response body truncated to 4 KB at a UTF-8 boundary before being
  formatted into errors or logs, so a misbehaving proxy that returns a
  giant body cannot OOM the connector or flood logs.

Robustness:
- Explicit `connect_timeout` (5 s) so an unreachable FE fails fast
  instead of consuming the full request timeout on the handshake alone.
- `send_stream_load` takes `bytes::Bytes`; clones inside the redirect
  loop are now refcount bumps instead of full `Vec<u8>` copies.

Observability:
- `warn!` when Doris reports `number_filtered_rows > 0` — schema drift
  in upstream messages was previously logged at `info!` and easy to miss.
- Per-batch success log demoted from `info!` to `debug!`.
- README documents `Expect: 100-continue`, `label_keep_max_second`
  guidance, and the filtered-row alert.

Tests: 21 unit tests pass (was 13, added 8 covering hash-suffix label
collision resistance, label length cap, severity ordering, identifier
validation, and log truncation). All 6 testcontainer integration tests
pass against a real Doris all-in-one image.
---
 Cargo.lock                                    |   5 +-
 core/connectors/sinks/doris_sink/Cargo.toml   |   2 +
 core/connectors/sinks/doris_sink/README.md    |  19 +-
 core/connectors/sinks/doris_sink/src/lib.rs   | 346 +++++++++++++++---
 core/integration/Cargo.toml                   |   3 +
 .../tests/connectors/doris/doris_sink.rs      |  26 +-
 6 files changed, 330 insertions(+), 71 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index f5080e3de0..3ee05b419d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6568,7 +6568,9 @@ name = "iggy_connector_doris_sink"
 version = "0.1.0"
 dependencies = [
  "async-trait",
- "base64 0.22.1",
+ "base64",
+ "blake3",
+ "bytes",
  "iggy_connector_sdk",
  "reqwest 0.13.3",
  "secrecy",
@@ -7046,6 +7048,7 @@ dependencies = [
  "iggy-cli",
  "iggy_binary_protocol",
  "iggy_common",
+ "iggy_connector_doris_sink",
  "iggy_connector_sdk",
  "jsonwebtoken",
  "keyring",
diff --git a/core/connectors/sinks/doris_sink/Cargo.toml b/core/connectors/sinks/doris_sink/Cargo.toml
index f2795b7ffe..385cbb50c3 100644
--- a/core/connectors/sinks/doris_sink/Cargo.toml
+++ b/core/connectors/sinks/doris_sink/Cargo.toml
@@ -34,6 +34,8 @@ crate-type = ["cdylib", "lib"]
 [dependencies]
 async-trait = { workspace = true }
 base64 = { workspace = true }
+blake3 = { workspace = true }
+bytes = { workspace = true }
 iggy_connector_sdk = { workspace = true }
 reqwest = { workspace = true }
 secrecy = { workspace = true }
diff --git a/core/connectors/sinks/doris_sink/README.md b/core/connectors/sinks/doris_sink/README.md
index d3633337f2..0f2efba475 100644
--- a/core/connectors/sinks/doris_sink/README.md
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -11,14 +11,14 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
 ## How it works
 
 1. For each batch of messages, the connector serializes the JSON payloads into a JSON array.
-2. It computes a deterministic Stream Load `label` of the form `{label_prefix}-{stream}-{topic}-{partition}-{first_offset}-{last_offset}`. Doris dedupes loads by label inside its `label_keep_max_second` window, so a replayed batch (after restart, retry, etc.) is silently absorbed instead of producing duplicates.
-3. It `PUT`s the batch to `{fe_url}/api/{database}/{table}/_stream_load` with HTTP Basic auth and the headers `format: json`, `strip_outer_array: true`, `label: <label>`.
+2. It computes a deterministic Stream Load `label` of the form `{label_prefix}-{stream}_{hash8}-{topic}_{hash8}-{partition}-{first_offset}-{last_offset}`. Each variable-length segment carries a 32-bit blake3 hash of the raw name so that names which sanitize to the same string (e.g. `events.v1` vs `events_v1`) cannot collide. The total label is bounded under Doris's 128-char cap regardless of input length. Doris dedupes loads by label inside its `label_keep_max_second` window, so a replayed batch (after restart, retry, etc.) is silently absorbed instead of producing duplicates.
+3. It `PUT`s the batch to `{fe_url}/api/{database}/{table}/_stream_load` with HTTP Basic auth and the headers `Expect: 100-continue`, `format: json`, `strip_outer_array: true`, `label: <label>`. (`Expect: 100-continue` lets Doris reject auth/4xx failures before the connector uploads the whole body — important for large batches and required if a reverse proxy sits in front of Doris.)
 4. The Doris frontend (FE) responds with a `307 Temporary Redirect` to a backend (BE). The connector follows the redirect manually so that the `Authorization` header is preserved across the hop (`reqwest`'s default policy strips it on cross-host redirects).
 5. The HTTP body is parsed as JSON and the `Status` field decides the outcome:
    - `Success` → batch accepted.
    - `Label Already Exists` → idempotent replay, treated as success.
-   - `Publish Timeout` or HTTP `5xx` → transient error (`Error::CannotStoreData`); the runtime can retry.
-   - `Fail` or HTTP `4xx` → permanent error (`Error::PermanentHttpError`); retrying is not useful.
+   - `Publish Timeout` or HTTP `5xx`/`408`/`429` → transient error (`Error::CannotStoreData`); the runtime can retry.
+   - `Fail`, any other `4xx`, or an unparseable response body → permanent error (`Error::PermanentHttpError`); retrying is not useful.
 
 ## Configuration
 
@@ -66,6 +66,17 @@ batch_size = 1000
 timeout_secs = 30
 ```
 
+## Security notes
+
+- **Use `https://` in production.** The connector accepts `http://` URLs and logs a `warn!` when `fe_url` points at a non-loopback host over plain HTTP, but it does not refuse. Over `http://`, the HTTP Basic credentials travel in cleartext.
+- **Trust boundary on the FE.** The connector intentionally preserves the `Authorization` header across the FE → BE 307 redirect (reqwest would otherwise strip it on cross-host redirects). A compromised or MITM'd FE could exfiltrate credentials by responding with `Location: http://attacker/`. The `http://` warning above is the primary mitigation; deploy Doris over TLS in hostile networks.
+- **`columns` and `where` are SQL-expression pass-throughs.** Whatever you put in those config fields is forwarded verbatim to Doris's Stream Load and evaluated as a SQL expression. Keep this config trusted.
+
+## Operational guidance
+
+- **`label_keep_max_second`.** Idempotent replay relies on Doris retaining each label for at least as long as it could take the Iggy runtime to redrive a failed batch. The Doris default is 3 days, which is conservative. If you set this lower on the Doris side, make sure your runtime retry budget fits inside the window — once a label expires, a replay re-loads instead of deduping, producing duplicate rows.
+- **Filtered-row alerts.** When Doris reports `number_filtered_rows > 0`, the connector emits a `warn!`. This is your signal that upstream message shapes have drifted from the table schema; alert on it.
+
 ## Limitations (v1)
 
 - JSON payload only. CSV and raw-text payloads are not supported yet.
diff --git a/core/connectors/sinks/doris_sink/src/lib.rs b/core/connectors/sinks/doris_sink/src/lib.rs
index ff5a39ac84..d5589f7b74 100644
--- a/core/connectors/sinks/doris_sink/src/lib.rs
+++ b/core/connectors/sinks/doris_sink/src/lib.rs
@@ -18,6 +18,7 @@
 
 use async_trait::async_trait;
 use base64::{Engine as _, engine::general_purpose};
+use bytes::Bytes;
 use iggy_connector_sdk::{
     ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector,
 };
@@ -32,10 +33,22 @@ sink_connector!(DorisSink);
 const DEFAULT_LABEL_PREFIX: &str = "iggy";
 const DEFAULT_BATCH_SIZE: u32 = 1000;
 const DEFAULT_TIMEOUT_SECS: u64 = 30;
+// A bounded TCP handshake timeout so an unreachable FE fails fast instead of
+// eating the entire request-timeout budget on connect alone.
+const CONNECT_TIMEOUT_SECS: u64 = 5;
 // Doris's FE Stream Load returns 307 to a BE; reqwest's default policy strips
 // the Authorization header on cross-host redirects, so we follow redirects
 // manually. This cap protects against a misbehaving cluster looping.
 const MAX_REDIRECTS: u8 = 5;
+// Per Doris docs, Stream Load labels must be 1..=128 chars matching
+// `[A-Za-z0-9_-]`. These caps keep the worst-case label well under that limit.
+const MAX_LABEL_PREFIX_LEN: usize = 16;
+const MAX_LABEL_NAME_LEN: usize = 16;
+const LABEL_HASH_HEX_LEN: usize = 8;
+// Doris Stream Load responses are normally tiny JSON blobs (~200 B). Cap the
+// portion we keep for logs/errors so a misbehaving proxy that returns a giant
+// body cannot OOM the connector or flood logs.
+const MAX_RESPONSE_LOG_BYTES: usize = 4096;
 
 #[derive(Debug)]
 pub struct DorisSink {
@@ -100,6 +113,7 @@ impl DorisSink {
         reqwest::Client::builder()
             .redirect(reqwest::redirect::Policy::none())
             .timeout(timeout)
+            .connect_timeout(Duration::from_secs(CONNECT_TIMEOUT_SECS))
             .build()
             .map_err(|e| Error::InitError(format!("Failed to build Doris HTTP client: {e}")))
     }
@@ -124,9 +138,9 @@ impl DorisSink {
         self.config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1) as usize
     }
 
-    /// Build the deterministic label for one batch. Same inputs always produce
-    /// the same label, so a replayed batch (after restart, retry, etc.) gets
-    /// deduplicated by Doris within its label-keep window.
+    /// Build the deterministic label for one batch. Delegates to the pure
+    /// `build_label` free function so integration tests can reproduce the
+    /// exact same label without instantiating a `DorisSink`.
     fn build_label(
         &self,
         topic_metadata: &TopicMetadata,
@@ -134,11 +148,10 @@ impl DorisSink {
         first_offset: u64,
         last_offset: u64,
     ) -> String {
-        format!(
-            "{}-{}-{}-{}-{}-{}",
-            sanitize_label(self.label_prefix()),
-            sanitize_label(&topic_metadata.stream),
-            sanitize_label(&topic_metadata.topic),
+        build_label(
+            self.label_prefix(),
+            &topic_metadata.stream,
+            &topic_metadata.topic,
             partition_id,
             first_offset,
             last_offset,
@@ -148,7 +161,7 @@ impl DorisSink {
     async fn send_stream_load(
         &self,
         label: &str,
-        body: Vec<u8>,
+        body: Bytes,
     ) -> Result<StreamLoadResponse, Error> {
         let client = self.client.as_ref().ok_or_else(|| {
             Error::InitError(format!(
@@ -212,17 +225,22 @@ impl DorisSink {
             }
 
             let response_text = response.text().await.unwrap_or_default();
+            let response_for_log = truncate_for_log(&response_text, MAX_RESPONSE_LOG_BYTES);
 
             if !status.is_success() {
                 let msg = format!(
-                    "Doris sink ID {} stream load returned HTTP {status}: {response_text}",
+                    "Doris sink ID {} stream load returned HTTP {status}: {response_for_log}",
                     self.id
                 );
                 error!("{msg}");
-                return Err(if status.is_server_error() {
-                    Error::CannotStoreData(msg)
-                } else {
-                    Error::PermanentHttpError(msg)
+                // 408 (Request Timeout) and 429 (Too Many Requests) are 4xx
+                // but transient — they should be retried, not DLQ'd.
+                return Err(match status {
+                    StatusCode::REQUEST_TIMEOUT | StatusCode::TOO_MANY_REQUESTS => {
+                        Error::CannotStoreData(msg)
+                    }
+                    s if s.is_server_error() => Error::CannotStoreData(msg),
+                    _ => Error::PermanentHttpError(msg),
                 });
             }
 
@@ -231,9 +249,9 @@ impl DorisSink {
     }
 }
 
-/// Replace Doris-label-illegal characters with `_`. Doris allows
-/// [A-Za-z0-9_-]. Anything else -> `_`.
-fn sanitize_label(value: &str) -> String {
+/// Replace Doris-label-illegal characters with `_` and cap the result at
+/// `max_len` chars. Doris labels allow `[A-Za-z0-9_-]` only.
+fn sanitize_segment(value: &str, max_len: usize) -> String {
     value
         .chars()
         .map(|c| {
@@ -243,16 +261,113 @@ fn sanitize_label(value: &str) -> String {
                 '_'
             }
         })
+        .take(max_len)
         .collect()
 }
 
+/// Sanitize + truncate a stream/topic name and append a short blake3 hash
+/// of the *raw* (unsanitized) name. The hash disambiguates names that
+/// sanitize to the same string (e.g. `events.v1` vs `events_v1`), which
+/// would otherwise produce identical labels and cause silent data loss
+/// via Doris's server-side label dedupe.
+fn label_segment_with_hash(value: &str) -> String {
+    let san = sanitize_segment(value, MAX_LABEL_NAME_LEN);
+    let hash = blake3::hash(value.as_bytes()).to_hex();
+    format!("{san}_{}", &hash.as_str()[..LABEL_HASH_HEX_LEN])
+}
+
+/// Pure label builder. Format:
+/// `{prefix}-{stream_san}_{hash8}-{topic_san}_{hash8}-{partition}-{first}-{last}`.
+///
+/// Exposed as a `pub` free function so integration tests can produce the
+/// exact same label without instantiating a `DorisSink`. The format is
+/// engineered to:
+///   1. Bound the total length under Doris's 128-char label cap regardless
+///      of input name length.
+///   2. Eliminate the silent-collision risk between names that sanitize to
+///      the same string (the hash is over the raw name).
+pub fn build_label(
+    prefix: &str,
+    stream: &str,
+    topic: &str,
+    partition_id: u32,
+    first_offset: u64,
+    last_offset: u64,
+) -> String {
+    format!(
+        "{}-{}-{}-{}-{}-{}",
+        sanitize_segment(prefix, MAX_LABEL_PREFIX_LEN),
+        label_segment_with_hash(stream),
+        label_segment_with_hash(topic),
+        partition_id,
+        first_offset,
+        last_offset,
+    )
+}
+
+/// Truncate `s` at the largest char boundary `<= max_bytes` and append a
+/// marker that records the original size. Used to bound the portion of an
+/// HTTP response body that lands in logs or error variants.
+fn truncate_for_log(s: &str, max_bytes: usize) -> String {
+    if s.len() <= max_bytes {
+        return s.to_string();
+    }
+    let mut end = max_bytes;
+    while end > 0 && !s.is_char_boundary(end) {
+        end -= 1;
+    }
+    format!("{}...(truncated, total {} bytes)", &s[..end], s.len())
+}
+
 fn parse_stream_load_response(body: &str) -> Result<StreamLoadResponse, Error> {
-    let parsed: StreamLoadResponse = serde_json::from_str(body).map_err(|e| {
-        Error::CannotStoreData(format!(
-            "Failed to parse Doris stream load response: {e}. Body: {body}"
+    // An unparseable 200-OK body is almost always a Doris bug, a proxy that
+    // injected HTML, or a future schema change — none of those are cured by
+    // retrying the same bytes. Default to permanent so the runtime DLQs the
+    // batch instead of looping forever.
+    serde_json::from_str(body).map_err(|e| {
+        Error::PermanentHttpError(format!(
+            "Failed to parse Doris stream load response: {e}. Body: {}",
+            truncate_for_log(body, MAX_RESPONSE_LOG_BYTES)
         ))
-    })?;
-    Ok(parsed)
+    })
+}
+
+/// True for errors the runtime must NOT retry (schema mismatch, bad config,
+/// permanent HTTP error, etc.). Used by `consume` to pick the most-severe
+/// error across chunks: a transient error from chunk N must NOT shadow a
+/// permanent error from chunk M, or the runtime would retry forever instead
+/// of routing the batch to its DLQ path.
+fn is_permanent_error(e: &Error) -> bool {
+    matches!(
+        e,
+        Error::PermanentHttpError(_) | Error::InvalidConfigValue(_) | Error::InvalidPayloadType
+    )
+}
+
+/// Replace `slot` with `e` either when it's still empty or when `e` is
+/// permanent and the existing entry is transient.
+fn record_error(slot: &mut Option<Error>, e: Error) {
+    let replace = match slot.as_ref() {
+        None => true,
+        Some(existing) => is_permanent_error(&e) && !is_permanent_error(existing),
+    };
+    if replace {
+        *slot = Some(e);
+    }
+}
+
+fn validate_identifier(name: &str, field: &str, id: u32) -> Result<(), Error> {
+    if name.is_empty() {
+        return Err(Error::InvalidConfigValue(format!(
+            "Doris sink ID {id}: {field} must not be empty"
+        )));
+    }
+    if !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+        return Err(Error::InvalidConfigValue(format!(
+            "Doris sink ID {id}: {field} '{name}' must match [A-Za-z0-9_]+ (Doris identifier rules)"
+        )));
+    }
+    Ok(())
 }
 
 fn classify_status(response: &StreamLoadResponse) -> Result<(), Error> {
@@ -291,13 +406,41 @@ impl Sink for DorisSink {
     async fn open(&mut self) -> Result<(), Error> {
         // Validate the URL up front so a bad config fails at startup rather
         // than on the first batch.
-        reqwest::Url::parse(&self.stream_load_url()).map_err(|e| {
+        let parsed_url = reqwest::Url::parse(&self.stream_load_url()).map_err(|e| {
             Error::InvalidConfigValue(format!(
                 "Doris sink ID {} has invalid fe_url '{}': {e}",
                 self.id, self.config.fe_url
             ))
         })?;
 
+        // Reject `database`/`table` outside Doris's identifier alphabet.
+        // Doris would reject them server-side anyway, but rejecting here
+        // also prevents path traversal in the constructed
+        // `/api/{db}/{table}/_stream_load` URL.
+        validate_identifier(&self.config.database, "database", self.id)?;
+        validate_identifier(&self.config.table, "table", self.id)?;
+
+        // Warn when credentials would travel in cleartext.
+        //
+        // NOTE on trust model: this connector intentionally preserves the
+        // Authorization header across the FE -> BE 307 redirect (reqwest
+        // would otherwise strip it on cross-host redirects). That means a
+        // compromised or MITM'd FE could exfiltrate credentials by responding
+        // with `Location: http://attacker/`. The cleartext warning is the
+        // primary mitigation; for hostile networks, deploy Doris over TLS.
+        if parsed_url.scheme().eq_ignore_ascii_case("http") {
+            let host = parsed_url.host_str().unwrap_or("");
+            let is_loopback = host == "localhost" || host == "127.0.0.1" || host == "::1";
+            if !is_loopback {
+                warn!(
+                    "Doris sink ID {} is configured with http:// to non-loopback host '{}'; \
+                     credentials and message data will be transmitted in cleartext. \
+                     Use https:// in production.",
+                    self.id, host
+                );
+            }
+        }
+
         self.client = Some(self.build_client()?);
 
         info!(
@@ -324,7 +467,7 @@ impl Sink for DorisSink {
         );
 
         let batch_size = self.batch_size();
-        let mut first_error: Option<Error> = None;
+        let mut worst_error: Option<Error> = None;
 
         for chunk in messages.chunks(batch_size) {
             let json_values: Vec<&simd_json::OwnedValue> = chunk
@@ -346,13 +489,12 @@ impl Sink for DorisSink {
             }
 
             let body = match simd_json::to_vec(&json_values) {
-                Ok(b) => b,
+                Ok(b) => Bytes::from(b),
                 Err(e) => {
-                    let err =
-                        Error::CannotStoreData(format!("Failed to serialize batch for Doris: {e}"));
-                    if first_error.is_none() {
-                        first_error = Some(err);
-                    }
+                    record_error(
+                        &mut worst_error,
+                        Error::CannotStoreData(format!("Failed to serialize batch for Doris: {e}")),
+                    );
                     continue;
                 }
             };
@@ -369,32 +511,42 @@ impl Sink for DorisSink {
             match self.send_stream_load(&label, body).await {
                 Ok(response) => match classify_status(&response) {
                     Ok(()) => {
-                        info!(
-                            "Doris sink ID {} loaded {} rows (filtered={}) into {}.{} (label={label})",
-                            self.id,
-                            response.number_loaded_rows,
-                            response.number_filtered_rows,
-                            self.config.database,
-                            self.config.table,
-                        );
+                        if response.number_filtered_rows > 0 {
+                            // Filtered rows usually indicate schema drift in
+                            // upstream messages. Surface above debug so
+                            // operators can alert on it.
+                            warn!(
+                                "Doris sink ID {} loaded {} rows but FILTERED {} rows for {}.{} (label={label}); \
+                                 likely schema drift upstream",
+                                self.id,
+                                response.number_loaded_rows,
+                                response.number_filtered_rows,
+                                self.config.database,
+                                self.config.table,
+                            );
+                        } else {
+                            debug!(
+                                "Doris sink ID {} loaded {} rows into {}.{} (label={label})",
+                                self.id,
+                                response.number_loaded_rows,
+                                self.config.database,
+                                self.config.table,
+                            );
+                        }
                     }
                     Err(e) => {
                         error!("Doris sink ID {} batch failed: {e}", self.id);
-                        if first_error.is_none() {
-                            first_error = Some(e);
-                        }
+                        record_error(&mut worst_error, e);
                     }
                 },
                 Err(e) => {
                     error!("Doris sink ID {} HTTP failed: {e}", self.id);
-                    if first_error.is_none() {
-                        first_error = Some(e);
-                    }
+                    record_error(&mut worst_error, e);
                 }
             }
         }
 
-        if let Some(err) = first_error {
+        if let Some(err) = worst_error {
             return Err(err);
         }
         Ok(())
@@ -453,7 +605,17 @@ mod tests {
         let a = sink.build_label(&topic, 7, 100, 199);
         let b = sink.build_label(&topic, 7, 100, 199);
         assert_eq!(a, b);
-        assert_eq!(a, "iggy-events-orders-7-100-199");
+        // Format: {prefix}-{stream_san}_{hash8}-{topic_san}_{hash8}-{partition}-{first}-{last}
+        let parts: Vec<&str> = a.split('-').collect();
+        assert_eq!(parts.len(), 6);
+        assert_eq!(parts[0], "iggy");
+        assert!(parts[1].starts_with("events_"));
+        assert_eq!(parts[1].len(), "events_".len() + LABEL_HASH_HEX_LEN);
+        assert!(parts[2].starts_with("orders_"));
+        assert_eq!(parts[2].len(), "orders_".len() + LABEL_HASH_HEX_LEN);
+        assert_eq!(parts[3], "7");
+        assert_eq!(parts[4], "100");
+        assert_eq!(parts[5], "199");
     }
 
     #[test]
@@ -467,7 +629,48 @@ mod tests {
         // dots and slashes are not allowed in Doris labels.
         assert!(!label.contains('.'));
         assert!(!label.contains('/'));
-        assert_eq!(label, "iggy-events_v1-orders_inbound-0-0-0");
+    }
+
+    #[test]
+    fn label_disambiguates_names_that_sanitize_identically() {
+        // The whole point of the hash suffix: `events.v1` and `events_v1`
+        // collapse to the same sanitized form, but the hash is over the raw
+        // name so the final labels differ. Without this, two streams could
+        // silently dedupe against each other in Doris.
+        let sink = DorisSink::new(1, make_config());
+        let dotted = TopicMetadata {
+            stream: "events.v1".into(),
+            topic: "orders".into(),
+        };
+        let underscored = TopicMetadata {
+            stream: "events_v1".into(),
+            topic: "orders".into(),
+        };
+        assert_ne!(
+            sink.build_label(&dotted, 0, 0, 0),
+            sink.build_label(&underscored, 0, 0, 0),
+            "labels must NOT collide for names that sanitize to the same string"
+        );
+    }
+
+    #[test]
+    fn label_stays_under_doris_128_char_cap() {
+        // Doris caps Stream Load labels at 128 chars. Build the worst case
+        // permitted by the connector: 100-char prefix/stream/topic (all of
+        // which get truncated), u64::MAX offsets, u32::MAX partition.
+        let mut cfg = make_config();
+        cfg.label_prefix = Some("p".repeat(100));
+        let sink = DorisSink::new(1, cfg);
+        let topic = TopicMetadata {
+            stream: "s".repeat(100),
+            topic: "t".repeat(100),
+        };
+        let label = sink.build_label(&topic, u32::MAX, u64::MAX, u64::MAX);
+        assert!(
+            label.len() <= 128,
+            "label exceeds Doris's 128-char cap: {} chars: {label}",
+            label.len()
+        );
     }
 
     #[test]
@@ -551,9 +754,54 @@ mod tests {
     }
 
     #[test]
-    fn parse_stream_load_response_rejects_garbage() {
+    fn parse_stream_load_response_rejects_garbage_as_permanent() {
+        // An unparseable body must surface as PermanentHttpError so the
+        // runtime DLQs the batch instead of retrying the same garbage forever.
         let body = "not json";
-        assert!(parse_stream_load_response(body).is_err());
+        assert!(matches!(
+            parse_stream_load_response(body).unwrap_err(),
+            Error::PermanentHttpError(_)
+        ));
+    }
+
+    #[test]
+    fn validate_identifier_rejects_path_traversal() {
+        assert!(validate_identifier("../admin", "database", 1).is_err());
+        assert!(validate_identifier("foo/bar", "table", 1).is_err());
+        assert!(validate_identifier("", "database", 1).is_err());
+        assert!(validate_identifier("ok_name_1", "database", 1).is_ok());
+    }
+
+    #[test]
+    fn truncate_for_log_caps_long_input() {
+        let long = "x".repeat(10_000);
+        let truncated = truncate_for_log(&long, 100);
+        assert!(truncated.len() <= 100 + "...(truncated, total 10000 bytes)".len());
+        assert!(truncated.contains("(truncated"));
+    }
+
+    #[test]
+    fn truncate_for_log_passes_short_input_through() {
+        let short = "hello";
+        assert_eq!(truncate_for_log(short, 100), "hello");
+    }
+
+    #[test]
+    fn record_error_prefers_permanent_over_transient() {
+        let mut slot: Option<Error> = None;
+        record_error(&mut slot, Error::CannotStoreData("transient".into()));
+        record_error(&mut slot, Error::PermanentHttpError("permanent".into()));
+        // Permanent must shadow transient — otherwise the runtime retries
+        // forever instead of routing to DLQ.
+        assert!(matches!(slot, Some(Error::PermanentHttpError(_))));
+    }
+
+    #[test]
+    fn record_error_keeps_permanent_when_transient_arrives_later() {
+        let mut slot: Option<Error> = None;
+        record_error(&mut slot, Error::PermanentHttpError("p".into()));
+        record_error(&mut slot, Error::CannotStoreData("t".into()));
+        assert!(matches!(slot, Some(Error::PermanentHttpError(_))));
     }
 
     #[test]
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index 9795e41daf..d88031ff37 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -48,6 +48,9 @@ iggy-cli = { workspace = true }
 iggy_binary_protocol = { workspace = true }
 iggy_common = { workspace = true }
 iggy_connector_sdk = { workspace = true, features = ["api"] }
+# Path-dep only so the Doris integration test can reuse the connector's pure
+# `build_label` function — keeping the test and production label format in lock-step.
+iggy_connector_doris_sink = { path = "../connectors/sinks/doris_sink" }
 jsonwebtoken = { workspace = true }
 keyring = { workspace = true }
 lazy_static = { workspace = true }
diff --git a/core/integration/tests/connectors/doris/doris_sink.rs b/core/integration/tests/connectors/doris/doris_sink.rs
index d5d69f2b41..31b075efb7 100644
--- a/core/integration/tests/connectors/doris/doris_sink.rs
+++ b/core/integration/tests/connectors/doris/doris_sink.rs
@@ -254,11 +254,15 @@ async fn given_replayed_label_should_dedupe(harness: &TestHarness, fixture: Dori
     assert_eq!(row_before, message_count as i64);
 
     // Replay the same offsets => same label => Doris dedupes server-side.
-    let label = format!(
-        "iggy_test-{}-{}-0-0-{}",
-        sanitize(seeds::names::STREAM),
-        sanitize(seeds::names::TOPIC),
-        message_count - 1,
+    // Reuse the connector's own `build_label` so the test cannot drift from
+    // the production label format.
+    let label = iggy_connector_doris_sink::build_label(
+        "iggy_test",
+        seeds::names::STREAM,
+        seeds::names::TOPIC,
+        0,
+        0,
+        (message_count - 1) as u64,
     );
 
     let body = serde_json::to_vec(&test_messages).expect("serialize replay body");
@@ -484,15 +488,3 @@ async fn given_columns_config_should_apply_derived_expression(
     );
 }
 
-fn sanitize(value: &str) -> String {
-    value
-        .chars()
-        .map(|c| {
-            if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
-                c
-            } else {
-                '_'
-            }
-        })
-        .collect()
-}

From 61327c01ced7722284062f098b50279e4eada024 Mon Sep 17 00:00:00 2001
From: ryankert01 <ryankert01@gmail.com>
Date: Mon, 11 May 2026 21:20:09 +0800
Subject: [PATCH 3/6] docs(doris): cover identifier validation, non-JSON
 payload behavior, 308 redirects
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Three small README gaps surfaced during a re-read against the post-review
code:

- `database` / `table` must match `[A-Za-z0-9_]+`. The connector rejects
  anything else at startup with `Error::InvalidConfigValue` — surface the
  constraint where operators look for it (Requirements + Configuration
  table).
- Non-JSON payloads are dropped with `warn!` and the offset advances past
  them. That is silent data loss, not a recoverable skip, so the README
  now spells it out instead of glossing it as "skipped with a warning".
- `308 Permanent Redirect` is followed in addition to `307` (defensive),
  and the redirect cap of 5 is documented.
---
 core/connectors/sinks/doris_sink/README.md | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/core/connectors/sinks/doris_sink/README.md b/core/connectors/sinks/doris_sink/README.md
index 0f2efba475..d2e4839894 100644
--- a/core/connectors/sinks/doris_sink/README.md
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -5,7 +5,8 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
 ## Requirements
 
 - The target Doris **database and table must be pre-created** before enabling the sink. The connector never issues DDL.
-- Messages must arrive with `Payload::Json` (i.e. the configured stream schema is `json`). Other payload types are skipped with a warning.
+- `database` and `table` config values must match `[A-Za-z0-9_]+`. Anything else is rejected at startup with `Error::InvalidConfigValue` — this also prevents path traversal in the constructed `/api/{db}/{table}/_stream_load` URL.
+- Messages must arrive with `Payload::Json` (i.e. the configured stream schema is `json`). Other payload types are dropped with a `warn!` and the consumer offset advances past them — this is silent data loss for any non-JSON message, so the upstream schema must be guaranteed JSON.
 - The Iggy message JSON shape must match the target table columns. Use the optional `columns` plugin setting if the column order differs from the JSON keys.
 
 ## How it works
@@ -13,7 +14,7 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
 1. For each batch of messages, the connector serializes the JSON payloads into a JSON array.
 2. It computes a deterministic Stream Load `label` of the form `{label_prefix}-{stream}_{hash8}-{topic}_{hash8}-{partition}-{first_offset}-{last_offset}`. Each variable-length segment carries a 32-bit blake3 hash of the raw name so that names which sanitize to the same string (e.g. `events.v1` vs `events_v1`) cannot collide. The total label is bounded under Doris's 128-char cap regardless of input length. Doris dedupes loads by label inside its `label_keep_max_second` window, so a replayed batch (after restart, retry, etc.) is silently absorbed instead of producing duplicates.
 3. It `PUT`s the batch to `{fe_url}/api/{database}/{table}/_stream_load` with HTTP Basic auth and the headers `Expect: 100-continue`, `format: json`, `strip_outer_array: true`, `label: <label>`. (`Expect: 100-continue` lets Doris reject auth/4xx failures before the connector uploads the whole body — important for large batches and required if a reverse proxy sits in front of Doris.)
-4. The Doris frontend (FE) responds with a `307 Temporary Redirect` to a backend (BE). The connector follows the redirect manually so that the `Authorization` header is preserved across the hop (`reqwest`'s default policy strips it on cross-host redirects).
+4. The Doris frontend (FE) responds with a `307 Temporary Redirect` to a backend (BE). The connector follows the redirect manually so that the `Authorization` header is preserved across the hop (`reqwest`'s default policy strips it on cross-host redirects). `308 Permanent Redirect` is also followed as a defensive measure; redirects beyond a hard cap of 5 are rejected as `HttpRequestFailed`.
 5. The HTTP body is parsed as JSON and the `Status` field decides the outcome:
    - `Success` → batch accepted.
    - `Label Already Exists` → idempotent replay, treated as success.
@@ -25,8 +26,8 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
 | Field | Required | Default | Description |
 | --- | --- | --- | --- |
 | `fe_url` | yes | — | Doris frontend HTTP base URL, e.g. `http://localhost:8030`. |
-| `database` | yes | — | Target database. |
-| `table` | yes | — | Target table. |
+| `database` | yes | — | Target database. Must match `[A-Za-z0-9_]+`. |
+| `table` | yes | — | Target table. Must match `[A-Za-z0-9_]+`. |
 | `username` | yes | — | Doris user with `LOAD_PRIV` on the table. |
 | `password` | yes | — | Doris user password. Stored as a `secrecy::SecretString` and never logged. |
 | `label_prefix` | no | `iggy` | Prefix for the deterministic Stream Load label. |

From 2b01642dc9ad6e4f67ab259540b1893bc13853d9 Mon Sep 17 00:00:00 2001
From: ryankert01 <ryankert01@gmail.com>
Date: Mon, 11 May 2026 21:56:50 +0800
Subject: [PATCH 4/6] docs(doris): reframe 'Limitations (v1)' as 'Limitations
 (todo)'

---
 core/connectors/sinks/doris_sink/README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/core/connectors/sinks/doris_sink/README.md b/core/connectors/sinks/doris_sink/README.md
index d2e4839894..522eb83be4 100644
--- a/core/connectors/sinks/doris_sink/README.md
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -78,7 +78,7 @@ timeout_secs = 30
 - **`label_keep_max_second`.** Idempotent replay relies on Doris retaining each label for at least as long as it could take the Iggy runtime to redrive a failed batch. The Doris default is 3 days, which is conservative. If you set this lower on the Doris side, make sure your runtime retry budget fits inside the window — once a label expires, a replay re-loads instead of deduping, producing duplicate rows.
 - **Filtered-row alerts.** When Doris reports `number_filtered_rows > 0`, the connector emits a `warn!`. This is your signal that upstream message shapes have drifted from the table schema; alert on it.
 
-## Limitations (v1)
+## Limitations (todo)
 
 - JSON payload only. CSV and raw-text payloads are not supported yet.
 - HTTP Basic auth only.

From 470c85ca7b4491b024e191ec946e56a6d6b4f936 Mon Sep 17 00:00:00 2001
From: ryankert01 <ryankert01@gmail.com>
Date: Tue, 12 May 2026 05:24:30 +0800
Subject: [PATCH 5/6] =?UTF-8?q?chore(doris):=20fix=20CI=20lints=20?=
 =?UTF-8?q?=E2=80=94=20typos,=20line=20length,=20fmt,=20dep=20sort?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Four pre-merge check failures from the previous commit, all mechanical:

- typos: `unparseable` → `unparsable` (1 in README, 2 in lib.rs comments).
- markdown lint MD013: README's label-format bullet was 583 chars; split
  into a parent bullet + 3 sub-bullets, all within the 500-char cap.
- rustfmt: trailing blank line in the integration test after the recent
  removal of the local `sanitize` helper.
- cargo sort: `iggy_connector_doris_sink` was added under
  `iggy_connector_sdk` in core/integration/Cargo.toml; reordered so the
  dependency list stays alphabetical.

No behavior change. 21 unit tests still pass; `cargo fmt --check` and
`cargo sort --workspace --check` both clean locally.
---
 core/connectors/sinks/doris_sink/README.md            | 7 +++++--
 core/connectors/sinks/doris_sink/src/lib.rs           | 4 ++--
 core/integration/Cargo.toml                           | 2 +-
 core/integration/tests/connectors/doris/doris_sink.rs | 1 -
 4 files changed, 8 insertions(+), 6 deletions(-)

diff --git a/core/connectors/sinks/doris_sink/README.md b/core/connectors/sinks/doris_sink/README.md
index 522eb83be4..89864bc688 100644
--- a/core/connectors/sinks/doris_sink/README.md
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -12,14 +12,17 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
 ## How it works
 
 1. For each batch of messages, the connector serializes the JSON payloads into a JSON array.
-2. It computes a deterministic Stream Load `label` of the form `{label_prefix}-{stream}_{hash8}-{topic}_{hash8}-{partition}-{first_offset}-{last_offset}`. Each variable-length segment carries a 32-bit blake3 hash of the raw name so that names which sanitize to the same string (e.g. `events.v1` vs `events_v1`) cannot collide. The total label is bounded under Doris's 128-char cap regardless of input length. Doris dedupes loads by label inside its `label_keep_max_second` window, so a replayed batch (after restart, retry, etc.) is silently absorbed instead of producing duplicates.
+2. It computes a deterministic Stream Load `label` of the form `{label_prefix}-{stream}_{hash8}-{topic}_{hash8}-{partition}-{first_offset}-{last_offset}`.
+   - Each variable-length segment carries a 32-bit blake3 hash of the raw name, so names that sanitize to the same string (e.g. `events.v1` vs `events_v1`) cannot collide.
+   - The total label is bounded under Doris's 128-char cap regardless of input length.
+   - Doris dedupes loads by label inside its `label_keep_max_second` window, so a replayed batch (after restart, retry, etc.) is silently absorbed instead of producing duplicates.
 3. It `PUT`s the batch to `{fe_url}/api/{database}/{table}/_stream_load` with HTTP Basic auth and the headers `Expect: 100-continue`, `format: json`, `strip_outer_array: true`, `label: <label>`. (`Expect: 100-continue` lets Doris reject auth/4xx failures before the connector uploads the whole body — important for large batches and required if a reverse proxy sits in front of Doris.)
 4. The Doris frontend (FE) responds with a `307 Temporary Redirect` to a backend (BE). The connector follows the redirect manually so that the `Authorization` header is preserved across the hop (`reqwest`'s default policy strips it on cross-host redirects). `308 Permanent Redirect` is also followed as a defensive measure; redirects beyond a hard cap of 5 are rejected as `HttpRequestFailed`.
 5. The HTTP body is parsed as JSON and the `Status` field decides the outcome:
    - `Success` → batch accepted.
    - `Label Already Exists` → idempotent replay, treated as success.
    - `Publish Timeout` or HTTP `5xx`/`408`/`429` → transient error (`Error::CannotStoreData`); the runtime can retry.
-   - `Fail`, any other `4xx`, or an unparseable response body → permanent error (`Error::PermanentHttpError`); retrying is not useful.
+   - `Fail`, any other `4xx`, or an unparsable response body → permanent error (`Error::PermanentHttpError`); retrying is not useful.
 
 ## Configuration
 
diff --git a/core/connectors/sinks/doris_sink/src/lib.rs b/core/connectors/sinks/doris_sink/src/lib.rs
index d5589f7b74..a8b85058f1 100644
--- a/core/connectors/sinks/doris_sink/src/lib.rs
+++ b/core/connectors/sinks/doris_sink/src/lib.rs
@@ -320,7 +320,7 @@ fn truncate_for_log(s: &str, max_bytes: usize) -> String {
 }
 
 fn parse_stream_load_response(body: &str) -> Result<StreamLoadResponse, Error> {
-    // An unparseable 200-OK body is almost always a Doris bug, a proxy that
+    // An unparsable 200-OK body is almost always a Doris bug, a proxy that
     // injected HTML, or a future schema change — none of those are cured by
     // retrying the same bytes. Default to permanent so the runtime DLQs the
     // batch instead of looping forever.
@@ -755,7 +755,7 @@ mod tests {
 
     #[test]
     fn parse_stream_load_response_rejects_garbage_as_permanent() {
-        // An unparseable body must surface as PermanentHttpError so the
+        // An unparsable body must surface as PermanentHttpError so the
         // runtime DLQs the batch instead of retrying the same garbage forever.
         let body = "not json";
         assert!(matches!(
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index d88031ff37..3c84d9f1db 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -47,10 +47,10 @@ iggy = { workspace = true }
 iggy-cli = { workspace = true }
 iggy_binary_protocol = { workspace = true }
 iggy_common = { workspace = true }
-iggy_connector_sdk = { workspace = true, features = ["api"] }
 # Path-dep only so the Doris integration test can reuse the connector's pure
 # `build_label` function — keeping the test and production label format in lock-step.
 iggy_connector_doris_sink = { path = "../connectors/sinks/doris_sink" }
+iggy_connector_sdk = { workspace = true, features = ["api"] }
 jsonwebtoken = { workspace = true }
 keyring = { workspace = true }
 lazy_static = { workspace = true }
diff --git a/core/integration/tests/connectors/doris/doris_sink.rs b/core/integration/tests/connectors/doris/doris_sink.rs
index 31b075efb7..eee347287c 100644
--- a/core/integration/tests/connectors/doris/doris_sink.rs
+++ b/core/integration/tests/connectors/doris/doris_sink.rs
@@ -487,4 +487,3 @@ async fn given_columns_config_should_apply_derived_expression(
         "expected calculated = count + 1 per row (delta sum = {message_count}), got {delta}"
     );
 }
-

From fd99d7f7923c249b47222797fa93a395bf2ee88c Mon Sep 17 00:00:00 2001
From: Ryan <ryan980053@gmail.com>
Date: Mon, 18 May 2026 22:41:36 +0800
Subject: [PATCH 6/6] fix(doris): update payload handling in consume method to
 not accept non-json or empty json

---
 Cargo.lock                                  |  1 +
 core/connectors/sinks/doris_sink/Cargo.toml |  1 +
 core/connectors/sinks/doris_sink/src/lib.rs | 81 +++++++++++++++++++--
 3 files changed, 77 insertions(+), 6 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 742b34fd7c..11cc2bdba2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6590,6 +6590,7 @@ dependencies = [
  "serde",
  "serde_json",
  "simd-json",
+ "tokio",
  "tracing",
 ]
 
diff --git a/core/connectors/sinks/doris_sink/Cargo.toml b/core/connectors/sinks/doris_sink/Cargo.toml
index 385cbb50c3..21974fa1eb 100644
--- a/core/connectors/sinks/doris_sink/Cargo.toml
+++ b/core/connectors/sinks/doris_sink/Cargo.toml
@@ -42,4 +42,5 @@ secrecy = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 simd-json = { workspace = true }
+tokio = { workspace = true }
 tracing = { workspace = true }
diff --git a/core/connectors/sinks/doris_sink/src/lib.rs b/core/connectors/sinks/doris_sink/src/lib.rs
index a8b85058f1..c6eedd1653 100644
--- a/core/connectors/sinks/doris_sink/src/lib.rs
+++ b/core/connectors/sinks/doris_sink/src/lib.rs
@@ -472,17 +472,17 @@ impl Sink for DorisSink {
         for chunk in messages.chunks(batch_size) {
             let json_values: Vec<&simd_json::OwnedValue> = chunk
                 .iter()
-                .filter_map(|m| match &m.payload {
-                    Payload::Json(value) => Some(value),
+                .map(|m| match &m.payload {
+                    Payload::Json(value) => Ok(value),
                     _ => {
-                        warn!(
-                            "Doris sink ID {} dropping non-JSON payload (schema={})",
+                        error!(
+                            "Doris sink ID {} received non-JSON payload (schema={}); aborting batch",
                             self.id, messages_metadata.schema
                         );
-                        None
+                        Err(Error::InvalidPayloadType)
                     }
                 })
-                .collect();
+                .collect::<Result<_, _>>()?;
 
             if json_values.is_empty() {
                 continue;
@@ -810,4 +810,73 @@ mod tests {
         // base64("root:pw") = cm9vdDpwdw==
         assert_eq!(sink.auth_header.expose_secret(), "Basic cm9vdDpwdw==");
     }
+
+    fn text_msg(offset: u64) -> ConsumedMessage {
+        ConsumedMessage {
+            id: offset as u128,
+            offset,
+            checksum: 0,
+            timestamp: 0,
+            origin_timestamp: 0,
+            headers: None,
+            payload: Payload::Text("not json".into()),
+        }
+    }
+
+    fn json_msg(offset: u64) -> ConsumedMessage {
+        let mut bytes = br#"{"k":1}"#.to_vec();
+        let value = simd_json::to_owned_value(&mut bytes).unwrap();
+        ConsumedMessage {
+            id: offset as u128,
+            offset,
+            checksum: 0,
+            timestamp: 0,
+            origin_timestamp: 0,
+            headers: None,
+            payload: Payload::Json(value),
+        }
+    }
+
+    fn topic_meta() -> TopicMetadata {
+        TopicMetadata {
+            stream: "events".into(),
+            topic: "orders".into(),
+        }
+    }
+
+    fn messages_meta() -> MessagesMetadata {
+        MessagesMetadata {
+            partition_id: 0,
+            current_offset: 0,
+            schema: iggy_connector_sdk::Schema::Json,
+        }
+    }
+
+    #[tokio::test]
+    async fn consume_aborts_on_first_non_json_payload() {
+        let sink = DorisSink::new(1, make_config());
+        let result = sink
+            .consume(&topic_meta(), messages_meta(), vec![text_msg(0)])
+            .await;
+        assert!(
+            matches!(result, Err(Error::InvalidPayloadType)),
+            "expected InvalidPayloadType, got {result:?}",
+        );
+    }
+
+    #[tokio::test]
+    async fn consume_aborts_on_non_json_in_mixed_batch() {
+        let sink = DorisSink::new(1, make_config());
+        let result = sink
+            .consume(
+                &topic_meta(),
+                messages_meta(),
+                vec![json_msg(0), text_msg(1)],
+            )
+            .await;
+        assert!(
+            matches!(result, Err(Error::InvalidPayloadType)),
+            "expected InvalidPayloadType, got {result:?}",
+        );
+    }
 }