I/O: Unify on_error configuration across Kinesis relay and MongoDB CDC
- Dominant language
- Python
- Stars
- 10
- Forks
- 5
- Avg merge
- 7d 5h
- Merged PRs (30d)
- 7
Description
### Background
Write-failure handling varies across I/O adapters. Surfaced in [PR #718](https://github.com/crate/cratedb-toolkit/pull/718) review (Kinesis relay error propagation).
| Adapter | Current behaviour | Config |
|---|---|---|
| Kinesis relay (`io/kinesis/relay.py:173`) | Re-raise `ProgrammingError` / `OperationalError` | Hard-coded since #718 |
| MongoDB CDC (`io/mongodb/cdc.py:41`, `:90`) | `on_error="ignore"\|"raise"` parameter **declared but not wired into the write**. FIXME: *"does not perform any sensible error handling yet"* | Stub only |
| MongoDB bulk (`io/mongodb/api.py:182`) | Catches all, logs, sets `outcome=False`, continues | None (out of scope, see below) |
| DynamoDB reader (`io/dynamodb/adapter.py:29`) | `on_error="log"\|"raise"`, default `"log"` | Present |
| Lambda handler (`io/awslambda/kinesis.py`) | `ON_ERROR` env var | Present |
Two concrete problems:
1. **MongoDB CDC has a dead `on_error` parameter.** Accepts `"ignore"|"raise"` but ignores it. The FIXME is honest about this.
2. **Naming inconsistency.** DynamoDB and Lambda use `"log"`. MongoDB CDC uses `"ignore"`. Kinesis relay has no switch. Callers writing to multiple adapters see three different conventions.
### Proposal
1. **Wire up MongoDB CDC error handling.** Catch `ProgrammingError` / `OperationalError` around the write, honour the `on_error` parameter, resolve the FIXME.
2. **Add `on_error` to the Kinesis relay** with the same signature.
3. **Standardise on `"log"|"raise"`** across both adapters. Rename MongoDB CDC's `"ignore"` value to `"log"` (breaking change, `"ignore"` already didn't do anything meaningful).
4. **Defaults**: `"raise"` for Kinesis relay (preserves post-#718 behaviour, no silent data loss), `"log"` for MongoDB CDC (preserves current default intent, though it previously did nothing).
### Scope
- `cratedb_toolkit/io/kinesis/relay.py` — add `on_error` parameter
- `cratedb_toolkit/io/mongodb/cdc.py` — wire parameter, rename `"ignore"` → `"log"`
- Tests for both adapters covering `"log"` and `"raise"` modes
- Docs: mention `on_error` in the Kinesis and MongoDB CDC sections
### Out of scope (follow-ups)
- **MongoDB bulk**: different surface (`BulkProcessor`), keep for a separate tranche
- **Lambda handler**: already uses `ON_ERROR` env var; aligning env-var naming can be a follow-up
- **Custom exception hierarchy / error taxonomy** (recoverable/transient/permanent): larger design question, separate issue
- **Delta Lake / Iceberg / InfluxDB**: error handling lives in external libraries
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.