diff --git a/.github/workflows/pulsar-ci.yaml b/.github/workflows/pulsar-ci.yaml index 721a1d2eafc72..160188e1807f3 100644 --- a/.github/workflows/pulsar-ci.yaml +++ b/.github/workflows/pulsar-ci.yaml @@ -459,12 +459,85 @@ jobs: continue-on-error: true with: action: wait + pulsar-debezium-db2-test-image: + name: Build Pulsar debezium-db2-test-image docker image + runs-on: ubuntu-20.04 + timeout-minutes: 60 + needs: ['preconditions', 'build-and-license-check'] + if: ${{ needs.preconditions.outputs.docs_only != 'true'}} + steps: + - name: checkout + uses: actions/checkout@v3 + + - name: Tune Runner VM + uses: ./.github/actions/tune-runner-vm + + - name: Configure Gradle Enterprise + uses: ./.github/actions/gradle-enterprise + with: + token: ${{ secrets.GE_ACCESS_TOKEN }} + + - name: Setup ssh access to build runner VM + # ssh access is enabled for builds in own forks + if: ${{ github.repository != 'apache/pulsar' && github.event_name == 'pull_request' }} + uses: ./.github/actions/ssh-access + continue-on-error: true + with: + limit-access-to-actor: true + - name: Cache Maven dependencies + uses: actions/cache@v3 + timeout-minutes: 5 + with: + path: | + ~/.m2/repository/*/*/* + !~/.m2/repository/org/apache/pulsar + key: ${{ runner.os }}-m2-dependencies-core-modules-${{ hashFiles('**/pom.xml') }} + restore-keys: | + ${{ runner.os }}-m2-dependencies-core-modules- + + - name: Set up JDK 17 + uses: actions/setup-java@v3 + with: + distribution: 'temurin' + java-version: 17 + + - name: Install gh-actions-artifact-client.js + uses: apache/pulsar-test-infra/gh-actions-artifact-client/dist@master + + - name: restore maven build results from Github artifact cache + run: | + cd $HOME + $GITHUB_WORKSPACE/build/pulsar_ci_tool.sh restore_tar_from_github_actions_artifacts pulsar-maven-repository-binaries + + - name: Pick ubuntu mirror for the docker image build + run: | + # pick the closest ubuntu mirror and set it to UBUNTU_MIRROR environment variable + $GITHUB_WORKSPACE/build/pulsar_ci_tool.sh pick_ubuntu_mirror + + - name: Build debezium-db2-test-image docker image + run: | + # build docker image + mvn -B -am -pl tests/docker-images/debezium-db2-test-image install -Pcore-modules,-main,integrationTests,docker \ + -Dmaven.test.skip=true -Ddocker.squash=true -DskipSourceReleaseAssembly=true \ + -Dspotbugs.skip=true -Dlicense.skip=true -Dcheckstyle.skip=true -Drat.skip=true + + - name: save docker image apachepulsar/debezium-db2-test-image:latest to Github artifact cache + run: | + $GITHUB_WORKSPACE/build/pulsar_ci_tool.sh docker_save_image_to_github_actions_artifacts apachepulsar/debezium-db2-test-image:latest pulsar-debezium-db2-test-image + + - name: Wait for ssh connection when build fails + # ssh access is enabled for builds in own forks + uses: ./.github/actions/ssh-access + if: ${{ failure() && github.repository != 'apache/pulsar' && github.event_name == 'pull_request' }} + continue-on-error: true + with: + action: wait integration-tests: name: CI - Integration - ${{ matrix.name }} runs-on: ubuntu-20.04 timeout-minutes: ${{ matrix.timeout || 60 }} - needs: ['preconditions', 'pulsar-java-test-image'] + needs: ['preconditions', 'pulsar-java-test-image', 'pulsar-debezium-db2-test-image'] if: ${{ needs.preconditions.outputs.docs_only != 'true' }} env: JOB_NAME: CI - Integration - ${{ matrix.name }} @@ -556,6 +629,10 @@ jobs: run: | $GITHUB_WORKSPACE/build/pulsar_ci_tool.sh docker_load_image_from_github_actions_artifacts pulsar-java-test-image + - name: Load docker image apachepulsar/debezium-db2-test-image:latest from Github artifact cache + run: | + $GITHUB_WORKSPACE/build/pulsar_ci_tool.sh docker_load_image_from_github_actions_artifacts pulsar-debezium-db2-test-image + - name: Run setup commands if: ${{ matrix.setup }} run: | @@ -672,6 +749,10 @@ jobs: run: | $GITHUB_WORKSPACE/build/pulsar_ci_tool.sh docker_load_image_from_github_actions_artifacts pulsar-java-test-image + - name: Load docker image apachepulsar/debezium-db2-test-image:latest from Github artifact cache + run: | + $GITHUB_WORKSPACE/build/pulsar_ci_tool.sh docker_load_image_from_github_actions_artifacts pulsar-debezium-db2-test-image + - name: Restore coverage files from build artifacts and create Jacoco reports run: | $GITHUB_WORKSPACE/build/pulsar_ci_tool.sh restore_inttest_coverage_files @@ -705,7 +786,7 @@ jobs: action: wait delete-integration-test-docker-image-artifact: - name: "Delete integration test docker image artifact" + name: "Delete integration test docker image artifacts" runs-on: ubuntu-20.04 timeout-minutes: 10 needs: [ @@ -733,6 +814,10 @@ jobs: run: | gh-actions-artifact-client.js delete pulsar-java-test-image.zst + - name: Delete db2 docker image from GitHub Actions Artifacts + run: | + gh-actions-artifact-client.js delete pulsar-debezium-db2-test-image.zst + pulsar-test-latest-version-image: name: Build Pulsar docker image runs-on: ubuntu-20.04 @@ -845,7 +930,7 @@ jobs: name: CI - System - ${{ matrix.name }} runs-on: ubuntu-20.04 timeout-minutes: 60 - needs: ['preconditions', 'pulsar-test-latest-version-image'] + needs: ['preconditions', 'pulsar-test-latest-version-image', 'pulsar-debezium-db2-test-image'] if: ${{ needs.preconditions.outputs.docs_only != 'true' }} env: JOB_NAME: CI - System - ${{ matrix.name }} @@ -929,6 +1014,10 @@ jobs: run: | $GITHUB_WORKSPACE/build/pulsar_ci_tool.sh docker_load_image_from_github_actions_artifacts pulsar-test-latest-version-image + - name: Load docker image apachepulsar/debezium-db2-test-image:latest from Github artifact cache + run: | + $GITHUB_WORKSPACE/build/pulsar_ci_tool.sh docker_load_image_from_github_actions_artifacts pulsar-debezium-db2-test-image + - name: Run setup commands if: ${{ matrix.setup }} run: | diff --git a/build/run_integration_group.sh b/build/run_integration_group.sh index bc1255d8d68aa..87d0bbeafb2e5 100755 --- a/build/run_integration_group.sh +++ b/build/run_integration_group.sh @@ -215,6 +215,10 @@ test_group_pulsar_io_ora() { mvn_run_integration_test "$@" -DintegrationTestSuiteFile=pulsar-io-ora-source.xml -DintegrationTests -Dgroups=source -DtestRetryCount=0 } +test_group_pulsar_io_db2() { + mvn_run_integration_test "$@" -DintegrationTestSuiteFile=pulsar-io-db2-source.xml -DintegrationTests -Dgroups=source -DtestRetryCount=0 +} + list_test_groups() { declare -F | awk '{print $NF}' | sort | grep -E '^test_group_' | sed 's/^test_group_//g' | tr '[:lower:]' '[:upper:]' } diff --git a/distribution/io/src/assemble/io.xml b/distribution/io/src/assemble/io.xml index 5b652170fdbb5..c215e3b323684 100644 --- a/distribution/io/src/assemble/io.xml +++ b/distribution/io/src/assemble/io.xml @@ -73,6 +73,7 @@ ${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 + ${basedir}/../../pulsar-io/debezium/db2/target/pulsar-io-debezium-db2-${project.version}.nar ${basedir}/../../pulsar-io/debezium/mssql/target/pulsar-io-debezium-mssql-${project.version}.nar ${basedir}/../../pulsar-io/debezium/mongodb/target/pulsar-io-debezium-mongodb-${project.version}.nar ${basedir}/../../pulsar-io/influxdb/target/pulsar-io-influxdb-${project.version}.nar diff --git a/pulsar-io/debezium/db2/pom.xml b/pulsar-io/debezium/db2/pom.xml new file mode 100644 index 0000000000000..081e83fe31ec7 --- /dev/null +++ b/pulsar-io/debezium/db2/pom.xml @@ -0,0 +1,65 @@ + + + 4.0.0 + + org.apache.pulsar + pulsar-io-debezium + 3.0.0-SNAPSHOT + + + pulsar-io-debezium-db2 + Pulsar IO :: Debezium :: DB2 + + + + + ${project.groupId} + pulsar-io-core + ${project.version} + provided + + + + ${project.groupId} + pulsar-io-debezium-core + ${project.version} + + + + io.debezium + debezium-connector-db2 + ${debezium.version} + + + + + + + + org.apache.nifi + nifi-nar-maven-plugin + + + + + diff --git a/pulsar-io/debezium/db2/src/main/java/org/apache/pulsar/io/debezium/db2/DebeziumDB2Source.java b/pulsar-io/debezium/db2/src/main/java/org/apache/pulsar/io/debezium/db2/DebeziumDB2Source.java new file mode 100644 index 0000000000000..d59d1d456a0d5 --- /dev/null +++ b/pulsar-io/debezium/db2/src/main/java/org/apache/pulsar/io/debezium/db2/DebeziumDB2Source.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.db2; + +import java.util.Map; +import org.apache.kafka.connect.runtime.TaskConfig; +import org.apache.pulsar.io.debezium.DebeziumSource; + + +/** + * A pulsar source that runs debezium oracle source. + */ +public class DebeziumDB2Source extends DebeziumSource { + private static final String DEFAULT_TASK = "io.debezium.connector.db2.Db2ConnectorTask"; + + @Override + public void setDbConnectorTask(Map config) throws Exception { + throwExceptionIfConfigNotMatch(config, TaskConfig.TASK_CLASS_CONFIG, DEFAULT_TASK); + } +} diff --git a/pulsar-io/debezium/db2/src/main/java/org/apache/pulsar/io/debezium/db2/package-info.java b/pulsar-io/debezium/db2/src/main/java/org/apache/pulsar/io/debezium/db2/package-info.java new file mode 100644 index 0000000000000..89a015103457c --- /dev/null +++ b/pulsar-io/debezium/db2/src/main/java/org/apache/pulsar/io/debezium/db2/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.db2; \ No newline at end of file diff --git a/pulsar-io/debezium/db2/src/main/resources/META-INF/services/pulsar-io.yaml b/pulsar-io/debezium/db2/src/main/resources/META-INF/services/pulsar-io.yaml new file mode 100644 index 0000000000000..9aeba132fea99 --- /dev/null +++ b/pulsar-io/debezium/db2/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-db2 +description: Debezium DB2 Source +sourceClass: org.apache.pulsar.io.debezium.db2.DebeziumDB2Source diff --git a/pulsar-io/debezium/db2/src/main/resources/debezium-db2-source-config.yaml b/pulsar-io/debezium/db2/src/main/resources/debezium-db2-source-config.yaml new file mode 100644 index 0000000000000..59842cd851b78 --- /dev/null +++ b/pulsar-io/debezium/db2/src/main/resources/debezium-db2-source-config.yaml @@ -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. +# + +## config for db2, docker image: ibmcom/db2, which was moved to icr.io/db2_community/db2 +## As per note at https://hub.docker.com/r/ibmcom/db2 +## docs are being moved to: https://www.ibm.com/docs/en/db2/11.5?topic=deployments-db2-community-edition-docker +## For testing, you can use this image that was configured with the required libraries: apachepulsar/debezium-db2-test-image:latest +## docker run -itd --name mydb2 --privileged=true -p 50000:50000 -e LICENSE=accept -e DB2INST1_PASSWORD=admin -e DBNAME=testdb -v :/database ibmcom/db2 + +tenant: "public" +namespace: "default" +name: "debezium-db2-source" +inputs: [ "db2-connect-topic" ] +topicName: "db2-connect-topic" +archive: "connectors/pulsar-io-debezium-db2-2.11.0-SNAPSHOT.nar" +parallelism: 1 +configs: + database.hostname: "localhost" + database.port: "50000" + database.user: "db2inst1" + database.password: "admin" + database.dbname: "mydb2" + database.server.name: "db2inst1" + topic.prefix: "stores" # Example for hypothetical stores topics + table.exclude.list: "DB2INST1.EXCLUDEDSTORES" # Example for hypothetical stores table + + database.history: "org.apache.pulsar.io.debezium.PulsarDatabaseHistory" + database.history.pulsar.topic: "debezium-db2-source-history-topic" + database.history.pulsar.service.url: "pulsar://127.0.0.1:6650" + ## PULSAR_SERVICE_URL_CONFIG + pulsar.service.url: "pulsar://127.0.0.1:6650" + topic.namespace: "public/default" + # CONVERTERS + key.converter: "org.apache.kafka.connect.json.JsonConverter" + value.converter: "org.apache.kafka.connect.json.JsonConverter" + typeClassName: "org.apache.pulsar.common.schema.KeyValue" + + task.class: "io.debezium.connector.db2.Db2ConnectorTask" + + offset.storage.topic: "offset-topic" + snapshot.mode: "initial" + database.tcpKeepAlive: "true" + decimal.handling.mode: "double" diff --git a/pulsar-io/debezium/pom.xml b/pulsar-io/debezium/pom.xml index 050da1af45330..d70d325497c81 100644 --- a/pulsar-io/debezium/pom.xml +++ b/pulsar-io/debezium/pom.xml @@ -78,6 +78,7 @@ mongodb oracle mssql + db2 diff --git a/pulsar-io/docs/pom.xml b/pulsar-io/docs/pom.xml index 305c7f1473077..2693104c46934 100644 --- a/pulsar-io/docs/pom.xml +++ b/pulsar-io/docs/pom.xml @@ -97,6 +97,11 @@ pulsar-io-debezium-mssql ${project.version} + + ${project.groupId} + pulsar-io-debezium-db2 + ${project.version} + ${project.groupId} pulsar-io-dynamodb diff --git a/tests/docker-images/debezium-db2-test-image/Dockerfile b/tests/docker-images/debezium-db2-test-image/Dockerfile new file mode 100644 index 0000000000000..723ea04fe1f08 --- /dev/null +++ b/tests/docker-images/debezium-db2-test-image/Dockerfile @@ -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. +# + +FROM ibmcom/db2:11.5.0.0a + +# As of 3/31/2023: +# "If the Program is designated as "Non-Production", the Program can only be deployed as part of the Licensee's internal development and test environment +# for internal non-production activities, including but not limited to testing, performance +# tuning, fault diagnosis, internal benchmarking, staging, quality assurance activity +# and/or developing internally used additions or extensions to the Program using published +# application programming interfaces. Licensee is not authorized to use any part of the +# Program for any other purposes without acquiring the appropriate production entitlements." +# (https://www.ibm.com/support/customer/csol/terms/?id=L-GMET-8KJN5X&lc=en#detail-document) +# Additionally: +# "Client license files are not required for connections to Db2, Cloudscape, +# or IBM Informix® databases from the IBM Data Server Driver for JDBC and SQLJ version +# 3.50 or later. Client license files are not required for direct connections to Db2 for +# z/OS if Db2 Connect Unlimited Edition for System z® server license activation is performed." +# (https://www.ibm.com/docs/en/db2/11.5?topic=apis-installing-data-server-driver-jdbc-sqlj) + +MAINTAINER Devin Bost +# Originally from https://github.com/debezium/debezium-connector-db2/blob/bee92787b1e02bb3e0813a6d8b7f62d34ad6d9e2/src/test/docker/db2-cdc-docker/Dockerfile + +RUN mkdir -p /asncdctools/src + +RUN cd /asncdctools/src && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/asncdc.c" +RUN cd /asncdctools/src && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/dbsetup.sh" +RUN cd /asncdctools/src && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/asncdc_UDF.sql" +RUN cd /asncdctools/src && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/asncdcaddremove.sql" +RUN cd /asncdctools/src && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/asncdctables.sql" + +RUN chmod -R 777 /asncdctools + +RUN mkdir /var/custom +RUN chmod -R 777 /var/custom + +RUN cd /var/custom && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/cdcsetup.sh" + +RUN chmod 777 /var/custom/cdcsetup.sh + +RUN chmod 554 /opt/ibm/db2/V11.5/adm/db2start \ No newline at end of file diff --git a/tests/docker-images/debezium-db2-test-image/pom.xml b/tests/docker-images/debezium-db2-test-image/pom.xml new file mode 100644 index 0000000000000..6d7588f11d06c --- /dev/null +++ b/tests/docker-images/debezium-db2-test-image/pom.xml @@ -0,0 +1,80 @@ + + + + org.apache.pulsar.tests + docker-images + 3.0.0-SNAPSHOT + + 4.0.0 + debezium-db2-test-image + Apache Pulsar :: Tests :: Docker Images :: Debezium DB2 Test Image + pom + + + + docker + + target/pulsar-server-distribution-bin.tar.gz + ${env.UBUNTU_MIRROR} + ${env.UBUNTU_SECURITY_MIRROR} + + + + integrationTests + + + + + + io.fabric8 + docker-maven-plugin + + + default + package + + build + + + + + ${docker.organization}/debezium-db2-test-image + + ${project.basedir} + + latest + ${project.version} + + true + + + + + + + + + + + + diff --git a/tests/docker-images/latest-version-image/Dockerfile b/tests/docker-images/latest-version-image/Dockerfile index f0093fa1eafc5..43c2ab190c8a2 100644 --- a/tests/docker-images/latest-version-image/Dockerfile +++ b/tests/docker-images/latest-version-image/Dockerfile @@ -117,17 +117,28 @@ COPY --from=pulsar-all /pulsar/connectors/pulsar-io-kinesis-*.nar /pulsar/connec # download Oracle JDBC driver for Oracle Debezium Connector tests RUN mkdir -p META-INF/bundled-dependencies -RUN cd META-INF/bundled-dependencies && curl -sSLO https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/ojdbc8/19.3.0.0/ojdbc8-19.3.0.0.jar -RUN cd META-INF/bundled-dependencies && curl -sSLO https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/ucp/19.3.0.0/ucp-19.3.0.0.jar -RUN cd META-INF/bundled-dependencies && curl -sSLO https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/oraclepki/19.3.0.0/oraclepki-19.3.0.0.jar -RUN cd META-INF/bundled-dependencies && curl -sSLO https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/osdt_cert/19.3.0.0/osdt_cert-19.3.0.0.jar -RUN cd META-INF/bundled-dependencies && curl -sSLO https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/osdt_core/19.3.0.0/osdt_core-19.3.0.0.jar -RUN cd META-INF/bundled-dependencies && curl -sSLO https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/simplefan/19.3.0.0/simplefan-19.3.0.0.jar -RUN cd META-INF/bundled-dependencies && curl -sSLO https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/orai18n/19.3.0.0/orai18n-19.3.0.0.jar -RUN cd META-INF/bundled-dependencies && curl -sSLO https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/xdb/19.3.0.0/xdb-19.3.0.0.jar -RUN cd META-INF/bundled-dependencies && curl -sSLO https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/xmlparserv2/19.3.0.0/xmlparserv2-19.3.0.0.jar +RUN cd META-INF/bundled-dependencies && curl -sSLO "https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/ojdbc8/19.3.0.0/ojdbc8-19.3.0.0.jar" +RUN cd META-INF/bundled-dependencies && curl -sSLO "https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/ucp/19.3.0.0/ucp-19.3.0.0.jar" +RUN cd META-INF/bundled-dependencies && curl -sSLO "https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/oraclepki/19.3.0.0/oraclepki-19.3.0.0.jar" +RUN cd META-INF/bundled-dependencies && curl -sSLO "https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/osdt_cert/19.3.0.0/osdt_cert-19.3.0.0.jar" +RUN cd META-INF/bundled-dependencies && curl -sSLO "https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/osdt_core/19.3.0.0/osdt_core-19.3.0.0.jar" +RUN cd META-INF/bundled-dependencies && curl -sSLO "https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/simplefan/19.3.0.0/simplefan-19.3.0.0.jar" +RUN cd META-INF/bundled-dependencies && curl -sSLO "https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/orai18n/19.3.0.0/orai18n-19.3.0.0.jar" +RUN cd META-INF/bundled-dependencies && curl -sSLO "https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/xdb/19.3.0.0/xdb-19.3.0.0.jar" +RUN cd META-INF/bundled-dependencies && curl -sSLO "https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/xmlparserv2/19.3.0.0/xmlparserv2-19.3.0.0.jar" RUN jar uf connectors/pulsar-io-debezium-oracle-*.nar META-INF/bundled-dependencies/ojdbc8-19.3.0.0.jar META-INF/bundled-dependencies/ucp-19.3.0.0.jar META-INF/bundled-dependencies/oraclepki-19.3.0.0.jar META-INF/bundled-dependencies/osdt_cert-19.3.0.0.jar META-INF/bundled-dependencies/osdt_core-19.3.0.0.jar META-INF/bundled-dependencies/simplefan-19.3.0.0.jar META-INF/bundled-dependencies/orai18n-19.3.0.0.jar META-INF/bundled-dependencies/xdb-19.3.0.0.jar META-INF/bundled-dependencies/xmlparserv2-19.3.0.0.jar - +# Download IBM DB2 JDBC driver for DB2 Debezium Connector tests: +RUN cd META-INF/bundled-dependencies && curl -sSLO "https://search.maven.org/remotecontent?filepath=com/ibm/db2/jcc/11.5.8.0/jcc-11.5.8.0.jar" +RUN cd META-INF/bundled-dependencies && curl -sSLO "https://search.maven.org/remotecontent?filepath=com/ibm/db2/jcc/db2jcc/db2jcc4/db2jcc-db2jcc4.jar" +RUN mkdir -p /asncdctools/src +RUN cd /asncdctools/src && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/asncdc.c" +RUN cd /asncdctools/src && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/dbsetup.sh" +RUN cd /asncdctools/src && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/asncdc_UDF.sql" +RUN cd /asncdctools/src && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/asncdcaddremove.sql" +RUN cd /asncdctools/src && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/asncdctables.sql" +RUN cd /asncdctools/src && curl -sSLO "https://raw.githubusercontent.com/debezium/debezium-connector-db2/main/src/test/docker/db2-cdc-docker/cdcsetup.sh" + +RUN jar uf connectors/pulsar-io-debezium-db2-*.nar META-INF/bundled-dependencies/jcc-11.5.8.0.jar META-INF/bundled-dependencies/db2jcc-db2jcc4.jar CMD bash diff --git a/tests/docker-images/pom.xml b/tests/docker-images/pom.xml index ade69e0778b04..c05f7d7ed47b6 100644 --- a/tests/docker-images/pom.xml +++ b/tests/docker-images/pom.xml @@ -51,6 +51,7 @@ java-test-plugins latest-version-image java-test-image + debezium-db2-test-image diff --git a/tests/integration/pom.xml b/tests/integration/pom.xml index d9b817ca2e2b5..574870076fbd7 100644 --- a/tests/integration/pom.xml +++ b/tests/integration/pom.xml @@ -101,6 +101,18 @@ ${project.version} test + + + com.ibm.db2 + jcc + 11.5.8.0 + test + org.testcontainers diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumDB2DbContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumDB2DbContainer.java new file mode 100644 index 0000000000000..d31a3e081c729 --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumDB2DbContainer.java @@ -0,0 +1,148 @@ +/* + * 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 java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.time.Duration; +import java.time.temporal.ChronoUnit; +import lombok.extern.slf4j.Slf4j; +import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy; + +@Slf4j +public class DebeziumDB2DbContainer extends ChaosContainer { + + public static final String NAME = "debezium-db2"; + static final Integer[] PORTS = { 50000 }; + + // + private static final String IMAGE_NAME = "apachepulsar/debezium-db2-test-image:latest"; + /* + "If the Program is designated as "Non-Production", the Program can only be deployed as part + of the Licensee's internal development and test environment for internal non-production activities, + including but not limited to testing, performance tuning, fault diagnosis, internal benchmarking, + staging, quality assurance activity and/or developing internally used additions or extensions to + the Program using published application programming interfaces. Licensee is not authorized to use + any part of the Program for any other purposes without acquiring the appropriate production + entitlements." + (https://www.ibm.com/support/customer/csol/terms/?id=L-GMET-8KJN5X&lc=en#detail-document) + */ + private static final String DBNAME = "mydb2"; + private static final String USERNAME = "db2inst1"; + private static final String PASS = "admin"; + + public DebeziumDB2DbContainer(String clusterName) { + super(clusterName, IMAGE_NAME); + } + + @Override + public String getContainerName() { + return clusterName; + } + + @Override + protected void configure() { + super.configure(); + this.withNetworkAliases(NAME) + .withExposedPorts(PORTS) + .withEnv("LICENSE", "accept") + .withEnv("DB2INSTANCE", USERNAME) + .withEnv("DB2INST1_PASSWORD", PASS) + .withEnv("DBNAME", DBNAME) + .withEnv("BLU", "false") + .withEnv("ENABLE_ORACLE_COMPATIBILITY", "false") + .withEnv("UPDATEAVAIL", "NO") + .withEnv("TO_CREATE_SAMPLEDB", "false") + .withEnv("REPODB", "false") + .withEnv("IS_OSXFS", "false") + .withEnv("PERSISTENT_HOME", "true") + .withEnv("HADR_ENABLED", "false") + .withPrivilegedMode(true) + .withStartupTimeout(Duration.of(900, ChronoUnit.SECONDS)) + .withCreateContainerCmdModifier(createContainerCmd -> { + createContainerCmd.withHostName(NAME); + createContainerCmd.withName(getContainerName()); + }).waitingFor((new LogMessageWaitStrategy()).withRegEx(".*Setup has completed\\..*") + .withStartupTimeout(Duration.of(10L, ChronoUnit.MINUTES))); + } + public String getDriverClassName() { + return "com.ibm.db2.jcc.DB2Driver"; + } + public String getUsername(){ + return USERNAME.toUpperCase(); + } + public String getPassword(){ + return PASS; + } + + public String getJdbcUrl() { + return "jdbc:db2://" + this.getHost() + ":" + this.getMappedPort(50000) + "/" + DBNAME; + } + + public String createTableStatement(){ + return String.format("CREATE TABLE %s.STORES(store_id INT GENERATED BY DEFAULT AS IDENTITY NOT NULL,store_name VARCHAR(150) NOT NULL,state_id INT NOT NULL,zip_code VARCHAR(6),PRIMARY KEY (store_id))", + getUsername()); + } + public String insertStatement(){ + return String.format("INSERT INTO %s.STORES(store_name, state_id, zip_code) VALUES ('mystore', 12, '11111')", + getUsername()); + } + public String selectStatement(){ + return String.format("SELECT * FROM %s.STORES", + getUsername()); + } + private Connection connection; + public Connection getConnection() throws SQLException, ClassNotFoundException { + log.info(String.format("JDBC Connection URL is: %s", this.getJdbcUrl())); + if(connection == null){ + Class.forName(this.getDriverClassName()); + var conn = DriverManager.getConnection(this.getJdbcUrl(), getUsername(), getPassword()); + this.connection = conn; + } + return connection; + } + public String enableCdcStatement(){ + return "VALUES ASNCDC.ASNCDCSERVICES('start','asncdc')"; + } + public String addCdcTableStatement(){ + return String.format("CALL ASNCDC.ADDTABLE('%s','STORES')", getUsername()); + } + public PreparedStatement getPreparedStatement(String sqlQuery) throws SQLException, ClassNotFoundException { + var myConnection = this.getConnection(); + return myConnection.prepareStatement(sqlQuery); + } + + public String getStoreNameResult() throws SQLException, ClassNotFoundException { + var statement = this.getPreparedStatement(this.selectStatement()); + var result = statement.execute(); + if(result == true) + { + var resultSet = statement.getResultSet(); + while(resultSet.next()) + { + var storeName = resultSet.getString("store_name"); + return storeName; + } + } + return null; + } +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/SourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/SourceTester.java index 37e241a873b6c..7f2d7a5916a30 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/SourceTester.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/SourceTester.java @@ -129,7 +129,7 @@ public void validateSourceResultJson(Consumer> consumer consumer.acknowledge(msg); msg = consumer.receive(1, TimeUnit.SECONDS); } - + log.info(String.format("Running Assert.assertEquals(recordsNumber, number) for (%d, %d)", recordsNumber, number)); Assert.assertEquals(recordsNumber, number); log.info("Stop {} server container. topic: {} has {} records.", getSourceType(), consumer.getTopic(), recordsNumber); } diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumDB2DbSourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumDB2DbSourceTester.java new file mode 100644 index 0000000000000..047abb18dc2e1 --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumDB2DbSourceTester.java @@ -0,0 +1,176 @@ +/* + * 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.io.sources.debezium; + +import com.google.common.base.Preconditions; +import java.util.Map; +import lombok.Getter; +import lombok.SneakyThrows; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.tests.integration.containers.DebeziumDB2DbContainer; +import org.apache.pulsar.tests.integration.containers.PulsarContainer; +import org.apache.pulsar.tests.integration.io.sources.SourceTester; +import org.apache.pulsar.tests.integration.topologies.PulsarCluster; + +/** + * A tester for testing Debezium DB2 source. + */ +@Slf4j +public class DebeziumDB2DbSourceTester extends SourceTester { + + private static final String NAME = "debezium-db2"; + private static final long SLEEP_AFTER_COMMAND_MS = 30_000; + + private final String pulsarServiceUrl; + + @Getter + private DebeziumDB2DbContainer debeziumDB2DbContainer; + + private final PulsarCluster pulsarCluster; + + public DebeziumDB2DbSourceTester(PulsarCluster cluster) { + super(NAME); + this.pulsarCluster = cluster; + this.numEntriesToInsert = 1; + this.numEntriesExpectAfterStart = 0; + + pulsarServiceUrl = "pulsar://pulsar-proxy:" + PulsarContainer.BROKER_PORT; + //sourceConfig.put("connector.class", "io.debezium.connector.db2.Db2Connector"); + sourceConfig.put("database.hostname", DebeziumDB2DbContainer.NAME); + sourceConfig.put("database.port", "50000"); + sourceConfig.put("database.user", "db2inst1"); + sourceConfig.put("database.password", "admin"); + sourceConfig.put("database.dbname", "mydb2"); + sourceConfig.put("database.server.name", "db2inst1"); + sourceConfig.put("topic.prefix", "stores"); + sourceConfig.put("table.include.list", "DB2INST1.STORES"); + sourceConfig.put("database.history","org.apache.pulsar.io.debezium.PulsarDatabaseHistory"); + sourceConfig.put("database.history.pulsar.topic", "debezium-db2-source-history-topic"); + sourceConfig.put("database.history.pulsar.service.url", pulsarServiceUrl); + sourceConfig.put("topic.namespace", "debezium/db2"); + sourceConfig.put("key.converter", "org.apache.kafka.connect.json.JsonConverter"); + sourceConfig.put("value.converter", "org.apache.kafka.connect.json.JsonConverter"); + sourceConfig.put("typeClassName", "org.apache.pulsar.common.schema.KeyValue"); + sourceConfig.put("task.class", "io.debezium.connector.db2.Db2ConnectorTask"); + + sourceConfig.put("offset.storage.topic", "offset-topic"); + sourceConfig.put("snapshot.mode", "initial"); + sourceConfig.put("database.tcpKeepAlive", "true"); + sourceConfig.put("decimal.handling.mode", "double"); + + } + + @Override + public void setServiceContainer(DebeziumDB2DbContainer container) { + log.info("start debezium db2 server container."); + Preconditions.checkState(debeziumDB2DbContainer == null); + debeziumDB2DbContainer = container; + pulsarCluster.startService(DebeziumDB2DbContainer.NAME, debeziumDB2DbContainer); + } + + @SneakyThrows + @Override + public void prepareSource() { + log.info("Starting DB2. Running cdcsetup.sh next"); + var lsResult = debeziumDB2DbContainer.execInContainer("/bin/bash", "-c", "/var/custom/cdcsetup.sh"); + String stdout = lsResult.getStdout(); + log.info(String.format("Output of running cdcsetup.sh is: %s", stdout)); + int exitCode = lsResult.getExitCode(); + + log.info("Running schema bind next"); + var lsResult2 = debeziumDB2DbContainer.execInContainer("/bin/bash", "-c", "/opt/ibm/db2/V11.5/bin/db2 bind db2schema.bnd blocking all grant public sqlerror continue"); + String stdout2 = lsResult2.getStdout(); + log.info(String.format("Output of running schema bind is: %s", stdout2)); + int exitCode2 = lsResult2.getExitCode(); + + log.info(String.format("Running : %s", debeziumDB2DbContainer.enableCdcStatement())); + debeziumDB2DbContainer.getPreparedStatement(debeziumDB2DbContainer.enableCdcStatement()) + .execute(); + + log.info(String.format("Running : %s", debeziumDB2DbContainer.createTableStatement())); + debeziumDB2DbContainer.getPreparedStatement(debeziumDB2DbContainer.createTableStatement()) + .execute(); + + log.info(String.format("Running : %s", debeziumDB2DbContainer.addCdcTableStatement())); + debeziumDB2DbContainer.getPreparedStatement(debeziumDB2DbContainer.addCdcTableStatement()) + .execute(); + + log.info(String.format("Running : %s", debeziumDB2DbContainer.insertStatement())); + debeziumDB2DbContainer.getPreparedStatement(debeziumDB2DbContainer.insertStatement()) + .execute(); + + log.info(String.format("Running : %s", debeziumDB2DbContainer.selectStatement())); + var result = debeziumDB2DbContainer.getStoreNameResult(); + log.info("debeziumDB2DbContainer.getStoreNameResult() returned: %s", result); + // Could add an assertion here that result == "store_name" + } + + @Override + public void prepareInsertEvent() throws Exception { + debeziumDB2DbContainer.getPreparedStatement("INSERT INTO DB2INST1.STORES(store_name, state_id, zip_code) VALUES ('mystore2', 2, '22222')") + .execute(); + //debeziumDB2DbContainer.getPreparedStatement("SELECT * FROM DB2INST1.STORES WHERE store_name='mystore2'"); + } + + @Override + public void prepareDeleteEvent() throws Exception { + debeziumDB2DbContainer.getPreparedStatement("DELETE FROM DB2INST1.STORES WHERE store_name='mystore2'") + .execute(); + //debeziumDB2DbContainer.getPreparedStatement("SELECT * FROM DB2INST1.STORES WHERE store_name='mystore2'"); + } + + @Override + public void prepareUpdateEvent() throws Exception { + debeziumDB2DbContainer.getPreparedStatement("UPDATE DB2INST1.STORES SET zip_code='33333' WHERE store_name='mystore2'") + .execute(); + //debeziumDB2DbContainer.getPreparedStatement("SELECT * FROM DB2INST1.STORES WHERE store_name='mystore2'"); + } + + @Override + public Map produceSourceMessages(int numMessages) { + log.info("debezium db2 server already contains preconfigured data."); + return null; + } + + @Override + public int initialDelayForMsgReceive() { + return 30; + } + + @Override + public String keyContains() { + return "mydb2.DB2INST1.STORES.Key"; + } + + @Override + public String valueContains() { + return "mydb2.DB2INST1.STORES.Value"; + } + + @Override + public void close() { + if (pulsarCluster != null) { + if (debeziumDB2DbContainer != null) { + PulsarCluster.stopService(DebeziumDB2DbContainer.NAME, debeziumDB2DbContainer); + debeziumDB2DbContainer = null; + } + } + } + +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java index 5c57c904fc77f..5724b14a2db4d 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java @@ -28,6 +28,7 @@ import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.policies.data.TenantInfoImpl; import org.apache.pulsar.common.schema.SchemaInfo; +import org.apache.pulsar.tests.integration.containers.DebeziumDB2DbContainer; import org.apache.pulsar.tests.integration.containers.DebeziumMongoDbContainer; import org.apache.pulsar.tests.integration.containers.DebeziumMsSqlContainer; import org.apache.pulsar.tests.integration.containers.DebeziumMySQLContainer; @@ -78,6 +79,10 @@ public void testDebeziumMongoDbSource() throws Exception{ public void testDebeziumMsSqlSource() throws Exception{ testDebeziumMsSqlConnect("org.apache.kafka.connect.json.JsonConverter", true); } + @Test(groups = "source") + public void testDebeziumDb2Source() throws Exception{ + testDebeziumDb2Connect("org.apache.kafka.connect.json.JsonConverter", true); + } private void testDebeziumMySqlConnect(String converterClassName, boolean jsonWithEnvelope, boolean testWithClientBuilder) throws Exception { @@ -242,6 +247,43 @@ private void testDebeziumMsSqlConnect(String converterClassName, boolean jsonWit runner.testSource(sourceTester); } + private void testDebeziumDb2Connect(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-" + testId.getAndIncrement(); + final String consumeTopicName = "debezium/db2/mydb2.DB2INST1.STORES"; + final String sourceName = "test-source-debezium-db2-" + functionRuntimeType + "-" + randomName(8); + + final int numMessages = 2; + + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(pulsarCluster.getPlainTextServiceUrl()) + .build(); + + @Cleanup + PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build(); + initNamespace(admin); + + admin.topics().createNonPartitionedTopic(consumeTopicName); + admin.topics().createNonPartitionedTopic(outputTopicName); + + admin.namespaces().setRetention("debezium/db2", new RetentionPolicies(-1, 1)); + admin.namespaces().setRetention("public/default", new RetentionPolicies(-1, 1)); + @Cleanup + DebeziumDB2DbSourceTester sourceTester = new DebeziumDB2DbSourceTester(pulsarCluster); + sourceTester.getSourceConfig().put("json-with-envelope", jsonWithEnvelope); + + DebeziumDB2DbContainer db2Container = new DebeziumDB2DbContainer(pulsarCluster.getClusterName()); + sourceTester.setServiceContainer(db2Container); + + PulsarIODebeziumSourceRunner runner = new PulsarIODebeziumSourceRunner(pulsarCluster, functionRuntimeType.toString(), + converterClassName, tenant, namespace, sourceName, outputTopicName, numMessages, jsonWithEnvelope, + consumeTopicName, client); + + runner.testSource(sourceTester); + } protected void initNamespace(PulsarAdmin admin) { log.info("[initNamespace] start."); try { @@ -253,6 +295,7 @@ protected void initNamespace(PulsarAdmin admin) { "debezium/mongodb", "debezium/postgresql", "debezium/mssql", + "debezium/db2" }; Policies policies = new Policies(); policies.retention_policies = new RetentionPolicies(-1, 50); diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarIODebeziumSourceRunner.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarIODebeziumSourceRunner.java index 762dd34e17c91..d34e70df44618 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarIODebeziumSourceRunner.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarIODebeziumSourceRunner.java @@ -94,8 +94,9 @@ public void testSource(SourceTester sourceTester .subscriptionType(SubscriptionType.Exclusive) .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) .subscribe(); - log.info("[debezium mysql test] create consumer finish. converterName: {}", converterClassName); + log.info("[debezium test] create consumer finish. converterName: {}", converterClassName); + log.info("sourceTester.getNumEntriesExpectAfterStart() is {}", sourceTester.getNumEntriesExpectAfterStart()); // validate the source result sourceTester.validateSourceResult(consumer, sourceTester.getNumEntriesExpectAfterStart(), null, converterClassName);