Debezium PostgresSQL Connector native support #1191

apache · May 20, 2020 · 34298be · 34298be
1 parent a97b3d1
commit 34298be
Show file tree

Hide file tree

Showing 19 changed files with 511 additions and 102 deletions.
diff --git a/.github/test-categories.yaml b/.github/test-categories.yaml
@@ -32,6 +32,7 @@ main:
   - main-xml-jaxb
 database:
   - couchdb
+  - debezium-postgres
   - influxdb
   - jdbc
   - kudu

diff --git a/docs/modules/ROOT/pages/extensions/debezium-postgres.adoc b/docs/modules/ROOT/pages/extensions/debezium-postgres.adoc
@@ -0,0 +1,30 @@
+[[debezium-postgres]]
+= Debezium Postgres Extension
+
+*Since Camel Quarkus 1.0.0-M6*
+
+*Only consumer is supported*
+
+The Debezium PostgresSQL component is wrapper around https://debezium.io/[Debezium] using
+https://debezium.io/documentation/reference/0.9/operations/embedded.html[Debezium Embedded], which enables Change Data
+Capture from PostgresSQL database using Debezium without the need for Kafka or Kafka Connect.
+
+[source,xml]
+------------------------------------------------------------
+<dependency>
+    <groupId>org.apache.camel.quarkus</groupId>
+    <artifactId>camel-debezium-postgres</artifactId>
+</dependency>
+------------------------------------------------------------
+
+== Usage
+
+The extension provides support for the Camel https://camel.apache.org/components/latest/debezium-postgres-component.html[Debezium Postgres Connector].
+
+=== Limitations
+
+Not all offset stores are supported in the native mode. Supported ones are:
+
+* org.apache.kafka.connect.storage.FileOffsetBackingStore
+* org.apache.kafka.connect.storage.MemoryOffsetBackingStore
+
diff --git a/docs/modules/ROOT/pages/list-of-camel-quarkus-extensions.adoc b/docs/modules/ROOT/pages/list-of-camel-quarkus-extensions.adoc
@@ -226,8 +226,8 @@ Level | Since | Description
  Preview | 1.0.0-M6 | Capture changes from a MySQL database.
 
 | link:https://camel.apache.org/components/latest/debezium-postgres-component.html[Debezium PostgresSQL Connector] (camel-quarkus-debezium-postgres) +
-`debezium-postgres:name` | JVM +
- Preview | 1.0.0-M6 | Capture changes from a PostgresSQL database.
+`debezium-postgres:name` | Native +
+ Stable | 1.0.0-M6 | Capture changes from a PostgresSQL database.
 
 | link:https://camel.apache.org/components/latest/debezium-sqlserver-component.html[Debezium SQL Server Connector] (camel-quarkus-debezium-sqlserver) +
 `debezium-sqlserver:name` | JVM +

diff --git a/...pache/camel/quarkus/component/debezium/postgres/deployment/DebeziumPostgresProcessor.java b/...pache/camel/quarkus/component/debezium/postgres/deployment/DebeziumPostgresProcessor.java
diff --git a/extensions-jvm/pom.xml b/extensions-jvm/pom.xml
@@ -49,7 +49,6 @@
         <module>couchbase</module>
         <module>debezium-mongodb</module>
         <module>debezium-mysql</module>
-        <module>debezium-postgres</module>
         <module>debezium-sqlserver</module>
         <module>google-bigquery</module>
         <module>google-pubsub</module>

diff --git a/...-jvm/debezium-postgres/deployment/pom.xml → ...ions/debezium-postgres/deployment/pom.xml b/...-jvm/debezium-postgres/deployment/pom.xml → ...ions/debezium-postgres/deployment/pom.xml
@@ -17,7 +17,9 @@
     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">
+<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.camel.quarkus</groupId>
@@ -48,16 +50,12 @@
         </dependency>
         <dependency>
             <groupId>org.apache.camel.quarkus</groupId>
-            <artifactId>camel-quarkus-support-debezium-deployment</artifactId>
+            <artifactId>camel-quarkus-debezium-postgres</artifactId>
         </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-jdbc-postgresql-deployment</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.apache.camel.quarkus</groupId>
-            <artifactId>camel-quarkus-debezium-postgres</artifactId>
-        </dependency>
     </dependencies>
 
     <build>

