Donghua's Blog - DBAGlobe

A hands-on, end-to-end walkthrough of building a change-data-capture (CDC) pipeline that continuously replicates an Oracle schema into Aurora PostgreSQL — covering the initial full load (snapshot) and every subsequent INSERT / UPDATE / DELETE.

Note on identifiers. All account IDs, endpoints, VPC/subnet/security-group IDs, bucket names, passwords, and profile names in this article are placeholders. Replace <...> tokens and the example values (e.g. 111111111111, oracle.example.internal) with your own.

Why this pipeline?

Migrating or continuously syncing from Oracle to PostgreSQL is one of the most common "get off commercial database licensing" projects. You usually need two things at once:

A full load — copy the data that already exists.
Continuous CDC — keep the target current as the source keeps changing, ideally with no application downtime.

Debezium delivers both. Its Oracle connector uses Oracle LogMiner to read the redo/archive logs and turn every committed row change into a Kafka event; its JDBC sink connector applies those events to a target database as idempotent upserts (and deletes). Amazon MSK is the Kafka transport in the middle.

Oracle (schema DEMO, LogMiner)
      │  Debezium Oracle source connector
      ▼
Amazon MSK  (Kafka 3.7.x topics: oracdc.DEMO.<TABLE>)
      │  Debezium JDBC sink connector (upsert + RegexRouter)
      ▼
Aurora PostgreSQL (schema "demo", lowercase tables)

Source: io.debezium.connector.oracle.OracleConnector (adapter logminer)
Transport: Amazon MSK provisioned cluster, topics oracdc.DEMO.<TABLE>
Sink: io.debezium.connector.jdbc.JdbcSinkConnector (upsert + deletes)

We ran the same connectors two ways, and this post covers both:

Managed — Debezium on MSK Connect, with the databases in one AWS account and MSK + MSK Connect in another, joined by cross-account VPC peering.
Self-managed — Debezium on Kafka Connect running on an EC2 instance, which gives you maximum control over the Connect runtime.

Versions used

Component	Version	Why
Oracle	19c EE, non-CDB, ARCHIVELOG	LogMiner source
Aurora PostgreSQL	17.x	target
Amazon MSK	Kafka 3.7.x, 2× `kafka.m5.large`	newest version matching an MSK Connect worker
Debezium	2.7.3.Final	Oracle plugin bundles `ojdbc8` (works with 19c); JDBC plugin bundles the PostgreSQL driver
Kafka Connect (self-managed)	3.7.2 on Java 17	Debezium 2.7 is validated on Java 17

log.mining.strategy=online_catalog (the Debezium default) is the lowest-overhead LogMiner strategy: no extra archive-log generation. The trade-off is no live DDL tracking, which is fine for a stable schema.

Prerequisites

Oracle running in ARCHIVELOG mode (automated backups on RDS enable this).
A network path so the Kafka Connect runtime can reach:
- Oracle on 1521, PostgreSQL on 5432, and the MSK brokers on 9092.
A DB client (sqlplus, psql) to run the setup SQL.
Debezium plugins matching your Kafka Connect version.

Step 1 — Prepare the Oracle source (LogMiner)

Run these as the RDS master user. On Amazon RDS you cannot connect as SYSDBA, so archive-log retention and supplemental logging go through the rdsadmin package, and grants on SYS-owned V_$ views / DBMS_LOGMNR* packages use grant_sys_object.

1a. Confirm ARCHIVELOG

SELECT log_mode FROM v$database;   -- must return ARCHIVELOG

1b. Set archive-log retention (so LogMiner can find the resume SCN after a restart):

EXEC rdsadmin.rdsadmin_util.set_configuration('archivelog retention hours', 24);

1c. Enable supplemental logging — database-level minimal, plus ALL COLUMNS per captured table (needed for complete before/after images):

EXEC rdsadmin.rdsadmin_util.alter_supplemental_logging('ADD');

ALTER TABLE DEMO.CUSTOMERS ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
-- ...repeat for each captured table

