From de74ef1fc72fbb8348c043b07eeee9ae645a3d79 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Fri, 10 Jul 2026 19:10:08 -0700 Subject: [PATCH] [improve][test] Add Canal source integration test Fixes #48 Stand up a full MySQL-binlog CDC pipeline with Testcontainers and assert that CanalStringSource delivers the change event. The test runs two containers on a shared network: mysql:8.0 with ROW binlog and canal/canal-server:v1.1.7 (matching the module's canal.client/canal.protocol 1.1.7). canal reaches MySQL via a network alias; the source reaches canal via the mapped 11111 port. An INSERT made after the source subscribes is read back on a bounded worker thread with a deadline, so a missing event fails fast instead of hanging CI. The test-only mysql-connector-j dependency excludes protobuf-java: it drags protobuf 4.x (for X DevAPI, unused by plain JDBC) onto the test classpath, where it outranks the platform's 3.25.5. canal's generated CanalPacket code calls GeneratedMessageV3.makeExtensionsImmutable(), which protobuf 4.x removed, so the canal client throws NoSuchMethodError on connect() under protobuf 4. With the exclusion the test runs against the same protobuf line the Pulsar runtime provides to the NAR (which still has the method), matching production. See #99 for the analysis. --- canal/build.gradle.kts | 10 + .../CanalStringSourceIntegrationTest.java | 244 ++++++++++++++++++ .../test/resources/canal/instance.properties | 56 ++++ 3 files changed, 310 insertions(+) create mode 100644 canal/src/test/java/org/apache/pulsar/io/canal/CanalStringSourceIntegrationTest.java create mode 100644 canal/src/test/resources/canal/instance.properties diff --git a/canal/build.gradle.kts b/canal/build.gradle.kts index 29397beeb3..fa96bb00b6 100644 --- a/canal/build.gradle.kts +++ b/canal/build.gradle.kts @@ -35,4 +35,14 @@ dependencies { implementation("com.alibaba.otter:canal.protocol:1.1.7") implementation("com.alibaba.otter:canal.client:1.1.7") implementation(libs.log4j.core) + + testImplementation(libs.testcontainers) + // Exclude protobuf: mysql-connector-j drags protobuf-java 4.x (X DevAPI, unused by plain JDBC) + // onto the test classpath, where it outranks the platform's 3.25.5. canal's generated + // CanalPacket code calls GeneratedMessageV3.makeExtensionsImmutable(), which protobuf 4.x + // removed, so the canal client would throw NoSuchMethodError on connect(). With the exclusion + // the test runs against the same protobuf line the Pulsar runtime provides to the NAR. + testImplementation(libs.mysql.connector.j) { + exclude(group = "com.google.protobuf", module = "protobuf-java") + } } diff --git a/canal/src/test/java/org/apache/pulsar/io/canal/CanalStringSourceIntegrationTest.java b/canal/src/test/java/org/apache/pulsar/io/canal/CanalStringSourceIntegrationTest.java new file mode 100644 index 0000000000..c124b5ec23 --- /dev/null +++ b/canal/src/test/java/org/apache/pulsar/io/canal/CanalStringSourceIntegrationTest.java @@ -0,0 +1,244 @@ +/* + * 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.canal; + +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.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.core.SourceContext; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.containers.output.Slf4jLogConsumer; +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; + +/** + * Integration test for {@link CanalStringSource} that stands up a full MySQL-binlog CDC pipeline + * with Testcontainers: + * + * 
+ *   MySQL (mysql:8.0, ROW binlog)  <--dump--  canal-server (canal/canal-server:v1.1.7)  <--tcp--  CanalStringSource
+ *
+ * + *

