jiengup opened a new pull request, #3886:
URL: https://github.com/apache/iggy/pull/3886
Which issue does this PR address?
Closes #3869
Relates to #3764
## Rationale
Cross-cluster topic replication had no first-class path: the connectors
subsystem's sinks and sources target external systems (Postgres, Elasticsearch,
HTTP, etc.), and the runtime itself connects to a single Iggy cluster.
Cross-cluster sync (disaster recovery, data migration, federated setups)
therefore required external tooling. This PR adds `iggy_source`: a source
connector that replicates a topic from an **upstream** Iggy cluster into the
cluster the connectors runtime is connected to.
## What changed?
Previously there was no way to sync two Iggy clusters through connectors.
Now `core/connectors/sources/iggy_source` connects to the upstream cluster and
discovers all partitions in `open()`, polls each partition with an explicit
`poll_messages(offset)` call per cycle, and passes payloads and user headers
through as `Schema::Raw` while preserving upstream message IDs for downstream
deduplication. Sync progress (per-partition confirmed offsets, total synced,
error count) is persisted in connector state, and restarts resume from
`saved_offset + 1` without replaying history.
### Design tradeoffs
- **Low-level `poll_messages` API instead of the high-level
`IggyConsumer`**: crash recovery requires per-partition starting offsets, but
`IggyConsumer` accepts only one global `polling_strategy` at construction time,
which cannot express per-partition offsets. The low-level API allows passing
`PollingStrategy::offset(saved + 1)` per partition explicitly.
- **No consumer group**: server-side group offset storage is untrusted
(consume ack precedes downstream produce, so a crash loses messages). For
single-instance replication the group provides only fencing, no real benefit,
while introducing join/rebalance failure modes. `JoinConsumerGroupResponse` is
empty (no member id returned), so assignment-driven multi-instance scaling is
not expressible through the low-level API. Revisit once the server exposes
member ids.
- **Direct offset jump vs replay-and-filter**: recovery uses `offset(saved +
1)` directly rather than `first()` plus in-memory filtering, keeping restart
cost O(batch) instead of O(topic history).
- **`Schema::Raw` byte passthrough**: no decode/encode; payloads, user
headers, and message IDs are preserved verbatim, avoiding re-serialization cost
and format corruption.
- **`InvalidOffset` reset**: when upstream retention expires a saved offset,
the partition is warned and reset to `initial_offset` (`earliest` / `latest` /
numeric) without blocking other partitions.
- **Exponential backoff**: connection-level failures reuse the SDK's
`exponential_backoff` + `jitter` (starting at `retry_interval`, capped at
`max_retry_interval`); the failure counter is an `AtomicU64` and resets on any
successful cycle.
- **Missing upstream stream/topic**: warn and auto-create (topics with 1
partition).
### Persisted state design
```rust
#[derive(Debug, Serialize, Deserialize)]
struct State {
offsets: HashMap<u32, u64>, // partition_id -> offset of the last
message confirmed written downstream
messages_synced: u64, // cumulative synced message count
errors_count: u64, // cumulative errors (connection failures,
conversion failures, offset resets)
}
```
- **Encoding and durability**: serialized with MessagePack (`rmp_serde`) via
`ConnectorState::serialize/deserialize`; the runtime persists it with the
existing `FileStateProvider` to `{state_path}/source_{key}.state`, inheriting
its atomic-rename + fsync protocol and `0o600` permissions. **The state save
path is untouched.**
- **Confirmation semantics (the key invariant)**: the connector advances
offsets in state only for messages handed to the runtime, and the runtime saves
state only after a **successful downstream send**. The on-disk state is
therefore always the "confirmed delivered downstream" watermark and the single
authoritative source for crash recovery.
- **Bounded size**: state is one `u64` per partition plus two counters,
O(partition count), so rewriting the whole file every batch is cheap.
- **Failure tolerance**: deserialization failures (corruption, version
mismatch) log a warning and start from a fresh state (non-fatal).
`initial_offset` applies only to partitions with no saved entry (first run or
newly added partitions); existing entries always win.
- **Error-count persistence**: `errors_count` increments and offset resets
ride the same state channel, and state is returned even on empty-message cycles
(e.g., connection-failure cycles), so error accounting and offset resets
survive restarts.
### Crash recovery analysis
| Failure point | Behavior |
|---------------|----------|
| Connector process crash (upstream and downstream healthy) | The runtime
persists state only after a successful downstream send; offsets not yet
persisted are re-polled → at-least-once, with the preserved upstream message ID
enabling downstream dedup |
| Upstream cluster outage | Poll errors → `errors_count` incremented,
offsets not advanced, exponential backoff with jitter; the SDK client
reconnects automatically |
| Downstream cluster outage | Runtime `producer.send` fails → state not
saved; the connector's in-memory offsets have advanced, so that batch is
dropped for the lifetime of the process, this can be fixed when #3855 is
merged; a connector restart replays from the stale state file → at-least-once |
| Upstream retention expires data | `InvalidOffset` → partition reset to
`initial_offset`; expired messages are unrecoverable (inherent to offset-based
replication) |
### Sync semantics
- **At-least-once**: state records only offsets confirmed written
downstream; recovery always resumes from the confirmation point + 1, so no
message is lost, and duplicates within the crash window are deduplicated
downstream via the preserved upstream message IDs.
- The state mutex is acquired exactly twice per poll cycle (read offsets /
write offsets) and never held across I/O; empty polls do not count as errors.
## Local Execution
- Passed: `cargo fmt --all`, `cargo sort --no-format --workspace`, `cargo
clippy -p iggy_connector_iggy_source --all-targets --all-features -- -D
warnings`, `cargo test -p iggy_connector_iggy_source` (11 unit tests: state
restore, serialization round-trip, config defaults, initial_offset parsing,
next_strategy jump, connection-string redaction), taplo, markdownlint,
license-headers, cargo machete
- Pre-commit hooks ran / not ran: not ran
### End-to-end verification (two real server-ng instances + runtime)
- Upstream/downstream `iggy-server` (TCP 8090/8091) + `iggy-connectors`; a
CLI producer continuously emitted messages with headers
- All 542 messages synced, message counts identical on both clusters;
sampled payloads, user headers (`producer:string`, `seq:uint64`), and message
IDs are byte-identical
- Crash recovery: killed the connector, produced 3 more messages, restarted
→ `Restored state ... Offsets: {0: 4}, messages synced: 5` → only the 3 missing
messages were synced, the first 5 with zero duplicates
- The topic-size difference between the two clusters was traced to the
server's 256-byte per-save metadata blocks (CLI single-message sends vs runtime
batched sends), not to any data difference
## AI Usage
1. Claude (opencode)
2. Entire implementation (architecture, code generation, tests) was
AI-generated with human review of the design decisions
3. Verified via compilation, clippy `-D warnings`, 11 unit tests, and a real
two-server E2E run including header passthrough and crash recovery
4. Yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