-- verify
SELECT supplemental_log_data_min FROM v$database;                 -- YES
SELECT COUNT(*) FROM dba_log_groups
 WHERE owner='DEMO' AND log_group_type='ALL COLUMN LOGGING';

1d. Create the LogMiner user (non-CDB → no c## prefix):

CREATE USER debezium IDENTIFIED BY "<ORACLE_CDC_PASSWORD>"
  DEFAULT TABLESPACE USERS QUOTA UNLIMITED ON USERS;

GRANT CREATE SESSION, SELECT ANY TABLE, SELECT ANY TRANSACTION,
      LOGMINING, CREATE TABLE, LOCK ANY TABLE, CREATE SEQUENCE,
      FLASHBACK ANY TABLE TO debezium;
GRANT SELECT_CATALOG_ROLE, EXECUTE_CATALOG_ROLE TO debezium;

Grant the SYS-owned objects via grant_sys_object (RDS-specific):

BEGIN
  rdsadmin.rdsadmin_util.grant_sys_object('DBMS_LOGMNR',        'DEBEZIUM','EXECUTE');
  rdsadmin.rdsadmin_util.grant_sys_object('DBMS_LOGMNR_D',      'DEBEZIUM','EXECUTE');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$DATABASE',        'DEBEZIUM','SELECT');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$LOG',             'DEBEZIUM','SELECT');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$LOG_HISTORY',     'DEBEZIUM','SELECT');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$LOGMNR_LOGS',     'DEBEZIUM','SELECT');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$LOGMNR_CONTENTS', 'DEBEZIUM','SELECT');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$LOGMNR_PARAMETERS','DEBEZIUM','SELECT');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$LOGFILE',         'DEBEZIUM','SELECT');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$ARCHIVED_LOG',    'DEBEZIUM','SELECT');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$ARCHIVE_DEST_STATUS','DEBEZIUM','SELECT');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$TRANSACTION',     'DEBEZIUM','SELECT');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$MYSTAT',          'DEBEZIUM','SELECT');
  rdsadmin.rdsadmin_util.grant_sys_object('V_$STATNAME',        'DEBEZIUM','SELECT');
END;
/

1e. Smoke-test the user:

-- connect as debezium/<ORACLE_CDC_PASSWORD>
SELECT USER FROM dual;                 -- DEBEZIUM
SELECT log_mode FROM v$database;       -- ARCHIVELOG  (proves the V_$ grant works)
SELECT COUNT(*) FROM demo.customers;   -- proves SELECT ANY TABLE works

Gotcha: an inline -- comment after a ; breaks SQL*Plus with ORA-00933. Keep comments on their own line.
LOB caveat: tables with BLOB columns are excluded here. LOB streaming needs lob.enabled=true and rules out the hybrid strategy — see Limitations.

Step 2 — Prepare the PostgreSQL target

Two design decisions matter:

Lowercase identifiers. PostgreSQL folds unquoted identifiers to lowercase; Oracle reports names in UPPERCASE. The Debezium JDBC sink emits unquoted identifiers by default (quote.identifiers=false), so Oracle's CUSTOMERS / ID become customers / id. Create the target schema and tables in lowercase and everything lines up with no quoting.

Type mapping follows Debezium's Oracle numeric rule for NUMBER(p,s) with scale 0: p-s<3→INT8, <5→INT16, <10→INT32, <19→INT64, otherwise Decimal.

Oracle	Kafka Connect type	PostgreSQL
`NUMBER(19,0)` (IDs/FKs)	`Decimal(0)`	`numeric(19)`
`NUMBER(10,0)`	`INT64`	`bigint`
`NUMBER(1,0)` (flags)	`INT8`/`INT16`	`smallint`
`NUMBER(38,2)` (money)	`Decimal(2)`	`numeric(38,2)`
`VARCHAR2(n)`	`string`	`text`
`DATE`	`io.debezium.time.Timestamp`	`timestamp`
`TIMESTAMP(6)`	`io.debezium.time.MicroTimestamp`	`timestamp(6)`

