This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 3bfc41809 feat(connectors): add Apache Doris sink connector (#3215)
3bfc41809 is described below
commit 3bfc41809b692d046d0353a493e8e3614ad7fef9
Author: Ryan Huang <[email protected]>
AuthorDate: Fri May 29 13:53:11 2026 +0800
feat(connectors): add Apache Doris sink connector (#3215)
---
.config/nextest.toml | 16 +
.github/actions/rust/pre-merge/action.yml | 10 +
Cargo.lock | 22 +
Cargo.toml | 7 +
core/connectors/README.md | 1 +
.../example_config/connectors/doris_sink.toml | 54 +
core/connectors/sinks/README.md | 1 +
core/connectors/sinks/doris_sink/Cargo.toml | 51 +
core/connectors/sinks/doris_sink/README.md | 99 ++
core/connectors/sinks/doris_sink/config.toml | 54 +
core/connectors/sinks/doris_sink/src/lib.rs | 1545 ++++++++++++++++++++
core/integration/Cargo.toml | 6 +-
.../tests/connectors/doris/doris_sink.rs | 492 +++++++
core/integration/tests/connectors/doris/mod.rs | 20 +
.../integration/tests/connectors/doris/sink.toml | 14 +-
.../tests/connectors/fixtures/doris/container.rs | 717 +++++++++
.../tests/connectors/fixtures/doris/mod.rs | 25 +
core/integration/tests/connectors/fixtures/mod.rs | 5 +
core/integration/tests/connectors/mod.rs | 1 +
19 files changed, 3128 insertions(+), 12 deletions(-)
diff --git a/.config/nextest.toml b/.config/nextest.toml
index fb488b694..0f4f9a2c3 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -21,6 +21,22 @@
filter = 'package(integration) and
test(cli::system::test_cli_session_scenario::should_be_successful)'
threads-required = "num-cpus"
+# Doris tests are serialized among themselves, but no longer monopolize the
+# whole runner. The all-in-one image's BE advertises 127.0.0.1:8040 for the
+# FE→BE 307 redirect, so only one Doris container per process can bind
+# host:8040. Within a nextest binary the fixture caches one shared container in
+# `SHARED_DORIS` so the first doris test pays the ~40s boot and the rest reuse
+# it; `max-threads = 1` keeps a second nextest binary (or a re-run) from
+# racing the same host port, while still letting unrelated light tests fill
+# the rest of the runner.
+[test-groups.doris]
+max-threads = 1
+
+[[profile.default.overrides]]
+filter = 'package(integration) and test(/connectors::doris::/)'
+test-group = "doris"
+slow-timeout = { period = "60s", terminate-after = 8 }
+
[profile.default]
slow-timeout = { period = "30s", terminate-after = 4 }
diff --git a/.github/actions/rust/pre-merge/action.yml
b/.github/actions/rust/pre-merge/action.yml
index 574134dea..c7388b909 100644
--- a/.github/actions/rust/pre-merge/action.yml
+++ b/.github/actions/rust/pre-merge/action.yml
@@ -207,6 +207,16 @@ runs:
source <(cargo llvm-cov show-env --export-prefix)
+ # Doris 4.0.3's start_be.sh hard-`exit 1`s unless vm.max_map_count >=
2000000.
+ # This kernel param can only be set on the host (no container can
raise it),
+ # so we raise it here. Everything else about booting Doris — image,
heap/mem
+ # caps, port mappings, BE-alive wait — lives in the testcontainers
fixture
+ # at core/integration/tests/connectors/fixtures/doris/container.rs, so
+ # `cargo test` and CI follow exactly the same path.
+ if [[ "$RUNNER_OS" == "Linux" ]]; then
+ sudo sysctl -w vm.max_map_count=2000000 || true
+ fi
+
bins_start=$(date +%s)
if [[ -n "$PACKAGE_FLAGS" ]]; then
cargo build --locked $PACKAGE_FLAGS
diff --git a/Cargo.lock b/Cargo.lock
index 97007571e..4a44aa177 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6715,6 +6715,26 @@ dependencies = [
"url",
]
+[[package]]
+name = "iggy_connector_doris_sink"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "base64",
+ "blake3",
+ "bytes",
+ "humantime",
+ "iggy_connector_sdk",
+ "reqwest 0.13.4",
+ "secrecy",
+ "serde",
+ "serde_json",
+ "simd-json",
+ "tokio",
+ "tracing",
+ "wiremock",
+]
+
[[package]]
name = "iggy_connector_elasticsearch_sink"
version = "0.4.1-edge.1"
@@ -7185,6 +7205,7 @@ dependencies = [
"iggy-cli",
"iggy_binary_protocol",
"iggy_common",
+ "iggy_connector_doris_sink",
"iggy_connector_sdk",
"jsonwebtoken",
"keyring-core",
@@ -7209,6 +7230,7 @@ dependencies = [
"sysinfo 0.39.2",
"tempfile",
"test-case",
+ "testcontainers",
"testcontainers-modules",
"tokio",
"toml 1.1.2+spec-1.1.0",
diff --git a/Cargo.toml b/Cargo.toml
index 2b994b357..172193865 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",
@@ -278,6 +279,10 @@ sqlx = { version = "0.9.0", features = [
"runtime-tokio",
"tls-rustls",
"postgres",
+ # "mysql": Doris exposes a MySQL-wire frontend; the Doris sink's
+ # integration-test fixture talks to it over this driver. Cargo unifies
+ # features across the workspace, so it is declared once here.
+ "mysql",
"chrono",
"uuid",
"json",
@@ -290,6 +295,7 @@ sysinfo = "0.39.2"
tempfile = "3.27.0"
terminal_size = { version = "0.4.4" }
test-case = "3.3.1"
+testcontainers = { version = "0.27.3", features = ["reusable-containers"] }
testcontainers-modules = { version = "0.15.0", features = ["postgres",
"http_wait"] }
thiserror = "2.0.18"
tokio = { version = "1.52.3", features = ["full"] }
@@ -327,6 +333,7 @@ web-sys = { version = "0.3", features = [
] }
webpki-roots = "1.0.7"
windows-native-keyring-store = "1.1.0"
+wiremock = "0.6"
yew = { version = "0.23", features = ["csr"] }
yew-router = "0.20"
zbus-secret-service-keyring-store = { version = "1.0.0", features =
["rt-async-io-crypto-rust"] }
diff --git a/core/connectors/README.md b/core/connectors/README.md
index f955a4872..b7d24cfba 100644
--- a/core/connectors/README.md
+++ b/core/connectors/README.md
@@ -80,6 +80,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 000000000..656b0c2b4
--- /dev/null
+++ b/core/connectors/runtime/example_config/connectors/doris_sink.toml
@@ -0,0 +1,54 @@
+# 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 = "30s"
+# TCP connect timeout (default "5s"). Independent of timeout; raise it for
+# cross-region or cold-start FEs that are slow to accept the connection.
+# connect_timeout = "5s"
+# Stream Load redirect security. Doris's FE redirects (307) to a BE on another
+# host; credentials are re-attached across that hop. By default a redirect that
+# downgrades https -> http is refused (it would leak credentials in cleartext).
+# Set true only for a known-insecure FE -> BE topology.
+# allow_insecure_redirect = false
+# Optional allowlist of hosts a redirect may target. When set and non-empty,
+# any redirect to another host is refused (hard lockdown against a MITM'd FE).
+# allowed_redirect_hosts = ["be1.doris.internal", "be2.doris.internal"]
diff --git a/core/connectors/sinks/README.md b/core/connectors/sinks/README.md
index 77f15d8da..f97d7c6ef 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 000000000..1a5a4722e
--- /dev/null
+++ b/core/connectors/sinks/doris_sink/Cargo.toml
@@ -0,0 +1,51 @@
+# 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"
+publish = false
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+async-trait = { workspace = true }
+base64 = { workspace = true }
+blake3 = { workspace = true }
+bytes = { workspace = true }
+humantime = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+reqwest = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+simd-json = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+
+[dev-dependencies]
+wiremock = { 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 000000000..6da9937be
--- /dev/null
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -0,0 +1,99 @@
+# 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.
+- `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`). If a non-JSON payload reaches the connector it logs at `error!` and
aborts the whole poll; since the consumer offset is already committed at poll
time the batch is not replayed — effectively silent data loss — so the upstream
schema must be guaranteed JSON. (Under `schema = "json"` the SDK drops non-JSON
before the connector sees it, so this abort is a defensive guard.)
+- 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_san}-{topic_san}-{hash16}-{partition}-{first_offset}-{last_offset}`.
+ - `hash16` is a single 64-bit blake3 hash computed over the *raw*
(un-sanitized), length-prefixed `(label_prefix, stream, topic)` triple. So
identities that sanitize to the same string get distinct labels — whether the
collision is in the names (`events.v1` vs `events_v1`) or in two tenants'
prefixes that truncate alike (`prod_events_us_east_1` vs `..._2`) — and no
boundary-shift aliasing is possible (`("ab","c")` ≠ `("a","bc")`).
+ - The total label is bounded under Doris's 128-char cap regardless of input
length (worst case 120 chars).
+ - Doris dedupes loads by label inside its `label_keep_max_second` window.
The deterministic label is **forward-compatible scaffolding**: if the runtime
ever gains retry/redrive, a duplicate load would be absorbed, not doubled.
**Today it protects no production scenario** — there is no retry loop,
`consume()` runs once per poll, and the runtime discards its return value.
Delivery is at-most-once: the offset is committed before `consume()` runs, so a
failed load is never replayed.
+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` is
required by Doris's Stream Load endpoint, which rejects PUTs that omit it.
Where the HTTP stack negotiates the handshake it also lets Doris reject
auth/4xx before the body uploads — a secondary benefit, not relied on for
correctness.)
+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 (or a redirect with no usable `Location`) are rejected
as a permanent `PermanentHttpError`, since retrying a malformed/looping
redirect cannot help.
+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` → classified as a transient
error (`Error::CannotStoreData`) — retryable in principle, but per the
at-most-once note above the runtime does not currently act on it.
+ - `Fail`, any other `4xx`, or an unparsable response body → permanent error
(`Error::PermanentHttpError`); retrying would not help even if the runtime did
redrive.
+
+## Configuration
+
+| Field | Required | Default | Description |
+| --- | --- | --- | --- |
+| `fe_url` | yes | — | Doris frontend HTTP base URL, e.g.
`http://localhost:8030`. |
+| `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. |
+| `batch_size` | no | `1000` | Maximum number of messages per Stream Load
request. |
+| `timeout` | no | `30s` | Per-request HTTP timeout (total request budget), as
a human-readable duration (e.g. `30s`, `1m`). |
+| `connect_timeout` | no | `5s` | TCP connect timeout, independent of
`timeout`, as a human-readable duration. Raise it for cross-region or
cold-start FEs. |
+| `max_filter_ratio` | no | unset | Forwarded as the `max_filter_ratio` Stream
Load header. Must be a finite value in `[0.0, 1.0]`; an out-of-range value
fails `open()`. |
+| `columns` | no | unset | Forwarded as the `columns` Stream Load header.
Validated at startup; an invalid value fails `open()`. |
+| `where` | no | unset | Forwarded as the `where` Stream Load header.
Validated at startup; an invalid value fails `open()`. |
+| `allow_insecure_redirect` | no | `false` | Permit a Stream Load redirect
that downgrades `https://` → `http://`. Refused by default because it would
push credentials onto a cleartext hop. |
+| `allowed_redirect_hosts` | no | unset | Allowlist of redirect targets. Each
entry is `host` (pins the host, any port) or `host:port` (pins the exact
endpoint). When set and non-empty, any other redirect target is refused. |
+
+### 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 = "30s"
+```
+
+## 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 try to exfiltrate credentials by responding
with `Location: http://attacker/`. Before re-attaching credentials, the
connector validates the redirect target: it **refuses a scheme downgrade**
(`https://` → `http://`) unless `allow_insecure_redirect = true`, requires an
**absolute** `Location` (a relative one is rejected, not silently resolved),
and — if `allowed_redirect_hosts` is set — refuses any target outside that
allowlist.
+ **When `allowed_redirect_hosts` is unset (the default), any same-scheme host
is accepted** — that is the price of supporting the normal cross-host FE → BE
topology out of the box. For lockdown in hostile networks, set
`allowed_redirect_hosts` to your known BE endpoints and deploy Doris over TLS.
List a bare `host` to pin only the host, or `host:port` to pin the exact
endpoint — pinning the port closes the "allowlisted host, attacker port" vector.
+- **`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.
+- **Keep `batch_size` stable across a redrive.** The label includes the
chunk's `first_offset` and `last_offset`, which are a function of `batch_size`.
If you change `batch_size` between a failed load and its redrive, the chunk
boundaries shift, the offsets differ, and the new label no longer matches the
old one — so Doris 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.
+- **Multi-chunk batches are best-effort for operational failures.** A poll
larger than `batch_size` is split into chunks, each loaded as its own labelled
Stream Load. If a chunk fails *operationally* (serialize, HTTP, or
status-classification error), the connector still attempts the remaining chunks
and then returns the worst error — it does **not** stop at the first such
failure.
+ The runtime commits the consumer offset for the whole poll before
`consume()` runs, so a failed chunk is not replayed regardless; pushing the
other chunks through maximizes delivered data, and the worst error is surfaced
at the end (logged at `error!` for observability — the runtime currently
discards `consume()`'s return value, so there is no retry or DLQ).
+ The one deliberate exception is a **non-JSON payload**, which is treated as
a schema-contract violation and aborts the whole poll immediately (see the
Requirements note above). Under `schema = "json"` this is unreachable, so it is
a defensive guard rather than a normal path.
+
+## Limitations
+
+- 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 000000000..5aaf6b07d
--- /dev/null
+++ b/core/connectors/sinks/doris_sink/config.toml
@@ -0,0 +1,54 @@
+# 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 = "30s"
+# TCP connect timeout (default "5s"). Independent of timeout; raise it for
+# cross-region or cold-start FEs that are slow to accept the connection.
+# connect_timeout = "5s"
+# Stream Load redirect security. Doris's FE redirects (307) to a BE on another
+# host; credentials are re-attached across that hop. By default a redirect that
+# downgrades https -> http is refused (it would leak credentials in cleartext).
+# Set true only for a known-insecure FE -> BE topology.
+# allow_insecure_redirect = false
+# Optional allowlist of hosts a redirect may target. When set and non-empty,
+# any redirect to another host is refused (hard lockdown against a MITM'd FE).
+# allowed_redirect_hosts = ["be1.doris.internal", "be2.doris.internal"]
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 000000000..a2c8c2bda
--- /dev/null
+++ b/core/connectors/sinks/doris_sink/src/lib.rs
@@ -0,0 +1,1545 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use async_trait::async_trait;
+use base64::{Engine as _, engine::general_purpose};
+use bytes::Bytes;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
sink_connector,
+};
+use reqwest::{Method, StatusCode, header};
+use secrecy::zeroize::Zeroizing;
+use secrecy::{ExposeSecret, SecretString};
+use serde::Deserialize;
+use std::str::FromStr;
+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;
+// Total per-request budget. Human-readable (e.g. "30s") to match the sibling
+// web sinks (http_sink, influxdb_sink).
+const DEFAULT_TIMEOUT: &str = "30s";
+// Bounded TCP handshake timeout so an unreachable FE fails fast instead of
+// burning the whole request-timeout budget on connect alone.
+const DEFAULT_CONNECT_TIMEOUT: &str = "5s";
+// Doris's FE Stream Load 307s to a BE, and reqwest strips Authorization on
+// cross-host redirects, so we follow them manually. This caps a looping
cluster.
+const MAX_REDIRECTS: u8 = 5;
+// Doris Stream Load labels must be 1..=128 chars of `[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;
+// A single 64-bit (16-hex) joint hash over the raw (prefix, stream, topic)
+// triple. 64 bits keeps the adversarial birthday bound high enough that a
+// multi-tenant namer can't cheaply force the label collisions that Doris's
+// server-side dedupe would turn into silent data loss. One joint hash (not one
+// per segment) buys that for the same length budget, leaving the sanitized
names
+// full-length.
+const LABEL_HASH_HEX_LEN: usize = 16;
+// Cap the response-body slice kept for logs/errors so a misbehaving proxy that
+// returns a giant body can't flood the logs. Bounds only what we *log*, not
peak
+// memory — `response.text()` already buffers the full body first.
+const MAX_RESPONSE_LOG_BYTES: usize = 4096;
+
+#[derive(Debug)]
+pub struct DorisSink {
+ id: u32,
+ config: DorisSinkConfig,
+ // Precomputed in `new()` and marked sensitive so reqwest keeps it out of
any
+ // debug/trace output (and never HPACK-indexes it on HTTP/2).
+ auth_header: header::HeaderValue,
+ // Set in `open()`, `None` until then. Holds the HTTP client, the parsed
+ // Stream Load URL (which doubles as the redirect-validation baseline), the
+ // precomputed/validated optional headers, and the resolved redirect
policy.
+ connected: Option<Connected>,
+}
+
+#[derive(Debug)]
+struct Connected {
+ client: reqwest::Client,
+ base_url: reqwest::Url,
+ // Optional Stream Load headers, validated once at `open()` so a bad byte
+ // fails fast at startup rather than on every batch.
+ max_filter_ratio_header: Option<header::HeaderValue>,
+ columns_header: Option<header::HeaderValue>,
+ where_header: Option<header::HeaderValue>,
+ // Redirect policy resolved once at `open()` so `validate_redirect` reads
it
+ // off `self` instead of threading it through every call.
+ allow_insecure_redirect: bool,
+ allowed_redirect_hosts: Option<Vec<String>>,
+}
+
+#[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>,
+ /// Total per-request HTTP timeout as a human-readable duration, e.g. "30s"
+ /// (default 30s). Matches the `timeout` field on the http/influxdb sinks.
+ pub timeout: Option<String>,
+ /// TCP connect timeout as a human-readable duration, e.g. "5s".
Independent
+ /// of `timeout` (the total request budget). Defaults to 5s; raise it for
+ /// cross-region or cold-start FEs that are slow to accept the connection.
+ pub connect_timeout: Option<String>,
+ pub batch_size: Option<u32>,
+ /// Permit a redirect that downgrades the scheme (e.g. `https://` FE ->
+ /// `http://` BE). Off by default: a downgrade would push Basic-auth
+ /// credentials onto a cleartext hop, so we refuse it unless the operator
+ /// explicitly opts in for a known-insecure FE -> BE topology.
+ pub allow_insecure_redirect: Option<bool>,
+ /// Optional allowlist of hosts a Stream Load redirect may target. Each
entry
+ /// is `host` or `host:port`; a bare host pins only the host (any port),
while
+ /// `host:port` pins the exact endpoint. When set and non-empty, a
redirect to
+ /// any other target is refused — a hard lockdown against a
compromised/MITM'd
+ /// FE exfiltrating credentials via `Location`. When unset, cross-host
+ /// redirects are allowed (required for the normal FE -> BE topology)
subject
+ /// only to the scheme-downgrade rule above.
+ pub allowed_redirect_hosts: Option<Vec<String>>,
+}
+
+#[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 = Zeroizing::new(format!(
+ "{}:{}",
+ config.username,
+ config.password.expose_secret()
+ ));
+ let encoded =
Zeroizing::new(general_purpose::STANDARD.encode(credential.as_bytes()));
+ // `Basic <base64>` is always visible ASCII, so this conversion cannot
fail.
+ let auth_value = Zeroizing::new(format!("Basic {}", encoded.as_str()));
+ let mut auth_header = header::HeaderValue::from_str(&auth_value)
+ .expect("Basic auth header is always valid ASCII");
+ auth_header.set_sensitive(true);
+
+ DorisSink {
+ id,
+ config,
+ auth_header,
+ connected: None,
+ }
+ }
+
+ fn build_client(&self) -> Result<reqwest::Client, Error> {
+ let timeout = parse_duration(self.config.timeout.as_deref(),
DEFAULT_TIMEOUT);
+ let connect_timeout = parse_duration(
+ self.config.connect_timeout.as_deref(),
+ DEFAULT_CONNECT_TIMEOUT,
+ );
+ // `Policy::none()` so we follow redirects manually and keep
Authorization
+ // alive across the FE -> BE hop (reqwest strips it on cross-host
307s).
+ reqwest::Client::builder()
+ .redirect(reqwest::redirect::Policy::none())
+ .timeout(timeout)
+ .connect_timeout(connect_timeout)
+ .build()
+ .map_err(|e| Error::InitError(format!("Failed to build Doris HTTP
client: {e}")))
+ }
+
+ async fn send_stream_load(
+ &self,
+ label: &str,
+ body: Bytes,
+ ) -> Result<StreamLoadResponse, Error> {
+ let connected = self.connected.as_ref().ok_or_else(|| {
+ Error::InitError(format!(
+ "Doris sink ID {} called before open() — not connected",
+ self.id
+ ))
+ })?;
+ // `base_url` is the redirect-validation baseline (original FE
scheme/host)
+ // and the parsed first-hop target.
+ let mut url = connected.base_url.clone();
+ let mut redirects = 0u8;
+
+ loop {
+ let mut request = connected
+ .client
+ .request(Method::PUT, url.clone())
+ .header(header::AUTHORIZATION, self.auth_header.clone())
+ .header(header::EXPECT, "100-continue")
+ .header("format", "json")
+ .header("strip_outer_array", "true")
+ .header("label", label)
+ .body(body.clone());
+
+ // Headers were validated and built once in `open()`.
+ if let Some(value) = &connected.max_filter_ratio_header {
+ request = request.header("max_filter_ratio", value.clone());
+ }
+ if let Some(value) = &connected.columns_header {
+ request = request.header("columns", value.clone());
+ }
+ if let Some(value) = &connected.where_header {
+ request = request.header("where", value.clone());
+ }
+
+ 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 {
+ // A redirect loop is permanent, not transient: retrying
just
+ // re-walks the same loop, so surface it as such.
+ return Err(Error::PermanentHttpError(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 {
+ // A redirect with no usable Location is malformed;
retrying
+ // won't produce one.
+ return Err(Error::PermanentHttpError(format!(
+ "Doris sink ID {} got {status} with no Location
header",
+ self.id
+ )));
+ };
+ // Doris always emits an *absolute* Location (the BE endpoint).
+ // A relative one is outside that contract; resolving it
against
+ // the current URL would silently target a sibling path (a
near-
+ // certain 404) and give a false sense of safety, so reject it.
+ let target = reqwest::Url::parse(location).map_err(|e| {
+ Error::PermanentHttpError(format!(
+ "Doris sink ID {} got {status} with non-absolute or
unparsable Location '{location}': {e}",
+ self.id
+ ))
+ })?;
+ connected.validate_redirect(&target, self.id)?;
+ debug!("Doris sink ID {} following redirect to {target}",
self.id);
+ url = target;
+ continue;
+ }
+
+ let is_success = status.is_success();
+ let response_text = match response.text().await {
+ Ok(text) => text,
+ Err(e) if is_success => {
+ // 2xx but the body never fully arrived (mid-stream TCP
reset,
+ // decompression error, body-read timeout). Doris almost
+ // certainly persisted the load, but we can't read the row
+ // counts to confirm. Classify transient — not a fabricated
+ // parse failure — so a retry re-PUTs under the same label
and
+ // Doris's dedupe reveals the real outcome instead of
DLQing a
+ // success.
+ warn!(
+ "Doris sink ID {} failed to read 2xx response body:
{e}; treating as retryable",
+ self.id
+ );
+ return Err(Error::CannotStoreData(format!(
+ "Doris sink ID {} could not read 2xx Stream Load
response body: {e}",
+ self.id
+ )));
+ }
+ Err(e) => {
+ // Non-2xx with an unreadable body: log it, then fall back
to
+ // an empty body so the status-based handling below still
+ // classifies the outcome (empty body on a non-2xx =>
permanent).
+ warn!(
+ "Doris sink ID {} failed to read response body: {e}",
+ self.id
+ );
+ String::new()
+ }
+ };
+ let response_for_log = truncate_for_log(&response_text,
MAX_RESPONSE_LOG_BYTES);
+
+ if !is_success {
+ let msg = format!(
+ "Doris sink ID {} stream load returned HTTP {status}:
{response_for_log}",
+ self.id
+ );
+ error!("{msg}");
+ // 408/429 are 4xx but transient — retry them, don't DLQ.
+ 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),
+ });
+ }
+
+ return parse_stream_load_response(&response_text);
+ }
+ }
+}
+
+impl Connected {
+ /// Validate a Stream Load redirect target before re-attaching credentials.
+ ///
+ /// Doris's FE legitimately redirects (307) to a BE on a *different host*,
so
+ /// we can't require same-host. Instead we enforce three rules that close
the
+ /// credential-exfiltration vector a compromised/MITM'd FE would otherwise
have:
+ ///
+ /// 0. The target scheme must be `http` or `https`. A non-HTTP scheme
+ /// (`ftp`, `file`, ...) would slip past the downgrade rule when the
FE
+ /// itself is `http`, so reject it before re-attaching credentials.
+ /// 1. No scheme downgrade (`https` -> `http`) unless
`allow_insecure_redirect`
+ /// is set — a downgrade would push Basic-auth creds onto a cleartext
hop.
+ /// 2. If `allowed_redirect_hosts` is non-empty, the target must match an
+ /// entry. A bare-host entry pins only the host; a `host:port` entry
pins
+ /// the exact endpoint, refusing an allowlisted host on an attacker
port.
+ fn validate_redirect(&self, target: &reqwest::Url, id: u32) -> Result<(),
Error> {
+ // Only http(s) targets ever get credentials re-attached. Preventing
something like ftp://.
+ let scheme = target.scheme();
+ if !scheme.eq_ignore_ascii_case("http") &&
!scheme.eq_ignore_ascii_case("https") {
+ return Err(Error::PermanentHttpError(format!(
+ "Doris sink ID {id}: refusing redirect to non-HTTP(S) scheme
'{scheme}'"
+ )));
+ }
+
+ let downgraded = self.base_url.scheme().eq_ignore_ascii_case("https")
+ && !target.scheme().eq_ignore_ascii_case("https");
+ if downgraded && !self.allow_insecure_redirect {
+ return Err(Error::PermanentHttpError(format!(
+ "Doris sink ID {id}: refusing redirect that downgrades {} ->
{} \
+ (would leak credentials in cleartext; set
allow_insecure_redirect=true \
+ to permit a known-insecure FE -> BE topology)",
+ self.base_url.scheme(),
+ target.scheme(),
+ )));
+ }
+
+ if let Some(allowed) = self.allowed_redirect_hosts.as_deref()
+ && !allowed.is_empty()
+ && !redirect_target_allowed(allowed, target)
+ {
+ return Err(Error::PermanentHttpError(format!(
+ "Doris sink ID {id}: redirect target '{}:{}' is not in
allowed_redirect_hosts",
+ target.host_str().unwrap_or(""),
+ target
+ .port_or_known_default()
+ .map(|p| p.to_string())
+ .unwrap_or_default(),
+ )));
+ }
+
+ Ok(())
+ }
+}
+
+/// Match a redirect target against the allowlist. An entry of `host` matches
any
+/// port on that host; an entry of `host:port` pins the exact endpoint. DNS
names,
+/// IPv4 literals, and IPv6 literals (bare `::1` or bracketed
`[::1]`/`[::1]:8040`)
+/// all split cleanly.
+fn redirect_target_allowed(allowed: &[String], target: &reqwest::Url) -> bool {
+ let raw_host = target.host_str().unwrap_or("");
+ // `host_str()` brackets IPv6 literals (`[::1]`); strip them so a bare
(`::1`)
+ // or bracketed (`[::1]`) allowlist entry both compare equal.
+ let host = strip_brackets(raw_host);
+ let port = target.port_or_known_default();
+ allowed.iter().any(|entry| match split_host_port(entry) {
+ (entry_host, Some(entry_port)) => {
+ entry_host.eq_ignore_ascii_case(host) && Some(entry_port) == port
+ }
+ (entry_host, None) => entry_host.eq_ignore_ascii_case(host),
+ })
+}
+
+/// Strip a single pair of surrounding `[ ]` brackets from an IPv6 literal, so
a
+/// bracketed host compares equal to its bare form.
+fn strip_brackets(host: &str) -> &str {
+ host.strip_prefix('[')
+ .and_then(|h| h.strip_suffix(']'))
+ .unwrap_or(host)
+}
+
+/// Split an allowlist entry into `(host, optional port)`, with the host
returned
+/// *unbracketed* so it compares against a bracket-stripped `host_str()`.
+///
+/// - `[host]` / `[host]:port` — bracketed IPv6: the bracketed host splits
from an
+/// optional trailing `:<port>`.
+/// - A bare entry with more than one `:` is an unbracketed IPv6 literal
(`::1`,
+/// `fe80::1`); a port suffix on it would be ambiguous, so it is host-only.
Pin a
+/// port on an IPv6 host by bracketing it (`[::1]:8040`).
+/// - Otherwise a trailing `:<digits>` is the port; anything else is host-only.
+fn split_host_port(entry: &str) -> (&str, Option<u16>) {
+ if let Some(rest) = entry.strip_prefix('[') {
+ // Bracketed IPv6: `[host]` or `[host]:port`.
+ if let Some((host, after)) = rest.split_once(']') {
+ let port = after.strip_prefix(':').and_then(|p|
p.parse::<u16>().ok());
+ return (host, port);
+ }
+ // No closing `]`: malformed, treat the whole thing as host-only.
+ return (entry, None);
+ }
+ // A bare multi-colon entry is an unbracketed IPv6 literal: host-only.
+ if entry.matches(':').count() > 1 {
+ return (entry, None);
+ }
+ if let Some((host, port)) = entry.rsplit_once(':')
+ && !port.is_empty()
+ && let Ok(port) = port.parse::<u16>()
+ {
+ (host, Some(port))
+ } else {
+ (entry, None)
+ }
+}
+
+/// Parse a human-readable duration (e.g. "30s"), falling back to `default`
with
+/// a warning on a malformed *or zero* value. Mirrors the http/influxdb sinks.
+///
+/// A zero duration parses fine but is degenerate: reqwest treats a zero
+/// timeout/connect-timeout as an immediate deadline, so every request fails
with
+/// a `TimedOut` error before it can complete. Treat it like a malformed value.
+fn parse_duration(input: Option<&str>, default: &str) -> Duration {
+ let raw = input.unwrap_or(default);
+ let fallback = || *HumanDuration::from_str(default).expect("default
duration must be valid");
+ let parsed = HumanDuration::from_str(raw)
+ .map(|d| *d)
+ .unwrap_or_else(|e| {
+ warn!("Invalid duration '{raw}': {e}, using default '{default}'");
+ fallback()
+ });
+ if parsed.is_zero() {
+ warn!(
+ "Duration '{raw}' is zero, which would time out every request
immediately; \
+ using default '{default}'"
+ );
+ return fallback();
+ }
+ parsed
+}
+
+/// Build the Stream Load URL from `fe_url` and the (already
identifier-checked)
+/// `database`/`table`. Replaces the path wholesale so a trailing slash or
stray
+/// path on `fe_url` can't double up the path.
+fn build_stream_load_url(
+ id: u32,
+ fe_url: &str,
+ database: &str,
+ table: &str,
+) -> Result<reqwest::Url, Error> {
+ let mut url = reqwest::Url::parse(fe_url).map_err(|e| {
+ Error::InvalidConfigValue(format!(
+ "Doris sink ID {id} has invalid fe_url '{fe_url}': {e}"
+ ))
+ })?;
+ // Stream Load speaks HTTP. A `file://`/`ftp://`/... base parses fine but
would
+ // only fail later, per-batch, at `send()`; reject it here at startup
instead.
+ let scheme = url.scheme();
+ if scheme != "http" && scheme != "https" {
+ return Err(Error::InvalidConfigValue(format!(
+ "Doris sink ID {id} fe_url '{fe_url}' must use http or https, got
'{scheme}'"
+ )));
+ }
+ url.set_path(&format!("/api/{database}/{table}/_stream_load"));
+ Ok(url)
+}
+
+/// Effective batch size: the configured value floored at 1 so `chunks()` is
+/// never handed a 0 (which would panic).
+fn effective_batch_size(configured: Option<u32>) -> usize {
+ configured.unwrap_or(DEFAULT_BATCH_SIZE).max(1) as usize
+}
+
+/// Build a validated Stream Load header value, surfacing a bad byte (CR/LF,
+/// non-visible-ASCII) as a startup-time `InvalidConfigValue` instead of a
+/// per-batch `HttpRequestFailed` (reqwest defers `HeaderValue::try_from` to
+/// `.send()`, so an invalid `columns`/`where` would otherwise fail every
batch).
+fn validated_header(field: &str, value: &str, id: u32) ->
Result<header::HeaderValue, Error> {
+ header::HeaderValue::from_str(value).map_err(|e| {
+ Error::InvalidConfigValue(format!(
+ "Doris sink ID {id}: '{field}' header value is invalid (must be
visible ASCII, no CR/LF): {e}"
+ ))
+ })
+}
+
+/// 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| {
+ if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
+ c
+ } else {
+ '_'
+ }
+ })
+ .take(max_len)
+ .collect()
+}
+
+/// A single blake3 fingerprint over the *raw* (unsanitized) `prefix`,
`stream`,
+/// and `topic`, truncated to `LABEL_HASH_HEX_LEN` hex chars. This
disambiguates
+/// identities that sanitize+truncate to the same string (e.g. `events.v1` vs
+/// `events_v1`, or prefixes `prod_events_us_east_1` vs `..._2`), which would
+/// otherwise produce identical labels and cause silent data loss via Doris's
+/// server-side label dedupe. Inputs are length-prefixed so distinct triples
+/// can't alias into one digest (e.g. `("ab","c",..)` vs `("a","bc",..)`).
+fn identity_hash(prefix: &str, stream: &str, topic: &str) -> String {
+ let mut hasher = blake3::Hasher::new();
+ for part in [prefix, stream, topic] {
+ hasher.update(&(part.len() as u64).to_le_bytes());
+ hasher.update(part.as_bytes());
+ }
+ let hash = hasher.finalize().to_hex();
+ hash.as_str()[..LABEL_HASH_HEX_LEN].to_string()
+}
+
+/// Pure label builder. Format:
+///
`{prefix_san}-{stream_san}-{topic_san}-{hash16}-{partition}-{first}-{last}`.
+///
+/// The segment caps bound the total under Doris's 128-char label limit (worst
+/// case 120), and the joint `hash16` over the raw (prefix, stream, topic)
keeps
+/// labels distinct even when the sanitized segments collide.
+///
+/// `#[doc(hidden)]`: `pub` only so the integration test harness can reproduce
+/// labels; not part of the connector's supported API.
+#[doc(hidden)]
+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),
+ sanitize_segment(stream, MAX_LABEL_NAME_LEN),
+ sanitize_segment(topic, MAX_LABEL_NAME_LEN),
+ identity_hash(prefix, stream, topic),
+ partition_id,
+ first_offset,
+ last_offset,
+ )
+}
+
+/// Truncate `s` at the largest char boundary `<= max_bytes` and append a
marker
+/// recording the original size. Bounds 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>
{
+ // An unparsable 200-OK body (Doris bug, proxy-injected HTML, future schema
+ // change) isn't cured by retrying the same bytes — default to permanent so
+ // the runtime DLQs the batch instead of looping.
+ 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)
+ ))
+ })
+}
+
+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_]+
(iggy's stricter subset of Doris identifiers, used as a path-traversal guard)"
+ )));
+ }
+ Ok(())
+}
+
+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: surfacing an unrecognized
+ // failure (e.g. a future Doris error variant) and letting the runtime
DLQ
+ // it beats silently retrying it against 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> {
+ // Constrain database/table BEFORE building the URL — they flow into
the
+ // path, so [A-Za-z0-9_]+ (narrower than Doris's own identifier rules)
+ // blocks path traversal in `/api/{db}/{table}/_stream_load`.
+ validate_identifier(&self.config.database, "database", self.id)?;
+ validate_identifier(&self.config.table, "table", self.id)?;
+
+ let base_url = build_stream_load_url(
+ self.id,
+ &self.config.fe_url,
+ &self.config.database,
+ &self.config.table,
+ )?;
+
+ // Doris permits passwordless users (e.g. a fresh `root`), so an empty
+ // password is valid — but almost always a misconfiguration. Warn,
don't
+ // fail, so local/dev setups still work.
+ if self.config.password.expose_secret().is_empty() {
+ warn!(
+ "Doris sink ID {} is configured with an empty password for
user '{}'; \
+ this is accepted but is usually a misconfiguration.",
+ self.id, self.config.username
+ );
+ }
+
+ // Warn when credentials would travel in cleartext. The FE -> BE 307
hop
+ // itself is guarded by `validate_redirect` (scheme downgrade + host
+ // allowlist); this covers the case where the FE itself is plain http.
+ if base_url.scheme().eq_ignore_ascii_case("http") {
+ let host = base_url.host_str().unwrap_or("");
+ // `host_str()` brackets IPv6 literals (`[::1]`); strip them and
parse
+ // as `IpAddr` to catch `127.0.0.1`, `::1`, and any loopback
spelling.
+ let is_loopback = host == "localhost"
+ || host
+ .trim_start_matches('[')
+ .trim_end_matches(']')
+ .parse::<std::net::IpAddr>()
+ .is_ok_and(|ip| ip.is_loopback());
+ 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
+ );
+ }
+ }
+
+ // Validate + precompute the optional Stream Load headers once. A bad
byte
+ // in `columns`/`where` fails here at startup, not silently per batch.
+ let max_filter_ratio_header = match self.config.max_filter_ratio {
+ Some(ratio) => {
+ // Doris's max_filter_ratio is a fraction in [0.0, 1.0]. A
NaN/inf or
+ // out-of-range value formats to a header-valid string (so the
ASCII
+ // check below would pass) but Doris rejects it on every batch
—
+ // catch it here at startup instead.
+ if !ratio.is_finite() || !(0.0..=1.0).contains(&ratio) {
+ return Err(Error::InvalidConfigValue(format!(
+ "Doris sink ID {}: max_filter_ratio must be a finite
value in [0.0, 1.0], got {ratio}",
+ self.id
+ )));
+ }
+ Some(validated_header(
+ "max_filter_ratio",
+ &ratio.to_string(),
+ self.id,
+ )?)
+ }
+ None => None,
+ };
+ let columns_header = match self.config.columns.as_deref() {
+ Some(columns) => Some(validated_header("columns", columns,
self.id)?),
+ None => None,
+ };
+ let where_header = match self.config.where_clause.as_deref() {
+ Some(where_clause) => Some(validated_header("where", where_clause,
self.id)?),
+ None => None,
+ };
+
+ self.connected = Some(Connected {
+ client: self.build_client()?,
+ base_url,
+ max_filter_ratio_header,
+ columns_header,
+ where_header,
+ allow_insecure_redirect:
self.config.allow_insecure_redirect.unwrap_or(false),
+ allowed_redirect_hosts: self.config.allowed_redirect_hosts.clone(),
+ });
+
+ 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 = effective_batch_size(self.config.batch_size);
+ let label_prefix = self
+ .config
+ .label_prefix
+ .as_deref()
+ .unwrap_or(DEFAULT_LABEL_PREFIX);
+ let mut first_error: Option<Error> = None;
+
+ // Best-effort across chunks: on a per-chunk serialize/HTTP/status
failure
+ // we log it, keep the first error, and `continue` so later chunks
still
+ // land. The runtime commits this poll's consumer offset before
consume()
+ // runs, so returning early would drop the remaining chunks rather than
+ // replay them. The returned error is mapped to a 0/1 status at the FFI
+ // boundary (severity isn't propagated), and every chunk error is
already
+ // logged individually below — so first-error is sufficient.
+ //
+ // The lone hard-abort is a non-JSON payload (via `?`): a stream-wide
+ // schema-contract violation, not a transient chunk failure. Under the
+ // documented `schema = "json"` config the SDK drops non-JSON before
+ // consume() is called, so this stands as a defensive guard.
+ for chunk in messages.chunks(batch_size) {
+ let json_values: Vec<&simd_json::OwnedValue> = chunk
+ .iter()
+ .map(|m| match &m.payload {
+ Payload::Json(value) => Ok(value),
+ _ => {
+ error!(
+ "Doris sink ID {} received non-JSON payload
(schema={}); aborting poll",
+ self.id, messages_metadata.schema
+ );
+ Err(Error::InvalidPayloadType)
+ }
+ })
+ .collect::<Result<_, _>>()?;
+
+ // `chunks()` never yields an empty slice, so first/last are
present.
+ // Use `zip` + `continue` (not `.expect`) so that if a future
refactor
+ // ever breaks that invariant we neither fabricate offset 0 (which
+ // would alias the real offset-0 label and break idempotency) nor
+ // panic across the `extern "C"` FFI boundary (UB per the nomicon).
+ let Some((first_msg, last_msg)) = chunk.first().zip(chunk.last())
else {
+ continue;
+ };
+
+ let body = match simd_json::to_vec(&json_values) {
+ Ok(b) => Bytes::from(b),
+ Err(e) => {
+ error!("Doris sink ID {} failed to serialize batch: {e}",
self.id);
+ first_error.get_or_insert(Error::CannotStoreData(format!(
+ "Failed to serialize batch for Doris: {e}"
+ )));
+ continue;
+ }
+ };
+
+ let label = build_label(
+ label_prefix,
+ &topic_metadata.stream,
+ &topic_metadata.topic,
+ messages_metadata.partition_id,
+ first_msg.offset,
+ last_msg.offset,
+ );
+
+ match self.send_stream_load(&label, body).await {
+ Ok(response) => match classify_status(&response) {
+ Ok(()) => {
+ if response.number_filtered_rows > 0 {
+ // Filtered rows usually mean schema drift
upstream.
+ // 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);
+ first_error.get_or_insert(e);
+ }
+ },
+ Err(e) => {
+ error!("Doris sink ID {} HTTP failed: {e}", self.id);
+ first_error.get_or_insert(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: None,
+ connect_timeout: None,
+ batch_size: None,
+ allow_insecure_redirect: None,
+ allowed_redirect_hosts: None,
+ }
+ }
+
+ #[test]
+ fn stream_load_url_is_well_formed() {
+ let url = build_stream_load_url(1, "http://localhost:8030";, "test_db",
"test_tbl").unwrap();
+ assert_eq!(
+ url.as_str(),
+ "http://localhost:8030/api/test_db/test_tbl/_stream_load";
+ );
+ }
+
+ #[test]
+ fn stream_load_url_handles_trailing_slash() {
+ let url =
+ build_stream_load_url(1, "http://localhost:8030/";, "test_db",
"test_tbl").unwrap();
+ assert_eq!(
+ url.as_str(),
+ "http://localhost:8030/api/test_db/test_tbl/_stream_load";
+ );
+ }
+
+ #[test]
+ fn stream_load_url_rejects_garbage_fe_url() {
+ assert!(matches!(
+ build_stream_load_url(1, "not a url", "db", "tbl"),
+ Err(Error::InvalidConfigValue(_))
+ ));
+ }
+
+ #[test]
+ fn stream_load_url_rejects_non_http_scheme() {
+ // A parseable but non-HTTP base would only fail later, per-batch, at
send().
+ for fe_url in ["file:///etc", "ftp://host/path";, "ws://host:8030"] {
+ assert!(
+ matches!(
+ build_stream_load_url(1, fe_url, "db", "tbl"),
+ Err(Error::InvalidConfigValue(_))
+ ),
+ "expected {fe_url} to be rejected at startup",
+ );
+ }
+ }
+
+ #[test]
+ fn label_is_deterministic() {
+ let a = build_label("iggy", "events", "orders", 7, 100, 199);
+ let b = build_label("iggy", "events", "orders", 7, 100, 199);
+ assert_eq!(a, b);
+ // Format:
{prefix}-{stream_san}-{topic_san}-{hash16}-{partition}-{first}-{last}
+ let parts: Vec<&str> = a.split('-').collect();
+ assert_eq!(parts.len(), 7);
+ assert_eq!(parts[0], "iggy");
+ assert_eq!(parts[1], "events");
+ assert_eq!(parts[2], "orders");
+ assert_eq!(parts[3].len(), LABEL_HASH_HEX_LEN);
+ assert!(parts[3].chars().all(|c| c.is_ascii_hexdigit()));
+ assert_eq!(parts[4], "7");
+ assert_eq!(parts[5], "100");
+ assert_eq!(parts[6], "199");
+ }
+
+ #[test]
+ fn label_sanitizes_illegal_chars() {
+ let label = build_label("iggy", "events.v1", "orders/inbound", 0, 0,
0);
+ // dots and slashes are not allowed in Doris labels.
+ assert!(!label.contains('.'));
+ assert!(!label.contains('/'));
+ }
+
+ #[test]
+ fn label_disambiguates_names_that_sanitize_identically() {
+ // The whole point of the joint hash: `events.v1` and `events_v1`
+ // collapse to the same sanitized form, but the hash is over the raw
+ // names so the final labels differ. Without this, two streams could
+ // silently dedupe against each other in Doris.
+ assert_ne!(
+ build_label("iggy", "events.v1", "orders", 0, 0, 0),
+ build_label("iggy", "events_v1", "orders", 0, 0, 0),
+ "labels must NOT collide for names that sanitize to the same
string"
+ );
+ }
+
+ #[test]
+ fn label_disambiguates_prefixes_that_sanitize_identically() {
+ // Two connectors writing the same stream/topic/partition/offset range
+ // but with prefixes that collapse to the same sanitized+truncated
+ // segment must still get distinct labels — otherwise Doris's label
+ // dedupe silently drops the second tenant's batch.
+ let a = build_label("prod_events_us_east_1", "events", "orders", 0, 0,
0);
+ let b = build_label("prod_events_us_east_2", "events", "orders", 0, 0,
0);
+ // Precondition: the sanitized prefix segments collide (both truncate
to
+ // the same 16 chars).
+ assert_eq!(
+ a.split('-').next(),
+ b.split('-').next(),
+ "precondition: sanitized prefixes should collide at 16 chars"
+ );
+ // ...but the full labels differ because the raw prefix is folded into
+ // the hash.
+ assert_ne!(
+ a, b,
+ "labels must NOT collide for prefixes that sanitize to the same
string"
+ );
+ }
+
+ #[test]
+ fn identity_hash_is_not_aliased_by_boundary_shift() {
+ // The joint hash is length-prefixed so shifting any boundary cannot
+ // produce the same digest: distinct (prefix, stream, topic) triples
must
+ // map to distinct hashes, otherwise two identities could share a label
+ // and silently dedupe in Doris.
+ assert_ne!(
+ identity_hash("iggy", "ab", "c"),
+ identity_hash("iggy", "a", "bc")
+ );
+ assert_ne!(
+ identity_hash("iggy", "events", "orders"),
+ identity_hash("iggy", "event", "sorders")
+ );
+ // The prefix participates too: shifting the prefix/stream boundary
must
+ // not alias.
+ assert_ne!(
+ identity_hash("ab", "c", "topic"),
+ identity_hash("a", "bc", "topic")
+ );
+ }
+
+ #[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 prefix = "p".repeat(100);
+ let stream = "s".repeat(100);
+ let topic = "t".repeat(100);
+ let label = build_label(&prefix, &stream, &topic, u32::MAX, u64::MAX,
u64::MAX);
+ assert!(
+ label.len() <= 128,
+ "label exceeds Doris's 128-char cap: {} chars: {label}",
+ label.len()
+ );
+ }
+
+ #[test]
+ fn effective_batch_size_floors_at_one() {
+ assert_eq!(effective_batch_size(Some(0)), 1);
+ assert_eq!(effective_batch_size(None), DEFAULT_BATCH_SIZE as usize);
+ assert_eq!(effective_batch_size(Some(500)), 500);
+ }
+
+ #[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_as_permanent() {
+ // 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!(
+ 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 parse_duration_parses_and_falls_back() {
+ assert_eq!(parse_duration(Some("10s"), "30s"),
Duration::from_secs(10));
+ assert_eq!(parse_duration(None, "30s"), Duration::from_secs(30));
+ // A malformed value falls back to the default rather than erroring.
+ assert_eq!(
+ parse_duration(Some("not_a_duration"), "30s"),
+ Duration::from_secs(30)
+ );
+ // A zero duration is degenerate (reqwest times out every request
+ // immediately) and falls back to the default.
+ assert_eq!(parse_duration(Some("0s"), "30s"), Duration::from_secs(30));
+ assert_eq!(parse_duration(Some("0ms"), "5s"), Duration::from_secs(5));
+ }
+
+ #[tokio::test]
+ async fn open_rejects_out_of_range_max_filter_ratio() {
+ for ratio in [1.5_f64, -0.1_f64, f64::INFINITY, f64::NAN] {
+ let mut cfg = make_config();
+ cfg.max_filter_ratio = Some(ratio);
+ let mut sink = DorisSink::new(1, cfg);
+ assert!(
+ matches!(sink.open().await, Err(Error::InvalidConfigValue(_))),
+ "expected InvalidConfigValue for max_filter_ratio={ratio}",
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn open_accepts_in_range_max_filter_ratio() {
+ for ratio in [0.0_f64, 0.5_f64, 1.0_f64] {
+ let mut cfg = make_config();
+ cfg.max_filter_ratio = Some(ratio);
+ let mut sink = DorisSink::new(1, cfg);
+ assert!(
+ sink.open().await.is_ok(),
+ "expected open() to accept max_filter_ratio={ratio}",
+ );
+ }
+ }
+
+ fn url(s: &str) -> reqwest::Url {
+ reqwest::Url::parse(s).unwrap()
+ }
+
+ /// Build a `Connected` for redirect-validation tests: a throwaway client
and
+ /// no precomputed headers, with the redirect policy under test.
+ fn connected(
+ base: &str,
+ allow_insecure: bool,
+ allowed_hosts: Option<Vec<String>>,
+ ) -> Connected {
+ Connected {
+ client: reqwest::Client::new(),
+ base_url: url(base),
+ max_filter_ratio_header: None,
+ columns_header: None,
+ where_header: None,
+ allow_insecure_redirect: allow_insecure,
+ allowed_redirect_hosts: allowed_hosts,
+ }
+ }
+
+ #[test]
+ fn redirect_refuses_https_to_http_downgrade_by_default() {
+ // A compromised FE redirecting https -> http would leak Basic creds in
+ // cleartext. Refuse it unless explicitly opted in.
+ let err = connected("https://fe.doris:8030";, false, None)
+ .validate_redirect(&url("http://attacker.evil/";), 1);
+ assert!(matches!(err, Err(Error::PermanentHttpError(_))));
+ }
+
+ #[test]
+ fn redirect_allows_downgrade_when_opted_in() {
+ // Known-insecure FE -> BE topology: operator accepts the risk.
+ assert!(
+ connected("https://fe.doris:8030";, true, None)
+ .validate_redirect(&url("http://be.doris:8040/";), 1)
+ .is_ok()
+ );
+ }
+
+ #[test]
+ fn redirect_allows_cross_host_same_scheme() {
+ // The normal FE -> BE hop: different host, same scheme, no allowlist.
+ assert!(
+ connected("https://fe.doris:8030";, false, None)
+ .validate_redirect(&url("https://be.doris:8040/";), 1)
+ .is_ok()
+ );
+ // http -> http is not a downgrade.
+ assert!(
+ connected("http://fe.doris:8030";, false, None)
+ .validate_redirect(&url("http://be.doris:8040/";), 1)
+ .is_ok()
+ );
+ }
+
+ #[test]
+ fn redirect_refuses_non_http_scheme() {
+ // An http FE redirecting to a non-HTTP scheme slips past the downgrade
+ // check (the base isn't https) and, with no allowlist, the host check
is
+ // skipped — so it must be rejected by the scheme gate before creds are
+ // re-attached. Covers the default (no allowlist) path.
+ for target in [
+ "ftp://be.doris/";,
+ "file:///etc/passwd",
+ "gopher://be.doris/";,
+ ] {
+ assert!(
+ matches!(
+ connected("http://fe.doris:8030";, false, None)
+ .validate_redirect(&url(target), 1),
+ Err(Error::PermanentHttpError(_))
+ ),
+ "scheme of {target} should be refused"
+ );
+ }
+ }
+
+ #[test]
+ fn redirect_enforces_host_allowlist_when_set() {
+ let allowed = vec!["be1.doris".to_string(), "be2.doris".to_string()];
+ // Target host not in the allowlist is refused.
+ assert!(matches!(
+ connected("http://fe.doris:8030";, false, Some(allowed.clone()))
+ .validate_redirect(&url("http://attacker.evil:8040/";), 1),
+ Err(Error::PermanentHttpError(_))
+ ));
+ // Target host in the allowlist passes (bare host pins host only).
+ assert!(
+ connected("http://fe.doris:8030";, false, Some(allowed))
+ .validate_redirect(&url("http://be2.doris:8040/";), 1)
+ .is_ok()
+ );
+ }
+
+ #[test]
+ fn redirect_allowlist_matches_ipv6_targets() {
+ // `host_str()` brackets IPv6 (`[::1]`), so a naive `rsplit_once(':')`
on a
+ // bare `::1` entry used to misparse to host ":" / port 1 and refuse a
+ // legitimate IPv6 BE redirect. Bare, bracketed, and port-pinned
entries
+ // must all match the same `http://[::1]:8040` target.
+ let target = "http://[::1]:8040/api/db/tbl/_stream_load";;
+ for entry in ["::1", "[::1]", "[::1]:8040"] {
+ assert!(
+ connected("http://fe.doris:8030";, false,
Some(vec![entry.to_string()]))
+ .validate_redirect(&url(target), 1)
+ .is_ok(),
+ "IPv6 allowlist entry {entry:?} should match {target}"
+ );
+ }
+ // A port-pinned IPv6 entry still refuses the wrong port.
+ assert!(matches!(
+ connected(
+ "http://fe.doris:8030";,
+ false,
+ Some(vec!["[::1]:8040".to_string()])
+ )
+ .validate_redirect(&url("http://[::1]:6379/exfil";), 1),
+ Err(Error::PermanentHttpError(_))
+ ));
+ // A different IPv6 host is refused.
+ assert!(matches!(
+ connected("http://fe.doris:8030";, false,
Some(vec!["::1".to_string()]))
+ .validate_redirect(&url("http://[fe80::1]:8040/";), 1),
+ Err(Error::PermanentHttpError(_))
+ ));
+ }
+
+ #[test]
+ fn redirect_allowlist_pins_port_when_specified() {
+ // A `host:port` entry pins the endpoint — an allowlisted host on a
+ // different (attacker) port is refused, closing the exfiltration
vector.
+ let allowed = vec!["be.doris:8040".to_string()];
+ assert!(
+ connected("http://fe.doris:8030";, false, Some(allowed.clone()))
+ .validate_redirect(&url("http://be.doris:8040/";), 1)
+ .is_ok()
+ );
+ assert!(matches!(
+ connected("http://fe.doris:8030";, false, Some(allowed))
+ .validate_redirect(&url("http://be.doris:6379/exfil";), 1),
+ Err(Error::PermanentHttpError(_))
+ ));
+ }
+
+ #[test]
+ fn auth_header_is_basic_b64() {
+ let sink = DorisSink::new(1, make_config());
+ // base64("root:pw") = cm9vdDpwdw==
+ assert_eq!(sink.auth_header.to_str().unwrap(), "Basic cm9vdDpwdw==");
+ // Marked sensitive so reqwest keeps it out of debug/trace output.
+ assert!(sink.auth_header.is_sensitive());
+ }
+
+ 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:?}",
+ );
+ }
+
+ #[tokio::test]
+ async fn open_rejects_columns_header_with_control_chars() {
+ // A CR/LF in `columns` is an invalid HeaderValue. reqwest would defer
the
+ // failure to every `.send()`; we must fail fast at open() instead.
+ let mut cfg = make_config();
+ cfg.columns = Some("c1,\nc2".into());
+ let mut sink = DorisSink::new(1, cfg);
+ assert!(
+ matches!(sink.open().await, Err(Error::InvalidConfigValue(_))),
+ "expected InvalidConfigValue for a columns header with a newline",
+ );
+ }
+
+ /// Drives a real 307 FE -> BE redirect through `send_stream_load` and
+ /// asserts the connector rebuilds the *full* Stream Load request on the
+ /// redirected hop. The BE mock only matches when every header is present
+ /// — crucially the `Authorization` header, which reqwest would otherwise
+ /// strip on a cross-host redirect — so a regression that drops a header
+ /// makes the BE mock miss, yielding a 404 and a failed assertion.
+ #[tokio::test]
+ async fn redirect_rebuilds_full_request_on_be() {
+ use wiremock::matchers::{header, method, path};
+ use wiremock::{Mock, MockServer, ResponseTemplate};
+
+ let server = MockServer::start().await;
+ let expected_auth = format!("Basic {}",
general_purpose::STANDARD.encode("root:pw"));
+ // Same host + scheme as the FE, so `validate_redirect` permits it —
+ // this is the normal Doris FE -> BE topology.
+ let be_url = format!("{}/be/_stream_load", server.uri());
+
+ Mock::given(method("PUT"))
+ .and(path("/api/test_db/test_tbl/_stream_load"))
+ .respond_with(ResponseTemplate::new(307).insert_header("Location",
be_url.as_str()))
+ .expect(1)
+ .mount(&server)
+ .await;
+
+ Mock::given(method("PUT"))
+ .and(path("/be/_stream_load"))
+ .and(header("authorization", expected_auth.as_str()))
+ .and(header("format", "json"))
+ .and(header("strip_outer_array", "true"))
+ .and(header("expect", "100-continue"))
+ .and(header("label", "iggy-test-label"))
+
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+ "Status": "Success",
+ "Message": "OK",
+ "NumberLoadedRows": 1,
+ "NumberFilteredRows": 0,
+ })))
+ .expect(1)
+ .mount(&server)
+ .await;
+
+ let mut cfg = make_config();
+ cfg.fe_url = server.uri();
+ let mut sink = DorisSink::new(1, cfg);
+ sink.open().await.expect("open should succeed");
+
+ let result = sink
+ .send_stream_load("iggy-test-label",
Bytes::from_static(b"[{\"a\":1}]"))
+ .await;
+
+ assert!(
+ matches!(&result, Ok(r) if r.status == "Success"),
+ "expected Ok(Success) after redirect, got {result:?}",
+ );
+ }
+
+ /// A redirect loop must surface as a *permanent* error: retrying just
+ /// re-walks the loop. (Regression guard for the HttpRequestFailed ->
+ /// PermanentHttpError reclassification.)
+ #[tokio::test]
+ async fn redirect_loop_is_permanent_error() {
+ use wiremock::matchers::{method, path};
+ use wiremock::{Mock, MockServer, ResponseTemplate};
+
+ let server = MockServer::start().await;
+ let mut cfg = make_config();
+ cfg.fe_url = server.uri();
+ // Self-redirect: same host + scheme, so validate_redirect permits it,
+ // and the connector loops until MAX_REDIRECTS is exceeded.
+ let self_url = format!("{}/api/test_db/test_tbl/_stream_load",
server.uri());
+
+ Mock::given(method("PUT"))
+ .and(path("/api/test_db/test_tbl/_stream_load"))
+ .respond_with(ResponseTemplate::new(307).insert_header("Location",
self_url.as_str()))
+ .mount(&server)
+ .await;
+
+ let mut sink = DorisSink::new(1, cfg);
+ sink.open().await.expect("open should succeed");
+ let result = sink
+ .send_stream_load("iggy-test-label",
Bytes::from_static(b"[{\"a\":1}]"))
+ .await;
+
+ assert!(
+ matches!(&result, Err(Error::PermanentHttpError(_))),
+ "expected PermanentHttpError on redirect loop, got {result:?}",
+ );
+ }
+
+ /// A redirect with no usable `Location` is malformed and permanent.
+ #[tokio::test]
+ async fn redirect_without_location_is_permanent_error() {
+ use wiremock::matchers::{method, path};
+ use wiremock::{Mock, MockServer, ResponseTemplate};
+
+ let server = MockServer::start().await;
+ let mut cfg = make_config();
+ cfg.fe_url = server.uri();
+
+ Mock::given(method("PUT"))
+ .and(path("/api/test_db/test_tbl/_stream_load"))
+ .respond_with(ResponseTemplate::new(307)) // no Location header
+ .mount(&server)
+ .await;
+
+ let mut sink = DorisSink::new(1, cfg);
+ sink.open().await.expect("open should succeed");
+ let result = sink
+ .send_stream_load("iggy-test-label",
Bytes::from_static(b"[{\"a\":1}]"))
+ .await;
+
+ assert!(
+ matches!(&result, Err(Error::PermanentHttpError(_))),
+ "expected PermanentHttpError on missing Location, got {result:?}",
+ );
+ }
+
+ /// A relative `Location` is outside Doris's absolute-Location contract and
+ /// must be rejected as permanent rather than silently joined.
+ #[tokio::test]
+ async fn redirect_with_relative_location_is_permanent_error() {
+ use wiremock::matchers::{method, path};
+ use wiremock::{Mock, MockServer, ResponseTemplate};
+
+ let server = MockServer::start().await;
+ let mut cfg = make_config();
+ cfg.fe_url = server.uri();
+
+ Mock::given(method("PUT"))
+ .and(path("/api/test_db/test_tbl/_stream_load"))
+ .respond_with(ResponseTemplate::new(307).insert_header("Location",
"be_endpoint"))
+ .mount(&server)
+ .await;
+
+ let mut sink = DorisSink::new(1, cfg);
+ sink.open().await.expect("open should succeed");
+ let result = sink
+ .send_stream_load("iggy-test-label",
Bytes::from_static(b"[{\"a\":1}]"))
+ .await;
+
+ assert!(
+ matches!(&result, Err(Error::PermanentHttpError(_))),
+ "expected PermanentHttpError on relative Location, got {result:?}",
+ );
+ }
+}
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index b0f30cbe5..81ba70ed9 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -53,6 +53,9 @@ iggy = { workspace = true }
iggy-cli = { workspace = true }
iggy_binary_protocol = { workspace = true }
iggy_common = { workspace = true }
+# 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-core = { workspace = true }
@@ -82,6 +85,7 @@ sqlx = { workspace = true }
sysinfo = { workspace = true }
tempfile = { workspace = true }
test-case = { workspace = true }
+testcontainers = { workspace = true }
testcontainers-modules = { workspace = true }
tokio = { workspace = true, features = ["full", "test-util"] }
toml = { workspace = true }
@@ -90,7 +94,7 @@ tracing-subscriber = { workspace = true }
twox-hash = { workspace = true }
url = { workspace = true }
uuid = { workspace = true }
-wiremock = "0.6"
+wiremock = { workspace = true }
zip = { workspace = true }
[target.'cfg(any(target_os = "linux", target_os = "freebsd", target_os =
"dragonfly", target_os = "netbsd", target_os = "openbsd"))'.dependencies]
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 000000000..bf067d141
--- /dev/null
+++ b/core/integration/tests/connectors/doris/doris_sink.rs
@@ -0,0 +1,492 @@
+/*
+ * 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_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(fixture.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(fixture.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(fixture.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) {
+ let db = fixture.database();
+ // 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(db, 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(db, 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(db, TEST_TABLE)
+ .await
+ .expect("count before");
+ assert_eq!(row_before, message_count as i64);
+
+ // Replay the same offsets => same label => Doris dedupes server-side.
+ // 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");
+
+ let client_http = reqwest::Client::builder()
+ .redirect(reqwest::redirect::Policy::none())
+ .build()
+ .unwrap();
+
+ let url = format!(
+ "{}/api/{db}/{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(¤t_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(db, 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;
+
+ let db = fixture.database();
+
+ // 1. The target table must NOT have been auto-created.
+ let exists = fixture
+ .table_exists(db, 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 the per-test database 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(sqlx::AssertSqlSafe(format!(
+ "SELECT TABLE_NAME FROM information_schema.tables \
+ WHERE TABLE_SCHEMA = '{db}'"
+ )))
+ .fetch_all(&pool)
+ .await
+ .expect("list tables in test database");
+ let tables: Vec<String> = rows
+ .iter()
+ .filter_map(|r| r.try_get::<String, _>("TABLE_NAME").ok())
+ .collect();
+ assert!(
+ tables.is_empty(),
+ "test database {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 db = fixture.database();
+ let count = fixture
+ .wait_for_rows(db, 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(sqlx::AssertSqlSafe(format!(
+ "SELECT SUM(calculated - `count`) AS delta FROM {db}.{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}"
+ );
+}
diff --git a/core/integration/tests/connectors/doris/mod.rs
b/core/integration/tests/connectors/doris/mod.rs
new file mode 100644
index 000000000..b75e42c97
--- /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/.config/nextest.toml
b/core/integration/tests/connectors/doris/sink.toml
similarity index 64%
copy from .config/nextest.toml
copy to core/integration/tests/connectors/doris/sink.toml
index fb488b694..59f191e0a 100644
--- a/.config/nextest.toml
+++ b/core/integration/tests/connectors/doris/sink.toml
@@ -15,14 +15,6 @@
# specific language governing permissions and limitations
# under the License.
-[[profile.default.overrides]]
-# This is a solution (or actually a workaround) for the problem that nextest
does not support
-# #[serial] macro which shall enforce sequential execution of the test case.
-filter = 'package(integration) and
test(cli::system::test_cli_session_scenario::should_be_successful)'
-threads-required = "num-cpus"
-
-[profile.default]
-slow-timeout = { period = "30s", terminate-after = 4 }
-
-[profile.ci]
-retries = 3
+[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 000000000..e18e53a60
--- /dev/null
+++ b/core/integration/tests/connectors/fixtures/doris/container.rs
@@ -0,0 +1,717 @@
+/*
+ * 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 fixture. One Doris container is shared across every doris test in this
+// CI job (or local `cargo test` session) via testcontainers'
reusable-containers
+// feature: the container is named `iggy-doris-test` and marked
+// `ReuseDirective::Always`, so the first test creates it and every subsequent
+// test (in this or any other test process on the same Docker daemon) attaches
+// to it. This is what makes `cargo test` and CI follow the same path — the old
+// orchestration in `.github/actions/rust/pre-merge` is gone.
+//
+// Cross-process sharing matters because nextest runs each test in its own
+// process. An in-process `OnceCell` would reboot Doris every time, and the new
+// container would race the previous one for the BE's 1:1-mapped 8040 port.
+//
+// The container outlives the test session by design (that's what enables
+// reuse). CI runners are ephemeral so it dies with them; locally, `docker rm
+// -f iggy-doris-test` forces a fresh boot.
+//
+// Two host-level prerequisites cannot live inside the container:
+// * `vm.max_map_count >= 2_000_000` — Doris 4.0.3's `start_be.sh` hard-exits
+// otherwise. `check_vm_max_map_count` reads `/proc/sys/vm/max_map_count`
+// and returns a self-describing error if the host is below the threshold.
+// * Routable Docker bridge IPs — the FE returns a 307 to a BE address that
+// is the BE container's internal Docker network IP. Those addresses are
+// routable from a Linux Docker host but not from macOS Docker Desktop, so
+// these tests are effectively Linux-only without extra networking work.
+
+use async_trait::async_trait;
+use integration::harness::{TestBinaryError, TestFixture};
+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, ReuseDirective,
+};
+use tokio::time::sleep;
+use tracing::info;
+use uuid::Uuid;
+
+const DORIS_IMAGE: &str = "apache/doris";
+// Apache's maintained single-container line is now tagged `<version>-all` /
+// `<version>-all-slim` (the old one-off `doris-all-in-one-2.1.0` was pushed
once
+// in 2024 and never refreshed). `-slim` is the smaller base, so it pulls
faster
+// on CI runners.
+const DORIS_TAG: &str = "4.0.3-all-slim";
+// Fixed name + `ReuseDirective::Always` is what makes the container survive
+// across nextest's per-test processes. Stable name means every test process
+// inspecting the Docker daemon finds the same one.
+const DORIS_CONTAINER_NAME: &str = "iggy-doris-test";
+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_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;
+
+// vm.max_map_count threshold Doris 4.0.3's start_be.sh enforces (`exit 1` if
+// the running kernel reports less). The number is Doris's, not ours. Only
+// referenced by the Linux precheck; gated to keep non-Linux builds
warning-free.
+#[cfg(target_os = "linux")]
+const REQUIRED_VM_MAX_MAP_COUNT: u64 = 2_000_000;
+
+// 4.0.3 sizes the FE for a dedicated host (8GB -Xms) and lets the BE consume
+// ~90% of RAM via `mem_limit = auto`. Both blow past a 16GB CI runner that
also
+// has to host the cargo build + the test binaries; cap both to fit beside that
+// workload. ~Matches the 2.1.0 image's effective footprint.
+const FE_HEAP_OVERRIDE_SED: &str = "s/-Xmx8192m -Xms8192m/-Xmx2048m
-Xms2048m/";
+const BE_MEM_LIMIT_OVERRIDE: &str = "mem_limit = 4096M";
+
+const ENV_SINK_FE_URL: &str =
"IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_FE_URL";
+const ENV_SINK_DATABASE: &str =
"IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_DATABASE";
+const ENV_SINK_TABLE: &str = "IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_TABLE";
+const ENV_SINK_USERNAME: &str =
"IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_USERNAME";
+const ENV_SINK_PASSWORD: &str =
"IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_PASSWORD";
+const ENV_SINK_LABEL_PREFIX: &str =
"IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_LABEL_PREFIX";
+const ENV_SINK_MAX_FILTER_RATIO: &str =
"IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_MAX_FILTER_RATIO";
+const ENV_SINK_COLUMNS: &str =
"IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_COLUMNS";
+const ENV_SINK_BATCH_SIZE: &str =
"IGGY_CONNECTORS_SINK_DORIS_PLUGIN_CONFIG_BATCH_SIZE";
+const ENV_SINK_STREAMS_0_STREAM: &str =
"IGGY_CONNECTORS_SINK_DORIS_STREAMS_0_STREAM";
+const ENV_SINK_STREAMS_0_TOPICS: &str =
"IGGY_CONNECTORS_SINK_DORIS_STREAMS_0_TOPICS";
+const ENV_SINK_STREAMS_0_SCHEMA: &str =
"IGGY_CONNECTORS_SINK_DORIS_STREAMS_0_SCHEMA";
+const ENV_SINK_STREAMS_0_CONSUMER_GROUP: &str =
+ "IGGY_CONNECTORS_SINK_DORIS_STREAMS_0_CONSUMER_GROUP";
+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. The
+/// table storage medium is left to Doris's default (the 2.1 all-in-one image
+/// advertised HDD-only and needed an explicit `storage_medium = HDD` override;
+/// the 4.0 image no longer does).
+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\"
+);
+";
+
+/// 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\"
+);
+";
+
+/// 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";
+
+/// Linux-only host precheck. macOS / Windows have no
`/proc/sys/vm/max_map_count`
+/// (and the all-in-one image is already unusable on macOS for the BE-redirect
+/// routing reason — see the file header). Treat the check as a no-op there;
the
+/// container start will surface any real issue.
+#[cfg(target_os = "linux")]
+fn check_vm_max_map_count() -> Result<(), TestBinaryError> {
+ let v: u64 = std::fs::read_to_string("/proc/sys/vm/max_map_count")
+ .ok()
+ .and_then(|s| s.trim().parse().ok())
+ .unwrap_or(0);
+ if v < REQUIRED_VM_MAX_MAP_COUNT {
+ return Err(TestBinaryError::FixtureSetup {
+ fixture_type: "SharedDoris".to_string(),
+ message: format!(
+ "Doris 4.0.3 BE refuses to start unless vm.max_map_count >= \
+ {REQUIRED_VM_MAX_MAP_COUNT} (current: {v}). \
+ Run: `sudo sysctl -w
vm.max_map_count={REQUIRED_VM_MAX_MAP_COUNT}`"
+ ),
+ });
+ }
+ Ok(())
+}
+
+#[cfg(not(target_os = "linux"))]
+fn check_vm_max_map_count() -> Result<(), TestBinaryError> {
+ Ok(())
+}
+
+pub struct DorisContainer {
+ // Held only so testcontainers' Drop runs on test exit.
ReuseDirective::Always
+ // makes that Drop a no-op (the container is left running for the next test
+ // to attach to), but keeping the handle around is still required to keep
+ // the bollard client connection alive.
+ _container: ContainerAsync<GenericImage>,
+ fe_url: String,
+ fe_mysql_host_port: u16,
+ // Unique per test, so many tests can share one cluster without their loads
+ // colliding. Created during setup; the connector writes into it.
+ database: String,
+}
+
+impl DorisContainer {
+ pub async fn start() -> Result<Self, TestBinaryError> {
+ check_vm_max_map_count()?;
+
+ // Custom entrypoint patches FE heap + BE mem_limit before handing off
+ // to the image's normal entry_point.sh; SKIP_CHECK_ULIMIT bypasses the
+ // image's swap/ulimit gates so we needn't swapoff the runner. Only
+ // runs on first boot; ignored on attach because the existing container
+ // already has its config applied.
+ let entrypoint_cmd = format!(
+ "sed -i '{FE_HEAP_OVERRIDE_SED}' /opt/apache-doris/fe/conf/fe.conf
&& \
+ echo '{BE_MEM_LIMIT_OVERRIDE}' >>
/opt/apache-doris/be/conf/be.conf && \
+ exec bash /usr/local/bin/entry_point.sh"
+ );
+
+ // 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.
+ //
+ // `with_container_name` + `with_reuse(Always)` is what makes the
+ // container survive across nextest's per-test processes: the first
+ // test creates `iggy-doris-test`, every later test (in any process)
+ // attaches to it. The 1:1 BE port is therefore held continuously by
+ // one container, never racing with itself across container restarts.
+ let container = GenericImage::new(DORIS_IMAGE, DORIS_TAG)
+ // GenericImage's own with_entrypoint/with_wait_for must come
before
+ // any ImageExt method, which turns GenericImage into
ContainerRequest.
+ .with_entrypoint("bash")
+ .with_wait_for(WaitFor::http(
+ HttpWaitStrategy::new(FE_HEALTH_ENDPOINT)
+ .with_port(FE_HTTP_PORT.tcp())
+ .with_expected_status_code(200u16),
+ ))
+ .with_env_var("SKIP_CHECK_ULIMIT", "true")
+ .with_cmd(["-c", entrypoint_cmd.as_str()])
+ .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())
+ .with_container_name(DORIS_CONTAINER_NAME)
+ .with_reuse(ReuseDirective::Always)
+ .start()
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "DorisContainer".to_string(),
+ message: format!("Failed to start container: {e}"),
+ })?;
+
+ 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 container ready (name={DORIS_CONTAINER_NAME}): FE HTTP ->
{fe_http_host_port}, FE MySQL -> {fe_mysql_host_port}"
+ );
+
+ // A fresh database per test so any number of tests can share one
+ // cluster without their Stream Loads colliding. `simple()` drops the
+ // hyphens so the name stays a valid Doris identifier ([A-Za-z0-9_]).
+ let database = format!("iggy_test_db_{}", Uuid::new_v4().simple());
+
+ let this = Self {
+ _container: container,
+ fe_url: format!("http://127.0.0.1:{fe_http_host_port}";),
+ fe_mysql_host_port,
+ database,
+ };
+
+ // Required on first boot; fast-paths in <1s once the BE is already
+ // alive on subsequent test-process attaches.
+ this.wait_for_be_alive().await?;
+ this.create_test_database().await?;
+ Ok(this)
+ }
+
+ pub fn fe_url(&self) -> String {
+ self.fe_url.clone()
+ }
+
+ /// The unique per-test database this fixture's connector writes into.
+ pub fn database(&self) -> &str {
+ &self.database
+ }
+
+ 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 0.9's `raw_sql` requires `SqlSafeStr`; the input here is a UUID
+ // we just generated, so `AssertSqlSafe` is appropriate.
+ sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
+ "CREATE DATABASE IF NOT EXISTS {}",
+ self.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;
+
+ /// The unique per-test database this fixture's connector writes into.
+ fn database(&self) -> &str {
+ self.container().database()
+ }
+
+ 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 {
+ // Borrow `ddl` (not move) — this runs inside the retry loop.
+ match sqlx::raw_sql(sqlx::AssertSqlSafe(ddl.as_str()))
+ .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(sqlx::AssertSqlSafe(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(sqlx::AssertSqlSafe(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);
+ }
+ // An overshoot means a duplicate landed (e.g. a label-window
+ // expiry replay). Surface it immediately with a clear message.
+ if c > expected {
+ return Err(TestBinaryError::InvalidState {
+ message: format!(
+ "Expected exactly {expected} rows in
{database}.{table} but observed {c} — a duplicate landed (label-window expiry
replay?)"
+ ),
+ });
+ }
+ }
+ 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, database: &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(), 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(),
self.container.database())
+ }
+}
+
+/// 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(inner.container.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(
+ inner.container.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 000000000..a29563655
--- /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 a020fa42c..f224c4213 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 adf97f1b7..bb4bcc69f 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;
