[Issue 6957][pulsar-io-debezium] Support CDC Connector for SqlServer

distribution/io/src/assemble/io.xml

@@ -70,6 +70,7 @@
     <file><source>${basedir}/../../pulsar-io/debezium/mysql/target/pulsar-io-debezium-mysql-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/debezium/postgres/target/pulsar-io-debezium-postgres-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/debezium/mongodb/target/pulsar-io-debezium-mongodb-${project.version}.nar</source></file>
+    <file><source>${basedir}/../../pulsar-io/debezium/sqlserver/target/pulsar-io-debezium-sqlserver-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/influxdb/target/pulsar-io-influxdb-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/redis/target/pulsar-io-redis-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/flume/target/pulsar-io-flume-${project.version}.nar</source></file>

pulsar-io/debezium/pom.xml

@@ -36,6 +36,7 @@
     <module>mysql</module>
     <module>postgres</module>
     <module>mongodb</module>
+    <module>sqlserver</module>
   </modules>
 
 </project>

pulsar-io/debezium/sqlserver/pom.xml

@@ -0,0 +1,59 @@
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+
+    <groupId>org.apache.pulsar</groupId>
+    <artifactId>pulsar-io-debezium</artifactId>
+    <version>2.7.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>pulsar-io-debezium-sqlserver</artifactId>
+  <name>Pulsar IO :: Debezium :: sqlserver</name>
+
+  <dependencies>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-io-debezium-core</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>io.debezium</groupId>
+      <artifactId>debezium-connector-sqlserver</artifactId>
+      <version>${debezium.version}</version>
+    </dependency>
+
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.nifi</groupId>
+        <artifactId>nifi-nar-maven-plugin</artifactId>
+      </plugin>
+    </plugins>
+  </build>
+
+</project>

pulsar-io/debezium/sqlserver/src/main/java/org/apache/pulsar/io/debezium/sqlserver/DebeziumSqlServerSource.java

@@ -0,0 +1,36 @@
+/**
+ * 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.sqlserver;
+
+import org.apache.kafka.connect.runtime.TaskConfig;
+import org.apache.pulsar.io.debezium.DebeziumSource;
+
+import java.util.Map;
+
+/**
+ * A pulsar source that runs debezium mongodb source
+ */
+public class DebeziumSqlServerSource extends DebeziumSource {
+    static private final String DEFAULT_TASK = "io.debezium.connector.sqlserver.SqlServerConnectorTask";
+
+    @Override
+    public void setDbConnectorTask(Map<String, Object> config) throws Exception {
+        throwExceptionIfConfigNotMatch(config, TaskConfig.TASK_CLASS_CONFIG, DEFAULT_TASK);
+    }
+}

pulsar-io/debezium/sqlserver/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-sqlserver
+description: Debezium SqlServer Source
+sourceClass: org.apache.pulsar.io.debezium.sqlserver.DebeziumSqlServerSource

pulsar-io/debezium/sqlserver/src/main/resources/debezium-sqlserver-source-config.yaml

@@ -0,0 +1,45 @@
+#
+# 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: "public"
+namespace: "default"
+name: "debezium-sqlserver-source"
+topicName: "debezium-sqlserver-topic"
+archive: "connectors/pulsar-io-debezium-sqlserver-2.4.0-SNAPSHOT.nar"
+
+parallelism: 1
+
+configs:
+  ## config for sqlserver, docker image:  microsoft/mssql-server-linux:2017-CU9-GDR2
+  database.server.name: "server1",
+  database.hostname: "sqlserver",
+  database.port: "1433",
+  database.user: "sa",
+  database.password: "Password!",
+  database.dbname: "testDB",
+  table.whitelist: "inventory",
+  tasks.max: 1
+
+
+  ## PULSAR_SERVICE_URL_CONFIG
+  pulsar.service.url: "pulsar://127.0.0.1:6650"
+  database.history.pulsar.topic: "sqlserver-history-topic"
+  offset.storage.topic: "sqlserver-offset-topic"
+
+

tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumSqlServerContainer.java