Using numeric for NUMBER(19,0) IDs exactly matches Debezium's Decimal type, avoiding a Decimal→integer bind mismatch on upsert. Using text for all strings sidesteps Oracle's byte-vs-char length ambiguity. Preserve primary keys so the sink can run in upsert / record_key mode.

Create the schema and tables:

psql "host=<PG_ENDPOINT> port=5432 dbname=demodb user=postgres sslmode=require" \
     -v ON_ERROR_STOP=1 -f sql/02-postgres-target-schema.sql
# The script does: DROP SCHEMA IF EXISTS demo CASCADE; CREATE SCHEMA demo; ...

The sink's JDBC URL sets currentSchema=demo and a RegexRouter rewrites oracdc.DEMO.CUSTOMERS → CUSTOMERS; unquoted, PostgreSQL resolves it to demo.customers. No per-table sink config is required.

Step 3 — Create the Amazon MSK cluster

aws kafka create-cluster --region <REGION> \
  --cluster-name debezium-msk \
  --kafka-version 3.7.x \
  --number-of-broker-nodes 2 \
  --broker-node-group-info '{
    "InstanceType":"kafka.m5.large",
    "ClientSubnets":["<SUBNET_A>","<SUBNET_B>"],
    "SecurityGroups":["<MSK_SG>"],
    "StorageInfo":{"EbsStorageInfo":{"VolumeSize":20}}
  }' \
  --client-authentication '{"Unauthenticated":{"Enabled":true}}' \
  --encryption-info '{"EncryptionInTransit":{"ClientBroker":"TLS_PLAINTEXT","InCluster":true}}'

Key choices:

2 brokers across 2 private subnets — one per AZ. Connect internal topics use replication factor 2.
TLS_PLAINTEXT + Unauthenticated — lets Kafka Connect use the simple PLAINTEXT bootstrap (port 9092) with authenticationType=NONE, keeping the Debezium config free of TLS/IAM plumbing. For production, prefer TLS + IAM auth and lock the security group down.

Creation takes ~20–30 minutes. Wait for State=ACTIVE, then grab the brokers:

aws kafka describe-cluster-v2 --region <REGION> --cluster-arn <ARN> \
  --query 'ClusterInfo.State'

aws kafka get-bootstrap-brokers --region <REGION> --cluster-arn <ARN>
# Use BootstrapBrokerString (PLAINTEXT, :9092) for Connect + Debezium.

Networking: MSK broker ENIs live in the private subnets. Any Kafka client (Connect on EC2, or the managed MSK Connect service) must be able to open TCP 9092 to those subnets. MSK Connect's connector ENIs also need a route to S3 (NAT gateway or S3 VPC endpoint) to download the plugins.

Step 4 — Get the Debezium plugins

Both connectors are Debezium 2.7.3.Final, and each plugin tarball already bundles its JDBC driver, so no extra jars are needed:

https://repo1.maven.org/maven2/io/debezium/debezium-connector-oracle/2.7.3.Final/debezium-connector-oracle-2.7.3.Final-plugin.tar.gz
https://repo1.maven.org/maven2/io/debezium/debezium-connector-jdbc/2.7.3.Final/debezium-connector-jdbc-2.7.3.Final-plugin.tar.gz

Self-managed Connect on EC2: you do not need S3 or an IAM role — just extract both tarballs into the worker's plugin.path.
Managed MSK Connect: zip each extracted connector directory, upload to S3, and create an IAM service-execution role (next).

For MSK Connect:

aws s3 mb s3://<PLUGIN_BUCKET> --region <REGION>
( cd oracle   && zip -r ../debezium-oracle-2.7.3.zip   debezium-connector-oracle )
( cd jdbcsink && zip -r ../debezium-jdbc-sink-2.7.3.zip debezium-connector-jdbc )
aws s3 cp debezium-oracle-2.7.3.zip    s3://<PLUGIN_BUCKET>/plugins/
aws s3 cp debezium-jdbc-sink-2.7.3.zip s3://<PLUGIN_BUCKET>/plugins/

