diff --git a/debezium/mariadb/build.gradle.kts b/debezium/mariadb/build.gradle.kts new file mode 100644 index 0000000000..6535ecf5cb --- /dev/null +++ b/debezium/mariadb/build.gradle.kts @@ -0,0 +1,32 @@ +/* + * 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. + */ + +plugins { + id("pulsar-connectors.java-conventions") + id("pulsar-connectors.nar-conventions") +} +dependencies { + implementation(libs.pulsar.io.core) + implementation(project(":debezium:pulsar-io-debezium-core")) + implementation(libs.debezium.connector.mariadb) + + testImplementation(libs.testcontainers) + testImplementation(libs.testcontainers.pulsar) + testImplementation(libs.pulsar.client) +} diff --git a/debezium/mariadb/src/main/java/org/apache/pulsar/io/debezium/mariadb/DebeziumMariaDbSource.java b/debezium/mariadb/src/main/java/org/apache/pulsar/io/debezium/mariadb/DebeziumMariaDbSource.java new file mode 100644 index 0000000000..809b894f72 --- /dev/null +++ b/debezium/mariadb/src/main/java/org/apache/pulsar/io/debezium/mariadb/DebeziumMariaDbSource.java @@ -0,0 +1,42 @@ +/* + * 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.mariadb; + +import java.util.Map; +import org.apache.kafka.connect.runtime.ConnectorConfig; +import org.apache.kafka.connect.runtime.TaskConfig; +import org.apache.pulsar.io.debezium.DebeziumSource; + +/** + * A pulsar source that runs debezium mariadb source. + */ +public class DebeziumMariaDbSource extends DebeziumSource { + private static final String DEFAULT_CONNECTOR = "io.debezium.connector.mariadb.MariaDbConnector"; + private static final String DEFAULT_TASK = "io.debezium.connector.mariadb.MariaDbConnectorTask"; + + @Override + public void setDbConnectorClass(Map config) throws Exception { + throwExceptionIfConfigNotMatch(config, ConnectorConfig.CONNECTOR_CLASS_CONFIG, DEFAULT_CONNECTOR); + } + + @Override + public void setDbConnectorTask(Map config) throws Exception { + throwExceptionIfConfigNotMatch(config, TaskConfig.TASK_CLASS_CONFIG, DEFAULT_TASK); + } +} diff --git a/debezium/mariadb/src/main/java/org/apache/pulsar/io/debezium/mariadb/package-info.java b/debezium/mariadb/src/main/java/org/apache/pulsar/io/debezium/mariadb/package-info.java new file mode 100644 index 0000000000..9c9a2abc9b --- /dev/null +++ b/debezium/mariadb/src/main/java/org/apache/pulsar/io/debezium/mariadb/package-info.java @@ -0,0 +1,19 @@ +/* + * 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.mariadb; diff --git a/debezium/mariadb/src/main/resources/META-INF/services/pulsar-io.yaml b/debezium/mariadb/src/main/resources/META-INF/services/pulsar-io.yaml new file mode 100644 index 0000000000..8586681c9f --- /dev/null +++ b/debezium/mariadb/src/main/resources/META-INF/services/pulsar-io.yaml @@ -0,0 +1,22 @@ +# +# 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. +# + +name: debezium-mariadb +description: Debezium MariaDb Source +sourceClass: org.apache.pulsar.io.debezium.mariadb.DebeziumMariaDbSource diff --git a/debezium/mariadb/src/main/resources/debezium-mariadb-source-config.yaml b/debezium/mariadb/src/main/resources/debezium-mariadb-source-config.yaml new file mode 100644 index 0000000000..4a4cbf68dd --- /dev/null +++ b/debezium/mariadb/src/main/resources/debezium-mariadb-source-config.yaml @@ -0,0 +1,41 @@ +# +# 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. +# + +tenant: "test" +namespace: "test-namespace" +name: "debezium-mariadb-source" +topicName: "debezium-mariadb-topic" +archive: "connectors/pulsar-io-debezium-mariadb-2.4.0-SNAPSHOT.nar" + +parallelism: 1 + +configs: + ## config for mariadb, docker image: mariadb:11.4 + database.hostname: "localhost" + database.port: "3306" + database.user: "debezium" + database.password: "dbz" + database.server.id: "184054" + database.include.list: "inventory" + topic.prefix: "dbserver1" + + schema.history.internal.pulsar.topic: "mariadb-history-topic" + schema.history.internal.pulsar.service.url: "pulsar://127.0.0.1:6650" + offset.storage.topic: "mariadb-offset-topic" + connector.class: "io.debezium.connector.mariadb.MariaDbConnector" diff --git a/debezium/mariadb/src/test/java/org/apache/pulsar/io/debezium/mariadb/DebeziumMariaDbSourceConfigTest.java b/debezium/mariadb/src/test/java/org/apache/pulsar/io/debezium/mariadb/DebeziumMariaDbSourceConfigTest.java new file mode 100644 index 0000000000..365f04727b --- /dev/null +++ b/debezium/mariadb/src/test/java/org/apache/pulsar/io/debezium/mariadb/DebeziumMariaDbSourceConfigTest.java @@ -0,0 +1,108 @@ +/* + * 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.mariadb; + +import static org.testng.Assert.assertEquals; +import java.util.HashMap; +import java.util.Map; +import org.apache.kafka.connect.runtime.ConnectorConfig; +import org.apache.kafka.connect.runtime.TaskConfig; +import org.testng.annotations.Test; + +/** + * Unit tests for the connector and task class wiring of {@link DebeziumMariaDbSource}. + * + *

These run without a database: they cover the defaulting and validation behavior that + * {@code DebeziumSource#throwExceptionIfConfigNotMatch} applies to the connector/task class + * config keys. Getting either class name wrong is a silent misconfiguration that would only + * surface at runtime, so it is pinned here. + */ +public class DebeziumMariaDbSourceConfigTest { + + private static final String MARIADB_CONNECTOR = "io.debezium.connector.mariadb.MariaDbConnector"; + private static final String MARIADB_TASK = "io.debezium.connector.mariadb.MariaDbConnectorTask"; + + @Test + public void testConnectorClassDefaultedWhenAbsent() throws Exception { + Map config = new HashMap<>(); + + new DebeziumMariaDbSource().setDbConnectorClass(config); + + assertEquals(config.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG), MARIADB_CONNECTOR); + } + + @Test + public void testTaskClassDefaultedWhenAbsent() throws Exception { + Map config = new HashMap<>(); + + new DebeziumMariaDbSource().setDbConnectorTask(config); + + assertEquals(config.get(TaskConfig.TASK_CLASS_CONFIG), MARIADB_TASK); + } + + @Test + public void testExplicitMatchingConnectorClassIsAccepted() throws Exception { + Map config = new HashMap<>(); + config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, MARIADB_CONNECTOR); + + new DebeziumMariaDbSource().setDbConnectorClass(config); + + assertEquals(config.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG), MARIADB_CONNECTOR); + } + + @Test + public void testExplicitMatchingTaskClassIsAccepted() throws Exception { + Map config = new HashMap<>(); + config.put(TaskConfig.TASK_CLASS_CONFIG, MARIADB_TASK); + + new DebeziumMariaDbSource().setDbConnectorTask(config); + + assertEquals(config.get(TaskConfig.TASK_CLASS_CONFIG), MARIADB_TASK); + } + + /** + * Guards against pointing the MariaDB source at the MySQL connector, which is an easy + * mistake now that both connectors share Debezium's binlog base. + */ + @Test(expectedExceptions = IllegalArgumentException.class) + public void testMismatchedConnectorClassIsRejected() throws Exception { + Map config = new HashMap<>(); + config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, "io.debezium.connector.mysql.MySqlConnector"); + + new DebeziumMariaDbSource().setDbConnectorClass(config); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testMismatchedTaskClassIsRejected() throws Exception { + Map config = new HashMap<>(); + config.put(TaskConfig.TASK_CLASS_CONFIG, "io.debezium.connector.mysql.MySqlConnectorTask"); + + new DebeziumMariaDbSource().setDbConnectorTask(config); + } + + /** + * The class names are string literals, so a typo would compile. Resolve them to prove + * they name real classes on the connector's runtime classpath. + */ + @Test + public void testConnectorAndTaskClassesResolve() throws Exception { + assertEquals(Class.forName(MARIADB_CONNECTOR).getName(), MARIADB_CONNECTOR); + assertEquals(Class.forName(MARIADB_TASK).getName(), MARIADB_TASK); + } +} diff --git a/debezium/mariadb/src/test/java/org/apache/pulsar/io/debezium/mariadb/DebeziumMariaDbSourceTest.java b/debezium/mariadb/src/test/java/org/apache/pulsar/io/debezium/mariadb/DebeziumMariaDbSourceTest.java new file mode 100644 index 0000000000..cc60d488ec --- /dev/null +++ b/debezium/mariadb/src/test/java/org/apache/pulsar/io/debezium/mariadb/DebeziumMariaDbSourceTest.java @@ -0,0 +1,173 @@ +/* + * 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.mariadb; + +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.sql.Connection; +import java.sql.DriverManager; +import java.sql.Statement; +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.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +@Slf4j +public class DebeziumMariaDbSourceTest { + + private static final String PULSAR_IMAGE = + System.getenv().getOrDefault("PULSAR_TEST_IMAGE", "apachepulsar/pulsar:4.1.3"); + + private static final int MARIADB_PORT = 3306; + + private GenericContainer mariadbContainer; + private PulsarContainer pulsarContainer; + private PulsarClient pulsarClient; + private DebeziumMariaDbSource source; + + @BeforeMethod + public void setup() throws Exception { + // GenericContainer rather than MariaDBContainer: debezium-bom pins Testcontainers core + // to 2.x, while the specialized modules still target 1.21.x and reference shaded classes + // removed in 2.x. Root access is also needed for Debezium's REPLICATION SLAVE/CLIENT + // privileges. Same approach as DebeziumMysqlSourceTest. + mariadbContainer = new GenericContainer<>(DockerImageName.parse("mariadb:11.4")) + .withExposedPorts(MARIADB_PORT) + .withEnv("MARIADB_ROOT_PASSWORD", "rootpw") + .withEnv("MARIADB_DATABASE", "testdb") + .withCommand( + "--log-bin=mariadb-bin", + "--binlog-format=ROW", + "--server-id=1" + ) + .waitingFor(Wait.forLogMessage(".*ready for connections.*\\s", 2)); + mariadbContainer.start(); + + pulsarContainer = new PulsarContainer(DockerImageName.parse(PULSAR_IMAGE)); + pulsarContainer.start(); + + pulsarClient = PulsarClient.builder() + .serviceUrl(pulsarContainer.getPulsarBrokerUrl()) + .build(); + + // The MariaDB 3.x driver only registers the jdbc:mariadb: scheme; it no longer + // handles jdbc:mysql: URLs. The driver arrives transitively with the connector. + String jdbcUrl = "jdbc:mariadb://" + mariadbContainer.getHost() + ":" + + mariadbContainer.getMappedPort(MARIADB_PORT) + "/testdb"; + + // Create test table and insert initial data (root has all privileges) + try (Connection conn = DriverManager.getConnection(jdbcUrl, "root", "rootpw"); + Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE products (" + + "id INT NOT NULL AUTO_INCREMENT PRIMARY KEY, " + + "name VARCHAR(255) NOT NULL, " + + "description VARCHAR(512))"); + stmt.execute("INSERT INTO products (name, description) VALUES ('widget', 'A small widget')"); + stmt.execute("INSERT INTO products (name, description) VALUES ('gadget', 'A fancy gadget')"); + } + + source = new DebeziumMariaDbSource(); + } + + @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 (mariadbContainer != null) { + mariadbContainer.stop(); + } + } + + /** + * Bounds the whole test: {@code source.read()} blocks, so a connector that never emits + * would otherwise hang until the CI job's own timeout. + */ + @Test(timeOut = 300_000) + public void testMariaDbCdcEvents() 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-mariadb-test"); + when(sourceContext.getSecret(anyString())).thenReturn(null); + + Map config = new HashMap<>(); + config.put("database.hostname", mariadbContainer.getHost()); + config.put("database.port", String.valueOf(mariadbContainer.getMappedPort(MARIADB_PORT))); + config.put("database.user", "root"); + config.put("database.password", "rootpw"); + config.put("database.server.id", "12345"); + config.put("topic.prefix", "dbserver1"); + config.put("include.schema.changes", "false"); + 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 we inserted. + Awaitility.await() + .atMost(60, TimeUnit.SECONDS) + .pollInterval(500, TimeUnit.MILLISECONDS) + .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/distribution/io/build.gradle.kts b/distribution/io/build.gradle.kts index 66f3a6c7fe..a43d4085d2 100644 --- a/distribution/io/build.gradle.kts +++ b/distribution/io/build.gradle.kts @@ -57,6 +57,7 @@ dependencies { connectorNars(project(":canal")) connectorNars(project(":netty")) connectorNars(project(":mongo")) + connectorNars(project(":debezium:pulsar-io-debezium-mariadb")) connectorNars(project(":debezium:pulsar-io-debezium-mysql")) connectorNars(project(":debezium:pulsar-io-debezium-postgres")) connectorNars(project(":debezium:pulsar-io-debezium-oracle")) diff --git a/distribution/io/src/assemble/io.xml b/distribution/io/src/assemble/io.xml index 3e4ed50d93..01ed22fbe7 100644 --- a/distribution/io/src/assemble/io.xml +++ b/distribution/io/src/assemble/io.xml @@ -68,6 +68,7 @@ ${basedir}/../../pulsar-io/canal/target/pulsar-io-canal-${project.version}.nar ${basedir}/../../pulsar-io/netty/target/pulsar-io-netty-${project.version}.nar ${basedir}/../../pulsar-io/mongo/target/pulsar-io-mongo-${project.version}.nar + ${basedir}/../../pulsar-io/debezium/mariadb/target/pulsar-io-debezium-mariadb-${project.version}.nar ${basedir}/../../pulsar-io/debezium/mysql/target/pulsar-io-debezium-mysql-${project.version}.nar ${basedir}/../../pulsar-io/debezium/postgres/target/pulsar-io-debezium-postgres-${project.version}.nar ${basedir}/../../pulsar-io/debezium/oracle/target/pulsar-io-debezium-oracle-${project.version}.nar diff --git a/docker/pulsar-all/build.gradle.kts b/docker/pulsar-all/build.gradle.kts index 448d3d1f3c..81a0f84858 100644 --- a/docker/pulsar-all/build.gradle.kts +++ b/docker/pulsar-all/build.gradle.kts @@ -60,6 +60,7 @@ dependencies { connectorNars(project(":canal")) connectorNars(project(":netty")) connectorNars(project(":mongo")) + connectorNars(project(":debezium:pulsar-io-debezium-mariadb")) connectorNars(project(":debezium:pulsar-io-debezium-mysql")) connectorNars(project(":debezium:pulsar-io-debezium-postgres")) connectorNars(project(":debezium:pulsar-io-debezium-oracle")) diff --git a/docs/build.gradle.kts b/docs/build.gradle.kts index 0f89397b3f..f8921971c2 100644 --- a/docs/build.gradle.kts +++ b/docs/build.gradle.kts @@ -29,6 +29,7 @@ dependencies { implementation(project(":aerospike")) implementation(project(":canal")) implementation(project(":cassandra")) + implementation(project(":debezium:pulsar-io-debezium-mariadb")) implementation(project(":debezium:pulsar-io-debezium-mysql")) implementation(project(":debezium:pulsar-io-debezium-postgres")) implementation(project(":debezium:pulsar-io-debezium-mongodb")) diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index b55f22d1aa..946640b620 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -461,6 +461,7 @@ elasticsearch-java = { module = "co.elastic.clients:elasticsearch-java", version debezium-bom = { module = "io.debezium:debezium-bom", version.ref = "debezium" } debezium-core = { module = "io.debezium:debezium-core" } debezium-connector-mysql = { module = "io.debezium:debezium-connector-mysql" } +debezium-connector-mariadb = { module = "io.debezium:debezium-connector-mariadb" } debezium-connector-mongodb = { module = "io.debezium:debezium-connector-mongodb" } debezium-connector-postgres = { module = "io.debezium:debezium-connector-postgres" } debezium-connector-oracle = { module = "io.debezium:debezium-connector-oracle" } diff --git a/settings.gradle.kts b/settings.gradle.kts index cdef210517..cd7176f346 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -111,6 +111,8 @@ include("debezium:pulsar-io-debezium-mongodb") project(":debezium:pulsar-io-debezium-mongodb").projectDir = file("debezium/mongodb") include("debezium:pulsar-io-debezium-mssql") project(":debezium:pulsar-io-debezium-mssql").projectDir = file("debezium/mssql") +include("debezium:pulsar-io-debezium-mariadb") +project(":debezium:pulsar-io-debezium-mariadb").projectDir = file("debezium/mariadb") include("debezium:pulsar-io-debezium-mysql") project(":debezium:pulsar-io-debezium-mysql").projectDir = file("debezium/mysql") include("debezium:pulsar-io-debezium-oracle")