You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@iggy.apache.org by "hubcio (via GitHub)" <gi...@apache.org> on 2026/05/22 18:00:34 UTC
Iggy Connectors: per-message routing for sources
GitHub user hubcio created a discussion: Iggy Connectors: per-message routing for sources
## Problem
Source connectors today write to one stream and one topic, picked at startup.
Say you have a Postgres table with 50 event types and you want each type in its own Iggy topic. today you either split into 50 tables (bad schema, no one uses databases like this) or run 50 connector copies. the 50-copy version means 50x DB load, 50 configs, 50 state files; for CDC, 50 replication slots, each pinning WAL until its `restart_lsn` advances - one stuck consumer pins WAL for the whole cluster.
Using iggy connectors should not force its users to bad schema or design choices.
The row should say where it goes; the connector should send it there.
## Proposal TL;DR
Messages carry `destination: Option<Destination>`; runtime groups a batch by `(stream, topic)` and dispatches each group via one unbound `IggyProducer::send_to`. Postgres source derives destination from a column or table name; any source can use the new `Route` transform to map a payload field through a static lookup. `Source` gains required `commit()` (runs after atomic state persist; owns slot-advance, deferred DELETE) and `mode()` (`Cdc | Polling { destructive } | Opaque`, drives commit-failure policy). Destination growth bounded by `max_destinations` cap, allowlist, and per-destination circuit breaker. Bundled: fix producer task leak on stream switch, atomic `FileStateProvider::save` (tmp+rename+fsync), switch Postgres CDC slot from `pgoutput` to `test_decoding` so the existing parser actually matches.
## Scope of changes
two changes that make ABI break - we're under 1.0.0 no migrators, etc. v1 favors simple-and-correct over flexible. things deliberately left for later live in Future so we don't accidentally close those doors.
## Step 1 - messages carry their own destination
- `ProducedMessage` and `DecodedMessage` get `destination: Option<Destination>` where `Destination { stream: Identifier, topic: Identifier, partition_key: Option<Partitioning> }`. `partition_key = None` = balanced (server picks).
- the runtime groups a batch by `(stream, topic)` and sends each group through one shared unbound `IggyProducer`. `partition_key` rides through to `send_to`. `IggyProducer::send_to(stream, topic, messages, partitioning)` already takes them per call, so the producer doesn't need to be bound.
- cap + allowlist checked before any send. a typo or a weird row value can't accidentally create thousands of topics.
fits CDC and any source that already knows where each row belongs (table name, tenant ID, event-type column). the Postgres source gains config keys for column-derived destinations, per-table mapping, and a flag to strip routing columns from the payload.
bonus: kills a real bug. configuring several streams for one source builds one producer per stream and keeps the last; each discarded producer leaves a spawned task at `producer.rs:153` that only exits on `DiagnosticEvent::Shutdown` from the underlying client. routing makes this impossible by design (one producer per source plugin), but the orphan leak should still be fixed independently with a cancellation token on `ProducerCore` - it bites any SDK user, not just connectors.
## Step 2 - a routing transform
- a new `Route` transform reads a field from the payload, looks the value up in a static mapping, and writes the destination onto the message (same field Step 1 added). the mapping is served by the existing config HTTP server (one new endpoint, same auth).
- `Route` replaces the whole `destination` atomically. mixing plugin-set `partition_key` with Route-set topic isn't allowed: different topic = different partition count = different key->partition mapping. one wins, all-or-nothing.
- `Route` always overwrites if it matches; no precedence option. a source configured with both any `routing.*` key and a `Route` transform fails init - explicit refusal at startup.
fits HTTP webhook sources, file tailers, bridges from other messaging systems - anything where the plugin dumps opaque data and the operator owns routing.
`topic_field` uses dotted-path syntax over JSON objects: `a.b.c` walks Object -> Object -> leaf. no arrays, no key names containing `.`. new grammar, scoped to `Route` only; the existing `filter_fields` transform stays top-level-key-only.
dispatch order:
1. plugin sets destination.
2. transforms run; `Route` may overwrite.
3. admission checks the result against cap and allowlist.
4. anything still missing follows `on_missing_destination`.
non-Route transforms mutate `DecodedMessage` in place today, so they preserve `destination` for free. only test helpers and trait docs need updating.
## SDK changes (ABI break)
Step 1 cannot land on the current SDK; batched into the same release:
- `IggyProducer::unbound()` constructor + builder. today `IggyProducerBuilder::new` requires stream/topic up front and `ProducerCore::init` creates them. unbound skips that and exposes only `send_to`; calling `send()` errors. the runtime enforces the split at startup: bound producer rejects messages with a `destination`, unbound rejects messages without one.
- `unbound()` is **direct-only**. builder refuses `SendMode::Background` - background ack means "queued into the linger buffer", not "server acked", and the commit story below needs real acks. losing the distinction silently breaks CDC. background batching for routed sources is a future design with explicit ack callbacks.
- `unbound()` forces `send_retries_count = None`. retries live in the per-destination circuit breaker.
- the `Source` trait gains two required methods, no defaults:
- `async fn commit(&self) -> Result<(), Error>` - runs **after** the runtime has atomically persisted state (tmp + rename + parent-dir fsync; today's `FileStateProvider::save` truncates-then-writes, leaving the file zero-byte on a mid-write crash - fix bundled in this release). plugin drains `Mutex<Option<PendingBatch>>` and runs side effects (slot-advance, deferred DELETE). if `commit` is skipped or fails, the next `poll` filters rows already covered by persisted state (CDC: drop rows where `lsn <= last_persisted_lsn`, then re-run slot-advance). required, no default impl - silent no-op corrupts state. no `state` arg: `PendingBatch` is the single typed home.
- `fn mode(&self) -> SourceMode` - returns `Cdc | Polling { destructive: bool } | Opaque`. drives `on_admission_failure` default and commit-failure policy. no default - plugin author picks at code time, not config time. runtime cross-checks at startup: `Polling { destructive: false }` with `delete_after_read = true` or `processed_column` set = init fails. config can't escalate non-destructive into destructive; the reverse (overdeclared destructive) is allowed.
- `ProducedMessage` and `DecodedMessage` gain `destination: Option<Destination>`. sink-side `ConsumedMessage` unchanged.
- `ConnectorState` stays `pub Vec<u8>`. the plugin puts a state-shape version as the first field of its own envelope. version mismatch -> typed error from `Source::init` -> connector refuses to start; operator clears the state file deliberately. silent reset is opt-in behind an explicit plugin flag (off by default). the runtime never interprets state contents.
- `ConnectorStatus` gains `Degraded`, slotted between `Running` and `Error`. today's enum (`Starting | Running | Stopping | Stopped | Error`) treats `Error` as terminal and can't carry "still retrying, not progressing" which the CDC commit-retry loop needs.
- runtime call site: `process_messages` returns `Vec<(DestinationKey, IggyMessage)>` instead of `Vec<IggyMessage>`. `source_forwarding_loop` replaces the single `producer.send(...)` with a loop that issues one `send_to` per destination group, waits for all to finish, persists state atomically, then calls `commit`.
- in-tree plugins (random, http, postgres, elasticsearch, influxdb) gain `commit: Ok(())` + correct `mode()` impls in the same release. plugins ship as independent `.so` files - the runtime gates `dlopen` on `iggy_source_abi_version() -> u32` matching its compiled-in constant and refuses load on mismatch with a typed error. without this gate, postcard would silently misdecode `ProducedMessage` (new `destination` field is non-self-describing wire-format change) and every batch from an old `.so` would drop.
## State and atomicity
this decides whether the design is safe to ship.
- batch = unit of state advance. `commit` runs only after every per-destination send has acked.
- strictly sequential per plugin: `poll -> send fan-out -> commit -> next poll`. the runtime holds one in-flight batch per plugin. pipelining gets added later behind `max_in_flight_batches` if anyone needs the throughput.
- per-destination sends inside a batch run in parallel, bounded by `max_concurrent_destinations` (default 16). this caps in-flight requests, not wire bandwidth - all sends share one client and one TCP/QUIC writer, so the knob mostly hides latency.
- a single failed destination fails the whole batch. state is not persisted, `commit` is skipped, same rows return on next poll.
- state persist happens only after every send acks, and `commit` runs only after state persist returns. a failed `commit` leaves state durable but side effects skipped - next `poll` filters covered rows and re-runs the side effect. crash between send-ack and state persist -> re-poll, no duplicates beyond the in-flight batch. crash between state persist and `commit` -> next `poll` sees state covering the batch, plugin filters and replays side effects. duplicates on restart accepted.
commit-failure policy by source mode:
| mode | policy |
|---|---|
| `Cdc` | retry with exp backoff capped at 30s, never gives up - stopping = WAL backs up on the server. after `commit_failure_threshold` (default 16) consecutive failures, status flips `Running -> Degraded`. one successful `commit` clears the counter and flips back. no hysteresis - flapping is visible in status-change history; alerts fire on entry to `Degraded`. |
| `Polling` | re-poll on next tick (today's behavior). rows stay in the source table, safe by construction. |
| `Opaque` (webhook/file) | stop the connector and surface the error. no buffer behind it. |
stop during CDC commit-retry: operator stop interrupts the backoff sleep and waits up to `commit_stop_grace` (default 30s) for the in-flight `commit` to finish. miss the window -> runtime exits without further commits; restart re-polls from last persisted state, duplicates for the in-flight batch accepted. no force-kill mid-`commit` in v1.
escape hatch for prolonged outage: `POST /sources/{key}/cdc/abandon-batch` advances the slot to `PendingBatch.last_lsn` without sending, clears the pending batch, surfaces `last_abandon_at` in status. documented data-loss path - the only mechanism that keeps WAL from filling PG storage if Iggy is unreachable for hours. restart cannot do this on its own (state still points at the stuck batch).
CDC requirement: switch `pg_logical_slot_get_changes` (auto-advances `confirmed_flush_lsn`) to `pg_logical_slot_peek_changes` plus explicit `pg_replication_slot_advance(slot, last_lsn)` from inside `commit`. `PendingBatch` carries `last_lsn`; persisted state carries `last_persisted_lsn`. on `poll`, drop rows with `lsn <= last_persisted_lsn` before forming the batch - closes the window where state is durable but the slot hasn't been advanced yet. requires PG >= 11.
polling requirement: every destructive op moves from `poll` to `commit`. that's `delete_after_read = true` (DELETE) and `processed_column` (UPDATE ... SET col = TRUE). today both run in `mark_or_delete_processed_rows` during poll; both move.
duplicates accepted as the cost of correctness. plugins should pick stable `id`s (row PK hash, LSN+xid) instead of `Uuid::v4` so downstream can dedupe. the Iggy server does not dedupe; the runtime treats `id` as opaque. filtering replays is a downstream consumer's job.
## Producer model
- sources without `routing.*` keep current behavior: the runtime builds a bound `IggyProducer` from `stream`/`topic` config and rejects messages that set `destination`. adding `routing.*` flips the source to unbound. one mode per source, picked at startup.
- one `IggyProducer::unbound()` per source plugin. direct-only.
- destinations resolved via `DashMap<(stream, topic), AdmissionRecord>`. `AdmissionRecord` holds a `tokio::sync::OnceCell<DestinationMeta>` (captures `partitions_count` during the one-time ensure call, used for `partition_id` bounds checks) plus circuit-breaker state. `DashMap::entry` makes get-or-insert atomic within the process.
- the ensure call is idempotent: `IggyClient::create_stream` then `create_topic`, treating `StreamNameAlreadyExists` / `TopicNameAlreadyExists` as success. server-side both errors leave no partial state behind, so concurrent first-touches are safe.
- allowlist destinations are pre-warmed at startup, in parallel. individual pre-warm failures log and continue (next `send_to` retries on the slow path). `startup_prewarm_deadline` (default 60s) bounds total time; on deadline, un-warmed destinations log warnings and the connector starts anyway. open / denylist modes pay the round-trip on first touch per new destination - documented tail-latency event.
- no eviction. `max_destinations` is a hard cap on live admission records; hitting it = admission failure regardless of `mode`. allowlist bounds the universe at config time; open mode is "set cap to expected universe; hitting it = config error, not a hot-path event". eviction would need breaker-pinning carveouts and re-touch semantics easy to get wrong.
- per-destination circuit breaker. counts send failures, opens after 5 consecutive failures (`circuit_breaker.failure_threshold`), holds open for 30s (`circuit_breaker.cool_down`), then half-opens (one probe; success closes, failure reopens for another full cool-down). hitting an open breaker = admission failure with reason `circuit_open` -> `on_admission_failure`. does **not** silently reroute to `default_*` - those are NULL/missing defaults applied before admission, not a fallback for healthy-but-busy destinations. breaker skips errors whose recursively-unwrapped cause is `CannotSendMessagesDueToClientDisconnection` (the SDK boxes every send error in `ProducerSendFailed { cause, ... }`, so this is a structural unwrap, not a variant match) - those hit every destination at once and would trip every breaker on a transient outage. proper transient-vs-permanent classification waits on `IggyError::is_retryable` (future).
## Config
### Postgres - destination from columns
```toml
[plugin_config]
mode = "polling"
tables = ["events"]
[plugin_config.routing]
topic_column = "event_type"
stream_column = "tenant"
default_topic = "events.unknown"
default_stream = "default"
strip_routing_columns = true
partition_key_column = "user_id"
partition_key_encoding = "bytes" # bytes | key_u32 | partition_id
```
`partition_key_encoding` maps to `Partitioning` variants: `bytes -> messages_key(&[u8])`; `key_u32 -> messages_key_u32(u32)` (server hashes, client doesn't pre-hash); `partition_id -> partition_id(u32)` (direct partition selection, no hashing). `partition_id` requires `routing.topic_defaults.partitions_count > 1`. admission bounds-checks each value against `partitions_count` from the cached `DestinationMeta` and rejects OOB rows as `partition_id_out_of_range` - counted separately, doesn't feed the breaker, so a bad ID column can't block a healthy topic.
NULL / missing values degrade gracefully: missing `partition_key_column` -> `Partitioning::None` (server picks); missing `stream_column` / `topic_column` -> `default_stream` / `default_topic`. config validation requires `default_*` whenever the matching `*_column` is set - otherwise the only alternative is silently dropping every NULL-keyed row.
### Postgres - destination from table name (CDC)
```toml
[plugin_config.routing.by_table]
"public.orders" = { stream = "commerce", topic = "orders" }
"public.users" = { stream = "identity", topic = "users" }
```
### any source - routing transform
```toml
[transforms.route]
topic_field = "event.type" # dotted-path into JSON objects
default_topic = "webhooks.unknown"
[transforms.route.mapping]
"user.created" = "users"
"order.placed" = "orders"
"payment.failed" = "alerts.payments"
```
the mapping is fetched once at startup from the config HTTP server. to reload, bump `config.version` and restart. hot-reload is future; v1 contract is "mapping is frozen for the connector's lifetime". if the fetch fails (HTTP error, 404, malformed JSON), the connector refuses to start - empty mapping is never an acceptable fallback because Route would silently route every message through `default_topic`.
### topic defaults (auto-created destinations)
```toml
[plugin_config.routing.topic_defaults]
partitions_count = 1
compression_algorithm = "none"
replication_factor = 1
message_expiry = "server_default"
max_topic_size = "server_default"
```
one block per source. every auto-created topic inherits. per-allowlist-entry overrides are future.
### admission
```toml
[plugin_config.routing.admission]
mode = "allowlist" # open | allowlist | denylist
max_destinations = 256 # per plugin instance, in-memory, resets on restart
max_concurrent_destinations = 16
on_missing_destination = "default" # default | drop | error
on_admission_failure = "error" # drop | error (default per source mode, see below)
allowlist = [
{ stream = "commerce", topic = "orders" },
{ stream = "commerce", topic = "*" },
{ stream = "identity", topic = "users" },
]
```
allowlist grammar is narrow: each entry is `{ stream: SEG, topic: SEG }` where `SEG` is either a literal identifier (`[a-zA-Z0-9._-]+`, dots literal) or `*` as the whole segment. no prefix/suffix globs, no regex. `{ stream = "*", topic = "*" }` is rejected by config validation - use `mode = "open"` instead, the same thing said deliberately.
denylist uses the same grammar. destinations matching any entry are rejected and follow `on_admission_failure`; everything else passes. `{ stream = "*", topic = "*" }` rejected here too - it would refuse every message, a config error not a routing decision.
`max_destinations` is a hard ceiling. hitting it = admission failure regardless of mode. true in `open` mode too: open means "no allowlist", it does not mean "unbounded".
`on_admission_failure` default by source mode:
| source mode | default | why |
|---|---|---|
| CDC | `error` | WAL already consumed; drop = silent data loss |
| polling + `delete_after_read` / `processed_column` | `error` | once `commit` runs the destructive op, drop = silent data loss |
| polling without destructive ops | `drop` | rows stay in source table; re-poll catches them |
| webhook / file / opaque | `drop` | upstream owns the buffer |
## Telemetry
hot-path counters stay low-cardinality (`connector_key` only, no stream/topic labels). per-destination detail is exposed via a pull admin endpoint instead of fanned out across Prometheus labels, so a runaway routing config can't blow up the metrics backend.
- `iggy_connector_messages_routed_total{connector_key}`
- `iggy_connector_destinations_active{connector_key}` (gauge, bounded by `max_destinations`)
- `iggy_connector_destinations_rejected_total{connector_key, reason}` - reason in {`cap`, `denylist`, `unknown`, `circuit_open`, `partition_id_out_of_range`}
- `iggy_connector_routing_unmatched_total{connector_key, action}` - action in {`default`, `drop`, `error`}
- `iggy_connector_destination_create_latency_seconds` (histogram, no destination labels)
- `iggy_connector_destination_circuit_open{connector_key}` (gauge, count of open breakers)
`GET /connectors/{key}/destinations` returns the live admission table: stream, topic, breaker state, last error, message counts. ops who need per-destination detail look there instead of in Prometheus.
## What this design is not
- not a generic SMT framework; `Route` is the only routing-aware transform.
- not a dynamic topic factory - cap + allowlist + circuit breaker bound destination growth and keep it observable.
- not a full fix for the broken Postgres CDC backend, but includes the smallest unblock: today's slot uses `'pgoutput'` (binary) but the parser expects `INSERT:` / `UPDATE:` / `DELETE:` prefixes that only `test_decoding` emits, so it never matches. the routing release switches the slot to `'test_decoding'` (one-line change) so CDC users can exercise routing. proper binary `pgoutput` decode lands behind `cdc_pg_replicate` later.
- not hot-reload of `Route` mapping; restart-connector is the v1 reload mechanism.
- not cross-destination ordering within a batch; parallel `send_to` calls ack out of order. operators who need cross-topic causality opt in with `max_concurrent_destinations = 1`.
- not background-dispatcher friendly; routed producers are direct-only.
- not per-destination schemas; one encoder per source plugin. revisit if anyone hits it.
- not sink-side fan-out (one sink writing to N tables/indices); separate design.
## Future, not in v1
deliberately out of scope:
- pipelined `poll -> commit` (`max_in_flight_batches`)
- hot-reload of `Route` mapping with epoch tag on routed messages
- transient-vs-permanent error classification (`IggyError::is_retryable`)
- proper binary `pgoutput` decode for the CDC backend
- bounded per-destination Prometheus labels under allowlist mode
- sink-side multi-destination dispatch (mirror of `Route` for sinks)
- per-allowlist-entry topic-defaults overrides
- background dispatch for routed producers (needs explicit ack callbacks)
## Verification
end-to-end checks before merging:
- existing single-stream sources still work unchanged (random_source, influxdb_source without routing config).
- Postgres column routing: configure `topic_column`, populate with three event types, assert three topics receive correctly.
- Postgres CDC routing: switch slot to `test_decoding`, insert into two tables, assert `by_table` routing lands rows and `confirmed_flush_lsn` advances only on `commit`.
- Route transform: webhook source feeding `{"event": {"type": "..."}}`, mapping resolves to three topics, default catches unknown.
- crash recovery: kill the runtime between send-ack and `commit`, restart, assert no data loss (duplicates expected for CDC and destructive-polling).
- atomic state file: kill the runtime mid `FileStateProvider::save` (between truncate and fsync), restart, assert state is the previous full version (tmp+rename leaves either old or new, never empty).
- ABI mismatch: load an `.so` built with old SDK, assert `dlopen` returns a typed `IncompatibleAbi` error before any FFI call - no postcard decode attempt.
- abandon-batch: black-hole Iggy, drive a CDC connector into `Degraded`, hit `/cdc/abandon-batch`, assert slot advances to `PendingBatch.last_lsn`, status returns to `Running`, `last_abandon_at` is set, next poll proceeds.
- partition_id OOB: configure `partition_id` encoding + `partitions_count = 4`, feed a row with id = 99, assert admission rejects with `partition_id_out_of_range`, breaker count is unchanged, other rows in the batch still send.
- admission: configure allowlist, feed non-listed destination, assert rejection counter increments and `on_admission_failure` drives drop-vs-error correctly.
- circuit breaker: black-hole one destination, assert N consecutive failures trip the breaker, traffic to other destinations is unaffected, breaker closes after cool-down.
- background-dispatcher refusal: try to build an unbound producer with `SendMode::Background`, assert builder errors at build time.
- config validation: source with both `routing.*` and a `Route` transform, assert init fails before start.
- CDC commit Degraded flip: black-hole Iggy during commit, assert retries with capped backoff, flips to `Degraded` after `commit_failure_threshold`, never silently sits as `Running`.
- cap-hit path: `max_destinations = 2`, feed three distinct destinations, assert the third routes through `on_admission_failure` regardless of admission `mode`.
- state version mismatch: write a state file with bumped version, restart, assert `Source::init` returns the typed error and status reports `Error` with version-seen-vs-expected - not silent reset.
- Route mapping fetch failure: point at config HTTP endpoint returning 500 (or 404, or malformed JSON), assert startup fails with upstream error surfaced - connector does not boot with empty mapping.
- partition_key NULL fallback: configure `partition_key_column`, feed a row with NULL, assert `Partitioning::None` and no error counter increment.
- Degraded visibility: drive a CDC connector past `commit_failure_threshold`, assert status API returns `Degraded` (not `Running`, not `Error`) and the alertable metric reflects it.
Reference discussion:
https://discord.com/channels/1144142576266530928/1423706109402808504/1507212593260134530
GitHub link: https://github.com/apache/iggy/discussions/3313
----
This is an automatically sent email for dev@iggy.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@iggy.apache.org
Re: Iggy Connectors: per-message routing for sources
Posted by "rohankumardubey (via GitHub)" <gi...@apache.org>.
GitHub user rohankumardubey added a comment to the discussion: Iggy Connectors: per-message routing for sources
Yeah, happy to take a stab at it! I’ll have a look at the current implementation and the requirements, and see if I can put together a PR. 👍
GitHub link: https://github.com/apache/iggy/discussions/3313#discussioncomment-18542780
----
This is an automatically sent email for dev@iggy.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@iggy.apache.org
Re: Iggy Connectors: per-message routing for sources
Posted by "rohankumardubey (via GitHub)" <gi...@apache.org>.
GitHub user rohankumardubey added a comment to the discussion: Iggy Connectors: per-message routing for sources
Hey @hubcio, following our Discord discussion, here’s the proposed design for the first part of #3313 before I start implementation.
My current idea includes the following:
- **Message destination**
- Add `destination: Option<Destination>` to `ProducedMessage` and `DecodedMessage`.
- `Destination` contains stream and topic names.
- Leave partition keys out of this version.
- Preserve the destination through decoding and existing transforms.
- **Runtime configuration**
- Make routed mode explicit in the source configuration.
- Configure the allowed stream/topic pairs and one encoder for the source.
- Validate at startup that the configured destinations exist and use persisted durability.
- Keep automatic topic creation for a follow-up.
- **Producer and dispatch**
- Use one direct, unbound producer per source.
- Validate the destinations in the entire batch before sending anything.
- Group messages by `(stream, topic)` and send each group through `send_to()` with balanced partitioning.
- Preserve message order within each group. There is no ordering guarantee across destinations.
- **State and acknowledgment**
- Keep the existing single in-flight batch and `on_batch_result()` contract.
- After every group succeeds, persist the staged state and ACK the source.
- If a destination is missing or disallowed, or any send fails, leave the checkpoint unchanged and NACK the batch.
- If some groups succeeded before another failed, replay may duplicate those messages.
- A state-save failure also produces a NACK.
- **Plugin compatibility**
- Include an explicit source ABI check with the message-format change.
- Reject incompatible plugins before opening them or decoding their batches.
- Update all in-tree sources to populate the new field, using `None` for sources without routing.
- **Verification**
- Confirm existing single-destination sources still work.
- Test one source batch reaching multiple topics correctly.
- Test missing, disallowed, and non-persisted destinations.
- Test a send failure after another group succeeds, verifying no checkpoint advance and replay.
- Test state-save failure and incompatible plugin rejection.
PostgreSQL column/table routing would follow once this foundation lands. Topic creation and the `Route` transform would come afterward.
Please flag anything you’d change in this first slice, particularly the config shape or failure handling.
GitHub link: https://github.com/apache/iggy/discussions/3313#discussioncomment-18833992
----
This is an automatically sent email for dev@iggy.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@iggy.apache.org
Re: Iggy Connectors: per-message routing for sources
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
GitHub user hubcio added a comment to the discussion: Iggy Connectors: per-message routing for sources
awesome! ping us on discord of anything is unclear or if you need help
GitHub link: https://github.com/apache/iggy/discussions/3313#discussioncomment-18542787
----
This is an automatically sent email for dev@iggy.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@iggy.apache.org
Re: Iggy Connectors: per-message routing for sources
Posted by "zaihuo88 (via GitHub)" <gi...@apache.org>.
GitHub user zaihuo88 added a comment to the discussion: Iggy Connectors: per-message routing for sources
is there any progress?
GitHub link: https://github.com/apache/iggy/discussions/3313#discussioncomment-18542679
----
This is an automatically sent email for dev@iggy.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@iggy.apache.org
Re: Iggy Connectors: per-message routing for sources
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
GitHub user hubcio added a comment to the discussion: Iggy Connectors: per-message routing for sources
hello @zaihuo88,
as far as I know no one is actively working on it right now. we recently just released 0.9.0 and right now core team focus is shifted towards kafka-gateway, performance improvements and post-release cleanup.
however in parallel we've got a few people from community working on connectors.
@mlevkov, @diegomrsantos, @mattp5657, @rohankumardubey - would any of you be willing to take a stab at implementing this?
GitHub link: https://github.com/apache/iggy/discussions/3313#discussioncomment-18542753
----
This is an automatically sent email for dev@iggy.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@iggy.apache.org