@@ -0,0 +1,57 @@
+/**
+ * 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.tests.integration.containers;
+
+
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+
+public class DebeziumSqlServerContainer extends ChaosContainer<DebeziumSqlServerContainer> {
+
+    public static final String NAME = "debezium-sqlserver-example";
+    static final Integer[] PORTS = { 1433 };
+
+    private static final String IMAGE_NAME = "microsoft/mssql-server-linux:2017-CU9-GDR2";
+
+    public DebeziumSqlServerContainer(String clusterName) {
+        super(clusterName, IMAGE_NAME);
+        this.withEnv("ACCEPT_EULA", "y");
+        this.withEnv("MSSQL_PID", "Standard");
+        this.withEnv("SA_PASSWORD", "Password!");
+        this.withEnv("MSSQL_AGENT_ENABLED", "true");
+
+    }
+
+    @Override
+    public String getContainerName() {
+        return clusterName;
+    }
+
+    @Override
+    protected void configure() {
+        super.configure();
+        this.withNetworkAliases(NAME)
+            .withExposedPorts(PORTS)
+            .withCreateContainerCmdModifier(createContainerCmd -> {
+                createContainerCmd.withHostName(NAME);
+                createContainerCmd.withName(getContainerName());
+            })
+            .waitingFor(new HostPortWaitStrategy());
+    }
+
+}

tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java

@@ -54,6 +54,7 @@
 import org.apache.pulsar.tests.integration.containers.DebeziumMongoDbContainer;
 import org.apache.pulsar.tests.integration.containers.DebeziumMySQLContainer;
 import org.apache.pulsar.tests.integration.containers.DebeziumPostgreSqlContainer;
+import org.apache.pulsar.tests.integration.containers.DebeziumSqlServerContainer;
 import org.apache.pulsar.tests.integration.docker.ContainerExecException;
 import org.apache.pulsar.tests.integration.docker.ContainerExecResult;
 import org.apache.pulsar.tests.integration.functions.utils.CommandGenerator;
@@ -62,6 +63,7 @@
 import org.apache.pulsar.tests.integration.io.DebeziumMongoDbSourceTester;
 import org.apache.pulsar.tests.integration.io.DebeziumMySqlSourceTester;
 import org.apache.pulsar.tests.integration.io.DebeziumPostgreSqlSourceTester;
+import org.apache.pulsar.tests.integration.io.DebeziumSqlServerSourceTester;
 import org.apache.pulsar.tests.integration.io.ElasticSearchSinkTester;
 import org.apache.pulsar.tests.integration.io.HdfsSinkTester;
 import org.apache.pulsar.tests.integration.io.JdbcPostgresSinkTester;
@@ -172,6 +174,110 @@ public void testDebeziumMongoDbSource() throws Exception{
         testDebeziumMongoDbConnect("org.apache.kafka.connect.json.JsonConverter", true);
     }
 
+    @Test(groups = "source")
+    public void testDebeziumSqlServerSource() throws Exception {
+        testDebeziumSqlServerConnect("org.apache.kafka.connect.json.JsonConverter", true);
+    }
+
+    private void testDebeziumSqlServerConnect(String converterClassName, boolean jsonWithEnvelope) throws Exception {
+        final String tenant = TopicName.PUBLIC_TENANT;
+        final String namespace = TopicName.DEFAULT_NAMESPACE;
+        final String outputTopicName = "debe-output-topic-name";
+        boolean isJsonConverter = converterClassName.endsWith("JsonConverter");
+        final String consumeTopicName = "debezium/sqlserver-"
+                + (isJsonConverter ? "json" : "avro")
+                + "/dbserver1.testDB.products";
+        final String sourceName = "test-source-debezium-sqlserver" + (isJsonConverter ? "json" : "avro")
+                + "-" + functionRuntimeType + "-" + randomName(8);
+
+        // This is the binlog count that contained in mysql container.
+        final int numMessages = 9;
+
+        if (pulsarCluster == null) {
+            super.setupCluster();
+            super.setupFunctionWorkers();
+        }
+
+        @Cleanup
+        PulsarClient client = PulsarClient.builder()
+                .serviceUrl(pulsarCluster.getPlainTextServiceUrl())
+                .build();
+
+        @Cleanup
+        PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build();
+        initNamespace(admin);
+
+        try {
+            SchemaInfo lastSchemaInfo = admin.schemas().getSchemaInfo(consumeTopicName);
+            log.info("lastSchemaInfo: {}", lastSchemaInfo == null ? "null" : lastSchemaInfo.toString());
+        } catch (Exception e) {
+            log.warn("failed to get schemaInfo for topic: {}, exceptions message: {}",
+                    consumeTopicName, e.getMessage());
+        }
+
+        admin.topics().createNonPartitionedTopic(outputTopicName);
+
+        @Cleanup
+        DebeziumSqlServerSourceTester sourceTester = new DebeziumSqlServerSourceTester(pulsarCluster, converterClassName);
+        sourceTester.getSourceConfig().put("json-with-envelope", jsonWithEnvelope);
+
+        // setup debezium sqlserver server
+        DebeziumSqlServerContainer sqlServerContainer = new DebeziumSqlServerContainer(pulsarCluster.getClusterName());
+        sourceTester.setServiceContainer(sqlServerContainer);
+
+        // prepare the testing environment for source
+        prepareSource(sourceTester);
+
+        // submit the source connector
+        submitSourceConnector(sourceTester, tenant, namespace, sourceName, outputTopicName);
+
+        // get source info
+        getSourceInfoSuccess(sourceTester, tenant, namespace, sourceName);
+
+        // get source status
+        Failsafe.with(statusRetryPolicy).run(() -> getSourceStatus(tenant, namespace, sourceName));
+
+        // wait for source to process messages
+        Failsafe.with(statusRetryPolicy).run(() ->
+                waitForProcessingSourceMessages(tenant, namespace, sourceName, numMessages));
+
+        @Cleanup
+        Consumer consumer = client.newConsumer(getSchema(jsonWithEnvelope))
+                .topic(consumeTopicName)
+                .subscriptionName("debezium-source-tester")
+                .subscriptionType(SubscriptionType.Exclusive)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscribe();
+        log.info("[debezium sqlserver test] create consumer finish. converterName: {}", converterClassName);
+
+        // validate the source result
+        sourceTester.validateSourceResult(consumer, 9, null, converterClassName);
+
+        // prepare insert event
+        sourceTester.prepareInsertEvent();
+
+        // validate the source insert event
+        sourceTester.validateSourceResult(consumer, 1, SourceTester.INSERT, converterClassName);
+
+        // prepare update event
+        sourceTester.prepareUpdateEvent();
+
+        // validate the source update event
+        sourceTester.validateSourceResult(consumer, 1, SourceTester.UPDATE, converterClassName);
+
+        // prepare delete event
+        sourceTester.prepareDeleteEvent();
+
+        // validate the source delete event
+        sourceTester.validateSourceResult(consumer, 1, SourceTester.DELETE, converterClassName);
+
+        // delete the source
+        deleteSource(tenant, namespace, sourceName);
+
+        // get source info (source should be deleted)
+        getSourceInfoNotFound(tenant, namespace, sourceName);
+    }
+
     private void testSink(SinkTester tester, boolean builtin) throws Exception {
         if (pulsarCluster == null) {
             super.setupCluster();