diff --git a/...pache/camel/quarkus/component/debezium/postgres/deployment/DebeziumPostgresProcessor.java b/...pache/camel/quarkus/component/debezium/postgres/deployment/DebeziumPostgresProcessor.java
@@ -0,0 +1,64 @@
+/*
+ * 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.camel.quarkus.component.debezium.postgres.deployment;
+
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.builditem.CombinedIndexBuildItem;
+import io.quarkus.deployment.builditem.FeatureBuildItem;
+import io.quarkus.deployment.builditem.IndexDependencyBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
+import org.jboss.jandex.IndexView;
+
+class DebeziumPostgresProcessor {
+
+    private static final String FEATURE = "camel-debezium-postgres";
+
+    @BuildStep
+    FeatureBuildItem feature() {
+        return new FeatureBuildItem(FEATURE);
+    }
+
+    @BuildStep
+    ReflectiveClassBuildItem registerForReflection(CombinedIndexBuildItem combinedIndex) {
+        IndexView index = combinedIndex.getIndex();
+
+        String[] dtos = index.getKnownClasses().stream().map(ci -> ci.name().toString())
+                .filter(n -> n.startsWith("org.apache.kafka.connect.json")
+                        || n.startsWith("io.debezium.embedded.spi"))
+                .sorted()
+                .toArray(String[]::new);
+
+        return new ReflectiveClassBuildItem(false, true, dtos);
+    }
+
+    @BuildStep
+    ReflectiveClassBuildItem reflectiveClasses() {
+        return new ReflectiveClassBuildItem(false, false,
+                new String[] { "org.apache.kafka.connect.storage.FileOffsetBackingStore",
+                        "org.apache.kafka.connect.storage.MemoryOffsetBackingStore",
+                        "io.debezium.connector.postgresql.PostgresConnector",
+                        "io.debezium.connector.postgresql.PostgresConnectorTask" });
+    }
+
+    @BuildStep
+    void addDependencies(BuildProducer<IndexDependencyBuildItem> indexDependency) {
+        indexDependency.produce(new IndexDependencyBuildItem("org.apache.kafka", "connect-json"));
+        indexDependency.produce(new IndexDependencyBuildItem("io.debezium", "debezium-connector-postgres"));
+        indexDependency.produce(new IndexDependencyBuildItem("io.debezium", "debezium-embedded"));
+    }
+}
diff --git a/extensions-jvm/debezium-postgres/pom.xml → extensions/debezium-postgres/pom.xml b/extensions-jvm/debezium-postgres/pom.xml → extensions/debezium-postgres/pom.xml
@@ -17,13 +17,15 @@
     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">
+<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.camel.quarkus</groupId>
-        <artifactId>camel-quarkus-build-parent-it</artifactId>
+        <artifactId>camel-quarkus-build-parent</artifactId>
         <version>1.1.0-SNAPSHOT</version>
-        <relativePath>../../poms/build-parent-it/pom.xml</relativePath>
+        <relativePath>../../poms/build-parent/pom.xml</relativePath>
     </parent>
 
     <artifactId>camel-quarkus-debezium-postgres-parent</artifactId>
@@ -33,6 +35,5 @@
     <modules>
         <module>deployment</module>
         <module>runtime</module>
-        <module>integration-test</module>
     </modules>
 </project>
diff --git a/...ons-jvm/debezium-postgres/runtime/pom.xml → extensions/debezium-postgres/runtime/pom.xml b/...ons-jvm/debezium-postgres/runtime/pom.xml → extensions/debezium-postgres/runtime/pom.xml
@@ -17,7 +17,9 @@
     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">
+<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.camel.quarkus</groupId>
@@ -46,22 +48,28 @@
     </dependencyManagement>
 
     <dependencies>
+        <dependency>
+            <groupId>org.graalvm.nativeimage</groupId>
+            <artifactId>svm</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.apache.camel.quarkus</groupId>
             <artifactId>camel-quarkus-core</artifactId>
         </dependency>
         <dependency>
-            <groupId>org.apache.camel.quarkus</groupId>
-            <artifactId>camel-quarkus-support-debezium</artifactId>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-debezium-postgres</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.maven</groupId>
+                    <artifactId>maven-artifact</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-jdbc-postgresql</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.apache.camel</groupId>
-            <artifactId>camel-debezium-postgres</artifactId>
-        </dependency>
     </dependencies>
 
     <build>

diff --git a/...in/java/org/apache/camel/quarkus/component/debezium/postgres/graal/SubstituteMetrics.java b/...in/java/org/apache/camel/quarkus/component/debezium/postgres/graal/SubstituteMetrics.java
@@ -0,0 +1,38 @@
+/*
+ * 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.camel.quarkus.component.debezium.postgres.graal;
+
+import com.oracle.svm.core.annotate.Substitute;
+import com.oracle.svm.core.annotate.TargetClass;
+import org.slf4j.Logger;
+
+@TargetClass(io.debezium.metrics.Metrics.class)
+final class SubstituteMetrics {
+
+    @Substitute
+    public synchronized void register(Logger logger) {
+        //JMX is not supported in the native mode
+        //because there is no API for avoiding MBean registration, substitution is used to skip registration
+        // enhancement in debezium:https://issues.redhat.com/browse/DBZ-2089
+        logger.warn("Metrics are not registered in native mode.");
+    }
+
+    @Substitute
+    public final void unregister(Logger logger) {
+        logger.debug("Metrics are not unregistered in native mode.");
+    }
+}
diff --git a/...resources/META-INF/quarkus-extension.yaml → ...resources/META-INF/quarkus-extension.yaml b/...resources/META-INF/quarkus-extension.yaml → ...resources/META-INF/quarkus-extension.yaml
@@ -24,8 +24,6 @@
 name: "Camel Debezium PostgresSQL Connector"
 description: "Capture changes from a PostgresSQL database"
 metadata:
-  unlisted: true
-  guide: "https://camel.apache.org/components/latest/debezium-postgres-component.html"
+  guide: "https://camel.apache.org/camel-quarkus/latest/extensions/debezium-postgres.html"
   categories:
   - "integration"
-  status: "preview"
diff --git a/extensions/pom.xml b/extensions/pom.xml
@@ -69,6 +69,7 @@
         <module>cron</module>
         <module>csv</module>
         <module>dataformat</module>
+        <module>debezium-postgres</module>
         <module>direct</module>
         <module>dozer</module>
         <module>elasticsearch-rest</module>