molt-replicator（蜕皮复制子）

molt replicator

Continuous change-data-capture (CDC) replication from source databases to CockroachDB. Run after molt fetch completes the initial bulk load.

Important: replicator is a separate binary from molt. It is not invoked by molt fetch. The data-load-and-replication mode in molt fetch is deprecated — use replicator directly instead.

Architecture

Source DB ──► [replicator] ──► Staging DB (_replicator schema) ──► Target CockroachDB
                  ▲
            Publication/
            Slot/BinLog/
            LogMiner

Replicator reads changes from the source, buffers them in a staging schema on the target CRDB cluster, and applies them to the target tables.

Subcommands by Source

Full Fetch → Replicator Workflow

Step 1: Initial bulk load with molt fetch

molt fetch \
  --source "postgresql://user:pass@source:5432/db" \
  --target "postgresql://root@crdb:26257/db" \
  --bucket-path "s3://mybucket/migration" \
  --table-handling drop-on-target-and-recreate

Step 2: Create publication on source (PostgreSQL)

-- Run on source PostgreSQL:
CREATE PUBLICATION molt_fetch FOR ALL TABLES;
-- (molt fetch may have already created this; check first)

Step 3: Create staging database on target

-- Run on target CockroachDB:
CREATE DATABASE _replicator;

Step 4: Test connectivity

replicator preflight \
  --sourceConn "postgresql://user:pass@source:5432/db" \
  --targetConn "postgresql://root@crdb:26257/db"

Step 5: Start replicator

replicator pglogical \
  --publicationName "molt_fetch" \
  --sourceConn "postgresql://user:pass@source:5432/db" \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public" \
  --metricsAddr "0.0.0.0:8080"

Step 6: Monitor lag

curl http://localhost:8080/metrics | grep replicator_
# Watch for: mutations applied, unapplied mutations, lag

Step 7: Cutover

1. When lag reaches ~0, redirect app writes to CockroachDB
2. Let replicator drain remaining changes
3. Confirm no new writes on source
4. Stop replicator
5. Decommission source

Source-Specific Setup

PostgreSQL (pglogical)

Source prerequisites:

• User with REPLICATION privilege
• Logical replication enabled (wal_level = logical)
• Publication exists (created by molt fetch or manually)

replicator pglogical \
  --publicationName "molt_fetch" \
  --slotName "replicator" \
  --sourceConn "postgresql://..." \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public"

MySQL (mylogical)

Source prerequisites:

• Binary logging enabled (binlog_format = ROW)
• GTID mode on (gtid_mode=ON, enforce_gtid_consistency=ON)
• User with REPLICATION CLIENT privilege

replicator mylogical \
  --sourceConn "mysql://root:pass@source:3306/db" \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public"

Oracle (oraclelogminer)

Source prerequisites:

• Archive log mode enabled
• Supplemental logging enabled
• LogMiner permissions granted

replicator oraclelogminer \
  --sourceConn "oracle://app_user:pass@oracle:1521/db" \
  --stagingConn "postgresql://root@crdb:26257/_replicator" \
  --stagingSchema "_replicator.public" \
  --targetConn "postgresql://root@crdb:26257/db" \
  --targetSchema "public"

Key Flags

# Performance
--parallelism 16          # concurrent DB transactions (default: 16)
--flushSize 1000          # rows per batch (default: 1000)
--flushPeriod 1s          # flush interval (default: 1s)

# Staging connection pool
--stagingMaxPoolSize 128
--stagingIdleTime 1m
--stagingMaxLifetime 5m

# Target connection pool
--targetMaxPoolSize 128
--targetStatementCacheSize 128

# Retry
--maxRetries 10
--retryInitialBackoff 25ms
--retryMaxBackoff 2s

# Monitoring
--metricsAddr "0.0.0.0:8080"    # Prometheus metrics endpoint
--schemaRefresh 1m               # refresh schema cache (0 = disabled)

# Dead letter queue (failed rows instead of stopping)
--dlqTableName "replicator_dlq"

# Logging
-v                               # debug
-vv                              # trace
--logFormat fluent                # for log aggregators
--logDestination "/var/log/replicator.log"

Gotchas

• Staging schema (_replicator.public) is auto-created by replicator, but the database (_replicator) must exist first
• --publicationName and --slotName must match what molt fetch created (default: molt_fetch / molt_slot)
• DLQ table grows over time — monitor and purge failed rows periodically
• Replicator holds an open replication slot on the source — this blocks WAL cleanup; monitor source disk usage
• Graceful shutdown respects --gracePeriod (default: 30s); don't SIGKILL without it
• No built-in alerting — set up external alerts on the Prometheus metrics endpoint
• Long cutover windows increase replication lag — plan for a maintenance window if needed

See flags reference for the full flag list.