diff --git a/debezium/oracle/build.gradle.kts b/debezium/oracle/build.gradle.kts index dd1ef02d83..d88bf9bcee 100644 --- a/debezium/oracle/build.gradle.kts +++ b/debezium/oracle/build.gradle.kts @@ -25,4 +25,8 @@ dependencies { implementation(libs.pulsar.io.core) implementation(project(":debezium:pulsar-io-debezium-core")) implementation(libs.debezium.connector.oracle) + + testImplementation(libs.testcontainers) + testImplementation(libs.testcontainers.pulsar) + testImplementation(libs.pulsar.client) } diff --git a/debezium/oracle/src/test/java/org/apache/pulsar/io/debezium/oracle/DebeziumOracleSourceTest.java b/debezium/oracle/src/test/java/org/apache/pulsar/io/debezium/oracle/DebeziumOracleSourceTest.java new file mode 100644 index 0000000000..db9d2b37a0 --- /dev/null +++ b/debezium/oracle/src/test/java/org/apache/pulsar/io/debezium/oracle/DebeziumOracleSourceTest.java @@ -0,0 +1,163 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.debezium.oracle; + +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.common.schema.KeyValue; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.core.SourceContext; +import org.awaitility.Awaitility; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.PulsarContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.utility.DockerImageName; +import org.testcontainers.utility.MountableFile; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +@Slf4j +public class DebeziumOracleSourceTest { + + private static final String PULSAR_IMAGE = + System.getenv().getOrDefault("PULSAR_TEST_IMAGE", "apachepulsar/pulsar:4.1.3"); + + private static final String ORACLE_IMAGE = "gvenzl/oracle-free:23-slim-faststart"; + + private static final int ORACLE_PORT = 1521; + + private GenericContainer oracleContainer; + private PulsarContainer pulsarContainer; + private PulsarClient pulsarClient; + private DebeziumOracleSource source; + + @BeforeMethod + public void setup() throws Exception { + // Use GenericContainer instead of the testcontainers oracle-free module: + // debezium-bom 3.4.x pins testcontainers core to 2.x, which is binary + // incompatible with the 1.x oracle-free module (and no 2.x release of it + // exists yet). + // All of the Oracle-side preparation (archive log mode, supplemental logging, + // the c##dbzuser LogMiner user and the seeded DEBEZIUM.PRODUCTS test table) + // is performed by the init script below, which the gvenzl image executes as + // SYSDBA on first startup, before it prints "DATABASE IS READY TO USE!". + oracleContainer = new GenericContainer<>(DockerImageName.parse(ORACLE_IMAGE)) + .withExposedPorts(ORACLE_PORT) + .withEnv("ORACLE_PASSWORD", "top_secret") + .withCopyFileToContainer( + MountableFile.forClasspathResource("debezium-oracle-setup.sql"), + "/container-entrypoint-initdb.d/01-debezium-setup.sql") + // The init script restarts the database to enable archive log mode, + // so allow a generous startup timeout. + .waitingFor(Wait.forLogMessage(".*DATABASE IS READY TO USE!.*\\s", 1) + .withStartupTimeout(Duration.ofMinutes(6))); + oracleContainer.start(); + + pulsarContainer = new PulsarContainer(DockerImageName.parse(PULSAR_IMAGE)); + pulsarContainer.start(); + + pulsarClient = PulsarClient.builder() + .serviceUrl(pulsarContainer.getPulsarBrokerUrl()) + .build(); + + source = new DebeziumOracleSource(); + } + + @AfterMethod(alwaysRun = true) + public void cleanup() throws Exception { + if (source != null) { + try { + source.close(); + } catch (Exception e) { + log.warn("Failed to close source", e); + } + } + if (pulsarClient != null) { + pulsarClient.close(); + } + if (pulsarContainer != null) { + pulsarContainer.stop(); + } + if (oracleContainer != null) { + oracleContainer.stop(); + } + } + + @Test(timeOut = 600_000) + public void testOracleCdcEvents() throws Exception { + String pulsarServiceUrl = pulsarContainer.getPulsarBrokerUrl(); + + SourceContext sourceContext = mock(SourceContext.class); + when(sourceContext.getPulsarClient()).thenReturn(pulsarClient); + when(sourceContext.getPulsarClientBuilder()).thenReturn( + PulsarClient.builder().serviceUrl(pulsarServiceUrl)); + when(sourceContext.getTenant()).thenReturn("public"); + when(sourceContext.getNamespace()).thenReturn("default"); + when(sourceContext.getSourceName()).thenReturn("debezium-oracle-test"); + when(sourceContext.getSecret(anyString())).thenReturn(null); + + Map config = new HashMap<>(); + config.put("database.hostname", oracleContainer.getHost()); + config.put("database.port", String.valueOf(oracleContainer.getMappedPort(ORACLE_PORT))); + config.put("database.user", "c##dbzuser"); + config.put("database.password", "dbz"); + config.put("database.dbname", "FREE"); + config.put("database.pdb.name", "FREEPDB1"); + config.put("topic.prefix", "oracle1"); + config.put("table.include.list", "DEBEZIUM.PRODUCTS"); + config.put("include.schema.changes", "false"); + // Read the data dictionary from the online catalog instead of mining it + // from the redo logs; this keeps connector startup lightweight for tests. + config.put("log.mining.strategy", "online_catalog"); + config.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl); + + source.open(config, sourceContext); + + // Debezium performs an initial snapshot of existing data. + // We should receive CDC records for the 2 rows seeded by the init script. + Awaitility.await() + .atMost(5, TimeUnit.MINUTES) + .pollInterval(1, TimeUnit.SECONDS) + .untilAsserted(() -> { + int recordCount = 0; + Record> record; + while ((record = source.read()) != null) { + assertNotNull(record.getValue()); + log.info("Received CDC record: key={}", record.getKey().orElse(null)); + recordCount++; + record.ack(); + if (recordCount >= 2) { + break; + } + } + assertTrue(recordCount >= 2, + "Expected at least 2 CDC records from initial snapshot, got " + recordCount); + }); + } +} diff --git a/debezium/oracle/src/test/resources/debezium-oracle-setup.sql b/debezium/oracle/src/test/resources/debezium-oracle-setup.sql new file mode 100644 index 0000000000..0c021ec56a --- /dev/null +++ b/debezium/oracle/src/test/resources/debezium-oracle-setup.sql @@ -0,0 +1,103 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- Prepares an Oracle Database Free (gvenzl/oracle-free) container for +-- Debezium LogMiner-based CDC. Executed once as SYSDBA from +-- /container-entrypoint-initdb.d/ on first container startup. +-- Based on the canonical Debezium Oracle connector setup: +-- https://debezium.io/documentation/reference/stable/connectors/oracle.html + +-- 1) Enable archive log mode (required by the LogMiner adapter). +ALTER SYSTEM SET db_recovery_file_dest_size = 5G; +ALTER SYSTEM SET db_recovery_file_dest = '/opt/oracle/oradata' SCOPE=spfile; +SHUTDOWN IMMEDIATE +STARTUP MOUNT +ALTER DATABASE ARCHIVELOG; +ALTER DATABASE OPEN; +ALTER PLUGGABLE DATABASE ALL OPEN; + +-- 2) Enable minimal supplemental logging at the CDB level. +ALTER DATABASE ADD SUPPLEMENTAL LOG DATA; + +-- 3) Tablespaces for the LogMiner user (CDB and PDB). +CREATE TABLESPACE LOGMINER_TBS DATAFILE '/opt/oracle/oradata/FREE/logminer_tbs.dbf' + SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED; + +ALTER SESSION SET CONTAINER=FREEPDB1; + +CREATE TABLESPACE LOGMINER_TBS DATAFILE '/opt/oracle/oradata/FREE/FREEPDB1/logminer_tbs.dbf' + SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED; + +ALTER SESSION SET CONTAINER=CDB$ROOT; + +-- 4) Common (CDB-wide) Debezium connector user with the LogMiner grant set. +CREATE USER c##dbzuser IDENTIFIED BY dbz + DEFAULT TABLESPACE LOGMINER_TBS + QUOTA UNLIMITED ON LOGMINER_TBS + CONTAINER=ALL; + +GRANT CREATE SESSION TO c##dbzuser CONTAINER=ALL; +GRANT SET CONTAINER TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$DATABASE TO c##dbzuser CONTAINER=ALL; +GRANT FLASHBACK ANY TABLE TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ANY TABLE TO c##dbzuser CONTAINER=ALL; +GRANT SELECT_CATALOG_ROLE TO c##dbzuser CONTAINER=ALL; +GRANT EXECUTE_CATALOG_ROLE TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ANY TRANSACTION TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ANY DICTIONARY TO c##dbzuser CONTAINER=ALL; +GRANT LOGMINING TO c##dbzuser CONTAINER=ALL; +GRANT CREATE TABLE TO c##dbzuser CONTAINER=ALL; +GRANT LOCK ANY TABLE TO c##dbzuser CONTAINER=ALL; +GRANT ALTER ANY TABLE TO c##dbzuser CONTAINER=ALL; +GRANT CREATE SEQUENCE TO c##dbzuser CONTAINER=ALL; +GRANT EXECUTE ON DBMS_LOGMNR TO c##dbzuser CONTAINER=ALL; +GRANT EXECUTE ON DBMS_LOGMNR_D TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$LOG TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$LOG_HISTORY TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$LOGMNR_LOGS TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$LOGMNR_CONTENTS TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$LOGMNR_PARAMETERS TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$LOGFILE TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$ARCHIVED_LOG TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$TRANSACTION TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$MYSTAT TO c##dbzuser CONTAINER=ALL; +GRANT SELECT ON V_$STATNAME TO c##dbzuser CONTAINER=ALL; + +-- 5) Test schema: a table with two seeded rows in the FREEPDB1 PDB. +ALTER SESSION SET CONTAINER=FREEPDB1; + +CREATE USER debezium IDENTIFIED BY dbz + DEFAULT TABLESPACE LOGMINER_TBS + QUOTA UNLIMITED ON LOGMINER_TBS; + +GRANT CONNECT TO debezium; +GRANT CREATE SESSION TO debezium; +GRANT CREATE TABLE TO debezium; +GRANT CREATE SEQUENCE TO debezium; + +CREATE TABLE debezium.products ( + id NUMBER(9,0) NOT NULL PRIMARY KEY, + name VARCHAR2(255) NOT NULL, + description VARCHAR2(512) +); + +INSERT INTO debezium.products VALUES (1, 'widget', 'A small widget'); +INSERT INTO debezium.products VALUES (2, 'gadget', 'A fancy gadget'); +COMMIT; + +ALTER TABLE debezium.products ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;