feat(connectors): Add Iggy source connector

.github/workflows/_build_rust_artifacts.yml

@@ -46,7 +46,7 @@ on:
       connector_plugins:
         type: string
         required: false
-        default: "iggy_connector_elasticsearch_sink,iggy_connector_elasticsearch_source,iggy_connector_iceberg_sink,iggy_connector_postgres_sink,iggy_connector_postgres_source,iggy_connector_quickwit_sink,iggy_connector_random_source,iggy_connector_s3_sink,iggy_connector_stdout_sink,iggy_connector_surrealdb_sink, iggy_connector_rabbitmq_sink"
+        default: "iggy_connector_elasticsearch_sink,iggy_connector_elasticsearch_source,iggy_connector_iceberg_sink,iggy_connector_iggy_source,iggy_connector_postgres_sink,iggy_connector_postgres_source,iggy_connector_quickwit_sink,iggy_connector_random_source,iggy_connector_s3_sink,iggy_connector_stdout_sink,iggy_connector_surrealdb_sink, iggy_connector_rabbitmq_sink"
         description: "Comma-separated list of connector plugin crates to build as shared libraries"
     outputs:
       artifact_name:

.github/workflows/edge-release.yml

@@ -98,6 +98,7 @@ jobs:
             - `iggy_connector_elasticsearch_sink`
             - `iggy_connector_elasticsearch_source`
             - `iggy_connector_iceberg_sink`
+            - `iggy_connector_iggy_source`
             - `iggy_connector_postgres_sink`
             - `iggy_connector_postgres_source`
             - `iggy_connector_quickwit_sink`

Cargo.toml