IAM service-execution role for MSK Connect — trust policy trusts kafkaconnect.amazonaws.com; the permissions policy grants s3:GetObject / s3:ListBucket on the plugin bucket, logs:* on /aws/mskconnect/*, kafka:DescribeCluster / kafka:GetBootstrapBrokers, and the ec2:*NetworkInterface* set MSK Connect needs to place ENIs in the VPC:

aws iam create-role --role-name debezium-mskconnect-role \
  --assume-role-policy-document file://iam/trust-policy.json
aws iam put-role-policy --role-name debezium-mskconnect-role \
  --policy-name debezium-mskconnect-permissions \
  --policy-document file://iam/mskconnect-permissions.json

Step 5 — The two connector configs (explained)

These configs are identical whether you deploy on MSK Connect or on self-managed Connect. Only the runtime differs.

Source — Debezium Oracle (LogMiner)

{
  "connector.class": "io.debezium.connector.oracle.OracleConnector",
  "tasks.max": "1",
  "database.hostname": "<ORACLE_ENDPOINT>",
  "database.port": "1521",
  "database.user": "debezium",
  "database.password": "<ORACLE_CDC_PASSWORD>",
  "database.dbname": "ORCL",
  "database.connection.adapter": "logminer",
  "log.mining.strategy": "online_catalog",
  "topic.prefix": "oracdc",
  "schema.include.list": "DEMO",
  "table.include.list": "DEMO.ADDRESSES,DEMO.BOOKS,...,DEMO.SHOPPING_CART_ITEMS",
  "snapshot.mode": "initial",
  "decimal.handling.mode": "precise",
  "tombstones.on.delete": "true",
  "include.schema.changes": "false",
  "schema.history.internal.kafka.bootstrap.servers": "<MSK_PLAINTEXT_BOOTSTRAP>",
  "schema.history.internal.kafka.topic": "schema-history.oracdc"
}

non-CDB → no database.pdb.name. (A CDB also needs database.pdb.name and a c## common user.)
snapshot.mode=initial gives the full load first, then streams CDC — this is what "full + CDC" means.
decimal.handling.mode=precise → NUMBER becomes Kafka Connect Decimal, which the sink maps to PostgreSQL numeric.
tombstones.on.delete=true + sink delete.enabled=true → Oracle deletes become row deletes in PostgreSQL.
Emits topics oracdc.DEMO.<TABLE> keyed by the table PK.

Sink — Debezium JDBC → PostgreSQL

{
  "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
  "tasks.max": "1",
  "topics.regex": "oracdc\\.DEMO\\..*",
  "connection.url": "jdbc:postgresql://<PG_ENDPOINT>:5432/demodb?currentSchema=demo",
  "connection.username": "postgres",
  "connection.password": "<PG_PASSWORD>",
  "insert.mode": "upsert",
  "primary.key.mode": "record_key",
  "delete.enabled": "true",
  "schema.evolution": "none",
  "quote.identifiers": "false",
  "transforms": "route",
  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "oracdc\\.DEMO\\.(.*)",
  "transforms.route.replacement": "$1"
}

RegexRouter rewrites oracdc.DEMO.CUSTOMERS → CUSTOMERS; with quote.identifiers=false the sink emits it unquoted → PostgreSQL resolves demo.customers.
insert.mode=upsert + primary.key.mode=record_key → idempotent writes keyed by the Debezium message key (the Oracle PK). Replaying events won't create duplicates.
schema.evolution=none — we pre-created the tables. Set to basic to let the sink ALTER TABLE/create tables as schemas change.

Store secrets properly. The passwords are inlined above for clarity. In production, pull them from a secrets manager rather than embedding them in connector JSON, and never commit real credentials.

Step 6 — Deploy (self-managed Kafka Connect on EC2)

This is the reliably-working path. Everything below can be driven over SSM (no SSH keys needed) against your EC2 instance.

6a. Install runtime + plugins (once):

sudo dnf install -y java-17-amazon-corretto-headless tar gzip
sudo mkdir -p /opt/connect && sudo chown ec2-user:ec2-user /opt/connect
cd /opt/connect
curl -fsSL -o kafka.tgz https://archive.apache.org/dist/kafka/3.7.2/kafka_2.13-3.7.2.tgz
tar -xzf kafka.tgz && mv kafka_2.13-3.7.2 kafka
mkdir -p plugins && cd plugins
curl -fsSL -o oracle.tgz https://repo1.maven.org/maven2/io/debezium/debezium-connector-oracle/2.7.3.Final/debezium-connector-oracle-2.7.3.Final-plugin.tar.gz && tar -xzf oracle.tgz
curl -fsSL -o jdbc.tgz   https://repo1.maven.org/maven2/io/debezium/debezium-connector-jdbc/2.7.3.Final/debezium-connector-jdbc-2.7.3.Final-plugin.tar.gz && tar -xzf jdbc.tgz

Pin Java 17 (export JAVA_HOME=/usr/lib/jvm/java-17-amazon-corretto) — Kafka 3.7 / Debezium 2.7 are validated on 17, and AL2023's default java may be 21.

6b. Worker config (connect-distributed.properties): set bootstrap.servers to the MSK PLAINTEXT bootstrap string, plugin.path=/opt/connect/plugins, replication factor 2 for the three internal topics, and JSON converters with schemas.enable=true (the JDBC sink needs the record schema).

6c. Start the worker (distributed mode, REST on :8083):

export JAVA_HOME=/usr/lib/jvm/java-17-amazon-corretto
cd /opt/connect
nohup kafka/bin/connect-distributed.sh connect-distributed.properties \
  > /opt/connect/connect.log 2>&1 &
# confirm the REST API answers and both plugins are present:
curl -s localhost:8083/connector-plugins | tr ',' '\n' | grep -i -E 'oracle|jdbc'

6d. Pre-create the Kafka topics — MSK sets auto.create.topics.enable=false, so the source cannot auto-create per-table data topics and will loop on UNKNOWN_TOPIC_OR_PARTITION. Create them first (RF=2 matches 2 brokers):

B=<MSK_PLAINTEXT_BOOTSTRAP>
for t in ADDRESSES BOOKS BOOK_TYPES CONDITIONS CUSTOMERS GENRES LISTINGS \
         ORDERS ORDER_ITEMS PASSWORD_RESET_TOKENS PERSISTENT_LOGINS \
         PUBLISHERS SHOPPING_CART_ITEMS; do
  kafka/bin/kafka-topics.sh --bootstrap-server $B \
    --create --if-not-exists --topic oracdc.DEMO.$t --partitions 1 --replication-factor 2
done

(Alternatively add topic.creation.default.replication.factor=2 and topic.creation.default.partitions=1 to the source connector so Kafka Connect creates the topics for you — no broker change needed.)

6e. Deploy the connectors:

curl -s -XPOST -H 'Content-Type: application/json' \
  localhost:8083/connectors -d @oracle-source.json
curl -s -XPOST -H 'Content-Type: application/json' \
  localhost:8083/connectors -d @postgres-sink.json

curl -s localhost:8083/connectors/oracle-source/status
curl -s localhost:8083/connectors/postgres-sink/status

Step 7 — Verify full load + CDC

Connector status — both should be RUNNING (connector + task 0).

Full load — compare row counts. In our run all 13 tables matched exactly (e.g. books 56/56, publishers 10/10, customers 3/3), and numeric, text, and timestamp(6) values all landed correctly.

CDC — INSERT / UPDATE / DELETE. Run on Oracle:

INSERT INTO demo.customers (id,username,email,first_name,last_name)
  VALUES (9001,'cdc_user1','cdc1@test.com','Cdc','One');
INSERT INTO demo.customers (id,username,email,first_name,last_name)
  VALUES (9002,'cdc_user2','cdc2@test.com','Cdc','Two');
COMMIT;
UPDATE demo.customers SET email='cdc1_UPDATED@test.com' WHERE id=9001; COMMIT;
DELETE FROM demo.customers WHERE id=9002; COMMIT;

Result in PostgreSQL (~20 s later):

SELECT id,email FROM demo.customers WHERE id IN (9001,9002);
 9001 | cdc1_UPDATED@test.com      ← INSERT + UPDATE captured
 (no row for 9002)                 ← DELETE captured

Insert, update, and delete all propagate Oracle → MSK → PostgreSQL.

Handy operational commands:

# consume raw change events for one table
kafka/bin/kafka-console-consumer.sh --bootstrap-server $B \
  --topic oracdc.DEMO.CUSTOMERS --from-beginning --max-messages 1

# connector health / restart
curl -s localhost:8083/connectors/oracle-source/status | jq
curl -s -XPOST localhost:8083/connectors/postgres-sink/restart?includeTasks=true

# worker log
tail -f /opt/connect/connect.log

The managed path: Debezium on MSK Connect (cross-account, hardened)

If your environment allows MSK Connect, you can run the connectors as a managed service. In our setup the databases stayed in one account and MSK + MSK Connect ran in a separate account, joined by cross-account VPC peering — with the second account fully hardened.

Account A (databases)                       Account B (streaming)
VPC 10.1.0.0/16                              VPC 10.2.0.0/16  (private-only)
┌───────────────────────────┐   VPC peering  ┌──────────────────────────────────┐
│ Oracle RDS  10.1.x:1521    │◀──────────────▶│ MSK  debezium-msk                 │
│ Aurora PG   10.1.x:5432    │                │ MSK Connect (source + sink)       │
│ (RDS SG allows 10.2.0.0/16)│                │ private subnets 10.2.1/24,10.2.2/24│
└───────────────────────────┘                └──────────────────────────────────┘

Security controls worth copying

Requirement	How it's met
No long-lived keys	Account accessed only through federation (temporary STS creds). MSK Connect runs under an IAM role; no IAM users or access keys.
No `0.0.0.0/0` open ports	MSK security-group ingress allows only the peer CIDR (`10.1.0.0/16`) on Kafka ports plus self-reference. Egress is restricted to the local VPC, the peer CIDR, and the S3 prefix list on 443 — the default `0.0.0.0/0` egress rule was revoked.
No public S3 bucket	The plugin bucket has Block Public Access = ON (all four flags). Plugins are pulled privately through an S3 gateway VPC endpoint.
Private-only networking	The streaming VPC has no Internet Gateway and no NAT. Connectors reach S3 via a gateway endpoint, CloudWatch Logs via an interface endpoint, and the databases only across the peering.

Outline of the managed build

Hardened VPC (10.2.0.0/16) with two private subnets, no IGW/NAT.
Cross-account VPC peering to the database VPC (10.1.0.0/16); add routes on both sides, and open the database-side RDS security group to the streaming CIDR (still a specific CIDR, never 0.0.0.0/0).
Private endpoints: S3 gateway endpoint (plugin download) + CloudWatch Logs interface endpoint (connector logs), and a Block-Public-Access plugin bucket.
MSK cluster with topic auto-create via a cluster configuration (auto.create.topics.enable=true) — so the Debezium producer creates the 13 oracdc.DEMO.* topics on demand, no manual pre-creation.
IAM role, custom plugins, worker configuration, and both connectors via aws kafkaconnect create-*. The connector configurations are identical to the self-managed ones above.

# cross-account VPC peering (requester = streaming account)
aws ec2 create-vpc-peering-connection --profile <STREAMING_PROFILE> \
  --vpc-id <STREAMING_VPC> --peer-vpc-id <DB_VPC> \
  --peer-owner-id <DB_ACCOUNT_ID> --peer-region <REGION>
aws ec2 accept-vpc-peering-connection --profile <DB_PROFILE> \
  --vpc-peering-connection-id <PCX_ID>

# MSK cluster config enabling broker-side auto topic creation
printf 'auto.create.topics.enable=true\ndefault.replication.factor=2\nmin.insync.replicas=1\nnum.partitions=1\ndelete.topic.enable=true\n' > msk-server.properties
aws kafka create-configuration --name debezium-autocreate --kafka-versions 3.7.x \
  --server-properties fileb://msk-server.properties --profile <STREAMING_PROFILE>

DNS note: the RDS endpoints are not publicly accessible, so their public DNS names resolve to the private 10.1.x addresses from any VPC. With the peering routes in place, MSK Connect reaches them directly — no private hosted zone sharing required.

⚠ Known issue: Debezium JDBC sink on the managed MSK Connect worker

The source connector runs cleanly on MSK Connect. The Debezium JDBC sink (2.7.3), however, repeatedly FAILED on the managed worker with a Jackson classpath clash — its Hibernate layer calls ObjectMapper.findAndRegisterModules(), which discovers the worker's own Jackson modules and errors:

Scala module 2.16.0 requires Jackson Databind >= 2.16.0 - Found 2.12.7-1
...then: Jdk8Module not a subtype ...then: JaxbAnnotationModule not a subtype

Bundling matching Jackson 2.16 jars fixes each module in turn, but the managed worker keeps contributing more — a whack-a-mole specific to that runtime. It does not occur on a self-managed Kafka Connect worker.

Resolution: run the source on MSK Connect and the sink on the self-managed EC2 worker, both pointed at the same MSK bootstrap. They share the same topics and together deliver the full pipeline. If you need the sink on MSK Connect too, use the Confluent JDBC sink (no Hibernate/Jackson dependency) or a Debezium JDBC build aligned to the managed worker's Jackson version.

Capturing whole schemas, exclusions, and scaling

Capture an entire schema — table.include.list is optional. Provide only schema.include.list and the connector captures every non-system table:

"schema.include.list": "DEMO",
"snapshot.mode": "initial"

Capture all-but-exclude some tables — combine schema.include.list with table.exclude.list (don't set both table.include.list and table.exclude.list):

"schema.include.list": "DEMO",
"table.exclude.list": "DEMO\\.DR\\$.*,DEMO\\.BOOKS_COVER,DEMO\\.AUDIT_.*"

Include/exclude lists are POSIX regexes, matched anchored (whole name) and case-sensitive (Oracle names are UPPERCASE); escape . as \\. and $ as \\$ in JSON. You can also drop specific columns with column.exclude.list.

Multiple schemas — schema.include.list is a comma-separated list of regexes. Topics become <topic.prefix>.<SCHEMA>.<TABLE>, so the schema name is preserved and the sink's RegexRouter must account for it. Enable supplemental logging for every captured table — for whole-schema capture it's easier to enable it once at the database level:

-- full before-images (higher redo volume)
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
-- or PK-only where before-images aren't required (less redo)
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY) COLUMNS;

Scaling. The Oracle connector is single-task — tasks.max is ignored; one connector runs one single-threaded LogMiner session.

Scale	Approach
up to a few hundred tables	single connector is fine
~1000+ tables / high change volume	shard across multiple connectors + tune

Tuning for large capture sets: use PK-only supplemental logging where possible; push filtering to the database (log.mining.query.filter.mode=in or regex); consider the hybrid mining strategy; raise JVM heap / MCUs; and budget MSK partitions (1000 tables ≈ 1000+ topics). To shard, run several connectors over disjoint table subsets, each with its own topic.prefix and schema.history.internal.kafka.topic. Safe with online_catalog or hybrid; not with redo_log_catalog.

Sink side for multiple schemas — the JDBC sink is a normal sink connector (scale it with tasks.max > 1 up to the topic partition count). Either flatten all schemas into one target schema, or preserve the schema in the target table name:

"transforms.route.regex": "oracdc\\.(.*)\\.(.*)",
"transforms.route.replacement": "$1_$2"

→ sales_orders, inventory_orders. Set schema.evolution=basic to let the sink auto-create/alter target tables.

Sizing MSK for many topics — the constraint is partitions per broker (including replicas), not "topics." For CDC each table topic is usually 1 partition:

total partition-replicas = (sum of topic partitions) × replication_factor
per-broker load          = total partition-replicas ÷ number_of_brokers

Keep per-broker load under AWS's recommended maximum for the instance size (e.g. ~1000 for m5.large/m7g.large, ~2000 for .2xlarge), and target ≤ 60–70% of that to leave headroom. For production durability use 3 brokers across 3 AZs, RF=3, min.insync.replicas=2, prefer Graviton (m7g) for price/perf, and validate with the official Amazon MSK Sizing spreadsheet.

Learnings, gotchas & limitations

Type-mapping learnings

Oracle NUMBER(19,0) does not map to INT64 — per Debezium's rule it falls through to Decimal. Target those columns as PostgreSQL numeric, not bigint, to avoid Decimal→integer bind mismatches on upsert.
NUMBER(1,0) flags become INT8/INT16 → smallint (not boolean). Use Debezium's NumberOneToBooleanConverter for real booleans.
Oracle DATE carries a time component → io.debezium.time.Timestamp (ms) → PostgreSQL timestamp, not date.
VARCHAR2(n) lengths are in bytes in all_tab_columns; mapping every string to text sidesteps the byte-vs-char ambiguity.

Common gotchas

Symptom	Cause	Fix
`ORA-00933` on a `GRANT`	inline `-- comment` after `;` in SQL*Plus	put comments on their own line
`ORA-01555 snapshot too old` on large snapshots	snapshot exceeds `UNDO_RETENTION`	raise `UNDO_RETENTION`, or `snapshot.mode=no_data` + incremental snapshot
Connector stalls after ~350s	idle-timeout on LogMiner JDBC calls behind a load balancer	enable TCP keepalives / `(ENABLE=broken)` in `database.url`
Sink "connection refused" to brokers	Connect host not allowed on the MSK SG, or wrong bootstrap type	use the PLAINTEXT bootstrap (:9092); ensure the SG allows it
Source loops on `UNKNOWN_TOPIC_OR_PARTITION`, no data in PG	MSK `auto.create.topics.enable=false`; data topics never created	pre-create the topics or set `topic.creation.default.*`, then restart the source

Limitations / out of scope

BLOB columns are not replicated in this setup. LOB streaming needs lob.enabled=true, which forces online_catalog/redo_log_catalog (not hybrid) and adds redo volume and edge cases.
DDL changes aren't tracked with log.mining.strategy=online_catalog. For evolving schemas use hybrid (no LOBs) and sink schema.evolution=basic.
Security posture here is workshop-grade (PLAINTEXT + unauthenticated MSK, broad security group). For production: TLS + IAM auth, least-privilege security groups, and secrets from a secrets manager rather than inlined in connector JSON.
Single task per connector. The Oracle connector always uses one LogMiner task. For higher sink throughput, raise the sink's tasks.max and the Kafka topic partition count.

Conclusion

Debezium + Amazon MSK gives you a robust Oracle → PostgreSQL pipeline that does the initial full load and then keeps the target continuously in sync with every insert, update, and delete. The connector configurations are identical whether you run on managed MSK Connect or self-managed Kafka Connect on EC2 — so you can start self-managed for maximum control (and to sidestep the managed-worker Jackson issue for the JDBC sink), and adopt the managed service where your environment allows it. The details that make or break the build are the Oracle-side supplemental logging and grants, the Oracle→PostgreSQL type mapping (especially NUMBER(19,0) → numeric), the lowercase identifier convention, and MSK topic creation when broker auto-create is disabled.