From 0b5b1be632faf6f511bacdb5a0e036544097eef5 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 12 Aug 2026 06:13:33 +0000 Subject: [PATCH] Update autogenerated docs for RDI 1.19.1 --- .../reference/api-migration.md | 56 +- .../reference/api-reference.md | 3 +- .../reference/api-reference/openapi.json | 14 +- .../reference/cli/previous-cli-comparison.md | 74 +- .../reference/cli/redis-di-delete-context.md | 2 +- .../reference/cli/redis-di-delete.md | 2 +- .../reference/cli/redis-di-deploy.md | 2 +- .../reference/cli/redis-di-describe-job.md | 2 +- .../cli/redis-di-dump-support-package.md | 2 +- .../reference/cli/redis-di-list-contexts.md | 2 +- .../reference/cli/redis-di-list-jobs.md | 2 +- .../reference/cli/redis-di-reset.md | 2 +- .../reference/cli/redis-di-scaffold.md | 2 +- .../reference/cli/redis-di-set-secret.md | 2 +- .../reference/cli/redis-di-start.md | 2 +- .../reference/cli/redis-di-stop.md | 2 +- .../reference/cli/redis-di.md | 2 +- .../reference/config-yaml-reference.md | 1522 ++++++++--------- 18 files changed, 816 insertions(+), 879 deletions(-) diff --git a/content/integrate/redis-data-integration/reference/api-migration.md b/content/integrate/redis-data-integration/reference/api-migration.md index eed803b967..a253e7c874 100644 --- a/content/integrate/redis-data-integration/reference/api-migration.md +++ b/content/integrate/redis-data-integration/reference/api-migration.md @@ -29,28 +29,28 @@ The API version is part of the URL. Update `/api/v1` requests to use `/api/v2` w ## Endpoint mapping -| API v1 | API v2 | -| --- | --- | -| `GET /api/v1/pipelines` | `GET /api/v2/pipelines` | -| `POST /api/v1/pipelines` | `POST /api/v2/pipelines` | -| `PATCH /api/v1/pipelines` | `PATCH /api/v2/pipelines/{name}` | -| `GET /api/v1/status` | `GET /api/v2/pipelines/{name}/status` | -| `POST /api/v1/pipelines/start` | `POST /api/v2/pipelines/{name}/start` | -| `POST /api/v1/pipelines/stop` | `POST /api/v2/pipelines/{name}/stop` | -| `POST /api/v1/pipelines/reset` | `POST /api/v2/pipelines/{name}/reset` | -| `GET /api/v1/monitoring/statistics` | `GET /api/v2/pipelines/{name}/metric-collections/{collection_name}` | -| `GET /api/v1/pipelines/config/schemas` | `GET /api/v2/schemas/config` | -| `GET /api/v1/pipelines/jobs/functions` | `GET /api/v2/functions` | -| `GET /api/v1/pipelines/jobs/schemas` | `GET /api/v2/schemas/jobs` | -| `PUT /api/v1/pipelines/sources` and source subresources | `PATCH /api/v2/pipelines/{name}` with `sources` in the payload | -| `PUT /api/v1/pipelines/targets` and target subresources | `PATCH /api/v2/pipelines/{name}` with `targets` in the payload | -| `PUT /api/v1/pipelines/processors` and `PUT /api/v1/pipelines/processors/{prop}` | `PATCH /api/v2/pipelines/{name}` with `processors` in the payload | -| Secret provider endpoints | `POST`, `PUT`, or `DELETE /api/v2/pipelines/{name}/secrets[/{key}]` | -| Source metadata, schemas, databases, tables, and columns endpoints | `GET /api/v2/pipelines/{name}/source-schemas/{source_name}` with the appropriate filters | -| `POST /api/v1/pipelines/sources/dry-run` | `POST /api/v2/pipelines?dry_run=true` | -| `POST /api/v1/pipelines/targets/dry-run` | `POST /api/v2/pipelines?dry_run=true` | -| `POST /api/v1/pipelines/undeploy` | `DELETE /api/v2/pipelines/{name}` | -| `POST /api/v1/trace/start` | `POST /api/v2/pipelines/{name}/traces` | +| API v1 | API v2 | +| -------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | +| `GET /api/v1/pipelines` | `GET /api/v2/pipelines` | +| `POST /api/v1/pipelines` | `POST /api/v2/pipelines` | +| `PATCH /api/v1/pipelines` | `PATCH /api/v2/pipelines/{name}` | +| `GET /api/v1/status` | `GET /api/v2/pipelines/{name}/status` | +| `POST /api/v1/pipelines/start` | `POST /api/v2/pipelines/{name}/start` | +| `POST /api/v1/pipelines/stop` | `POST /api/v2/pipelines/{name}/stop` | +| `POST /api/v1/pipelines/reset` | `POST /api/v2/pipelines/{name}/reset` | +| `GET /api/v1/monitoring/statistics` | `GET /api/v2/pipelines/{name}/metric-collections/{collection_name}` | +| `GET /api/v1/pipelines/config/schemas` | `GET /api/v2/schemas/config` | +| `GET /api/v1/pipelines/jobs/functions` | `GET /api/v2/functions` | +| `GET /api/v1/pipelines/jobs/schemas` | `GET /api/v2/schemas/jobs` | +| `PUT /api/v1/pipelines/sources` and source subresources | `PATCH /api/v2/pipelines/{name}` with `sources` in the payload | +| `PUT /api/v1/pipelines/targets` and target subresources | `PATCH /api/v2/pipelines/{name}` with `targets` in the payload | +| `PUT /api/v1/pipelines/processors` and `PUT /api/v1/pipelines/processors/{prop}` | `PATCH /api/v2/pipelines/{name}` with `processors` in the payload | +| Secret provider endpoints | `POST`, `PUT`, or `DELETE /api/v2/pipelines/{name}/secrets[/{key}]` | +| Source metadata, schemas, databases, tables, and columns endpoints | `GET /api/v2/pipelines/{name}/source-schemas/{source_name}` with the appropriate filters | +| `POST /api/v1/pipelines/sources/dry-run` | `POST /api/v2/pipelines?dry_run=true` | +| `POST /api/v1/pipelines/targets/dry-run` | `POST /api/v2/pipelines?dry_run=true` | +| `POST /api/v1/pipelines/undeploy` | `DELETE /api/v2/pipelines/{name}` | +| `POST /api/v1/trace/start` | `POST /api/v2/pipelines/{name}/traces` | API v2 also adds endpoints for DLQ inspection, target flushing, metric collections, and API information. See the [API reference]({{< relref "/integrate/redis-data-integration/reference/api-reference" >}}) for the complete list. @@ -58,13 +58,13 @@ API v2 also adds endpoints for DLQ inspection, target flushing, metric collectio Most v1 endpoints have a v2 replacement. However, the following endpoints remain available under v1 because the current API v2 design does not define a corresponding endpoint: -| v1 endpoint | Notes | -| --- | --- | -| `GET /api/v1/me` | Returns the authenticated user. | -| `GET /api/v1/pipelines/strategies` | Returns pipeline strategies. | -| `POST /api/v1/login` | API v2 continues to use this endpoint for authentication. | +| v1 endpoint | Notes | +| --------------------------------------------------------- | ----------------------------------------------------------------------------- | +| `GET /api/v1/me` | Returns the authenticated user. | +| `GET /api/v1/pipelines/strategies` | Returns pipeline strategies. | +| `POST /api/v1/login` | API v2 continues to use this endpoint for authentication. | | `GET /api/v1/pipelines/config/templates/ingest/{db_type}` | Used by the CLI to scaffold pipeline configuration and will remain available. | -| `GET /api/v1/pipelines/jobs/templates/ingest` | Used to scaffold jobs and will remain available. | +| `GET /api/v1/pipelines/jobs/templates/ingest` | Used to scaffold jobs and will remain available. | ## Replace action polling with pipeline-status polling diff --git a/content/integrate/redis-data-integration/reference/api-reference.md b/content/integrate/redis-data-integration/reference/api-reference.md index f8d00f7504..50d6a8f0f5 100644 --- a/content/integrate/redis-data-integration/reference/api-reference.md +++ b/content/integrate/redis-data-integration/reference/api-reference.md @@ -4,7 +4,7 @@ Title: Redis Data Integration API layout: apireference type: page aliases: -- /integrate/redis-data-integration/reference/api/api-reference/ + - /integrate/redis-data-integration/reference/api/api-reference/ --- - diff --git a/content/integrate/redis-data-integration/reference/api-reference/openapi.json b/content/integrate/redis-data-integration/reference/api-reference/openapi.json index 8c3c37d013..dada8fcad7 100644 --- a/content/integrate/redis-data-integration/reference/api-reference/openapi.json +++ b/content/integrate/redis-data-integration/reference/api-reference/openapi.json @@ -2,8 +2,8 @@ "openapi": "3.1.0", "info": { "title": "Redis Data Integration API", - "description": "> **NOTE:** RDI API v1 is deprecated as of RDI 1.19.0. Use RDI API v2 for new integrations and migrate existing clients. API v1 will not be extended with new RDI features and may be removed in a future RDI version. See the [RDI API migration guide](/integrate/redis-data-integration/reference/api-migration/).\n\nAPI for Redis Data Integration services", - "version": "1.19.0" + "description": "API for Redis Data Integration services", + "version": "1.19.1" }, "paths": { "/": { @@ -287,9 +287,6 @@ } }, "security": [ - { - "JWTBearer": [] - }, { "JWTBearer": [] } @@ -7553,6 +7550,13 @@ "type": { "type": "string", "title": "Error Type" + }, + "input": { + "title": "Input" + }, + "ctx": { + "type": "object", + "title": "Context" } }, "type": "object", diff --git a/content/integrate/redis-data-integration/reference/cli/previous-cli-comparison.md b/content/integrate/redis-data-integration/reference/cli/previous-cli-comparison.md index 1bbaa51cf7..cc53d4dcc1 100644 --- a/content/integrate/redis-data-integration/reference/cli/previous-cli-comparison.md +++ b/content/integrate/redis-data-integration/reference/cli/previous-cli-comparison.md @@ -18,16 +18,16 @@ the RDI REST API. ## Compare the CLIs -| Area | Previous CLI (before RDI 1.19.0) | Current CLI (RDI 1.19.0 and later) | -| :--- | :------------------------------- | :---------------------------------- | -| Implementation | Python application | Self-contained Go binary | -| Installation types | VM installations | VM, Kubernetes, and Redis Cloud installations | -| Connection | RDI database and Kubernetes API | RDI REST API | -| Pipeline scope | The `default` pipeline | Multiple named pipelines; defaults to `default` | -| Connection options | `--rdi-host`, `--rdi-port`, `--rdi-user`, `--rdi-password`, and RDI database TLS options | `--api-url`, `--user`, `--password`, and API TLS options; Redis Cloud also supports `--account-key` and `--user-key` | -| Contexts | A list of RDI database connections in `~/.redis-di`, with an `is_active` field on each entry | A map of API connections in `~/.redis-di`, with one `current-context` | -| Output | Human-readable tables | Compact tables and sectioned descriptions; `list` and `get` commands also support JSON and YAML | -| Secrets | `set-secret` on VM installations; `rdi-secret.sh` for Kubernetes | Create, inspect, update, and delete operations through `redis-di` on every installation type | +| Area | Previous CLI (before RDI 1.19.0) | Current CLI (RDI 1.19.0 and later) | +| :----------------- | :------------------------------------------------------------------------------------------- | :------------------------------------------------------------------------------------------------------------------- | +| Implementation | Python application | Self-contained Go binary | +| Installation types | VM installations | VM, Kubernetes, and Redis Cloud installations | +| Connection | RDI database and Kubernetes API | RDI REST API | +| Pipeline scope | The `default` pipeline | Multiple named pipelines; defaults to `default` | +| Connection options | `--rdi-host`, `--rdi-port`, `--rdi-user`, `--rdi-password`, and RDI database TLS options | `--api-url`, `--user`, `--password`, and API TLS options; Redis Cloud also supports `--account-key` and `--user-key` | +| Contexts | A list of RDI database connections in `~/.redis-di`, with an `is_active` field on each entry | A map of API connections in `~/.redis-di`, with one `current-context` | +| Output | Human-readable tables | Compact tables and sectioned descriptions; `list` and `get` commands also support JSON and YAML | +| Secrets | `set-secret` on VM installations; `rdi-secret.sh` for Kubernetes | Create, inspect, update, and delete operations through `redis-di` on every installation type | ## Update the connection options @@ -54,14 +54,14 @@ redis-di describe \ The current CLI does not accept the previous RDI database connection options on API-based commands. Update scripts and environment variables as follows: -| Previous setting | Current setting | -| :--------------- | :-------------- | -| `--rdi-host` and `--rdi-port` | `--api-url` or `RDI_API_URL` | -| `--rdi-user` or `RDI_REDIS_USERNAME` | `--user` or `RDI_USER` | -| `--rdi-password` or `RDI_REDIS_PASSWORD` | `--password` or `RDI_PASSWORD` | -| `--rdi-cacert` | `--cacert` or `RDI_CACERT` | +| Previous setting | Current setting | +| :---------------------------------------------- | :-------------------------------------------------------------------------- | +| `--rdi-host` and `--rdi-port` | `--api-url` or `RDI_API_URL` | +| `--rdi-user` or `RDI_REDIS_USERNAME` | `--user` or `RDI_USER` | +| `--rdi-password` or `RDI_REDIS_PASSWORD` | `--password` or `RDI_PASSWORD` | +| `--rdi-cacert` | `--cacert` or `RDI_CACERT` | | `--rdi-key`, `--rdi-cert`, `--rdi-key-password` | No equivalent; the CLI authenticates to the API instead of the RDI database | -| `--rdi-namespace` or `RDI_NAMESPACE` | No equivalent; pipeline operations go through the API | +| `--rdi-namespace` or `RDI_NAMESPACE` | No equivalent; pipeline operations go through the API | For Redis Cloud, use `--account-key` with `--user-key` instead of `--user` with `--password`. See the [CLI reference overview]({{< relref "/integrate/redis-data-integration/reference/cli#connecting-to-the-api" >}}) @@ -89,12 +89,12 @@ redis-di use-context The context commands also changed meaning: -| Task | Previous CLI | Current CLI | -| :--- | :----------- | :---------- | -| Create a context | `redis-di add-context [connection options]` | `redis-di set-context [connection options]` | -| Select the active context | `redis-di set-context ` | `redis-di use-context ` | -| Update a context | Recreate the context | `redis-di set-context [options to update]` | -| Remove all contexts | `redis-di delete-all-contexts` | Delete contexts individually with `redis-di delete-context ` | +| Task | Previous CLI | Current CLI | +| :------------------------ | :------------------------------------------------- | :----------------------------------------------------------------- | +| Create a context | `redis-di add-context [connection options]` | `redis-di set-context [connection options]` | +| Select the active context | `redis-di set-context ` | `redis-di use-context ` | +| Update a context | Recreate the context | `redis-di set-context [options to update]` | +| Remove all contexts | `redis-di delete-all-contexts` | Delete contexts individually with `redis-di delete-context ` | Passwords and Redis Cloud user keys are not saved in current contexts. Supply them with an environment variable, a command option, or the interactive prompt. @@ -113,20 +113,20 @@ redis-di reset [pipeline] Other common tasks changed as follows: -| Task | Previous CLI | Current CLI | -| :--- | :----------- | :---------- | -| Inspect pipeline status | `redis-di status` | `redis-di describe [pipeline]`; `status` remains an alias | -| Continuously refresh status | `redis-di status --live` | `watch -n 1 redis-di describe [pipeline]` | -| List pipelines | Not available | `redis-di list` | -| Get a pipeline | Not available | `redis-di get [pipeline]` | -| Delete a pipeline | Not available | `redis-di delete [pipeline]` | -| Inspect rejected records | `redis-di get-rejected [options]` | `redis-di list-dlqs`, then `redis-di list-dlq-records `; `get-rejected` remains an alias | -| Inspect jobs | `redis-di list-jobs` and `redis-di describe-job ` | The same commands, with `--pipeline ` for a non-default pipeline | -| Manage secrets | `redis-di set-secret ` or `rdi-secret.sh` | `list-secrets`, `get-secret`, `describe-secret`, `set-secret`, and `delete-secret` | -| Install or upgrade RDI on a VM | `redis-di install` or `redis-di upgrade` | Run `install.sh` or `upgrade.sh` from the VM installation package | -| Install or upgrade RDI on Kubernetes | Not available | Use the RDI Helm chart | -| Trace pipeline records | `redis-di trace` | Removed | -| Create the RDI database | `redis-di create` | Removed; use the installation workflow | +| Task | Previous CLI | Current CLI | +| :----------------------------------- | :----------------------------------------------------- | :-------------------------------------------------------------------------------------------- | +| Inspect pipeline status | `redis-di status` | `redis-di describe [pipeline]`; `status` remains an alias | +| Continuously refresh status | `redis-di status --live` | `watch -n 1 redis-di describe [pipeline]` | +| List pipelines | Not available | `redis-di list` | +| Get a pipeline | Not available | `redis-di get [pipeline]` | +| Delete a pipeline | Not available | `redis-di delete [pipeline]` | +| Inspect rejected records | `redis-di get-rejected [options]` | `redis-di list-dlqs`, then `redis-di list-dlq-records `; `get-rejected` remains an alias | +| Inspect jobs | `redis-di list-jobs` and `redis-di describe-job ` | The same commands, with `--pipeline ` for a non-default pipeline | +| Manage secrets | `redis-di set-secret ` or `rdi-secret.sh` | `list-secrets`, `get-secret`, `describe-secret`, `set-secret`, and `delete-secret` | +| Install or upgrade RDI on a VM | `redis-di install` or `redis-di upgrade` | Run `install.sh` or `upgrade.sh` from the VM installation package | +| Install or upgrade RDI on Kubernetes | Not available | Use the RDI Helm chart | +| Trace pipeline records | `redis-di trace` | Removed | +| Create the RDI database | `redis-di create` | Removed; use the installation workflow | On VM installations, `configure-rdi` and `dump-support-package` remain available through `redis-di`. They are not available with a standalone CLI or on Kubernetes, Redis Cloud, diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-delete-context.md b/content/integrate/redis-data-integration/reference/cli/redis-di-delete-context.md index f627c4f6d0..667711f61c 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-delete-context.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-delete-context.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-delete-context/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-delete-context/ --- Deletes a context from the `~/.redis-di` context file. Because this is destructive, the command asks diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-delete.md b/content/integrate/redis-data-integration/reference/cli/redis-di-delete.md index 1ea013d9ef..131ed95c99 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-delete.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-delete.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-delete/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-delete/ --- Deletes a pipeline. Because this is destructive, the command asks for confirmation unless you pass diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-deploy.md b/content/integrate/redis-data-integration/reference/cli/redis-di-deploy.md index 9e66e27268..8cce7ba61f 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-deploy.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-deploy.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-deploy/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-deploy/ --- Deploys a pipeline, creating it or updating it from the configuration in the `--dir` directory. The diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-describe-job.md b/content/integrate/redis-data-integration/reference/cli/redis-di-describe-job.md index 237d52a924..03894d1a69 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-describe-job.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-describe-job.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-describe-job/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-describe-job/ --- Describes a single job of a pipeline, printing its source properties followed by tables that diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-dump-support-package.md b/content/integrate/redis-data-integration/reference/cli/redis-di-dump-support-package.md index c04e47ba17..6289e05a6d 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-dump-support-package.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-dump-support-package.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-dump-support-package/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-dump-support-package/ --- Dumps a comprehensive set of RDI forensics data that you can send to Redis support (see diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-list-contexts.md b/content/integrate/redis-data-integration/reference/cli/redis-di-list-contexts.md index a9b6abe46f..af6d1601ac 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-list-contexts.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-list-contexts.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-list-contexts/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-list-contexts/ --- Lists all contexts from the `~/.redis-di` context file and indicates which one is active. See the diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-list-jobs.md b/content/integrate/redis-data-integration/reference/cli/redis-di-list-jobs.md index 622054a59e..529597acdf 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-list-jobs.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-list-jobs.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-list-jobs/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-list-jobs/ --- Lists the jobs of a pipeline, one row per job with its source, its transformation and output counts, diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-reset.md b/content/integrate/redis-data-integration/reference/cli/redis-di-reset.md index 14d627c0a4..5b35a54587 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-reset.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-reset.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-reset/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-reset/ --- Resets a pipeline into initial full-sync mode, so it reloads a snapshot of the source data before diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-scaffold.md b/content/integrate/redis-data-integration/reference/cli/redis-di-scaffold.md index 927acb678d..7a604b64f4 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-scaffold.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-scaffold.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-scaffold/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-scaffold/ --- Generates a starter pipeline configuration for the given source database type. With `--dir`, the diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-set-secret.md b/content/integrate/redis-data-integration/reference/cli/redis-di-set-secret.md index f82e5978db..7afa1b67fe 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-set-secret.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-set-secret.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-set-secret/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-set-secret/ --- Creates or updates a secret of a pipeline. Secrets hold the credentials and certificates that the diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-start.md b/content/integrate/redis-data-integration/reference/cli/redis-di-start.md index 23a5b9b124..66fa29714e 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-start.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-start.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-start/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-start/ --- Starts a pipeline. By default, the command waits for the pipeline to reach the `started` state before diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di-stop.md b/content/integrate/redis-data-integration/reference/cli/redis-di-stop.md index 17b8dd5877..b0cb9b079c 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di-stop.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di-stop.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di-stop/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di-stop/ --- Stops a pipeline. By default, the command waits for the pipeline to reach the `stopped` state before diff --git a/content/integrate/redis-data-integration/reference/cli/redis-di.md b/content/integrate/redis-data-integration/reference/cli/redis-di.md index 608fbd8953..9de759dc64 100644 --- a/content/integrate/redis-data-integration/reference/cli/redis-di.md +++ b/content/integrate/redis-data-integration/reference/cli/redis-di.md @@ -6,7 +6,7 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/cli/redis-di/ + - /integrate/redis-data-integration/ingest/reference/cli/redis-di/ --- `redis-di` is the command line tool that manages Redis Data Integration (RDI). It is a thin client diff --git a/content/integrate/redis-data-integration/reference/config-yaml-reference.md b/content/integrate/redis-data-integration/reference/config-yaml-reference.md index a7a94416b1..87244be1f0 100644 --- a/content/integrate/redis-data-integration/reference/config-yaml-reference.md +++ b/content/integrate/redis-data-integration/reference/config-yaml-reference.md @@ -6,355 +6,476 @@ weight: 10 alwaysopen: false categories: ["redis-di"] aliases: -- /integrate/redis-data-integration/ingest/reference/config-yaml-reference/ --- Configuration file for Redis Data Integration (RDI) source collectors and target connections. - **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|[**sources**](#sources)
(Source collectors)|`object`|Source collectors that capture changes from upstream databases. Each key is a unique source identifier; the value configures one collector.
|| -|[**targets**](#targets)
(Target connections)|`object`|Target Redis databases where processed records are written. Each key is a target identifier; the value configures the connection.
|| -|[**processors**](#processors)
(Data processing configuration)|`object`, `null`|Settings that control how data is processed, including batch sizes, error handling, and performance tuning.
|| -|[**secret\-providers**](#secret-providers)
(Secret providers)|`object`|External secret providers used to resolve `${...}` references in the configuration.
|| -|[**metadata**](#metadata)
(Pipeline metadata)|`object`|Optional metadata describing this pipeline, such as a display name and description.
|| - -**Additional Properties:** not allowed - -## sources: Source collectors +| Name | Type | Description | Required | +| ----------------------------------------------------------------- | ---------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | -------- | +| [**sources**](#sources)
(Source collectors) | `object` | Source collectors that capture changes from upstream databases. Each key is a unique source identifier; the value configures one collector.
| | +| [**targets**](#targets)
(Target connections) | `object` | Target Redis databases where processed records are written. Each key is a target identifier; the value configures the connection.
| | +| [**processors**](#processors)
(Data processing configuration) | `object`, `null` | Settings that control how data is processed, including batch sizes, error handling, and performance tuning.
| | +| [**secret\-providers**](#secret-providers)
(Secret providers) | `object` | External secret providers used to resolve `${...}` references in the configuration.
| | +| [**metadata**](#metadata)
(Pipeline metadata) | `object` | Optional metadata describing this pipeline, such as a display name and description.
| | -Source collectors that capture changes from upstream databases. Each key is a unique source identifier; the value configures one collector. +**Additional Properties:** not allowed  +  + +Optional metadata describing this pipeline, such as a display name and description. -**Properties** (key: `.*`) +**Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|[**connection**](#sourcesconnection)
(Source database connection)|`object`|Connection configuration for a non-Redis source database. The exact set of properties depends on the database type.
|yes| -|**name**
(Source name)|`string`|Human-readable name for the source collector. Maximum 100 characters.
Maximal Length: `100`
|no| -|**type**
(Collector type)|`string`|Type of the source collector. Use `cdc` (default) for change data capture using [Debezium](https://debezium.io/). Use `flink` for Spanner change streams using the Apache Flink-based collector. Use `riotx` for Snowflake CDC using [RIOT-X](https://redis.github.io/riotx/).
Default: `"cdc"`
Enum: `"cdc"`, `"flink"`, `"riotx"`
|yes| -|**active**
(Collector enabled)|`boolean`|When `true`, the collector runs; when `false`, the collector is disabled and produces no events.
Default: `true`
|no| -|[**logging**](#sourceslogging)
(Logging configuration)|`object`|Logging settings for this source collector.
|no| -|[**tables**](#sourcestables)
(Tables to capture)|`object`|Tables to capture from the source database, keyed by table name. The value configures column selection and key handling for that table.
|no| -|[**schemas**](#sourcesschemas)
(Schema names)|`string[]`|Schema names to capture from the source database. Maps to the underlying connector's `schema.include.list`.
|no| -|[**databases**](#sourcesdatabases)
(Database names)|`string[]`|Database names to capture from the source database. Maps to the underlying connector's `database.include.list`. Applies only to MySQL, MariaDB, and MongoDB connections.
|no| -|[**advanced**](#sourcesadvanced)
(Advanced configuration)|`object`|Advanced configuration that overrides the underlying engine's defaults. Only required for non-standard tuning.
|no| +| Name | Type | Description | Required | +| --------------------------------------------- | ---------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **name**
(Pipeline name) | `string` | Human-readable name for the pipeline. Maximum 100 characters.
Maximal Length: `100`
| | +| **description**
(Pipeline description) | `string` | Free-form description of what the pipeline does. Maximum 500 characters.
Maximal Length: `500`
| | +| **revision**
(Pipeline revision) | `integer` | Pipeline revision number. Must be a non-negative integer.
Minimum: `0`
| | +| [**tags**](#metadatatags)
(Pipeline tags) | `string[]` | Array of pipeline tags. Each tag must be a string of up to 50 characters, containing only alphanumeric characters, dots, dashes, or underscores, and must start and end with an alphanumeric character. Tags must be unique.
| | +**Additional Properties:** not allowed  +  + - -### sources\.connection: Source database connection +Array of pipeline tags. Each tag must be a string of up to 50 characters, containing only alphanumeric characters, dots, dashes, or underscores, and must start and end with an alphanumeric character. Tags must be unique. -Connection configuration for a non-Redis source database. The exact set of properties depends on the database type. +**Unique Items:** yes  +  + +Settings that control how data is processed, including batch sizes, error handling, and performance tuning. **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|[**SQL database**](#sourcesconnectionsqldatabase)
(SQL database)|`object`|Connection configuration for a supported SQL database.
|| -|[**MongoDB**](#sourcesconnectionmongodb)|`object`|Connection configuration for a MongoDB database.
|yes| -|[**Spanner**](#sourcesconnectionspanner)|`object`|Connection configuration for a Google Cloud Spanner database.
|yes| -|[**Snowflake**](#sourcesconnectionsnowflake)|`object`|Connection configuration for a Snowflake database.
|yes| +| Name | Type | Description | Required | +| ------------------------------------------------------------------------------ | ------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -------- | +| **type**
(Processor type) | `string` | Processor implementation to run. `classic` runs the classic processor; `flink` runs the Apache Flink-based processor.
Default: `"classic"`
Enum: `"classic"`, `"flink"`
| | +| **read\_batch\_size** | `integer`, `string` | Maximum number of records read from the source streams in a single batch.
Default: `2000`
Pattern: `^\${.*}$`
Minimum: `1`
| | +| **read\_batch\_timeout\_ms**
(Read batch timeout) | `integer` | Maximum time in milliseconds to wait for a batch to fill before processing it.
Default: `100`
Minimum: `1`
| | +| **duration**
(Batch duration limit) | `integer`, `string` | (DEPRECATED)
This property has no effect; use `read_batch_timeout_ms` instead.
Default: `100`
Pattern: `^\${.*}$`
Minimum: `1`
| | +| **write\_batch\_size** | `integer`, `string` | Maximum number of records written to the target Redis database in a single batch.
Default: `200`
Pattern: `^\${.*}$`
Minimum: `1`
| | +| **enable\_async\_processing** | `boolean` | When `true`, the processor handles batches asynchronously to improve throughput. **Classic processor only.**
Default: `true`
| | +| **batch\_queue\_size** | `integer` | Maximum number of batches queued for processing. **Classic processor only.**
Default: `3`
Minimum: `1`
| | +| **ack\_queue\_size** | `integer` | Maximum number of batches queued for asynchronous acknowledgement. **Classic processor only.**
Default: `10`
Minimum: `1`
| | +| **dedup**
(Enable deduplication) | `boolean` | When `true`, the processor deduplicates incoming records. **Classic processor only.**
Default: `false`
| | +| **dedup\_max\_size**
(Deduplication set size) | `integer` | Maximum number of entries kept in the deduplication set. **Classic processor only.**
Default: `1024`
Minimum: `1`
| | +| **dedup\_strategy**
(Deduplication strategy) | `string` | (DEPRECATED)
This property has no effect — the only supported strategy is `ignore`. Remove it from the configuration. **Classic processor only.**
Default: `"ignore"`
Enum: `"reject"`, `"ignore"`
| | +| **error\_handling**
(Error handling strategy) | `string` | Strategy for handling failed records. `ignore` silently drops them; `dlq` writes them to the dead-letter queue.
Default: `"dlq"`
| | +| **dlq\_max\_messages**
(DLQ message limit) | `integer`, `string` | Maximum number of messages stored per dead-letter queue stream.
Default: `1000`
Pattern: `^\${.*}$`
Minimum: `1`
| | +| **target\_data\_type**
(Target Redis data type) | `string` | Data type used to store target records in Redis. `hash` writes a Redis Hash; `json` writes a RedisJSON document and requires the RedisJSON module.
Default: `"hash"`
| | +| **json\_update\_strategy** | `string` | Strategy for updating existing JSON documents in Redis. `replace` overwrites the entire document; `merge` merges incoming fields into it.
Default: `"replace"`
| | +| **use\_native\_json\_merge**
(Use native JSON merge from RedisJSON module) | `boolean` | Controls whether JSON merge operations use the native `JSON.MERGE` command (when `true`) or Lua scripts (when `false`). Introduced in RDI 1.15.0. The native command provides 2x performance improvement but handles null values differently:

**Previous behavior (Lua merge)**: When merging `{"field1": "value1", "field2": "value2"}` with `{"field2": null, "field3": "value3"}`, the result was `{"field1": "value1", "field2": null, "field3": "value3"}` (null value is preserved).

**New behavior (JSON.MERGE)**: The same merge produces `{"field1": "value1", "field3": "value3"}` (null value removes the field, following [RFC 7396](https://datatracker.ietf.org/doc/html/rfc7396)).

**Note**: The native `JSON.MERGE` command requires RedisJSON 2.6.0 or higher. If the target database has an older version of RedisJSON, RDI automatically falls back to Lua-based merge operations regardless of this setting.

**Impact**: If your application logic distinguishes between a field with a `null` value and a missing field, you may need to adjust your data handling. This follows the JSON Merge Patch RFC standard but differs from the previous Lua implementation. Set to `false` to revert to the previous Lua-based merge behavior if needed.

The Flink processor always uses the native `JSON.MERGE` command when the target database supports it. **Classic processor only.**
Default: `true`
| | +| **initial\_sync\_processes** | `integer`, `string` | Number of parallel processes used to perform the initial data synchronization. For the Flink processor, parallelism is controlled by Flink properties instead. **Classic processor only.**
Default: `4`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `32`
| | +| **idle\_sleep\_time\_ms**
(Idle sleep interval) | `integer`, `string` | Time in milliseconds to sleep between processing batches when idle. **Classic processor only.**
Default: `200`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `999999`
| | +| **idle\_streams\_check\_interval\_ms**
(Idle streams check interval) | `integer`, `string` | Time in milliseconds between checks for new streams when the processor is idle. For the Flink processor, use `processors.advanced.source.discovery.interval.ms` instead to configure a single discovery interval regardless of load. **Classic processor only.**
Default: `1000`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `999999`
| | +| **busy\_streams\_check\_interval\_ms**
(Busy streams check interval) | `integer`, `string` | Time in milliseconds between checks for new streams when the processor is busy. For the Flink processor, use `processors.advanced.source.discovery.interval.ms` instead to configure a single discovery interval regardless of load. **Classic processor only.**
Default: `5000`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `999999`
| | +| **retry\_max\_attempts**
(Maximum retry attempts) | `integer`, `string` | Maximum number of attempts for a failed write to the target Redis database before giving up.
Default: `5`
Pattern: `^\${.*}$`
Minimum: `1`
| | +| **retry\_initial\_delay\_ms**
(Initial retry delay) | `integer`, `string` | Initial delay in milliseconds before the first retry of a failed write.
Default: `1000`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `999999`
| | +| **retry\_max\_delay\_ms**
(Maximum retry delay) | `integer`, `string` | Maximum delay in milliseconds between retry attempts.
Default: `10000`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `999999`
| | +| **wait\_enabled**
(Enable replica wait) | `boolean` | When `true`, RDI verifies that each write has been replicated to the target database's replica shards before acknowledging it. Enable this only when target database replication is enabled and a healthy replica is available. For the Flink processor, `processors.advanced.target.wait.enabled` takes priority.
Default: `false`
| | +| **wait\_timeout**
(Replica wait timeout) | `integer`, `string` | Maximum time in milliseconds to wait for replica write verification on the target database.
Default: `1000`
Pattern: `^\${.*}$`
Minimum: `1`
| | +| **retry\_on\_replica\_failure** | `boolean` | When `true`, RDI keeps retrying a write until replica replication is confirmed; when `false`, it gives up after the first failure.
Default: `true`
| | +| **on\_failed\_retry\_interval**
(Retry interval on failure) | `integer`, `string` | (DEPRECATED)
This property has no effect; remove it from the configuration.
Default: `5`
Pattern: `^\${.*}$`
Minimum: `1`
| | +| [**logging**](#processorslogging)
(Logging configuration) | `object` | Logging settings for the processor. **Flink processor only.**
| | +| [**advanced**](#processorsadvanced)
(Advanced configuration) | `object` | Advanced configuration for fine-tuning the processor. **All properties under `advanced` apply to the Flink processor only and are silently ignored by the classic processor.**
| | + +**Additional Properties:** not allowed  +  + + +Advanced configuration for fine-tuning the processor. **All properties under `advanced` apply to the Flink processor only and are silently ignored by the classic processor.** + +**Properties** +| Name | Type | Description | Required | +| ------------------------------------------------------------------------------- | -------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| [**source**](#processorsadvancedsource)
(Advanced source settings) | `object` | Advanced configuration properties for the source Redis client and streams reader. **Flink processor only.**
| | +| [**target**](#processorsadvancedtarget)
(Advanced target settings) | `object` | Advanced configuration properties for the target Redis client and sink. **Flink processor only.**
| | +| [**dlq**](#processorsadvanceddlq)
(Advanced DLQ settings) | `object` | Advanced configuration properties for the DLQ Redis client and sink. **Flink processor only.**
| | +| [**processor**](#processorsadvancedprocessor)
(Advanced processor settings) | `object` | Advanced configuration properties for the processor. **Flink processor only.**
| | +| [**flink**](#processorsadvancedflink)
(Advanced Flink settings) | `object` | Advanced configuration properties forwarded to the underlying Flink runtime. Any property listed in the [Flink configuration documentation](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/config/) can be set here and will override the RDI default. **Flink processor only.**
| | +| [**resources**](#processorsadvancedresources)
(Advanced resource settings) | `object` | Compute resources allocated to the Flink job, such as the number of task manager pods. **Flink processor only.**
| | + +**Additional Properties:** not allowed  +**Minimal Properties:** 1  **Example** ```yaml -SQL database: - hr: - type: postgresql - host: localhost - port: 5432 - database: postgres - user: postgres - password: postgres -MongoDB: - mongodb-source: - type: mongodb - connection_string: mongodb://localhost:27017/?replicaSet=rs0 - user: debezium - password: dbz - database: db1,db2 -Spanner: - spanner-source: - type: spanner - project_id: example-12345 - instance_id: example - database_id: example - change_streams: - change_stream_all: - retention_period_hours: 24 -Snowflake: - snowflake: - type: snowflake - url: jdbc:snowflake://myaccount.snowflakecomputing.com/ - user: myuser - password: mypassword - database: MYDB - warehouse: COMPUTE_WH - +source: + stream.name.pattern: data:* + discovery.interval.ms: 1000 + batch.size: 2000 + batch.timeout.ms: 100 + connection.timeout.ms: 2000 + socket.timeout.ms: 2000 + retry.max.attempts: 5 + retry.initial.delay.ms: 100 + retry.max.delay.ms: 3000 + retry.backoff.multiplier: 2 +target: + batch.size: 200 + flush.interval.ms: 100 + connection.timeout.ms: 2000 + socket.timeout.ms: 2000 + retry.max.attempts: 5 + retry.initial.delay.ms: 1000 + retry.max.delay.ms: 10000 + retry.backoff.multiplier: 2 + wait.enabled: false + wait.write.timeout.ms: 1000 + wait.retry.enabled: true + wait.retry.delay.ms: 1000 +dlq: + max.len: 1000 + batch.size: 100 + flush.interval.ms: 100 + connection.timeout.ms: 2000 + socket.timeout.ms: 2000 + retry.max.attempts: 1 + retry.initial.delay.ms: 100 + retry.max.delay.ms: 3000 + retry.backoff.multiplier: 2 + wait.enabled: false + wait.write.timeout.ms: 1000 + wait.retry.enabled: false + wait.retry.delay.ms: 1000 +processor: + default.data.type: hash + default.json.update.strategy: replace + dlq.enabled: true +flink: + taskmanager.numberOfTaskSlots: 1 + taskmanager.memory.process.size: 2048m +resources: + taskManager: {} ``` - -#### sources\.connection\.SQL database: SQL database - -Connection configuration for a supported SQL database. +  + +Advanced configuration properties for the DLQ Redis client and sink. **Flink processor only.** **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**type**
(Database type)|`string`|SQL database engine.
Enum: `"mariadb"`, `"mysql"`, `"oracle"`, `"postgresql"`, `"sqlserver"`
|| -|**host**
(Database host)|`string`|Hostname or IP address of the SQL database server.
|| -|**port**
(Database port)|`integer`|Network port on which the SQL database server is listening.
Minimum: `1`
Maximum: `65535`
|| -|**database**
(Database name)|`string`|Name of the database to connect to.
|| -|**user**
(Database user)|`string`|Username for authentication to the SQL database.
|| -|**password**
(Database password)|`string`|Password for authentication to the SQL database.
|| +| Name | Type | Description | Required | +| ----------------------------------------------------------------- | --------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -------- | +| **max\.len**
(DLQ sink max length) | `integer` | Maximum number of messages stored per dead letter queue stream. Alias for `processors.dlq_max_messages`; takes priority when both are set.
Default: `1000`
Minimum: `1`
| | +| **batch\.size**
(DLQ sink batch size) | `integer` | Maximum number of records the DLQ sink writes in a single batch.
Default: `100`
Minimum: `1`
| | +| **flush\.interval\.ms**
(DLQ sink flush interval) | `integer` | Maximum time in milliseconds the DLQ sink waits to fill a batch before flushing it to Redis.
Default: `100`
Minimum: `1`
| | +| **connection\.timeout\.ms**
(DLQ connection timeout) | `integer` | Connection timeout in milliseconds for the DLQ Redis client.
Default: `2000`
Minimum: `1`
| | +| **socket\.timeout\.ms**
(DLQ socket timeout) | `integer` | Socket read/write timeout in milliseconds for the DLQ Redis client.
Default: `2000`
Minimum: `1`
| | +| **retry\.max\.attempts**
(DLQ retry max attempts) | `integer` | Maximum number of retry attempts for failed DLQ Redis operations.
Default: `1`
Minimum: `1`
| | +| **retry\.initial\.delay\.ms**
(DLQ retry initial delay) | `integer` | Initial delay in milliseconds before the first retry of a failed DLQ Redis operation.
Default: `100`
Minimum: `1`
| | +| **retry\.max\.delay\.ms**
(DLQ retry max delay) | `integer` | Maximum delay in milliseconds between retry attempts for DLQ Redis operations.
Default: `3000`
Minimum: `1`
| | +| **retry\.backoff\.multiplier**
(DLQ retry backoff multiplier) | `number` | Exponential backoff multiplier between retry attempts for DLQ Redis operations.
Default: `2`
Minimum: `1`
| | +| **wait\.enabled**
(DLQ replica wait enabled) | `boolean` | When `true`, RDI verifies that each DLQ write has been replicated to the DLQ database's replica shards before acknowledging it.
Default: `false`
| | +| **wait\.write\.timeout\.ms**
(DLQ replica wait timeout) | `integer` | Maximum time in milliseconds to wait for DLQ replica write verification.
Default: `1000`
Minimum: `1`
| | +| **wait\.retry\.enabled**
(DLQ replica wait retry enabled) | `boolean` | When `true`, RDI keeps retrying a DLQ write until replica replication is confirmed; when `false`, it gives up after the first failure.
Default: `false`
| | +| **wait\.retry\.delay\.ms**
(DLQ replica wait retry delay) | `integer` | Delay in milliseconds between DLQ replica wait retry attempts.
Default: `1000`
Minimum: `1`
| | + +**Additional Properties** + +| Name | Type | Description | Required | +| ------------------------- | ----------------------------- | ----------- | -------- | +| **Additional Properties** | `string`, `number`, `boolean` | | | -**Additional Properties:** not allowed +**Minimal Properties:** 1  **Example** ```yaml -hr: - type: postgresql - host: localhost - port: 5432 - database: postgres - user: postgres - password: postgres - +max.len: 1000 +batch.size: 100 +flush.interval.ms: 100 +connection.timeout.ms: 2000 +socket.timeout.ms: 2000 +retry.max.attempts: 1 +retry.initial.delay.ms: 100 +retry.max.delay.ms: 3000 +retry.backoff.multiplier: 2 +wait.enabled: false +wait.write.timeout.ms: 1000 +wait.retry.enabled: false +wait.retry.delay.ms: 1000 ``` +  + + +Advanced configuration properties forwarded to the underlying Flink runtime. Any property listed in the [Flink configuration documentation](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/config/) can be set here and will override the RDI default. **Flink processor only.**

The properties listed below are the ones most likely to require adjustment. **Changing any other Flink property is not recommended unless instructed by Redis support.** + +**Properties** + +| Name | Type | Description | Required | +| ------------------------------------------------------------------------ | --------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **parallelism\.default**
(Default parallelism) | `integer` | Default parallelism for jobs and operators. When unset, Flink uses the number of available task slots across all task managers (`taskManager.replicas × taskmanager.numberOfTaskSlots`). Increase to fan out work across more task slots; see [parallel execution](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/dev/datastream/execution/parallel/).
Minimum: `1`
| | +| **taskmanager\.numberOfTaskSlots**
(Task slots per task manager) | `integer` | Number of parallel task slots per task manager pod. Each slot can run one parallel pipeline instance, so this caps the parallelism a single task manager can absorb. See [task slots and resources](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/concepts/flink-architecture/#task-slots-and-resources).
Default: `1`
Minimum: `1`
| | +| **taskmanager\.memory\.process\.size**
(Task manager process memory) | `string` | Total memory budget for each task manager JVM process (heap + managed + network + metaspace + JVM overhead), expressed with a unit suffix such as `2048m` or `4g`. See [task manager memory configuration](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/memory/mem_setup_tm/).
Default: `"2048m"`
| | + +**Additional Properties** + +| Name | Type | Description | Required | +| ------------------------- | ----------------------------- | ----------- | -------- | +| **Additional Properties** | `string`, `number`, `boolean` | | | + +**Minimal Properties:** 1  **Example** ```yaml -my-oracle: - type: oracle - host: 172.17.0.4 - port: 1521 - user: c##dbzuser - password: dbz - +taskmanager.numberOfTaskSlots: 1 +taskmanager.memory.process.size: 2048m ``` - -#### sources\.connection\.MongoDB: MongoDB - -Connection configuration for a MongoDB database. +  + +Advanced configuration properties for the processor. **Flink processor only.** **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**type**
(Database type)|`string`|Database type identifier. Always `mongodb` for this connection.
Constant Value: `"mongodb"`
|yes| -|**connection\_string**|`string`|MongoDB connection URI including host, port, and any connection options.
|yes| -|**user**
(MongoDB user)|`string`|Username for authentication to MongoDB.
|no| -|**password**
(MongoDB password)|`string`|Password for authentication to MongoDB.
|no| -|**database**
(MongoDB databases)|`string`|Comma-separated list of MongoDB databases to monitor.
|no| +| Name | Type | Description | Required | +| ---------------------------------------------------------------------- | --------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **default\.data\.type**
(Default target data type) | `string` | Data type to use in Redis when not overridden per job: `hash` for Redis Hash, `json` for RedisJSON. Alias for `processors.target_data_type`; takes priority when both are set.
Default: `"hash"`
Enum: `"hash"`, `"json"`
| | +| **default\.json\.update\.strategy**
(Default JSON update strategy) | `string` | Strategy for updating JSON data in Redis: `replace` to overwrite the entire JSON object, `merge` to merge new data with the existing JSON object. Alias for `processors.json_update_strategy`; takes priority when both are set.
Default: `"replace"`
Enum: `"replace"`, `"merge"`
| | +| **dlq\.enabled**
(Enable DLQ) | `boolean` | When `true`, rejected messages are stored in the dead-letter queue; when `false`, errors are silently skipped. Alias for `processors.error_handling`; takes priority when both are set.
Default: `true`
| | + +**Additional Properties** + +| Name | Type | Description | Required | +| ------------------------- | ----------------------------- | ----------- | -------- | +| **Additional Properties** | `string`, `number`, `boolean` | | | -**Additional Properties:** not allowed +**Minimal Properties:** 1  **Example** ```yaml -mongodb-source: - type: mongodb - connection_string: mongodb://localhost:27017/?replicaSet=rs0 - user: debezium - password: dbz - database: db1,db2 - +default.data.type: hash +default.json.update.strategy: replace +dlq.enabled: true ``` - -#### sources\.connection\.Spanner: Spanner - -Connection configuration for a Google Cloud Spanner database. +  + +Compute resources allocated to the Flink job, such as the number of task manager pods. **Flink processor only.** **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**type**
(Database type)|`string`|Database type identifier. Always `spanner` for this connection.
Constant Value: `"spanner"`
|yes| -|**project\_id**
(Spanner project ID)|`string`|Google Cloud project ID that hosts the Spanner instance.
|yes| -|**instance\_id**
(Spanner instance ID)|`string`|Spanner instance identifier within the project.
|yes| -|**database\_id**
(Spanner database ID)|`string`|Spanner database identifier within the instance.
|yes| -|**emulator\_host**
(Spanner emulator host)|`string`|Host and port of the Spanner emulator. Used for local development; leave unset against real Spanner.
|no| -|**use\_credentials\_file**|`boolean`|When `true`, RDI authenticates using a service account credentials file; when `false`, it uses application default credentials.
Default: `false`
|no| -|[**change\_streams**](#sourcesconnectionspannerchange_streams)
(Change streams configuration)|`object`|Spanner change streams to capture, keyed by change stream name.
|yes| - -**Additional Properties:** not allowed +| Name | Type | Description | Required | +| ----------------------------------------------------------------------------------------------- | -------- | --------------------------------------------------- | -------- | +| [**taskManager**](#processorsadvancedresourcestaskmanager)
(Task manager resource settings) | `object` | Resource settings for Flink task manager pods.
| | + +**Additional Properties:** not allowed  +**Minimal Properties:** 1  **Example** ```yaml -spanner-source: - type: spanner - project_id: example-12345 - instance_id: example - database_id: example - change_streams: - change_stream_all: - retention_period_hours: 24 - +taskManager: {} ``` - -##### sources\.connection\.Spanner\.change\_streams: Change streams configuration +  + -Spanner change streams to capture, keyed by change stream name. +Resource settings for Flink task manager pods. +**Properties** -**Additional Properties** +| Name | Type | Description | Required | +| ---------------------------------------- | --------- | --------------------------------------------------------------- | -------- | +| **replicas**
(Task manager replicas) | `integer` | Number of Flink task manager pods to run.
Minimum: `1`
| | -|Name|Type|Description|Required| -|----|----|-----------|--------| -|[**Additional Properties**](#sourcesconnectionspannerchange_streamsadditionalproperties)|`object`, `null`||| +**Additional Properties:** not allowed  +**Minimal Properties:** 1  +  + -**Minimal Properties:** 1 - -###### sources\.connection\.Spanner\.change\_streams\.additionalProperties: object,null +Advanced configuration properties for the source Redis client and streams reader. **Flink processor only.** **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**retention\_period\_hours**
(Change stream retention period hours)|`integer`, `string`|Retention period for the change stream, in hours.
Pattern: `^\${.*}$`
Minimum: `1`
|| +| Name | Type | Description | Required | +| -------------------------------------------------------------------- | --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **stream\.name\.pattern**
(Source stream name pattern) | `string` | Glob pattern used to discover input streams in the source Redis database, for example `data:*`.
Default: `"data:*"`
| | +| **discovery\.interval\.ms**
(Stream discovery interval) | `integer` | Time in milliseconds between checks for new input streams. Replaces the classic `processors.idle_streams_check_interval_ms` and `processors.busy_streams_check_interval_ms` properties.
Default: `1000`
Minimum: `0`
| | +| **batch\.size**
(Source batch size) | `integer` | Maximum number of records the source operator reads in a single batch. Alias for `processors.read_batch_size`; takes priority when both are set.
Default: `2000`
Minimum: `1`
| | +| **batch\.timeout\.ms**
(Source batch timeout) | `integer` | Maximum time in milliseconds to wait for a source batch to fill before processing. Alias for `processors.read_batch_timeout_ms`; takes priority when both are set.
Default: `100`
Minimum: `1`
| | +| **connection\.timeout\.ms**
(Source connection timeout) | `integer` | Connection timeout in milliseconds for the source Redis client.
Default: `2000`
Minimum: `1`
| | +| **socket\.timeout\.ms**
(Source socket timeout) | `integer` | Socket read/write timeout in milliseconds for the source Redis client.
Default: `2000`
Minimum: `1`
| | +| **retry\.max\.attempts**
(Source retry max attempts) | `integer` | Maximum number of retry attempts for failed source Redis operations.
Default: `5`
Minimum: `1`
| | +| **retry\.initial\.delay\.ms**
(Source retry initial delay) | `integer` | Initial delay in milliseconds before the first retry of a failed source Redis operation.
Default: `100`
Minimum: `1`
| | +| **retry\.max\.delay\.ms**
(Source retry max delay) | `integer` | Maximum delay in milliseconds between retry attempts for source Redis operations.
Default: `3000`
Minimum: `1`
| | +| **retry\.backoff\.multiplier**
(Source retry backoff multiplier) | `number` | Exponential backoff multiplier between retry attempts for source Redis operations.
Default: `2`
Minimum: `1`
| | -**Additional Properties:** not allowed - -#### sources\.connection\.Snowflake: Snowflake +**Additional Properties** -Connection configuration for a Snowflake database. +| Name | Type | Description | Required | +| ------------------------- | ----------------------------- | ----------- | -------- | +| **Additional Properties** | `string`, `number`, `boolean` | | | + +**Minimal Properties:** 1  +**Example** + +```yaml +stream.name.pattern: data:* +discovery.interval.ms: 1000 +batch.size: 2000 +batch.timeout.ms: 100 +connection.timeout.ms: 2000 +socket.timeout.ms: 2000 +retry.max.attempts: 5 +retry.initial.delay.ms: 100 +retry.max.delay.ms: 3000 +retry.backoff.multiplier: 2 +``` +  + + +Advanced configuration properties for the target Redis client and sink. **Flink processor only.** **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**type**
(Database type)|`string`|Database type identifier. Always `snowflake` for this connection.
Constant Value: `"snowflake"`
|yes| -|**url**
(JDBC URL)|`string`|Snowflake JDBC connection URL, for example `jdbc:snowflake://account.snowflakecomputing.com/`.
|yes| -|**user**
(Snowflake user)|`string`|Username for authentication to Snowflake.
|yes| -|**password**
(Snowflake password)|`string`|Password for authentication to Snowflake. For key-pair authentication, omit this field and provide the private key via the `source-db-ssl` secret (`client.key` field).
|no| -|**database**
(Snowflake database)|`string`|Name of the Snowflake database to connect to.
|yes| -|**warehouse**
(Snowflake warehouse)|`string`|Name of the Snowflake warehouse used for compute.
|yes| -|**role**
(Snowflake role)|`string`|Snowflake role used for the connection.
|no| -|**cdcDatabase**
(CDC database)|`string`|Database hosting the CDC streams. Defaults to the main `database` if not set.
|no| -|**cdcSchema**
(CDC schema)|`string`|Schema hosting the CDC streams. Defaults to the main schema if not set.
|no| - -**Additional Properties:** not allowed +| Name | Type | Description | Required | +| -------------------------------------------------------------------- | --------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **batch\.size**
(Target sink batch size) | `integer` | Maximum number of records the target sink writes in a single batch. Alias for `processors.write_batch_size`; takes priority when both are set.
Default: `200`
Minimum: `1`
| | +| **flush\.interval\.ms**
(Target sink flush interval) | `integer` | Maximum time in milliseconds the target sink waits to fill a batch before flushing it to Redis.
Default: `100`
Minimum: `1`
| | +| **connection\.timeout\.ms**
(Target connection timeout) | `integer` | Connection timeout in milliseconds for the target Redis client.
Default: `2000`
Minimum: `1`
| | +| **socket\.timeout\.ms**
(Target socket timeout) | `integer` | Socket read/write timeout in milliseconds for the target Redis client.
Default: `2000`
Minimum: `1`
| | +| **retry\.max\.attempts**
(Target retry max attempts) | `integer` | Maximum number of retry attempts for failed target Redis operations. Alias for `processors.retry_max_attempts`; takes priority when both are set.
Default: `5`
Minimum: `1`
| | +| **retry\.initial\.delay\.ms**
(Target retry initial delay) | `integer` | Initial delay in milliseconds before the first retry of a failed target Redis operation. Alias for `processors.retry_initial_delay_ms`; takes priority when both are set.
Default: `1000`
Minimum: `1`
| | +| **retry\.max\.delay\.ms**
(Target retry max delay) | `integer` | Maximum delay in milliseconds between retry attempts for target Redis operations. Alias for `processors.retry_max_delay_ms`; takes priority when both are set.
Default: `10000`
Minimum: `1`
| | +| **retry\.backoff\.multiplier**
(Target retry backoff multiplier) | `number` | Exponential backoff multiplier between retry attempts for target Redis operations.
Default: `2`
Minimum: `1`
| | +| **wait\.enabled**
(Target replica wait enabled) | `boolean` | When `true`, RDI verifies that each write has been replicated to the target database's replica shards before acknowledging it. Enable this only when target database replication is enabled and a healthy replica is available. Alias for `processors.wait_enabled`; takes priority when both are set.
Default: `false`
| | +| **wait\.write\.timeout\.ms**
(Target replica wait timeout) | `integer` | Maximum time in milliseconds to wait for target replica write verification. Alias for `processors.wait_timeout`; takes priority when both are set.
Default: `1000`
Minimum: `1`
| | +| **wait\.retry\.enabled**
(Target replica wait retry enabled) | `boolean` | When `true`, RDI keeps retrying a target write until replica replication is confirmed; when `false`, it gives up after the first failure. Alias for `processors.retry_on_replica_failure`; takes priority when both are set. When enabled, the Flink processor retries indefinitely. Failed checkpoints can restart the job, after which the retries resume. The classic processor retries once.
Default: `true`
| | +| **wait\.retry\.delay\.ms**
(Target replica wait retry delay) | `integer` | Delay in milliseconds between target replica wait retry attempts.
Default: `1000`
Minimum: `1`
| | + +**Additional Properties** + +| Name | Type | Description | Required | +| ------------------------- | ----------------------------- | ----------- | -------- | +| **Additional Properties** | `string`, `number`, `boolean` | | | + +**Minimal Properties:** 1  **Example** ```yaml -snowflake: - type: snowflake - url: jdbc:snowflake://myaccount.snowflakecomputing.com/ - user: myuser - password: mypassword - database: MYDB - warehouse: COMPUTE_WH - +batch.size: 200 +flush.interval.ms: 100 +connection.timeout.ms: 2000 +socket.timeout.ms: 2000 +retry.max.attempts: 5 +retry.initial.delay.ms: 1000 +retry.max.delay.ms: 10000 +retry.backoff.multiplier: 2 +wait.enabled: false +wait.write.timeout.ms: 1000 +wait.retry.enabled: true +wait.retry.delay.ms: 1000 ``` - -### sources\.logging: Logging configuration - -Logging settings for this source collector. +  + +Logging settings for the processor. **Flink processor only.** **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**level**
(Logging level)|`string`|Log verbosity for the source collector.
Default: `"info"`
Enum: `"trace"`, `"debug"`, `"info"`, `"warn"`, `"error"`
|| +| Name | Type | Description | Required | +| ----------------------------- | -------- | ------------------------------------------------------------------------------------------------------------------------- | -------- | +| **level**
(Logging level) | `string` | Log verbosity for the processor.
Default: `"info"`
Enum: `"trace"`, `"debug"`, `"info"`, `"warn"`, `"error"`
| | -**Additional Properties:** not allowed +**Additional Properties:** not allowed  **Example** ```yaml level: info - ``` - -### sources\.tables: Tables to capture +  + -Tables to capture from the source database, keyed by table name. The value configures column selection and key handling for that table. +External secret providers used to resolve `${...}` references in the configuration. +**Properties** (key: `.*`) -**Additional Properties** +| Name | Type | Description | Required | +| ----------------------------------------------------------------------- | -------- | ------------------------------------------------------------------------------------------------------------------------ | -------- | +| **type**
(Provider type) | `string` | Secret provider backend. `aws` uses AWS Secrets Manager; `vault` uses HashiCorp Vault.
Enum: `"aws"`, `"vault"`
| yes | +| [**parameters**](#secret-providersparameters)
(Provider parameters) | `object` | Configuration parameters for the secret provider.
| yes | -|Name|Type|Description|Required| -|----|----|-----------|--------| -|[**Additional Properties**](#sourcestablesadditionalproperties)|`object`, `null`||| +  + -**Minimal Properties:** 1 - -#### sources\.tables\.additionalProperties: object,null +Configuration parameters for the secret provider. **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**snapshot\_sql**|`string`|Custom SQL statement used during the initial snapshot, giving fine-grained control over the data captured.
|| -|[**columns**](#sourcestablesadditionalpropertiescolumns)
(Columns to capture)|`string[]`|List of specific columns to capture for changes. If not specified, all columns will be captured. For RIOTX Snowflake sources, this is rendered as per-table `--table-columns` projection. Note: This property cannot be used for MongoDB connections
|| -|[**exclude\_columns**](#sourcestablesadditionalpropertiesexclude_columns)
(Columns to exclude)|`string[]`|Specific columns to exclude from capture. When omitted, no columns are excluded. Only supported for MongoDB connections.
|| -|[**keys**](#sourcestablesadditionalpropertieskeys)
(Message keys)|`string[]`|Optional list of columns to use as a composite unique identifier. For RIOTX Snowflake sources, this is rendered as per-table `--table-keys`. Only required when the table lacks a primary key or unique constraint. Must form a unique combination of fields
|| - -**Additional Properties:** not allowed - -##### sources\.tables\.additionalProperties\.columns\[\]: Columns to capture - -List of specific columns to capture for changes. If not specified, all columns will be captured. For RIOTX Snowflake sources, this is rendered as per-table `--table-columns` projection. Note: This property cannot be used for MongoDB connections +| Name | Type | Description | Required | +| ----------------------------------------------------------------------------- | ---------- | ----------------------------------------------- | -------- | +| [**objects**](#secret-providersparametersobjects)
(Secrets objects array) | `object[]` | Secret objects to fetch from the provider.
| yes | +**Example** - -##### sources\.tables\.additionalProperties\.exclude\_columns\[\]: Columns to exclude - -Specific columns to exclude from capture. When omitted, no columns are excluded. Only supported for MongoDB connections. - +```yaml +objects: + - {} +``` - -##### sources\.tables\.additionalProperties\.keys\[\]: Message keys +  + -Optional list of columns to use as a composite unique identifier. For RIOTX Snowflake sources, this is rendered as per-table `--table-keys`. Only required when the table lacks a primary key or unique constraint. Must form a unique combination of fields +Secret objects to fetch from the provider. +**Items: Secret object** - -### sources\.schemas\[\]: Schema names +**No properties.** -Schema names to capture from the source database. Maps to the underlying connector's `schema.include.list`. +**Example** +```yaml +- {} +``` - -### sources\.databases\[\]: Database names +  + -Database names to capture from the source database. Maps to the underlying connector's `database.include.list`. Applies only to MySQL, MariaDB, and MongoDB connections. +Source collectors that capture changes from upstream databases. Each key is a unique source identifier; the value configures one collector. +**Properties** (key: `.*`) +| Name | Type | Description | Required | +| ----------------------------------------------------------------------------------- | ---------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -------- | +| [**connection**](#definitionsnon_redis_connection)
(Source database connection) | `object` | Connection configuration for a non-Redis source database. The exact set of properties depends on the database type.
| | +| **name**
(Source name) | `string` | Human-readable name for the source collector. Maximum 100 characters.
Maximal Length: `100`
| no | +| **type**
(Collector type) | `string` | Type of the source collector. Use `cdc` (default) for change data capture using [Debezium](https://debezium.io/). Use `flink` for Spanner change streams using the Apache Flink-based collector. Use `riotx` for Snowflake CDC using [RIOT-X](https://redis.github.io/riotx/).
Default: `"cdc"`
Enum: `"cdc"`, `"flink"`, `"riotx"`
| yes | +| **active**
(Collector enabled) | `boolean` | When `true`, the collector runs; when `false`, the collector is disabled and produces no events.
Default: `true`
| no | +| [**logging**](#sourceslogging)
(Logging configuration) | `object` | Logging settings for this source collector.
| no | +| [**tables**](#sourcestables)
(Tables to capture) | `object` | Tables to capture from the source database, keyed by table name. The value configures column selection and key handling for that table.
| no | +| [**schemas**](#sourcesschemas)
(Schema names) | `string[]` | Schema names to capture from the source database. Maps to the underlying connector's `schema.include.list`.
| no | +| [**databases**](#sourcesdatabases)
(Database names) | `string[]` | Database names to capture from the source database. Maps to the underlying connector's `database.include.list`. Applies only to MySQL, MariaDB, and MongoDB connections.
| no | +| [**advanced**](#sourcesadvanced)
(Advanced configuration) | `object` | Advanced configuration that overrides the underlying engine's defaults. Only required for non-standard tuning.
| no | + +  -### sources\.advanced: Advanced configuration Advanced configuration that overrides the underlying engine's defaults. Only required for non-standard tuning. - **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|[**sink**](#sourcesadvancedsink)
(RDI Collector stream writer configuration)|`object`|Advanced configuration properties for the RDI Collector stream writer connection and behaviour. **Applies to the `cdc` and `flink` collector types.**
|| -|[**source**](#sourcesadvancedsource)
(Advanced source settings)|`object`|Advanced configuration properties for the source database connection and CDC behavior. **Applies to the `cdc` and `flink` collector types.**
|| -|[**quarkus**](#sourcesadvancedquarkus)
(Quarkus runtime settings)|`object`|Advanced configuration properties for the Quarkus runtime that hosts Debezium Server. **Only applies to the `cdc` collector type.** See the [Debezium Server documentation](https://debezium.io/documentation/reference/stable/operations/debezium-server.html) for runtime configuration options. When using a property from that page, omit the `quarkus.` prefix.
|| -|[**flink**](#sourcesadvancedflink)
(Advanced Flink settings)|`object`|Advanced configuration properties forwarded to the Flink runtime that hosts the collector. Any property listed in the [Flink configuration documentation](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/config/) can be set here and will override the RDI default. **Only applies to the `flink` collector type.**
|| -|[**resources**](#sourcesadvancedresources)
(Collector resource settings)|`object`|Compute resources allocated to the collector. **Only applies to the `cdc` collector type.**
|| -|[**riotx**](#sourcesadvancedriotx)
(Advanced RIOT\-X settings)|`object`|Advanced configuration properties for the RIOT-X Snowflake collector. **Only applies to the `riotx` collector type.**
|| -|**java\_options**
(Advanced Java options)|`string`|These Java options will be passed to the command line command when launching the source collector. **Only applies to the `cdc` collector type.**
|| - -**Additional Properties:** not allowed -**Minimal Properties:** 1 +| Name | Type | Description | Required | +| -------------------------------------------------------------------------------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| [**sink**](#sourcesadvancedsink)
(RDI Collector stream writer configuration) | `object` | Advanced configuration properties for the RDI Collector stream writer connection and behaviour. **Applies to the `cdc` and `flink` collector types.**
| | +| [**source**](#sourcesadvancedsource)
(Advanced source settings) | `object` | Advanced configuration properties for the source database connection and CDC behavior. **Applies to the `cdc` and `flink` collector types.**
| | +| [**quarkus**](#sourcesadvancedquarkus)
(Quarkus runtime settings) | `object` | Advanced configuration properties for the Quarkus runtime that hosts Debezium Server. **Only applies to the `cdc` collector type.** See the [Debezium Server documentation](https://debezium.io/documentation/reference/stable/operations/debezium-server.html) for runtime configuration options. When using a property from that page, omit the `quarkus.` prefix.
| | +| [**flink**](#sourcesadvancedflink)
(Advanced Flink settings) | `object` | Advanced configuration properties forwarded to the Flink runtime that hosts the collector. Any property listed in the [Flink configuration documentation](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/config/) can be set here and will override the RDI default. **Only applies to the `flink` collector type.**
| | +| [**resources**](#sourcesadvancedresources)
(Collector resource settings) | `object` | Compute resources allocated to the collector. **Only applies to the `cdc` collector type.**
| | +| [**riotx**](#sourcesadvancedriotx)
(Advanced RIOT\-X settings) | `object` | Advanced configuration properties for the RIOT-X Snowflake collector. **Only applies to the `riotx` collector type.**
| | +| **java\_options**
(Advanced Java options) | `string` | These Java options will be passed to the command line command when launching the source collector. **Only applies to the `cdc` collector type.**
| | + +**Additional Properties:** not allowed  +**Minimal Properties:** 1  **Example** ```yaml @@ -397,695 +518,508 @@ resources: {} riotx: poll: 30s snapshot: INITIAL - streamPrefix: 'data:' + streamPrefix: "data:" clearOffset: false count: 0 - -``` - - -#### sources\.advanced\.sink: RDI Collector stream writer configuration - -Advanced configuration properties for the RDI Collector stream writer connection and behaviour. **Applies to the `cdc` and `flink` collector types.**

For the `cdc` collector type, see the full list of properties at [Debezium Server — Redis Stream sink](https://debezium.io/documentation/reference/stable/operations/debezium-server.html#_redis_stream). When using a property from that page, omit the `debezium.sink.` prefix.

**The properties listed below only apply to the `flink` collector type.** - - -**Properties** - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**redis\.batch\.size**
(Sink batch size)|`integer`|Maximum number of records the collector sink writes to Redis in a single batch.
Default: `1000`
Minimum: `1`
|| -|**redis\.flush\.interval\.ms**
(Sink flush interval)|`integer`|Maximum time in milliseconds the collector sink waits to fill a batch before flushing it to Redis.
Default: `100`
Minimum: `1`
|| -|**redis\.connection\.timeout\.ms**
(Sink connection timeout)|`integer`|Connection timeout in milliseconds for the target Redis client used by the collector sink.
Default: `2000`
Minimum: `1`
|| -|**redis\.socket\.timeout\.ms**
(Sink socket timeout)|`integer`|Socket read/write timeout in milliseconds for the target Redis client used by the collector sink.
Default: `2000`
Minimum: `1`
|| -|**redis\.retry\.max\.attempts**
(Sink retry max attempts)|`integer`|Maximum number of retry attempts for failed Redis operations.
Default: `5`
Minimum: `1`
|| -|**redis\.retry\.initial\.delay\.ms**
(Sink retry initial delay)|`integer`|Initial delay in milliseconds before the first retry of a failed Redis operation.
Default: `100`
Minimum: `1`
|| -|**redis\.retry\.max\.delay\.ms**
(Sink retry max delay)|`integer`|Maximum delay in milliseconds between retry attempts for failed Redis operations.
Default: `3000`
Minimum: `1`
|| -|**redis\.retry\.backoff\.multiplier**
(Sink retry backoff multiplier)|`number`|Exponential backoff multiplier between retry attempts for failed Redis operations.
Default: `2`
Minimum: `1`
|| -|**redis\.oom\.retry\.initial\.delay\.ms**
(Sink OOM retry initial delay)|`integer`|Initial delay in milliseconds before the first retry after a Redis out-of-memory error.
Default: `1000`
Minimum: `1`
|| -|**redis\.oom\.retry\.max\.delay\.ms**
(Sink OOM retry max delay)|`integer`|Maximum delay in milliseconds between retry attempts after a Redis out-of-memory error.
Default: `10000`
Minimum: `1`
|| -|**redis\.oom\.retry\.backoff\.multiplier**
(Sink OOM retry backoff multiplier)|`number`|Exponential backoff multiplier between retry attempts after a Redis out-of-memory error.
Default: `2`
Minimum: `1`
|| -|**redis\.wait\.enabled**
(Sink replica wait enabled)|`boolean`|When `true`, the collector verifies that each write has been replicated to the configured number of Redis replica shards before acknowledging it.
Default: `false`
|| -|**redis\.wait\.write\.timeout\.ms**
(Sink replica wait timeout)|`integer`|Maximum time in milliseconds to wait for replica write acknowledgements.
Default: `1000`
Minimum: `1`
|| -|**redis\.wait\.retry\.enabled**
(Sink replica wait retry enabled)|`boolean`|When `true`, the collector keeps retrying a write until replica acknowledgement succeeds; when `false`, it gives up after the first failure.
Default: `false`
|| -|**redis\.wait\.retry\.delay\.ms**
(Sink replica wait retry delay)|`integer`|Delay in milliseconds between replica wait retry attempts.
Default: `1000`
Minimum: `1`
|| - -**Additional Properties** - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**Additional Properties**|`string`, `number`, `boolean`||| - -**Minimal Properties:** 1 -**Example** - -```yaml -redis.batch.size: 1000 -redis.flush.interval.ms: 100 -redis.connection.timeout.ms: 2000 -redis.socket.timeout.ms: 2000 -redis.retry.max.attempts: 5 -redis.retry.initial.delay.ms: 100 -redis.retry.max.delay.ms: 3000 -redis.retry.backoff.multiplier: 2 -redis.oom.retry.initial.delay.ms: 1000 -redis.oom.retry.max.delay.ms: 10000 -redis.oom.retry.backoff.multiplier: 2 -redis.wait.enabled: false -redis.wait.write.timeout.ms: 1000 -redis.wait.retry.enabled: false -redis.wait.retry.delay.ms: 1000 - -``` - - -#### sources\.advanced\.source: Advanced source settings - -Advanced configuration properties for the source database connection and CDC behavior. **Applies to the `cdc` and `flink` collector types.**

For the `cdc` collector type, available properties depend on the source database — refer to the relevant Debezium connector documentation: [MySQL](https://debezium.io/documentation/reference/stable/connectors/mysql.html), [MariaDB](https://debezium.io/documentation/reference/stable/connectors/mariadb.html), [PostgreSQL](https://debezium.io/documentation/reference/stable/connectors/postgresql.html), [Oracle](https://debezium.io/documentation/reference/stable/connectors/oracle.html), [SQL Server](https://debezium.io/documentation/reference/stable/connectors/sqlserver.html), [Db2](https://debezium.io/documentation/reference/stable/connectors/db2.html), [MongoDB](https://debezium.io/documentation/reference/stable/connectors/mongodb.html). When using a property from those pages, omit the `debezium.source.` prefix.

**The named properties below cover the most commonly tuned settings: `spanner.*` properties apply to the `flink` collector type, all others apply to the `cdc` collector type. Any other property from the Debezium documentation can still be set as a free-form key-value pair.** - - -**Properties** - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**record\.processing\.threads**
(Record processing threads)|`integer`|Controls how many worker threads process captured records before they are written downstream.
Minimum: `1`
|| -|**snapshot\.max\.threads**
(Snapshot max threads)|`integer`|Sets the maximum number of threads used while taking the initial snapshot.
Default: `1`
Minimum: `1`
|| -|**poll\.interval\.ms**
(Poll interval ms)|`integer`|Defines how often the collector polls the source for new changes.
Default: `500`
Minimum: `1`
|| -|**snapshot\.fetch\.size**
(Snapshot fetch size)|`integer`|Defines how many rows are fetched per batch during the initial snapshot.
Default: `10000`
Minimum: `1`
|| -|**max\.batch\.size**
(Max batch size)|`integer`|Caps how many records are processed together in a single batch.
Default: `2048`
Minimum: `1`
|| -|**max\.queue\.size**
(Max queue size)|`integer`|Limits how many records can be buffered in memory before processing catches up.
Default: `8192`
Minimum: `1`
|| -|**heartbeat\.interval\.ms**
(Heartbeat interval ms)|`integer`|Sets how often heartbeat events are emitted to keep change tracking active. Use 0 to disable them.
Default: `0`
Minimum: `0`
|| -|**heartbeat\.action\.query**
(Heartbeat action query)|`string`|SQL query executed on the source whenever a heartbeat is emitted.
|| -|**lob\.enabled**
(Lob enabled)|`boolean`|Determines whether large object columns are included in change capture.
Default: `false`
|| -|**gtid\.source\.includes**
(GTID source includes)|`string`|Restricts MySQL GTID processing to the listed source UUIDs.
|| -|**publication\.autocreate\.mode**
(Publication autocreate mode)|`string`|Controls whether and how the PostgreSQL publication is created or updated automatically.
Default: `"all_tables"`
Enum: `"all_tables"`, `"filtered"`, `"disabled"`
|| -|**publication\.name**
(Publication name)|`string`|Sets the PostgreSQL logical replication publication name used by the collector.
Default: `"dbz_publication"`
|| -|**slot\.name**
(Slot name)|`string`|Sets the PostgreSQL replication slot name the collector reads from.
Default: `"debezium"`
|| -|**spanner\.version\.retention\.period\.hours**
(Spanner version retention period)|`integer`|Retention period in hours for Spanner change stream versions. Determines how far back the collector can resume after an outage.
Default: `1`
Minimum: `1`
|| -|**spanner\.fetch\.timeout\.ms**
(Spanner fetch timeout)|`integer`|Timeout in milliseconds for a single change stream fetch request to Spanner.
Default: `500`
Minimum: `1`
|| -|**spanner\.fetch\.heartbeat\.ms**
(Spanner fetch heartbeat interval)|`integer`|Interval in milliseconds at which Spanner sends heartbeat records when no data changes are available.
Default: `100`
Minimum: `1`
|| -|**spanner\.max\.rows\.per\.partition**
(Spanner max rows per partition)|`integer`|Maximum number of rows the collector reads from a single Spanner change stream partition before yielding.
Default: `10000`
Minimum: `1`
|| -|**spanner\.dialect**
(Spanner SQL dialect)|`string`|SQL dialect of the Spanner database. Use `GOOGLESQL` for Google Standard SQL or `POSTGRESQL` for the PostgreSQL interface.
Default: `"GOOGLESQL"`
Enum: `"GOOGLESQL"`, `"POSTGRESQL"`
|| - -**Additional Properties** - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**Additional Properties**|`string`, `number`, `boolean`||| - -**Minimal Properties:** 1 -**Example** - -```yaml -snapshot.max.threads: 1 -poll.interval.ms: 500 -snapshot.fetch.size: 10000 -max.batch.size: 2048 -max.queue.size: 8192 -heartbeat.interval.ms: 0 -lob.enabled: false -publication.autocreate.mode: all_tables -publication.name: dbz_publication -slot.name: debezium -spanner.version.retention.period.hours: 1 -spanner.fetch.timeout.ms: 500 -spanner.fetch.heartbeat.ms: 100 -spanner.max.rows.per.partition: 10000 -spanner.dialect: GOOGLESQL - ``` - -#### sources\.advanced\.quarkus: Quarkus runtime settings - -Advanced configuration properties for the Quarkus runtime that hosts Debezium Server. **Only applies to the `cdc` collector type.** See the [Debezium Server documentation](https://debezium.io/documentation/reference/stable/operations/debezium-server.html) for runtime configuration options. When using a property from that page, omit the `quarkus.` prefix. - - -**Additional Properties** - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**Additional Properties**|`string`, `number`, `boolean`||| - -**Minimal Properties:** 1 +  -#### sources\.advanced\.flink: Advanced Flink settings Advanced configuration properties forwarded to the Flink runtime that hosts the collector. Any property listed in the [Flink configuration documentation](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/config/) can be set here and will override the RDI default. **Only applies to the `flink` collector type.**

The properties listed below are the ones most likely to require adjustment. **Changing any other Flink property is not recommended unless instructed by Redis support.** - **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**parallelism\.default**
(Default parallelism)|`integer`|Default parallelism for Flink jobs and operators. When unset, Flink uses the number of available task slots across all task managers (`taskManager.replicas × taskmanager.numberOfTaskSlots`). See [parallel execution](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/dev/datastream/execution/parallel/).
Minimum: `1`
|| -|**taskmanager\.numberOfTaskSlots**
(Task slots per task manager)|`integer`|Number of parallel task slots per task manager pod. Each slot can run one parallel pipeline instance, so this caps the parallelism a single task manager can absorb. See [task slots and resources](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/concepts/flink-architecture/#task-slots-and-resources).
Default: `1`
Minimum: `1`
|| -|**taskmanager\.memory\.process\.size**
(Task manager process memory)|`string`|Total memory budget for each task manager JVM process, expressed with a unit suffix such as `2048m` or `4g`. See [task manager memory configuration](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/memory/mem_setup_tm/).
|| +| Name | Type | Description | Required | +| ------------------------------------------------------------------------ | --------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **parallelism\.default**
(Default parallelism) | `integer` | Default parallelism for Flink jobs and operators. When unset, Flink uses the number of available task slots across all task managers (`taskManager.replicas × taskmanager.numberOfTaskSlots`). See [parallel execution](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/dev/datastream/execution/parallel/).
Minimum: `1`
| | +| **taskmanager\.numberOfTaskSlots**
(Task slots per task manager) | `integer` | Number of parallel task slots per task manager pod. Each slot can run one parallel pipeline instance, so this caps the parallelism a single task manager can absorb. See [task slots and resources](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/concepts/flink-architecture/#task-slots-and-resources).
Default: `1`
Minimum: `1`
| | +| **taskmanager\.memory\.process\.size**
(Task manager process memory) | `string` | Total memory budget for each task manager JVM process, expressed with a unit suffix such as `2048m` or `4g`. See [task manager memory configuration](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/memory/mem_setup_tm/).
| | **Additional Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**Additional Properties**|`string`, `number`, `boolean`||| +| Name | Type | Description | Required | +| ------------------------- | ----------------------------- | ----------- | -------- | +| **Additional Properties** | `string`, `number`, `boolean` | | | -**Minimal Properties:** 1 +**Minimal Properties:** 1  **Example** ```yaml taskmanager.numberOfTaskSlots: 1 - ``` - -#### sources\.advanced\.resources: Collector resource settings - -Compute resources allocated to the collector. **Only applies to the `cdc` collector type.** - - -**Properties** - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**cpu**
(CPU resource value)|`string`|CPU request for the collector container, for example `1` or `500m`.
|| -|**memory**
(Memory resource value)|`string`|Memory request for the collector container, for example `1024Mi` or `2Gi`.
|| - -**Additional Properties:** not allowed -**Minimal Properties:** 1 - -#### sources\.advanced\.riotx: Advanced RIOT\-X settings - -Advanced configuration properties for the RIOT-X Snowflake collector. **Only applies to the `riotx` collector type.** - - -**Properties** - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**poll**
(Polling interval)|`string`|Interval between polls for new stream changes, for example `30s` or `PT30S`.
Default: `"30s"`
|| -|**snapshot**
(Snapshot mode)|`string`|Initial-load behavior. `INITIAL` performs a one-time snapshot before streaming; `NEVER` skips the snapshot.
Default: `"INITIAL"`
Enum: `"INITIAL"`, `"NEVER"`
|| -|**streamPrefix**
(Redis stream key prefix)|`string`|Prefix used when constructing Redis stream keys, for example `data:`.
Default: `"data:"`
|| -|**streamLimit**
(Maximum stream length)|`integer`|Maximum number of entries kept in each Redis stream before older entries are trimmed.
Minimum: `1`
|| -|[**keyColumns**](#sourcesadvancedriotxkeycolumns)
(Key columns)|`string[]`|Deprecated RIOTX global fallback list of columns to use as message keys for every captured table. Prefer `tables..keys`
|| -|**clearOffset**
(Clear existing offset)|`boolean`|When `true`, the stored offset is cleared on startup, forcing a fresh read.
Default: `false`
|| -|**count**
(Record count limit)|`integer`|Maximum number of records to process. Set to `0` for unlimited.
Default: `0`
Minimum: `0`
|| - -**Additional Properties:** not allowed -**Minimal Properties:** 1 -**Example** - -```yaml -poll: 30s -snapshot: INITIAL -streamPrefix: 'data:' -clearOffset: false -count: 0 - -``` - - -##### sources\.advanced\.riotx\.keyColumns\[\]: Key columns - -Deprecated RIOTX global fallback list of columns to use as message keys for every captured table. Prefer `tables..keys` - - - -## targets: Target connections - -Target Redis databases where processed records are written. Each key is a target identifier; the value configures the connection. - - -**Properties** (key: `.*`) - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|[**connection**](#targetsconnection)
(Database connection)|`object`|Connection configuration for a Redis database.
|yes| -|**name**
(Target name)|`string`|Human-readable name for the target connection. Maximum 100 characters.
Maximal Length: `100`
|no| - - - -### targets\.connection: Database connection - -Connection configuration for a Redis database. - - -**Properties** - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**type**
(Database type)||Database type identifier. Always `redis` for this connection.
Constant Value: `"redis"`
|yes| -|**host**
(Database host)|`string`|Hostname or IP address of the Redis server.
|yes| -|**port**
(Database port)|`integer`|Network port on which the Redis server is listening.
Minimum: `1`
Maximum: `65535`
|yes| -|**user**
(Database user)|`string`|Username for authentication to the Redis database.
|no| -|**password**
(Database password)|`string`|Password for authentication to the Redis database.
|no| -|**key**
(Private key file)|`string`|Path to the private key file used for SSL/TLS client authentication.
|no| -|**key\_password**
(Private key password)|`string`|Password used to decrypt the private key file.
|no| -|**cert**
(Client certificate)|`string`|Path to the client certificate file used for SSL/TLS client authentication.
|no| -|**cacert**
(CA certificate)|`string`|Path to the Certificate Authority (CA) certificate file used to verify the server's TLS certificate.
|no| - -**Additional Properties:** not allowed -**Minimal Properties:** 3 -**If property *key* is defined**, property/ies *cert* is/are required. -**If property *cert* is defined**, property/ies *key* is/are required. -**If property *key_password* is defined**, property/ies *key* is/are required. - -## processors: Data processing configuration - -Settings that control how data is processed, including batch sizes, error handling, and performance tuning. - - -**Properties** - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**type**
(Processor type)|`string`|Processor implementation to run. `classic` runs the classic processor; `flink` runs the Apache Flink-based processor.
Default: `"classic"`
Enum: `"classic"`, `"flink"`
|| -|**read\_batch\_size**|`integer`, `string`|Maximum number of records read from the source streams in a single batch.
Default: `2000`
Pattern: `^\${.*}$`
Minimum: `1`
|| -|**read\_batch\_timeout\_ms**
(Read batch timeout)|`integer`|Maximum time in milliseconds to wait for a batch to fill before processing it.
Default: `100`
Minimum: `1`
|| -|**duration**
(Batch duration limit)|`integer`, `string`|(DEPRECATED)
This property has no effect; use `read_batch_timeout_ms` instead.
Default: `100`
Pattern: `^\${.*}$`
Minimum: `1`
|| -|**write\_batch\_size**|`integer`, `string`|Maximum number of records written to the target Redis database in a single batch.
Default: `200`
Pattern: `^\${.*}$`
Minimum: `1`
|| -|**enable\_async\_processing**|`boolean`|When `true`, the processor handles batches asynchronously to improve throughput. **Classic processor only.**
Default: `true`
|| -|**batch\_queue\_size**|`integer`|Maximum number of batches queued for processing. **Classic processor only.**
Default: `3`
Minimum: `1`
|| -|**ack\_queue\_size**|`integer`|Maximum number of batches queued for asynchronous acknowledgement. **Classic processor only.**
Default: `10`
Minimum: `1`
|| -|**dedup**
(Enable deduplication)|`boolean`|When `true`, the processor deduplicates incoming records. **Classic processor only.**
Default: `false`
|| -|**dedup\_max\_size**
(Deduplication set size)|`integer`|Maximum number of entries kept in the deduplication set. **Classic processor only.**
Default: `1024`
Minimum: `1`
|| -|**dedup\_strategy**
(Deduplication strategy)|`string`|(DEPRECATED)
This property has no effect — the only supported strategy is `ignore`. Remove it from the configuration. **Classic processor only.**
Default: `"ignore"`
Enum: `"reject"`, `"ignore"`
|| -|**error\_handling**
(Error handling strategy)|`string`|Strategy for handling failed records. `ignore` silently drops them; `dlq` writes them to the dead-letter queue.
Default: `"dlq"`
|| -|**dlq\_max\_messages**
(DLQ message limit)|`integer`, `string`|Maximum number of messages stored per dead-letter queue stream.
Default: `1000`
Pattern: `^\${.*}$`
Minimum: `1`
|| -|**target\_data\_type**
(Target Redis data type)|`string`|Data type used to store target records in Redis. `hash` writes a Redis Hash; `json` writes a RedisJSON document and requires the RedisJSON module.
Default: `"hash"`
|| -|**json\_update\_strategy**|`string`|Strategy for updating existing JSON documents in Redis. `replace` overwrites the entire document; `merge` merges incoming fields into it.
Default: `"replace"`
|| -|**use\_native\_json\_merge**
(Use native JSON merge from RedisJSON module)|`boolean`|Controls whether JSON merge operations use the native `JSON.MERGE` command (when `true`) or Lua scripts (when `false`). Introduced in RDI 1.15.0. The native command provides 2x performance improvement but handles null values differently:

**Previous behavior (Lua merge)**: When merging `{"field1": "value1", "field2": "value2"}` with `{"field2": null, "field3": "value3"}`, the result was `{"field1": "value1", "field2": null, "field3": "value3"}` (null value is preserved).

**New behavior (JSON.MERGE)**: The same merge produces `{"field1": "value1", "field3": "value3"}` (null value removes the field, following [RFC 7396](https://datatracker.ietf.org/doc/html/rfc7396)).

**Note**: The native `JSON.MERGE` command requires RedisJSON 2.6.0 or higher. If the target database has an older version of RedisJSON, RDI automatically falls back to Lua-based merge operations regardless of this setting.

**Impact**: If your application logic distinguishes between a field with a `null` value and a missing field, you may need to adjust your data handling. This follows the JSON Merge Patch RFC standard but differs from the previous Lua implementation. Set to `false` to revert to the previous Lua-based merge behavior if needed.

The Flink processor always uses the native `JSON.MERGE` command when the target database supports it. **Classic processor only.**
Default: `true`
|| -|**initial\_sync\_processes**|`integer`, `string`|Number of parallel processes used to perform the initial data synchronization. For the Flink processor, parallelism is controlled by Flink properties instead. **Classic processor only.**
Default: `4`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `32`
|| -|**idle\_sleep\_time\_ms**
(Idle sleep interval)|`integer`, `string`|Time in milliseconds to sleep between processing batches when idle. **Classic processor only.**
Default: `200`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `999999`
|| -|**idle\_streams\_check\_interval\_ms**
(Idle streams check interval)|`integer`, `string`|Time in milliseconds between checks for new streams when the processor is idle. For the Flink processor, use `processors.advanced.source.discovery.interval.ms` instead to configure a single discovery interval regardless of load. **Classic processor only.**
Default: `1000`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `999999`
|| -|**busy\_streams\_check\_interval\_ms**
(Busy streams check interval)|`integer`, `string`|Time in milliseconds between checks for new streams when the processor is busy. For the Flink processor, use `processors.advanced.source.discovery.interval.ms` instead to configure a single discovery interval regardless of load. **Classic processor only.**
Default: `5000`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `999999`
|| -|**retry\_max\_attempts**
(Maximum retry attempts)|`integer`, `string`|Maximum number of attempts for a failed write to the target Redis database before giving up.
Default: `5`
Pattern: `^\${.*}$`
Minimum: `1`
|| -|**retry\_initial\_delay\_ms**
(Initial retry delay)|`integer`, `string`|Initial delay in milliseconds before the first retry of a failed write.
Default: `1000`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `999999`
|| -|**retry\_max\_delay\_ms**
(Maximum retry delay)|`integer`, `string`|Maximum delay in milliseconds between retry attempts.
Default: `10000`
Pattern: `^\${.*}$`
Minimum: `1`
Maximum: `999999`
|| -|**wait\_enabled**
(Enable replica wait)|`boolean`|When `true`, RDI verifies that each write has been replicated to the target database's replica shards before acknowledging it. Enable this only when target database replication is enabled and a healthy replica is available. For the Flink processor, `processors.advanced.target.wait.enabled` takes priority.
Default: `false`
|| -|**wait\_timeout**
(Replica wait timeout)|`integer`, `string`|Maximum time in milliseconds to wait for replica write verification on the target database.
Default: `1000`
Pattern: `^\${.*}$`
Minimum: `1`
|| -|**retry\_on\_replica\_failure**|`boolean`|When `true`, RDI keeps retrying a write until replica replication is confirmed; when `false`, it gives up after the first failure.
Default: `true`
|| -|**on\_failed\_retry\_interval**
(Retry interval on failure)|`integer`, `string`|(DEPRECATED)
This property has no effect; remove it from the configuration.
Default: `5`
Pattern: `^\${.*}$`
Minimum: `1`
|| -|[**logging**](#processorslogging)
(Logging configuration)|`object`|Logging settings for the processor. **Flink processor only.**
|| -|[**advanced**](#processorsadvanced)
(Advanced configuration)|`object`|Advanced configuration for fine-tuning the processor. **All properties under `advanced` apply to the Flink processor only and are silently ignored by the classic processor.**
|| - -**Additional Properties:** not allowed - -### processors\.logging: Logging configuration - -Logging settings for the processor. **Flink processor only.** - - -**Properties** - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**level**
(Logging level)|`string`|Log verbosity for the processor.
Default: `"info"`
Enum: `"trace"`, `"debug"`, `"info"`, `"warn"`, `"error"`
|| - -**Additional Properties:** not allowed -**Example** - -```yaml -level: info - -``` - - -### processors\.advanced: Advanced configuration - -Advanced configuration for fine-tuning the processor. **All properties under `advanced` apply to the Flink processor only and are silently ignored by the classic processor.** - - -**Properties** - -|Name|Type|Description|Required| -|----|----|-----------|--------| -|[**source**](#processorsadvancedsource)
(Advanced source settings)|`object`|Advanced configuration properties for the source Redis client and streams reader. **Flink processor only.**
|| -|[**target**](#processorsadvancedtarget)
(Advanced target settings)|`object`|Advanced configuration properties for the target Redis client and sink. **Flink processor only.**
|| -|[**dlq**](#processorsadvanceddlq)
(Advanced DLQ settings)|`object`|Advanced configuration properties for the DLQ Redis client and sink. **Flink processor only.**
|| -|[**processor**](#processorsadvancedprocessor)
(Advanced processor settings)|`object`|Advanced configuration properties for the processor. **Flink processor only.**
|| -|[**flink**](#processorsadvancedflink)
(Advanced Flink settings)|`object`|Advanced configuration properties forwarded to the underlying Flink runtime. Any property listed in the [Flink configuration documentation](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/config/) can be set here and will override the RDI default. **Flink processor only.**
|| -|[**resources**](#processorsadvancedresources)
(Advanced resource settings)|`object`|Compute resources allocated to the Flink job, such as the number of task manager pods. **Flink processor only.**
|| - -**Additional Properties:** not allowed -**Minimal Properties:** 1 -**Example** +  + -```yaml -source: - stream.name.pattern: data:* - discovery.interval.ms: 1000 - batch.size: 2000 - batch.timeout.ms: 100 - connection.timeout.ms: 2000 - socket.timeout.ms: 2000 - retry.max.attempts: 5 - retry.initial.delay.ms: 100 - retry.max.delay.ms: 3000 - retry.backoff.multiplier: 2 -target: - batch.size: 200 - flush.interval.ms: 100 - connection.timeout.ms: 2000 - socket.timeout.ms: 2000 - retry.max.attempts: 5 - retry.initial.delay.ms: 1000 - retry.max.delay.ms: 10000 - retry.backoff.multiplier: 2 - wait.enabled: false - wait.write.timeout.ms: 1000 - wait.retry.enabled: true - wait.retry.delay.ms: 1000 -dlq: - max.len: 1000 - batch.size: 100 - flush.interval.ms: 100 - connection.timeout.ms: 2000 - socket.timeout.ms: 2000 - retry.max.attempts: 1 - retry.initial.delay.ms: 100 - retry.max.delay.ms: 3000 - retry.backoff.multiplier: 2 - wait.enabled: false - wait.write.timeout.ms: 1000 - wait.retry.enabled: false - wait.retry.delay.ms: 1000 -processor: - default.data.type: hash - default.json.update.strategy: replace - dlq.enabled: true -flink: - taskmanager.numberOfTaskSlots: 1 - taskmanager.memory.process.size: 2048m -resources: - taskManager: {} +Advanced configuration properties for the Quarkus runtime that hosts Debezium Server. **Only applies to the `cdc` collector type.** See the [Debezium Server documentation](https://debezium.io/documentation/reference/stable/operations/debezium-server.html) for runtime configuration options. When using a property from that page, omit the `quarkus.` prefix. -``` +**Additional Properties** - -#### processors\.advanced\.source: Advanced source settings +| Name | Type | Description | Required | +| ------------------------- | ----------------------------- | ----------- | -------- | +| **Additional Properties** | `string`, `number`, `boolean` | | | -Advanced configuration properties for the source Redis client and streams reader. **Flink processor only.** +**Minimal Properties:** 1  +  + +Compute resources allocated to the collector. **Only applies to the `cdc` collector type.** **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**stream\.name\.pattern**
(Source stream name pattern)|`string`|Glob pattern used to discover input streams in the source Redis database, for example `data:*`.
Default: `"data:*"`
|| -|**discovery\.interval\.ms**
(Stream discovery interval)|`integer`|Time in milliseconds between checks for new input streams. Replaces the classic `processors.idle_streams_check_interval_ms` and `processors.busy_streams_check_interval_ms` properties.
Default: `1000`
Minimum: `0`
|| -|**batch\.size**
(Source batch size)|`integer`|Maximum number of records the source operator reads in a single batch. Alias for `processors.read_batch_size`; takes priority when both are set.
Default: `2000`
Minimum: `1`
|| -|**batch\.timeout\.ms**
(Source batch timeout)|`integer`|Maximum time in milliseconds to wait for a source batch to fill before processing. Alias for `processors.read_batch_timeout_ms`; takes priority when both are set.
Default: `100`
Minimum: `1`
|| -|**connection\.timeout\.ms**
(Source connection timeout)|`integer`|Connection timeout in milliseconds for the source Redis client.
Default: `2000`
Minimum: `1`
|| -|**socket\.timeout\.ms**
(Source socket timeout)|`integer`|Socket read/write timeout in milliseconds for the source Redis client.
Default: `2000`
Minimum: `1`
|| -|**retry\.max\.attempts**
(Source retry max attempts)|`integer`|Maximum number of retry attempts for failed source Redis operations.
Default: `5`
Minimum: `1`
|| -|**retry\.initial\.delay\.ms**
(Source retry initial delay)|`integer`|Initial delay in milliseconds before the first retry of a failed source Redis operation.
Default: `100`
Minimum: `1`
|| -|**retry\.max\.delay\.ms**
(Source retry max delay)|`integer`|Maximum delay in milliseconds between retry attempts for source Redis operations.
Default: `3000`
Minimum: `1`
|| -|**retry\.backoff\.multiplier**
(Source retry backoff multiplier)|`number`|Exponential backoff multiplier between retry attempts for source Redis operations.
Default: `2`
Minimum: `1`
|| +| Name | Type | Description | Required | +| -------------------------------------- | -------- | ------------------------------------------------------------------------------- | -------- | +| **cpu**
(CPU resource value) | `string` | CPU request for the collector container, for example `1` or `500m`.
| | +| **memory**
(Memory resource value) | `string` | Memory request for the collector container, for example `1024Mi` or `2Gi`.
| | -**Additional Properties** +**Additional Properties:** not allowed  +**Minimal Properties:** 1  +  + -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**Additional Properties**|`string`, `number`, `boolean`||| +Advanced configuration properties for the RIOT-X Snowflake collector. **Only applies to the `riotx` collector type.** + +**Properties** -**Minimal Properties:** 1 +| Name | Type | Description | Required | +| ------------------------------------------------------------------- | ---------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **poll**
(Polling interval) | `string` | Interval between polls for new stream changes, for example `30s` or `PT30S`.
Default: `"30s"`
| | +| **snapshot**
(Snapshot mode) | `string` | Initial-load behavior. `INITIAL` performs a one-time snapshot before streaming; `NEVER` skips the snapshot.
Default: `"INITIAL"`
Enum: `"INITIAL"`, `"NEVER"`
| | +| **streamPrefix**
(Redis stream key prefix) | `string` | Prefix used when constructing Redis stream keys, for example `data:`.
Default: `"data:"`
| | +| **streamLimit**
(Maximum stream length) | `integer` | Maximum number of entries kept in each Redis stream before older entries are trimmed.
Minimum: `1`
| | +| [**keyColumns**](#sourcesadvancedriotxkeycolumns)
(Key columns) | `string[]` | Deprecated RIOTX global fallback list of columns to use as message keys for every captured table. Prefer `tables..keys`
| | +| **clearOffset**
(Clear existing offset) | `boolean` | When `true`, the stored offset is cleared on startup, forcing a fresh read.
Default: `false`
| | +| **count**
(Record count limit) | `integer` | Maximum number of records to process. Set to `0` for unlimited.
Default: `0`
Minimum: `0`
| | + +**Additional Properties:** not allowed  +**Minimal Properties:** 1  **Example** ```yaml -stream.name.pattern: data:* -discovery.interval.ms: 1000 -batch.size: 2000 -batch.timeout.ms: 100 -connection.timeout.ms: 2000 -socket.timeout.ms: 2000 -retry.max.attempts: 5 -retry.initial.delay.ms: 100 -retry.max.delay.ms: 3000 -retry.backoff.multiplier: 2 - +poll: 30s +snapshot: INITIAL +streamPrefix: "data:" +clearOffset: false +count: 0 ``` - -#### processors\.advanced\.target: Advanced target settings +  + -Advanced configuration properties for the target Redis client and sink. **Flink processor only.** +Deprecated RIOTX global fallback list of columns to use as message keys for every captured table. Prefer `tables..keys` + +  + +Advanced configuration properties for the RDI Collector stream writer connection and behaviour. **Applies to the `cdc` and `flink` collector types.**

For the `cdc` collector type, see the full list of properties at [Debezium Server — Redis Stream sink](https://debezium.io/documentation/reference/stable/operations/debezium-server.html#_redis_stream). When using a property from that page, omit the `debezium.sink.` prefix.

**The properties listed below only apply to the `flink` collector type.** **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**batch\.size**
(Target sink batch size)|`integer`|Maximum number of records the target sink writes in a single batch. Alias for `processors.write_batch_size`; takes priority when both are set.
Default: `200`
Minimum: `1`
|| -|**flush\.interval\.ms**
(Target sink flush interval)|`integer`|Maximum time in milliseconds the target sink waits to fill a batch before flushing it to Redis.
Default: `100`
Minimum: `1`
|| -|**connection\.timeout\.ms**
(Target connection timeout)|`integer`|Connection timeout in milliseconds for the target Redis client.
Default: `2000`
Minimum: `1`
|| -|**socket\.timeout\.ms**
(Target socket timeout)|`integer`|Socket read/write timeout in milliseconds for the target Redis client.
Default: `2000`
Minimum: `1`
|| -|**retry\.max\.attempts**
(Target retry max attempts)|`integer`|Maximum number of retry attempts for failed target Redis operations. Alias for `processors.retry_max_attempts`; takes priority when both are set.
Default: `5`
Minimum: `1`
|| -|**retry\.initial\.delay\.ms**
(Target retry initial delay)|`integer`|Initial delay in milliseconds before the first retry of a failed target Redis operation. Alias for `processors.retry_initial_delay_ms`; takes priority when both are set.
Default: `1000`
Minimum: `1`
|| -|**retry\.max\.delay\.ms**
(Target retry max delay)|`integer`|Maximum delay in milliseconds between retry attempts for target Redis operations. Alias for `processors.retry_max_delay_ms`; takes priority when both are set.
Default: `10000`
Minimum: `1`
|| -|**retry\.backoff\.multiplier**
(Target retry backoff multiplier)|`number`|Exponential backoff multiplier between retry attempts for target Redis operations.
Default: `2`
Minimum: `1`
|| -|**wait\.enabled**
(Target replica wait enabled)|`boolean`|When `true`, RDI verifies that each write has been replicated to the target database's replica shards before acknowledging it. Enable this only when target database replication is enabled and a healthy replica is available. Alias for `processors.wait_enabled`; takes priority when both are set.
Default: `false`
|| -|**wait\.write\.timeout\.ms**
(Target replica wait timeout)|`integer`|Maximum time in milliseconds to wait for target replica write verification. Alias for `processors.wait_timeout`; takes priority when both are set.
Default: `1000`
Minimum: `1`
|| -|**wait\.retry\.enabled**
(Target replica wait retry enabled)|`boolean`|When `true`, RDI keeps retrying a target write until replica replication is confirmed; when `false`, it gives up after the first failure. Alias for `processors.retry_on_replica_failure`; takes priority when both are set. When enabled, the Flink processor retries indefinitely. Failed checkpoints can restart the job, after which the retries resume. The classic processor retries once.
Default: `true`
|| -|**wait\.retry\.delay\.ms**
(Target replica wait retry delay)|`integer`|Delay in milliseconds between target replica wait retry attempts.
Default: `1000`
Minimum: `1`
|| +| Name | Type | Description | Required | +| ---------------------------------------------------------------------------------- | --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **redis\.batch\.size**
(Sink batch size) | `integer` | Maximum number of records the collector sink writes to Redis in a single batch.
Default: `1000`
Minimum: `1`
| | +| **redis\.flush\.interval\.ms**
(Sink flush interval) | `integer` | Maximum time in milliseconds the collector sink waits to fill a batch before flushing it to Redis.
Default: `100`
Minimum: `1`
| | +| **redis\.connection\.timeout\.ms**
(Sink connection timeout) | `integer` | Connection timeout in milliseconds for the target Redis client used by the collector sink.
Default: `2000`
Minimum: `1`
| | +| **redis\.socket\.timeout\.ms**
(Sink socket timeout) | `integer` | Socket read/write timeout in milliseconds for the target Redis client used by the collector sink.
Default: `2000`
Minimum: `1`
| | +| **redis\.retry\.max\.attempts**
(Sink retry max attempts) | `integer` | Maximum number of retry attempts for failed Redis operations.
Default: `5`
Minimum: `1`
| | +| **redis\.retry\.initial\.delay\.ms**
(Sink retry initial delay) | `integer` | Initial delay in milliseconds before the first retry of a failed Redis operation.
Default: `100`
Minimum: `1`
| | +| **redis\.retry\.max\.delay\.ms**
(Sink retry max delay) | `integer` | Maximum delay in milliseconds between retry attempts for failed Redis operations.
Default: `3000`
Minimum: `1`
| | +| **redis\.retry\.backoff\.multiplier**
(Sink retry backoff multiplier) | `number` | Exponential backoff multiplier between retry attempts for failed Redis operations.
Default: `2`
Minimum: `1`
| | +| **redis\.oom\.retry\.initial\.delay\.ms**
(Sink OOM retry initial delay) | `integer` | Initial delay in milliseconds before the first retry after a Redis out-of-memory error.
Default: `1000`
Minimum: `1`
| | +| **redis\.oom\.retry\.max\.delay\.ms**
(Sink OOM retry max delay) | `integer` | Maximum delay in milliseconds between retry attempts after a Redis out-of-memory error.
Default: `10000`
Minimum: `1`
| | +| **redis\.oom\.retry\.backoff\.multiplier**
(Sink OOM retry backoff multiplier) | `number` | Exponential backoff multiplier between retry attempts after a Redis out-of-memory error.
Default: `2`
Minimum: `1`
| | +| **redis\.wait\.enabled**
(Sink replica wait enabled) | `boolean` | When `true`, the collector verifies that each write has been replicated to the configured number of Redis replica shards before acknowledging it.
Default: `false`
| | +| **redis\.wait\.write\.timeout\.ms**
(Sink replica wait timeout) | `integer` | Maximum time in milliseconds to wait for replica write acknowledgements.
Default: `1000`
Minimum: `1`
| | +| **redis\.wait\.retry\.enabled**
(Sink replica wait retry enabled) | `boolean` | When `true`, the collector keeps retrying a write until replica acknowledgement succeeds; when `false`, it gives up after the first failure.
Default: `false`
| | +| **redis\.wait\.retry\.delay\.ms**
(Sink replica wait retry delay) | `integer` | Delay in milliseconds between replica wait retry attempts.
Default: `1000`
Minimum: `1`
| | **Additional Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**Additional Properties**|`string`, `number`, `boolean`||| +| Name | Type | Description | Required | +| ------------------------- | ----------------------------- | ----------- | -------- | +| **Additional Properties** | `string`, `number`, `boolean` | | | -**Minimal Properties:** 1 +**Minimal Properties:** 1  **Example** ```yaml -batch.size: 200 -flush.interval.ms: 100 -connection.timeout.ms: 2000 -socket.timeout.ms: 2000 -retry.max.attempts: 5 -retry.initial.delay.ms: 1000 -retry.max.delay.ms: 10000 -retry.backoff.multiplier: 2 -wait.enabled: false -wait.write.timeout.ms: 1000 -wait.retry.enabled: true -wait.retry.delay.ms: 1000 - +redis.batch.size: 1000 +redis.flush.interval.ms: 100 +redis.connection.timeout.ms: 2000 +redis.socket.timeout.ms: 2000 +redis.retry.max.attempts: 5 +redis.retry.initial.delay.ms: 100 +redis.retry.max.delay.ms: 3000 +redis.retry.backoff.multiplier: 2 +redis.oom.retry.initial.delay.ms: 1000 +redis.oom.retry.max.delay.ms: 10000 +redis.oom.retry.backoff.multiplier: 2 +redis.wait.enabled: false +redis.wait.write.timeout.ms: 1000 +redis.wait.retry.enabled: false +redis.wait.retry.delay.ms: 1000 ``` - -#### processors\.advanced\.dlq: Advanced DLQ settings - -Advanced configuration properties for the DLQ Redis client and sink. **Flink processor only.** +  + +Advanced configuration properties for the source database connection and CDC behavior. **Applies to the `cdc` and `flink` collector types.**

For the `cdc` collector type, available properties depend on the source database — refer to the relevant Debezium connector documentation: [MySQL](https://debezium.io/documentation/reference/stable/connectors/mysql.html), [MariaDB](https://debezium.io/documentation/reference/stable/connectors/mariadb.html), [PostgreSQL](https://debezium.io/documentation/reference/stable/connectors/postgresql.html), [Oracle](https://debezium.io/documentation/reference/stable/connectors/oracle.html), [SQL Server](https://debezium.io/documentation/reference/stable/connectors/sqlserver.html), [Db2](https://debezium.io/documentation/reference/stable/connectors/db2.html), [MongoDB](https://debezium.io/documentation/reference/stable/connectors/mongodb.html). When using a property from those pages, omit the `debezium.source.` prefix.

**The named properties below cover the most commonly tuned settings: `spanner.*` properties apply to the `flink` collector type, all others apply to the `cdc` collector type. Any other property from the Debezium documentation can still be set as a free-form key-value pair.** **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**max\.len**
(DLQ sink max length)|`integer`|Maximum number of messages stored per dead letter queue stream. Alias for `processors.dlq_max_messages`; takes priority when both are set.
Default: `1000`
Minimum: `1`
|| -|**batch\.size**
(DLQ sink batch size)|`integer`|Maximum number of records the DLQ sink writes in a single batch.
Default: `100`
Minimum: `1`
|| -|**flush\.interval\.ms**
(DLQ sink flush interval)|`integer`|Maximum time in milliseconds the DLQ sink waits to fill a batch before flushing it to Redis.
Default: `100`
Minimum: `1`
|| -|**connection\.timeout\.ms**
(DLQ connection timeout)|`integer`|Connection timeout in milliseconds for the DLQ Redis client.
Default: `2000`
Minimum: `1`
|| -|**socket\.timeout\.ms**
(DLQ socket timeout)|`integer`|Socket read/write timeout in milliseconds for the DLQ Redis client.
Default: `2000`
Minimum: `1`
|| -|**retry\.max\.attempts**
(DLQ retry max attempts)|`integer`|Maximum number of retry attempts for failed DLQ Redis operations.
Default: `1`
Minimum: `1`
|| -|**retry\.initial\.delay\.ms**
(DLQ retry initial delay)|`integer`|Initial delay in milliseconds before the first retry of a failed DLQ Redis operation.
Default: `100`
Minimum: `1`
|| -|**retry\.max\.delay\.ms**
(DLQ retry max delay)|`integer`|Maximum delay in milliseconds between retry attempts for DLQ Redis operations.
Default: `3000`
Minimum: `1`
|| -|**retry\.backoff\.multiplier**
(DLQ retry backoff multiplier)|`number`|Exponential backoff multiplier between retry attempts for DLQ Redis operations.
Default: `2`
Minimum: `1`
|| -|**wait\.enabled**
(DLQ replica wait enabled)|`boolean`|When `true`, RDI verifies that each DLQ write has been replicated to the DLQ database's replica shards before acknowledging it.
Default: `false`
|| -|**wait\.write\.timeout\.ms**
(DLQ replica wait timeout)|`integer`|Maximum time in milliseconds to wait for DLQ replica write verification.
Default: `1000`
Minimum: `1`
|| -|**wait\.retry\.enabled**
(DLQ replica wait retry enabled)|`boolean`|When `true`, RDI keeps retrying a DLQ write until replica replication is confirmed; when `false`, it gives up after the first failure.
Default: `false`
|| -|**wait\.retry\.delay\.ms**
(DLQ replica wait retry delay)|`integer`|Delay in milliseconds between DLQ replica wait retry attempts.
Default: `1000`
Minimum: `1`
|| +| Name | Type | Description | Required | +| ------------------------------------------------------------------------------------- | --------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **record\.processing\.threads**
(Record processing threads) | `integer` | Controls how many worker threads process captured records before they are written downstream.
Minimum: `1`
| | +| **snapshot\.max\.threads**
(Snapshot max threads) | `integer` | Sets the maximum number of threads used while taking the initial snapshot.
Default: `1`
Minimum: `1`
| | +| **poll\.interval\.ms**
(Poll interval ms) | `integer` | Defines how often the collector polls the source for new changes.
Default: `500`
Minimum: `1`
| | +| **snapshot\.fetch\.size**
(Snapshot fetch size) | `integer` | Defines how many rows are fetched per batch during the initial snapshot.
Default: `10000`
Minimum: `1`
| | +| **max\.batch\.size**
(Max batch size) | `integer` | Caps how many records are processed together in a single batch.
Default: `2048`
Minimum: `1`
| | +| **max\.queue\.size**
(Max queue size) | `integer` | Limits how many records can be buffered in memory before processing catches up.
Default: `8192`
Minimum: `1`
| | +| **heartbeat\.interval\.ms**
(Heartbeat interval ms) | `integer` | Sets how often heartbeat events are emitted to keep change tracking active. Use 0 to disable them.
Default: `0`
Minimum: `0`
| | +| **heartbeat\.action\.query**
(Heartbeat action query) | `string` | SQL query executed on the source whenever a heartbeat is emitted.
| | +| **lob\.enabled**
(Lob enabled) | `boolean` | Determines whether large object columns are included in change capture.
Default: `false`
| | +| **gtid\.source\.includes**
(GTID source includes) | `string` | Restricts MySQL GTID processing to the listed source UUIDs.
| | +| **publication\.autocreate\.mode**
(Publication autocreate mode) | `string` | Controls whether and how the PostgreSQL publication is created or updated automatically.
Default: `"all_tables"`
Enum: `"all_tables"`, `"filtered"`, `"disabled"`
| | +| **publication\.name**
(Publication name) | `string` | Sets the PostgreSQL logical replication publication name used by the collector.
Default: `"dbz_publication"`
| | +| **slot\.name**
(Slot name) | `string` | Sets the PostgreSQL replication slot name the collector reads from.
Default: `"debezium"`
| | +| **spanner\.version\.retention\.period\.hours**
(Spanner version retention period) | `integer` | Retention period in hours for Spanner change stream versions. Determines how far back the collector can resume after an outage.
Default: `1`
Minimum: `1`
| | +| **spanner\.fetch\.timeout\.ms**
(Spanner fetch timeout) | `integer` | Timeout in milliseconds for a single change stream fetch request to Spanner.
Default: `500`
Minimum: `1`
| | +| **spanner\.fetch\.heartbeat\.ms**
(Spanner fetch heartbeat interval) | `integer` | Interval in milliseconds at which Spanner sends heartbeat records when no data changes are available.
Default: `100`
Minimum: `1`
| | +| **spanner\.max\.rows\.per\.partition**
(Spanner max rows per partition) | `integer` | Maximum number of rows the collector reads from a single Spanner change stream partition before yielding.
Default: `10000`
Minimum: `1`
| | +| **spanner\.dialect**
(Spanner SQL dialect) | `string` | SQL dialect of the Spanner database. Use `GOOGLESQL` for Google Standard SQL or `POSTGRESQL` for the PostgreSQL interface.
Default: `"GOOGLESQL"`
Enum: `"GOOGLESQL"`, `"POSTGRESQL"`
| | **Additional Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**Additional Properties**|`string`, `number`, `boolean`||| +| Name | Type | Description | Required | +| ------------------------- | ----------------------------- | ----------- | -------- | +| **Additional Properties** | `string`, `number`, `boolean` | | | -**Minimal Properties:** 1 +**Minimal Properties:** 1  **Example** ```yaml -max.len: 1000 -batch.size: 100 -flush.interval.ms: 100 -connection.timeout.ms: 2000 -socket.timeout.ms: 2000 -retry.max.attempts: 1 -retry.initial.delay.ms: 100 -retry.max.delay.ms: 3000 -retry.backoff.multiplier: 2 -wait.enabled: false -wait.write.timeout.ms: 1000 -wait.retry.enabled: false -wait.retry.delay.ms: 1000 - +snapshot.max.threads: 1 +poll.interval.ms: 500 +snapshot.fetch.size: 10000 +max.batch.size: 2048 +max.queue.size: 8192 +heartbeat.interval.ms: 0 +lob.enabled: false +publication.autocreate.mode: all_tables +publication.name: dbz_publication +slot.name: debezium +spanner.version.retention.period.hours: 1 +spanner.fetch.timeout.ms: 500 +spanner.fetch.heartbeat.ms: 100 +spanner.max.rows.per.partition: 10000 +spanner.dialect: GOOGLESQL ``` - -#### processors\.advanced\.processor: Advanced processor settings - -Advanced configuration properties for the processor. **Flink processor only.** +  + +Database names to capture from the source database. Maps to the underlying connector's `database.include.list`. Applies only to MySQL, MariaDB, and MongoDB connections. -**Properties** +  + -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**default\.data\.type**
(Default target data type)|`string`|Data type to use in Redis when not overridden per job: `hash` for Redis Hash, `json` for RedisJSON. Alias for `processors.target_data_type`; takes priority when both are set.
Default: `"hash"`
Enum: `"hash"`, `"json"`
|| -|**default\.json\.update\.strategy**
(Default JSON update strategy)|`string`|Strategy for updating JSON data in Redis: `replace` to overwrite the entire JSON object, `merge` to merge new data with the existing JSON object. Alias for `processors.json_update_strategy`; takes priority when both are set.
Default: `"replace"`
Enum: `"replace"`, `"merge"`
|| -|**dlq\.enabled**
(Enable DLQ)|`boolean`|When `true`, rejected messages are stored in the dead-letter queue; when `false`, errors are silently skipped. Alias for `processors.error_handling`; takes priority when both are set.
Default: `true`
|| +Logging settings for this source collector. -**Additional Properties** +**Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**Additional Properties**|`string`, `number`, `boolean`||| +| Name | Type | Description | Required | +| ----------------------------- | -------- | -------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **level**
(Logging level) | `string` | Log verbosity for the source collector.
Default: `"info"`
Enum: `"trace"`, `"debug"`, `"info"`, `"warn"`, `"error"`
| | -**Minimal Properties:** 1 +**Additional Properties:** not allowed  **Example** ```yaml -default.data.type: hash -default.json.update.strategy: replace -dlq.enabled: true - +level: info ``` - -#### processors\.advanced\.flink: Advanced Flink settings +  + -Advanced configuration properties forwarded to the underlying Flink runtime. Any property listed in the [Flink configuration documentation](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/config/) can be set here and will override the RDI default. **Flink processor only.**

The properties listed below are the ones most likely to require adjustment. **Changing any other Flink property is not recommended unless instructed by Redis support.** +Schema names to capture from the source database. Maps to the underlying connector's `schema.include.list`. + +  + +Tables to capture from the source database, keyed by table name. The value configures column selection and key handling for that table. + +**Additional Properties** + +| Name | Type | Description | Required | +| --------------------------------------------------------------- | ---------------- | ----------- | -------- | +| [**Additional Properties**](#sourcestablesadditionalproperties) | `object`, `null` | | | + +**Minimal Properties:** 1  +  + **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**parallelism\.default**
(Default parallelism)|`integer`|Default parallelism for jobs and operators. When unset, Flink uses the number of available task slots across all task managers (`taskManager.replicas × taskmanager.numberOfTaskSlots`). Increase to fan out work across more task slots; see [parallel execution](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/dev/datastream/execution/parallel/).
Minimum: `1`
|| -|**taskmanager\.numberOfTaskSlots**
(Task slots per task manager)|`integer`|Number of parallel task slots per task manager pod. Each slot can run one parallel pipeline instance, so this caps the parallelism a single task manager can absorb. See [task slots and resources](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/concepts/flink-architecture/#task-slots-and-resources).
Default: `1`
Minimum: `1`
|| -|**taskmanager\.memory\.process\.size**
(Task manager process memory)|`string`|Total memory budget for each task manager JVM process (heap + managed + network + metaspace + JVM overhead), expressed with a unit suffix such as `2048m` or `4g`. See [task manager memory configuration](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/deployment/memory/mem_setup_tm/).
Default: `"2048m"`
|| +| Name | Type | Description | Required | +| -------------------------------------------------------------------------------------------------- | ---------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **snapshot\_sql** | `string` | Custom SQL statement used during the initial snapshot, giving fine-grained control over the data captured.
| | +| [**columns**](#sourcestablesadditionalpropertiescolumns)
(Columns to capture) | `string[]` | List of specific columns to capture for changes. If not specified, all columns will be captured. For RIOTX Snowflake sources, this is rendered as per-table `--table-columns` projection. Note: This property cannot be used for MongoDB connections
| | +| [**exclude\_columns**](#sourcestablesadditionalpropertiesexclude_columns)
(Columns to exclude) | `string[]` | Specific columns to exclude from capture. When omitted, no columns are excluded. Only supported for MongoDB connections.
| | +| [**keys**](#sourcestablesadditionalpropertieskeys)
(Message keys) | `string[]` | Optional list of columns to use as a composite unique identifier. For RIOTX Snowflake sources, this is rendered as per-table `--table-keys`. Only required when the table lacks a primary key or unique constraint. Must form a unique combination of fields
| | -**Additional Properties** +**Additional Properties:** not allowed  +  + -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**Additional Properties**|`string`, `number`, `boolean`||| +List of specific columns to capture for changes. If not specified, all columns will be captured. For RIOTX Snowflake sources, this is rendered as per-table `--table-columns` projection. Note: This property cannot be used for MongoDB connections -**Minimal Properties:** 1 -**Example** +  + -```yaml -taskmanager.numberOfTaskSlots: 1 -taskmanager.memory.process.size: 2048m +Specific columns to exclude from capture. When omitted, no columns are excluded. Only supported for MongoDB connections. -``` +  + - -#### processors\.advanced\.resources: Advanced resource settings +Optional list of columns to use as a composite unique identifier. For RIOTX Snowflake sources, this is rendered as per-table `--table-keys`. Only required when the table lacks a primary key or unique constraint. Must form a unique combination of fields -Compute resources allocated to the Flink job, such as the number of task manager pods. **Flink processor only.** +  + +Connection configuration for a non-Redis source database. The exact set of properties depends on the database type. **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|[**taskManager**](#processorsadvancedresourcestaskmanager)
(Task manager resource settings)|`object`|Resource settings for Flink task manager pods.
|| +| Name | Type | Description | Required | +| ---------------------------------------------------------------------------------- | -------- | ------------------------------------------------------------------ | -------- | +| [**SQL database**](#definitionsnon_redis_connectionsqldatabase)
(SQL database) | `object` | Connection configuration for a supported SQL database.
| | +| [**MongoDB**](#definitionsnon_redis_connectionmongodb) | `object` | Connection configuration for a MongoDB database.
| yes | +| [**Spanner**](#definitionsnon_redis_connectionspanner) | `object` | Connection configuration for a Google Cloud Spanner database.
| yes | +| [**Snowflake**](#definitionsnon_redis_connectionsnowflake) | `object` | Connection configuration for a Snowflake database.
| yes | -**Additional Properties:** not allowed -**Minimal Properties:** 1 **Example** ```yaml -taskManager: {} - +SQL database: + hr: + type: postgresql + host: localhost + port: 5432 + database: postgres + user: postgres + password: postgres +MongoDB: + mongodb-source: + type: mongodb + connection_string: mongodb://localhost:27017/?replicaSet=rs0 + user: debezium + password: dbz + database: db1,db2 +Spanner: + spanner-source: + type: spanner + project_id: example-12345 + instance_id: example + database_id: example + change_streams: + change_stream_all: + retention_period_hours: 24 +Snowflake: + snowflake: + type: snowflake + url: jdbc:snowflake://myaccount.snowflakecomputing.com/ + user: myuser + password: mypassword + database: MYDB + warehouse: COMPUTE_WH ``` - -##### processors\.advanced\.resources\.taskManager: Task manager resource settings - -Resource settings for Flink task manager pods. +  + +Connection configuration for a MongoDB database. **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**replicas**
(Task manager replicas)|`integer`|Number of Flink task manager pods to run.
Minimum: `1`
|| +| Name | Type | Description | Required | +| ------------------------------------ | -------- | ---------------------------------------------------------------------------------------------------- | -------- | +| **type**
(Database type) | `string` | Database type identifier. Always `mongodb` for this connection.
Constant Value: `"mongodb"`
| yes | +| **connection\_string** | `string` | MongoDB connection URI including host, port, and any connection options.
| yes | +| **user**
(MongoDB user) | `string` | Username for authentication to MongoDB.
| no | +| **password**
(MongoDB password) | `string` | Password for authentication to MongoDB.
| no | +| **database**
(MongoDB databases) | `string` | Comma-separated list of MongoDB databases to monitor.
| no | -**Additional Properties:** not allowed -**Minimal Properties:** 1 - -## secret\-providers: Secret providers +**Additional Properties:** not allowed  +**Example** -External secret providers used to resolve `${...}` references in the configuration. +```yaml +mongodb-source: + type: mongodb + connection_string: mongodb://localhost:27017/?replicaSet=rs0 + user: debezium + password: dbz + database: db1,db2 +``` +  + -**Properties** (key: `.*`) +Connection configuration for a supported SQL database. -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**type**
(Provider type)|`string`|Secret provider backend. `aws` uses AWS Secrets Manager; `vault` uses HashiCorp Vault.
Enum: `"aws"`, `"vault"`
|yes| -|[**parameters**](#secret-providersparameters)
(Provider parameters)|`object`|Configuration parameters for the secret provider.
|yes| +**Properties** +| Name | Type | Description | Required | +| ------------------------------------ | --------- | ------------------------------------------------------------------------------------------------------ | -------- | +| **type**
(Database type) | `string` | SQL database engine.
Enum: `"mariadb"`, `"mysql"`, `"oracle"`, `"postgresql"`, `"sqlserver"`
| | +| **host**
(Database host) | `string` | Hostname or IP address of the SQL database server.
| | +| **port**
(Database port) | `integer` | Network port on which the SQL database server is listening.
Minimum: `1`
Maximum: `65535`
| | +| **database**
(Database name) | `string` | Name of the database to connect to.
| | +| **user**
(Database user) | `string` | Username for authentication to the SQL database.
| | +| **password**
(Database password) | `string` | Password for authentication to the SQL database.
| | - -### secret\-providers\.parameters: Provider parameters +**Additional Properties:** not allowed  +**Example** -Configuration parameters for the secret provider. +```yaml +hr: + type: postgresql + host: localhost + port: 5432 + database: postgres + user: postgres + password: postgres +``` +**Example** -**Properties** +```yaml +my-oracle: + type: oracle + host: 172.17.0.4 + port: 1521 + user: c##dbzuser + password: dbz +``` + +  + -|Name|Type|Description|Required| -|----|----|-----------|--------| -|[**objects**](#secret-providersparametersobjects)
(Secrets objects array)|`object[]`|Secret objects to fetch from the provider.
|yes| +Connection configuration for a Snowflake database. + +**Properties** +| Name | Type | Description | Required | +| --------------------------------------- | -------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **type**
(Database type) | `string` | Database type identifier. Always `snowflake` for this connection.
Constant Value: `"snowflake"`
| yes | +| **url**
(JDBC URL) | `string` | Snowflake JDBC connection URL, for example `jdbc:snowflake://account.snowflakecomputing.com/`.
| yes | +| **user**
(Snowflake user) | `string` | Username for authentication to Snowflake.
| yes | +| **password**
(Snowflake password) | `string` | Password for authentication to Snowflake. For key-pair authentication, omit this field and provide the private key via the `source-db-ssl` secret (`client.key` field).
| no | +| **database**
(Snowflake database) | `string` | Name of the Snowflake database to connect to.
| yes | +| **warehouse**
(Snowflake warehouse) | `string` | Name of the Snowflake warehouse used for compute.
| yes | +| **role**
(Snowflake role) | `string` | Snowflake role used for the connection.
| no | +| **cdcDatabase**
(CDC database) | `string` | Database hosting the CDC streams. Defaults to the main `database` if not set.
| no | +| **cdcSchema**
(CDC schema) | `string` | Schema hosting the CDC streams. Defaults to the main schema if not set.
| no | + +**Additional Properties:** not allowed  **Example** ```yaml -objects: - - {} - +snowflake: + type: snowflake + url: jdbc:snowflake://myaccount.snowflakecomputing.com/ + user: myuser + password: mypassword + database: MYDB + warehouse: COMPUTE_WH ``` - -#### secret\-providers\.parameters\.objects\[\]: Secrets objects array - -Secret objects to fetch from the provider. - +  + -**Items: Secret object** +Connection configuration for a Google Cloud Spanner database. -**No properties.** +**Properties** +| Name | Type | Description | Required | +| --------------------------------------------------------------------------------------------------------------- | --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | +| **type**
(Database type) | `string` | Database type identifier. Always `spanner` for this connection.
Constant Value: `"spanner"`
| yes | +| **project\_id**
(Spanner project ID) | `string` | Google Cloud project ID that hosts the Spanner instance.
| yes | +| **instance\_id**
(Spanner instance ID) | `string` | Spanner instance identifier within the project.
| yes | +| **database\_id**
(Spanner database ID) | `string` | Spanner database identifier within the instance.
| yes | +| **emulator\_host**
(Spanner emulator host) | `string` | Host and port of the Spanner emulator. Used for local development; leave unset against real Spanner.
| no | +| **use\_credentials\_file** | `boolean` | When `true`, RDI authenticates using a service account credentials file; when `false`, it uses application default credentials.
Default: `false`
| no | +| [**change\_streams**](#definitionsnon_redis_connectionspannerchange_streams)
(Change streams configuration) | `object` | Spanner change streams to capture, keyed by change stream name.
| yes | + +**Additional Properties:** not allowed  **Example** ```yaml -- {} - +spanner-source: + type: spanner + project_id: example-12345 + instance_id: example + database_id: example + change_streams: + change_stream_all: + retention_period_hours: 24 ``` - -## metadata: Pipeline metadata +  + -Optional metadata describing this pipeline, such as a display name and description. +Spanner change streams to capture, keyed by change stream name. + +**Additional Properties** +| Name | Type | Description | Required | +| ------------------------------------------------------------------------------------------------------ | ---------------- | ----------- | -------- | +| [**Additional Properties**](#definitionsnon_redis_connectionspannerchange_streamsadditionalproperties) | `object`, `null` | | | + +**Minimal Properties:** 1  +  + **Properties** -|Name|Type|Description|Required| -|----|----|-----------|--------| -|**name**
(Pipeline name)|`string`|Human-readable name for the pipeline. Maximum 100 characters.
Maximal Length: `100`
|| -|**description**
(Pipeline description)|`string`|Free-form description of what the pipeline does. Maximum 500 characters.
Maximal Length: `500`
|| -|**revision**
(Pipeline revision)|`integer`|Pipeline revision number. Must be a non-negative integer.
Minimum: `0`
|| -|[**tags**](#metadatatags)
(Pipeline tags)|`string[]`|Array of pipeline tags. Each tag must be a string of up to 50 characters, containing only alphanumeric characters, dots, dashes, or underscores, and must start and end with an alphanumeric character. Tags must be unique.
|| +| Name | Type | Description | Required | +| ----------------------------------------------------------------------- | ------------------- | ----------------------------------------------------------------------------------------------- | -------- | +| **retention\_period\_hours**
(Change stream retention period hours) | `integer`, `string` | Retention period for the change stream, in hours.
Pattern: `^\${.*}$`
Minimum: `1`
| | -**Additional Properties:** not allowed - -### metadata\.tags\[\]: Pipeline tags +**Additional Properties:** not allowed  +  + -Array of pipeline tags. Each tag must be a string of up to 50 characters, containing only alphanumeric characters, dots, dashes, or underscores, and must start and end with an alphanumeric character. Tags must be unique. +Target Redis databases where processed records are written. Each key is a target identifier; the value configures the connection. + +**Properties** (key: `.*`) + +| Name | Type | Description | Required | +| ------------------------------------------------------------------------ | -------- | ----------------------------------------------------------------------------------------------------- | -------- | +| [**connection**](#definitionsredis_connection)
(Database connection) | `object` | Connection configuration for a Redis database.
| yes | +| **name**
(Target name) | `string` | Human-readable name for the target connection. Maximum 100 characters.
Maximal Length: `100`
| no | +  + + +Connection configuration for a Redis database. + +**Properties** -**Unique Items:** yes +| Name | Type | Description | Required | +| -------------------------------------------- | --------- | --------------------------------------------------------------------------------------------------------- | -------- | +| **type**
(Database type) | | Database type identifier. Always `redis` for this connection.
Constant Value: `"redis"`
| yes | +| **host**
(Database host) | `string` | Hostname or IP address of the Redis server.
| yes | +| **port**
(Database port) | `integer` | Network port on which the Redis server is listening.
Minimum: `1`
Maximum: `65535`
| yes | +| **user**
(Database user) | `string` | Username for authentication to the Redis database.
| no | +| **password**
(Database password) | `string` | Password for authentication to the Redis database.
| no | +| **key**
(Private key file) | `string` | Path to the private key file used for SSL/TLS client authentication.
| no | +| **key\_password**
(Private key password) | `string` | Password used to decrypt the private key file.
| no | +| **cert**
(Client certificate) | `string` | Path to the client certificate file used for SSL/TLS client authentication.
| no | +| **cacert**
(CA certificate) | `string` | Path to the Certificate Authority (CA) certificate file used to verify the server's TLS certificate.
| no | + +**Additional Properties:** not allowed  +**Minimal Properties:** 3  +**If property _key_ is defined**, property/ies _cert_ is/are required.  +**If property _cert_ is defined**, property/ies _key_ is/are required.  +**If property _key_password_ is defined**, property/ies _key_ is/are required.