@@ -52,6 +52,7 @@ members = [
     "core/connectors/sinks/surrealdb_sink",
     "core/connectors/sources/elasticsearch_source",
     "core/connectors/sources/http_source",
+    "core/connectors/sources/iggy_source",
     "core/connectors/sources/influxdb_source",
     "core/connectors/sources/mqtt_source",
     "core/connectors/sources/postgres_source",

core/connectors/runtime/example_config/connectors/iggy_source.toml

@@ -0,0 +1,49 @@
+# 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 = "source"
+key = "iggy"
+enabled = true
+version = 0
+name = "Iggy source"
+path = "target/release/libiggy_connector_iggy_source"
+verbose = false # Log message processing at info level instead of debug (default: false)
+
+[[streams]]
+stream = "downstream_stream"
+topic = "downstream_topic"
+schema = "raw"
+batch_length = 100
+linger_time = "5ms"
+
+[plugin_config]
+# Connection string of the UPSTREAM Iggy cluster to replicate from
+connection_string = "iggy+tcp://iggy:iggy@127.0.0.1:8090"
+upstream_stream = "upstream_stream"
+upstream_topic = "upstream_topic"
+poll_interval = "1s"
+batch_size = 100
+# "earliest", "latest", or a numeric offset; used for partitions without a
+# saved offset, including after the upstream topic is recreated with the same name.
+initial_offset = "earliest"
+include_user_headers = true
+# "block" preserves malformed messages for retry; "drop_headers" forwards
+# their payload without user headers.
+malformed_message_policy = "block"
+retry_interval = "1s"
+max_retry_interval = "60s"
+verbose_logging = false

core/connectors/sources/README.md

@@ -10,6 +10,7 @@ Source connectors are responsible for ingesting data from external sources into
 | ------ | ----------- |
 | **elasticsearch_source** | Polls documents from Elasticsearch indices with timestamp-based tracking |
 | **http_source** | Webhook gateway: an embedded HTTP server shared by every instance, with per-endpoint bearer/HMAC auth and a management API for endpoints registered at runtime |
+| **iggy_source** | Replicates a topic from an upstream Iggy cluster with per-partition offset tracking |
 | **influxdb_source** | Polls InfluxDB with cursor-based timestamp tracking; supports V2 (Flux, annotated CSV) and V3 (SQL, JSONL) |
 | **mqtt_source** | Subscribes to MQTT 3.1.1 or MQTT 5 brokers with QoS 0, 1, or 2 |
 | **postgres_source** | Reads rows from PostgreSQL tables with multiple strategies: delete after read, mark as processed, or timestamp tracking |

core/connectors/sources/iggy_source/Cargo.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.
+
+[package]
+name = "iggy_connector_iggy_source"
+version = "0.5.0-edge.1"
+description = "Iggy source connector replicating a topic from an upstream Apache Iggy cluster"
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming", "replication"]
+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
+
+[package.metadata.cargo-machete]
+ignored = ["dashmap"]
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+async-trait = { workspace = true }
+dashmap = { workspace = true }
+iggy = { workspace = true }
+iggy_common = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+rmp-serde = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+
+[dev-dependencies]
+serde_json = { workspace = true }
+
+[lints]
+workspace = true

core/connectors/sources/iggy_source/README.md

@@ -0,0 +1,137 @@
+# Iggy Source Connector
+
+The Iggy source connector replicates a topic from an upstream Apache Iggy
+cluster into a stream and topic managed by the connectors runtime. It polls
+every upstream partition and persists the last acknowledged offset for each
+partition so replication can resume after a restart.
+
+## Features
+
+- Polls every partition of one upstream topic over successive cycles.
+- Preserves payload bytes and non-zero message IDs.
+- Preserves user headers by default, with an option to omit them.
+- Stores per-partition offsets in the connector runtime's state storage.
+- Retries fully failed polling cycles with exponential backoff and jitter.
+- Creates the configured upstream stream or topic when it does not exist.
+
+## Configuration
+
+```toml
+type = "source"
+key = "iggy"
+enabled = true
+version = 0
+name = "Iggy source"
+path = "target/release/libiggy_connector_iggy_source"
+verbose = false
+benchmark = false
+
+[[streams]]
+stream = "downstream_stream"
+topic = "downstream_topic"
+schema = "raw"
+batch_length = 100
+linger_time = "5ms"
+
+[plugin_config]
+connection_string = "iggy+tcp://iggy:iggy@127.0.0.1:8090"
+upstream_stream = "upstream_stream"
+upstream_topic = "upstream_topic"
+poll_interval = "1s"
+batch_size = 100
+initial_offset = "earliest"
+include_user_headers = true
+malformed_message_policy = "block"
+retry_interval = "1s"
+max_retry_interval = "60s"
+verbose_logging = false
+```
+
+Use `schema = "raw"` on the downstream stream to replicate payload bytes
+without decoding or re-encoding them. The runtime-level `batch_length` controls
+downstream producer batching, while `plugin_config.batch_size` limits each
+upstream partition poll. A poll cycle forwards at most 10,000 messages across
+all partitions and resumes with another partition on the next cycle.
+
+### Plugin Fields
+
+| Field | Required | Default | Description |
+| --- | --- | --- | --- |
+| `connection_string` | yes | none | Connection string for the upstream Iggy cluster. |
+| `upstream_stream` | yes | none | Name of the stream to replicate from. |
+| `upstream_topic` | yes | none | Name of the topic to replicate from. |
+| `poll_interval` | no | `2s` | Delay before each upstream poll cycle. |
+| `batch_size` | no | `100` | Maximum messages requested from each upstream partition per poll. Must be between `1` and `10,000`. |
+| `initial_offset` | no | `earliest` | Starting position for a partition without a saved offset, including after the upstream topic is recreated with the same name. Accepts `earliest`, `latest`, or an absolute numeric offset. |
+| `include_user_headers` | no | `true` | Copy user headers to downstream messages. |
+| `malformed_message_policy` | no | `block` | Handling for messages with unparsable user headers: `block` retries from the failed offset; `drop_headers` forwards the payload without those headers. |
+| `retry_interval` | no | `1s` | Base delay used for exponential backoff when every partition poll in a cycle fails. |
+| `max_retry_interval` | no | `60s` | Maximum delay between poll retries. |
+| `verbose_logging` | no | `false` | Log per-cycle connector details at info level instead of debug level. |
+
+The `connection_string` can use any transport supported by the Rust client,
+including TCP, QUIC, HTTP, and WebSocket. It may contain credentials, so keep
+the connector configuration private or supply the value through the
+`IGGY_CONNECTORS_SOURCE_IGGY_PLUGIN_CONFIG_CONNECTION_STRING` environment
+variable.
+
+## Offset and Delivery Semantics
+
+For each partition, the connector resumes at one offset after the last saved
+offset. `initial_offset` is used only when that partition has no saved offset:
+
+- `earliest` starts with the oldest available message.
+- `latest` requests up to `batch_size` of the most recent messages.
+- A numeric value starts at that absolute offset.
+
+If the upstream topic is recreated with the same name, the connector clears
+its saved offsets and uses `initial_offset` for the new topic.
+
+The connector stages new offsets while polling. It commits them only after the
+runtime sends the complete downstream batch and saves the connector state. If
+the batch is rejected, the previous offsets remain committed and the messages
+are eligible for replay. Consumers should therefore tolerate duplicate
+delivery after failures or crashes.
+
+After five consecutive downstream send or state-save failures, the source
+stops. An operator must restart it before it can replay the unacknowledged
+messages.
+
+> When `initial_offset = "latest"` and no offset has been committed yet, each
+> retry recalculates the current tail of the upstream partition. If a first
+> batch is rejected while new messages arrive, messages from that failed batch
+> can be skipped. Use `earliest` or a numeric offset when replay continuity is
+> required.
+
+## Message Mapping
+
+| Upstream field | Downstream behavior |
+| --- | --- |
+| Payload | Copied without modification when the downstream schema is `raw`. |
+| Message ID | Copied when the upstream ID is non-zero. |
+| User headers | Copied when `include_user_headers = true`. |
+| Offset and partition | Reassigned by the downstream Iggy topic. |
+| Timestamp and checksum | Generated for the downstream message. |
+
+## Operational Notes
+
+- The connector discovers new upstream partitions automatically on each poll cycle.
+- A `batch_size` outside `1..=10,000` prevents the connector from starting.
+- The 10,000-message cycle cap bounds message count, not payload bytes. Large
+  messages can still use substantial memory while the batch crosses the runtime.
+- If the upstream stream is missing, the connector creates it. If the topic is
+  missing, it creates the topic with one partition, no compression, persisted
+  durability, and a one-message save threshold.
+- Invalid `initial_offset` values produce a warning and fall back to
+  `earliest`.
+- If user headers cannot be decoded, `malformed_message_policy = "block"`
+  forwards the valid prefix before the malformed message, leaves that
+  partition at the failed offset, and continues polling other partitions at
+  the normal `poll_interval`. Record conversion errors do not activate the
+  connector-wide polling backoff.
+- `malformed_message_policy = "drop_headers"` forwards a malformed message's
+  payload without its user headers and advances the partition offset. Setting
+  `include_user_headers = false` bypasses header parsing for every message.
+- A polling error on one partition is logged and counted without skipping the
+  remaining partitions. Connector-wide backoff is activated only when no
+  partition poll succeeds during the cycle.

core/connectors/sources/iggy_source/config.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.
+
+type = "source"
+key = "iggy"
+enabled = true
+version = 0
+name = "Iggy source"
+path = "../../target/release/libiggy_connector_iggy_source"
+verbose = false
+
+[[streams]]
+stream = "downstream_stream"
+topic = "downstream_topic"
+schema = "raw"
+batch_length = 100
+
+[plugin_config]
+connection_string = "iggy+tcp://iggy:iggy@127.0.0.1:8090"
+upstream_stream = "upstream_stream"
+upstream_topic = "upstream_topic"
+poll_interval = "1s"
+batch_size = 100
+# Starting position when a partition has no committed offset, including after
+# the upstream topic is recreated with the same name:
+# "earliest", "latest", or an absolute numeric offset.
+# With "latest", if the first batch is NACKed before an offset is committed,
+# retrying recalculates the moving tail and may skip messages from the failed
+# batch when the upstream topic advances.
+initial_offset = "earliest"
+include_user_headers = true
+# "block" preserves malformed messages for retry; "drop_headers" forwards
+# their payload without user headers.
+malformed_message_policy = "block"
+retry_interval = "1s"
+max_retry_interval = "60s"
+verbose_logging = false