This is an automated email from the ASF dual-hosted git repository.
david-streamlio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new 5cecc7a5 [improve][test] Add MariaDB JDBC sink integration test (#59)
5cecc7a5 is described below
commit 5cecc7a597a626fdac019652c3ff020500486dec
Author: David Kjerrumgaard <[email protected]>
AuthorDate: Thu Jul 9 13:18:33 2026 -0700
[improve][test] Add MariaDB JDBC sink integration test (#59)
The jdbc/mariadb module had no tests. Add a Testcontainers-based
integration test that runs the MariadbJdbcAutoSchemaSink against a
real MariaDB 11.4 server: the table is created over plain JDBC, a
record is written through the sink's Avro schema path with
transactions enabled, and the row is read back over JDBC to verify
the write.
Also adds a testcontainers-mariadb entry to the version catalog
(version resolved via the testcontainers BOM imported by the
pulsar-connectors-dependencies platform).
---
gradle/libs.versions.toml | 1 +
jdbc/mariadb/build.gradle.kts | 5 +
.../io/jdbc/MariadbJdbcSinkIntegrationTest.java | 180 +++++++++++++++++++++
3 files changed, 186 insertions(+)
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index b55f22d1..ab230c87 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -423,6 +423,7 @@ testcontainers-localstack = { module =
"org.testcontainers:localstack" }
testcontainers-rabbitmq = { module = "org.testcontainers:rabbitmq" }
testcontainers-kafka = { module = "org.testcontainers:kafka" }
testcontainers-mysql = { module = "org.testcontainers:mysql" }
+testcontainers-mariadb = { module = "org.testcontainers:mariadb" }
testcontainers-clickhouse = { module = "org.testcontainers:clickhouse" }
testcontainers-postgresql = { module = "org.testcontainers:postgresql" }
testcontainers-mongodb = { module = "org.testcontainers:mongodb" }
diff --git a/jdbc/mariadb/build.gradle.kts b/jdbc/mariadb/build.gradle.kts
index ce5fe962..a2d43e96 100644
--- a/jdbc/mariadb/build.gradle.kts
+++ b/jdbc/mariadb/build.gradle.kts
@@ -24,4 +24,9 @@ plugins {
dependencies {
implementation(project(":jdbc:pulsar-io-jdbc-core"))
runtimeOnly(libs.mariadb.jdbc)
+
+ testImplementation(libs.testcontainers.mariadb)
+ testImplementation(libs.pulsar.client)
+ testImplementation(libs.pulsar.functions.instance)
+ testImplementation(libs.avro)
}
diff --git
a/jdbc/mariadb/src/test/java/org/apache/pulsar/io/jdbc/MariadbJdbcSinkIntegrationTest.java
b/jdbc/mariadb/src/test/java/org/apache/pulsar/io/jdbc/MariadbJdbcSinkIntegrationTest.java
new file mode 100644
index 00000000..1039dc1c
--- /dev/null
+++
b/jdbc/mariadb/src/test/java/org/apache/pulsar/io/jdbc/MariadbJdbcSinkIntegrationTest.java
@@ -0,0 +1,180 @@
+/*
+ * 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.jdbc;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.SchemaDefinition;
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.schema.AutoConsumeSchema;
+import org.apache.pulsar.client.impl.schema.AvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.functions.source.PulsarRecord;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+import org.testcontainers.containers.MariaDBContainer;
+import org.testcontainers.utility.DockerImageName;
+
+/**
+ * Integration test for the MariaDB JDBC sink, exercised against a real
MariaDB server
+ * via Testcontainers.
+ *
+ * <p>The sink is opened against a pre-created table, a record is written
through the
+ * connector's Avro schema path, and the resulting row is read back over plain
JDBC
+ * to verify that the write actually reached the database.
+ */
+@Slf4j
+public class MariadbJdbcSinkIntegrationTest {
+
+ private static final DockerImageName MARIADB_IMAGE =
DockerImageName.parse("mariadb:11.4");
+
+ private static final String TABLE_NAME = "pulsar_messages";
+
+ private MariaDBContainer<?> mariadb;
+ private BaseJdbcAutoSchemaSink jdbcSink;
+
+ /**
+ * A simple record class matching the {@link #TABLE_NAME} columns.
+ */
+ @Data
+ @NoArgsConstructor
+ @AllArgsConstructor
+ public static class Foo {
+ private String field1;
+ private String field2;
+ private int field3;
+ }
+
+ @BeforeClass(alwaysRun = true)
+ public void setUp() throws Exception {
+ mariadb = new MariaDBContainer<>(MARIADB_IMAGE);
+ mariadb.start();
+
+ // Create the destination table over plain JDBC.
+ try (Connection connection = openConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE " + TABLE_NAME + " ("
+ + " field1 VARCHAR(255),"
+ + " field2 VARCHAR(255),"
+ + " field3 INT PRIMARY KEY"
+ + ")");
+ }
+
+ Map<String, Object> conf = new HashMap<>();
+ conf.put("jdbcUrl", mariadb.getJdbcUrl());
+ conf.put("userName", mariadb.getUsername());
+ conf.put("password", mariadb.getPassword());
+ conf.put("tableName", TABLE_NAME);
+ conf.put("key", "field3");
+ conf.put("nonKey", "field1,field2");
+ // MariaDB supports JDBC transactions; commit on each flush.
+ conf.put("useTransactions", true);
+ // Flush on each write.
+ conf.put("batchSize", 1);
+
+ jdbcSink = new MariadbJdbcAutoSchemaSink();
+ jdbcSink.open(conf, null);
+ }
+
+ @AfterClass(alwaysRun = true)
+ public void tearDown() throws Exception {
+ if (jdbcSink != null) {
+ jdbcSink.close();
+ }
+ if (mariadb != null) {
+ mariadb.stop();
+ }
+ }
+
+ @Test
+ public void testOpenDiscoversTableAndWrites() throws Exception {
+ // prepare a foo Record
+ Foo insertObj = new Foo("ValueOfField1", "ValueOfField2", 3);
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ final Record<GenericObject> record = createMockFooRecord(insertObj,
future);
+
+ jdbcSink.write(record);
+ log.info("executed write");
+
+ // wait for the backend flush to complete and the record to be acked
+ Assert.assertTrue(future.get(30, TimeUnit.SECONDS), "record should be
acknowledged");
+
+ // value has been written to MariaDB, read it back over JDBC and
verify.
+ try (Connection connection = openConnection();
+ PreparedStatement statement = connection.prepareStatement(
+ "SELECT field1, field2, field3 FROM " + TABLE_NAME + "
WHERE field3 = ?")) {
+ statement.setInt(1, insertObj.getField3());
+ try (ResultSet resultSet = statement.executeQuery()) {
+ Assert.assertTrue(resultSet.next(), "a matching row should
have been written");
+ Assert.assertEquals(resultSet.getString("field1"),
insertObj.getField1());
+ Assert.assertEquals(resultSet.getString("field2"),
insertObj.getField2());
+ Assert.assertEquals(resultSet.getInt("field3"),
insertObj.getField3());
+ Assert.assertFalse(resultSet.next(), "exactly one matching row
should have been written");
+ }
+ }
+ }
+
+ private Connection openConnection() throws Exception {
+ return DriverManager.getConnection(
+ mariadb.getJdbcUrl(), mariadb.getUsername(),
mariadb.getPassword());
+ }
+
+ @SuppressWarnings("unchecked")
+ private Record<GenericObject> createMockFooRecord(Foo record,
CompletableFuture<Boolean> future) {
+ Message<GenericRecord> insertMessage = mock(MessageImpl.class);
+ AvroSchema<Foo> schema = AvroSchema.of(SchemaDefinition.<Foo>builder()
+ .withPojo(Foo.class).withAlwaysAllowNull(true).build());
+ AutoConsumeSchema autoConsumeSchema = new AutoConsumeSchema();
+ autoConsumeSchema.setSchema(schema);
+
+ byte[] insertBytes = schema.encode(record);
+
+ Record<? extends GenericObject> insertRecord =
PulsarRecord.<GenericRecord>builder()
+ .message(insertMessage)
+ .topicName("fake_topic_name")
+ .schema(autoConsumeSchema)
+ .ackFunction(() -> future.complete(true))
+ .failFunction(() -> future.complete(false))
+ .build();
+
+ GenericAvroSchema genericAvroSchema = new
GenericAvroSchema(schema.getSchemaInfo());
+
when(insertMessage.getValue()).thenReturn(genericAvroSchema.decode(insertBytes));
+ when(insertMessage.getProperties()).thenReturn(new HashMap<>());
+ return (Record<GenericObject>) insertRecord;
+ }
+}
