Migrate from the classic processor to the Flink processor

Learn how to migrate an existing RDI pipeline from the classic processor to the Apache Flink-based processor.

RDI ships with two stream processor implementations. The classic processor is implemented in Python. The default Flink processor is built on top of Apache Flink. Both run on VM and Kubernetes installations. The Flink processor can achieve much higher throughput during snapshots, scales horizontally by changing the number of TaskManager replicas, and uses Flink checkpointing for fault tolerance. See Stream processor implementations for an overview.

The classic processor is the default in RDI 1.19.0. The Flink processor is the default starting with RDI 2.0.0.

This page describes how to migrate an existing pipeline from the classic processor to the Flink processor. The steps are the same on VMs and Kubernetes, except for the optional Helm-level tuning in Step 1, which applies to Kubernetes only.

Before you migrate

This procedure migrates the pipeline processor on RDI 1.19.0. It does not upgrade RDI.

Before you start, save your existing configuration and jobs. Wait for the initial snapshot to finish. Interrupting it causes the snapshot to restart from the beginning.

Warning:

Switching processors with records still in the RDI input streams can leave records unprocessed. Stop the collector and let the classic processor empty the streams before switching.

Confirm that your pipeline is compatible with the Flink processor:

  • JSON.MERGE semantics differ from the classic processor's Lua-based merge when null values are involved (see use_native_json_merge). The Flink processor always uses the native JSON.MERGE command when the target database supports it.
  • Ensure your Kubernetes cluster or VM has enough capacity for the Flink JobManager and TaskManager pods (see Configure the Flink processor for the default sizing).

This step applies to Kubernetes installations only. On VM installations, continue with Step 2.

The Flink processor is always available — no opt-in is required at the Helm chart level. The defaults are sized for typical workloads, so you can skip this step if you don't need to override them. To adjust the JobManager and TaskManager defaults, add an operator.dataPlane.flinkProcessor block to your rdi-values.yaml file and run helm upgrade as described in Configure the Flink processor. Existing pipelines continue to run on the classic processor until you switch them in Step 4.

For VM installations, skip this step. You can configure per-pipeline Flink resources in Step 6.

Step 2: Disable source collection

In your existing config.yaml, add active: false under the source and set processors.type to classic. Use your existing source name and preserve all other source, target, processor, and job settings. This example shows only the fields to change:

sources:
  <existing-source-name>:
    active: false
processors:
  type: classic

Deploy the complete configuration directory, including the existing jobs:

redis-di deploy --dir <pipeline-config-directory>

Wait for the deployment to finish and the source collector to stop. Keep the pipeline active so the classic processor can process the remaining input records.

Applications can continue writing to the source database while collection is disabled. When the collector restarts, it resumes from the saved source position and processes those changes.

Step 3: Wait for the input streams to empty

After the collector has stopped, wait for every input stream to have a length of 0 in three complete checks, five seconds apart. An error, missing statistics, or an unexpectedly empty stream list does not count as 0. Do not include DLQ streams. Use any of the following methods.

Check with redis-di

Run:

redis-di describe

In the Statistics table, the Pending value for each classic processor stream is its current length. Confirm that every input stream is listed.

Check with Redis commands

Connect an authenticated Redis client to the RDI database that stores the pipeline's input streams, not the target database. Find the input stream keys with SCAN:

SCAN 0 MATCH data:{rdi}:* COUNT 1000 TYPE stream

If the returned cursor is not 0, pass it to the next command:

SCAN <returned-cursor> MATCH data:{rdi}:* COUNT 1000 TYPE stream

Repeat with each new cursor until the returned cursor is 0, even if an intermediate result contains no keys.

For every input stream returned, run XLEN:

XLEN <input-stream-key>

Run a complete SCAN and all XLEN commands in each of the three checks.

Step 4: Switch processors and resume collection

After the drain check passes, remove the source's active: false setting from the existing config.yaml and set processors.type to flink (or remove this property, since flink is the default):

processors:
  type: flink

RDI 1.19.x requires this setting because its default processor is classic. On RDI 2.0.0 the default processor is flink, so this setting is no longer required.

Keep the remaining configuration and jobs, then redeploy the complete configuration directory:

redis-di deploy --dir <pipeline-config-directory>

Wait for the classic processor to terminate and the Flink JobManager and TaskManager workloads to become healthy. Confirm that collection resumes from the saved source position and changes committed during the pause reach the target. Verify new inserts, updates, and deletes. The processor migration is complete after these checks pass.

Step 5: Adapt deprecated and classic-only properties

Some processors properties are no-ops, classic-only, or have moved to processors.advanced for the Flink processor. The following table lists the properties that need attention when migrating.

Property Action when migrating to Flink
on_failed_retry_interval No-op. Remove.
duration No-op. Use read_batch_timeout_ms instead.
dedup, dedup_max_size, dedup_strategy Classic-only. Remove.
enable_async_processing, batch_queue_size, ack_queue_size Classic-only. Remove.
initial_sync_processes Classic-only. Configure parallelism through advanced.flink.taskmanager.numberOfTaskSlots and advanced.resources.taskManager.replicas instead.
idle_streams_check_interval_ms, busy_streams_check_interval_ms Classic-only. Use processors.advanced.source.discovery.interval.ms for a single discovery interval.
idle_sleep_time_ms Classic-only. Remove.
use_native_json_merge Classic-only. The Flink processor always uses JSON.MERGE when the target supports it.

The classic processor silently ignores processors.advanced, and the Flink processor silently ignores classic-only top-level properties, so keeping both top-level properties and their processors.advanced equivalents lets you switch back without further edits.

Fine-tune the Flink processor through the processors.advanced section. For example:

processors:
  type: flink
  advanced:
    source:
      # Time between checks for new input streams.
      discovery.interval.ms: 1000
    flink:
      # Number of parallel task slots per TaskManager pod.
      taskmanager.numberOfTaskSlots: 2
      # Total memory budget for each TaskManager JVM process.
      taskmanager.memory.process.size: 4096m
    resources:
      taskManager:
        # Number of TaskManager pods.
        replicas: 2

See the processors.advanced reference for the full set of available properties.

Step 7: Update observability

The Flink processor exposes Prometheus metrics directly from the Flink JobManager and TaskManager pods. See Flink processor metrics for the ServiceMonitor configuration and the available metrics.

Rolling back

To revert a pipeline to the classic processor, set processors.type to classic and redeploy the pipeline. This setting is required on RDI 2.0.0, where the default is flink. The classic processor silently ignores processors.advanced, so you don't need to remove it before switching back.

RATE THIS PAGE
Back to top ↑