The two containers share a {@link Network}; canal reaches MySQL through the {@code mysql} + * network alias, and the source (running in this JVM) reaches canal through the mapped 11111 port. + * canal-server v1.1.7 matches the {@code canal.client}/{@code canal.protocol} 1.1.7 dependency of + * the module. The image does no env-var substitution, so the instance configuration is mounted from + * {@code src/test/resources/canal/instance.properties}. + */ +@Slf4j +public class CanalStringSourceIntegrationTest { + + private static final DockerImageName MYSQL_IMAGE = DockerImageName.parse("mysql:8.0"); + private static final DockerImageName CANAL_IMAGE = DockerImageName.parse("canal/canal-server:v1.1.7"); + + private static final int MYSQL_PORT = 3306; + private static final int CANAL_PORT = 11111; + + /** + * Deadline for a single bounded {@code read()}. {@code PushSource.read()} blocks until a record + * is pushed, so it must run on a worker thread with a deadline: an under-delivering pipeline then + * surfaces as a prompt, diagnosable failure instead of hanging the CI job to its own timeout. + */ + private static final int READ_TIMEOUT_SECONDS = 180; + + private Network network; + private GenericContainer mysqlContainer; + private GenericContainer canalContainer; + private CanalStringSource source; + private ExecutorService readerExecutor; + + @BeforeMethod + public void setup() throws Exception { + readerExecutor = Executors.newSingleThreadExecutor(r -> { + Thread t = new Thread(r, "canal-test-reader"); + t.setDaemon(true); + return t; + }); + + network = Network.newNetwork(); + + // GenericContainer (not MySQLContainer) so we keep full root access with the native + // password plugin canal 1.1.x expects for the binlog dump handshake. + mysqlContainer = new GenericContainer<>(MYSQL_IMAGE) + .withNetwork(network) + .withNetworkAliases("mysql") + .withExposedPorts(MYSQL_PORT) + .withEnv("MYSQL_ROOT_PASSWORD", "rootpw") + .withEnv("MYSQL_DATABASE", "testdb") + .withCommand( + "--default-authentication-plugin=mysql_native_password", + "--log-bin=mysql-bin", + "--binlog-format=ROW", + "--server-id=1") + .waitingFor(Wait.forLogMessage(".*ready for connections.*\\s", 2) + .withStartupTimeout(Duration.ofMinutes(3))); + mysqlContainer.start(); + + // Create the tracked table BEFORE canal starts, so its DDL is not part of the binlog + // stream canal reads; only the INSERT made after subscription should be delivered. + try (Connection conn = openMysql(); + Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE testdb.products (" + + "id INT NOT NULL AUTO_INCREMENT PRIMARY KEY, " + + "name VARCHAR(255) NOT NULL, " + + "description VARCHAR(512))"); + } + + canalContainer = new GenericContainer<>(CANAL_IMAGE) + .withNetwork(network) + .withExposedPorts(CANAL_PORT) + .withCopyFileToContainer( + MountableFile.forClasspathResource("canal/instance.properties"), + "/home/admin/canal-server/conf/example/instance.properties") + .withLogConsumer(new Slf4jLogConsumer(log).withPrefix("canal-server")) + .waitingFor(Wait.forLogMessage(".*START SUCCESSFUL.*", 1) + .withStartupTimeout(Duration.ofMinutes(3))); + canalContainer.start(); + + source = new CanalStringSource(); + } + + @AfterMethod(alwaysRun = true) + public void cleanup() throws Exception { + if (readerExecutor != null) { + // shutdownNow: a reader may still be blocked in read(), which never returns null. + readerExecutor.shutdownNow(); + } + if (source != null) { + try { + source.close(); + } catch (Exception e) { + log.warn("Failed to close source", e); + } + } + if (canalContainer != null) { + canalContainer.stop(); + } + if (mysqlContainer != null) { + mysqlContainer.stop(); + } + if (network != null) { + network.close(); + } + } + + @Test(timeOut = 600_000) + public void testCanalCdcEvents() throws Exception { + SourceContext sourceContext = mock(SourceContext.class); + when(sourceContext.getSourceName()).thenReturn("canal-string-source-test"); + // Sensitive fields (username/password) are looked up as secrets first; return null so the + // loader falls back to the plain config values below. + when(sourceContext.getSecret(anyString())).thenReturn(null); + + Map config = new HashMap<>(); + config.put("cluster", false); + // zkServers is a @FieldDoc(required = true) with an empty default, so it must be present in + // the map even for single (non-cluster) mode or the config loader rejects it. + config.put("zkServers", ""); + config.put("destination", "example"); + config.put("singleHostname", canalContainer.getHost()); + config.put("singlePort", canalContainer.getMappedPort(CANAL_PORT)); + // canal.user/canal.passwd are unset in the server, so no client auth is required. + config.put("username", ""); + config.put("password", ""); + config.put("batchSize", 1000); + + source.open(config, sourceContext); + + // Give the source thread time to connect + subscribe to canal before the mutation, so the + // INSERT lands in the stream canal delivers to this subscriber. + Thread.sleep(5000); + + try (Connection conn = openMysql(); + Statement stmt = conn.createStatement()) { + stmt.execute("INSERT INTO testdb.products (name, description) " + + "VALUES ('canal-widget', 'inserted after subscription')"); + } + + Record record = readOne(); + assertNotNull(record.getValue(), "CDC record value should not be null"); + String message = record.getValue().getMessage(); + log.info("Received CDC record: key={}, message={}", record.getKey().orElse(null), message); + assertNotNull(message, "CDC message payload should not be null"); + // The FlatMessage JSON carries the table name and the inserted row values. + assertTrue(message.contains("products"), + "CDC message should reference the products table, but was: " + message); + assertTrue(message.contains("canal-widget"), + "CDC message should carry the inserted row value, but was: " + message); + record.ack(); + } + + private String jdbcUrl() { + return "jdbc:mysql://" + mysqlContainer.getHost() + ":" + + mysqlContainer.getMappedPort(MYSQL_PORT) + "/testdb"; + } + + /** + * Opens a MySQL connection, retrying briefly: the container logs "ready for connections" during + * its temporary init server before the real TCP listener is fully up, so the first attempts can + * hit an EOF mid-handshake. + */ + private Connection openMysql() throws Exception { + Exception last = null; + for (int attempt = 0; attempt < 30; attempt++) { + try { + return DriverManager.getConnection(jdbcUrl(), "root", "rootpw"); + } catch (Exception e) { + last = e; + Thread.sleep(2000); + } + } + throw new AssertionError("Could not connect to MySQL after retries", last); + } + + /** + * Reads a single record on a worker thread, failing (not hanging) if none arrives in time. + */ + private Record readOne() throws Exception { + Future> future = readerExecutor.submit(() -> source.read()); + try { + return future.get(READ_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (TimeoutException e) { + future.cancel(true); + throw new AssertionError("Timed out after " + READ_TIMEOUT_SECONDS + + "s waiting for a CDC record from canal. The pipeline produced no record; " + + "see the canal-server logs above.", e); + } + } +} diff --git a/canal/src/test/resources/canal/instance.properties b/canal/src/test/resources/canal/instance.properties new file mode 100644 index 0000000000..373c7aaa23 --- /dev/null +++ b/canal/src/test/resources/canal/instance.properties @@ -0,0 +1,56 @@ +# +# 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. +# +################################################# +## Canal instance configuration for the integration test. +## Mounted into /home/admin/canal-server/conf/example/instance.properties. +## The v1.1.7 docker image performs no env-var substitution, so the MySQL +## coordinates are baked in here and point at the "mysql" network alias. +################################################# + +# enable gtid use true/false +canal.instance.gtidon=false + +# position info - empty journal name makes canal start from the current +# binlog position when the instance boots, so only mutations made after +# start-up are captured. +canal.instance.master.address=mysql:3306 +canal.instance.master.journal.name= +canal.instance.master.position= +canal.instance.master.timestamp= +canal.instance.master.gtid= + +# table meta tsdb - disabled to avoid the extra h2/mysql tsdb wiring; canal +# fetches table metadata live from MySQL instead. +canal.instance.tsdb.enable=false + +# username/password used by canal to open the binlog dump connection +canal.instance.dbUsername=root +canal.instance.dbPassword=rootpw +canal.instance.connectionCharset=UTF-8 +canal.instance.enableDruid=false + +# only track the test schema +canal.instance.filter.regex=testdb\\..* +canal.instance.filter.black.regex=mysql\\.slave_.* + +# mq config (unused in tcp mode, kept for parity with the shipped default) +canal.mq.topic=example +canal.mq.partition=0 + +canal.instance.multi.stream.on=false