Debezium date format + trustServerCertificate flag (#104)

* debezium date formating * trustServerCertificate * Update sourceProxy.js * license headers
drasi-project · Nov 5, 2024 · 1003342 · 1003342
1 parent cce5fc8
commit 1003342
Show file tree

Hide file tree

Showing 11 changed files with 202 additions and 93 deletions.
diff --git a/.gitignore b/.gitignore
@@ -1,4 +1,4 @@
 **/.DS_Store
 **/.vscode
 bin
-obj
+obj
diff --git a/cli/service/resources/default-source-providers.yaml b/cli/service/resources/default-source-providers.yaml
@@ -1,17 +1,17 @@
-# Copyright 2024 The Drasi Authors.
-#
-# Licensed 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.
-
+# Copyright 2024 The Drasi Authors.
+#
+# Licensed 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.
+
 apiVersion: v1
 kind: SourceProvider
 name: PostgreSQL
@@ -104,6 +104,9 @@ spec:
       encrypt:
         type: boolean
         default: false
+      trustServerCertificate:
+        type: boolean
+        default: false
       user:
         type: string
       tables:

diff --git a/sources/relational/debezium-reactivator/Dockerfile b/sources/relational/debezium-reactivator/Dockerfile
@@ -1,21 +1,21 @@
-# Copyright 2024 The Drasi Authors.
-#
-# Licensed 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.
-
+# Copyright 2024 The Drasi Authors.
+#
+# Licensed 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.
+
 #
 # Build stage
 #
-FROM --platform=$BUILDPLATFORM maven@sha256:202d9532fc5519f7df02f4b9a6450af93a97a48276bf7975d25f9a10284d366b AS build
+FROM --platform=$BUILDPLATFORM maven:3.9.9-amazoncorretto-21 AS build
 COPY src /home/app/src
 COPY pom.xml /home/app
 RUN mvn -f /home/app/pom.xml clean package
@@ -26,6 +26,6 @@ RUN cd /home/app && mvn dependency:tree
 #
 # Package stage
 #
-FROM --platform=$TARGETPLATFORM mcr.microsoft.com/openjdk/jdk@sha256:8f1350862e40b24ba7af80e9f0226990bfadd16eb1b92fd3859f92c9a54b7d30
+FROM --platform=$TARGETPLATFORM mcr.microsoft.com/openjdk/jdk:21-distroless
 COPY --from=build /home/app/target /home/app
 CMD ["-jar","/home/app/debezium-reactivator-1.0.jar"]
diff --git a/sources/relational/debezium-reactivator/pom.xml b/sources/relational/debezium-reactivator/pom.xml
@@ -1,20 +1,20 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<!--
- Copyright 2024 The Drasi Authors.
-
- Licensed 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.
--->
-
+<!--
+ Copyright 2024 The Drasi Authors.
+
+ Licensed 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>
@@ -28,8 +28,8 @@
 
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    <maven.compiler.source>17</maven.compiler.source>
-    <maven.compiler.target>17</maven.compiler.target>
+    <maven.compiler.source>21</maven.compiler.source>
+    <maven.compiler.target>21</maven.compiler.target>
     <version.debezium>2.6.2.Final</version.debezium>
   </properties>
 
@@ -181,8 +181,8 @@
               <groupId>org.apache.maven.plugins</groupId>
               <artifactId>maven-compiler-plugin</artifactId>
               <configuration>
-                  <source>17</source>
-                  <target>17</target>
+                  <source>21</source>
+                  <target>21</target>
               </configuration>
           </plugin>
         <plugin>

diff --git a/sources/relational/debezium-reactivator/src/main/java/com/drasi/DeprovisionController.java b/sources/relational/debezium-reactivator/src/main/java/com/drasi/DeprovisionController.java
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2024 The Drasi Authors.
+ *
+ * Licensed 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 com.drasi;
 
 import io.dapr.client.DaprClient;

diff --git a/sources/relational/debezium-reactivator/src/main/java/com/drasi/PostgresChangeMonitor.java b/sources/relational/debezium-reactivator/src/main/java/com/drasi/PostgresChangeMonitor.java
@@ -76,6 +76,9 @@ public void run() throws IOException, SQLException {
                 .with("publication.autocreate.mode", "filtered")
                 .with("snapshot.mode", "never")
                 .with("decimal.handling.mode", "double")
+                .with("time.precision.mode", "adaptive_time_microseconds")
+                .with("converters", "temporalConverter")
+                .with("temporalConverter.type", "com.drasi.TemporalConverter")
                 .with("table.include.list", tableListStr).build();
 
         var sr = new SchemaReader(config);

diff --git a/sources/relational/debezium-reactivator/src/main/java/com/drasi/SchemaReader.java b/sources/relational/debezium-reactivator/src/main/java/com/drasi/SchemaReader.java
@@ -1,19 +1,19 @@
-/*
- * Copyright 2024 The Drasi Authors.
- *
- * Licensed 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.
- */
-
+/*
+ * Copyright 2024 The Drasi Authors.
+ *
+ * Licensed 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 com.drasi;
 
 import com.drasi.models.NodeMapping;
@@ -25,6 +25,7 @@
 import io.debezium.jdbc.JdbcConnection;
 import io.debezium.relational.RelationalDatabaseConnectorConfig;
 
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.*;
 

diff --git a/sources/relational/debezium-reactivator/src/main/java/com/drasi/SqlServerChangeMonitor.java b/sources/relational/debezium-reactivator/src/main/java/com/drasi/SqlServerChangeMonitor.java
@@ -67,11 +67,16 @@ public void run() throws IOException, SQLException {
                 .with("database.port", System.getenv("port"))
                 .with("database.user", System.getenv("user"))
                 .with("database.password", System.getenv("password"))
+                .with("database.encrypt", System.getenv("encrypt"))
+                .with("database.trustServerCertificate", System.getenv("trustServerCertificate"))
                 .with("database.names", System.getenv("database"))
                 .with("tombstones.on.delete", false)
                 .with("snapshot.mode", "no_data")
                 .with("schema.history.internal", "com.drasi.NoOpSchemaHistory")
                 .with("decimal.handling.mode", "double")
+                .with("time.precision.mode", "adaptive_time_microseconds")
+                .with("converters", "temporalConverter")
+                .with("temporalConverter.type", "com.drasi.TemporalConverter")
                 .with("table.include.list", tableListStr).build();
 
         var sr = new SchemaReader(config);

diff --git a/sources/relational/debezium-reactivator/src/main/java/com/drasi/TemporalConverter.java b/sources/relational/debezium-reactivator/src/main/java/com/drasi/TemporalConverter.java
@@ -0,0 +1,80 @@
+/*
+ * Copyright 2024 The Drasi Authors.
+ *
+ * Licensed 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 com.drasi;
+
+import io.debezium.spi.converter.CustomConverter;
+import io.debezium.spi.converter.RelationalColumn;
+import org.apache.kafka.connect.data.SchemaBuilder;
+
+import java.sql.Types;
+import java.time.*;
+import java.time.format.DateTimeFormatter;
+import java.util.Properties;
+
+public class TemporalConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {
+
+    @Override
+    public void configure(Properties props) {
+    }
+
+    @Override
+    public void converterFor(RelationalColumn column,
+                             CustomConverter.ConverterRegistration<SchemaBuilder> registration) {
+
+        switch (column.jdbcType()) {
+            case Types.TIMESTAMP:
+                registration.register(SchemaBuilder.string().name(column.name()), x -> {
+                    switch (x) {
+                        case null:
+                            return null;
+                        case Instant instant:
+                            return DateTimeFormatter.ISO_INSTANT.format(instant);
+                        case LocalDateTime localDateTime:
+                            return DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(localDateTime);
+                        default:
+                            return x.toString();
+                    }
+                });
+                break;
+            case Types.DATE:
+                registration.register(SchemaBuilder.string().name(column.name()), x -> {
+                    switch (x) {
+                        case null:
+                            return null;
+                        case LocalDate localDate:
+                            return DateTimeFormatter.ISO_LOCAL_DATE.format(localDate);
+                        default:
+                            return x.toString();
+                    }
+                });
+                break;
+            case Types.TIME:
+                registration.register(SchemaBuilder.string().name(column.name()), x -> {
+                    switch (x) {
+                        case null:
+                            return null;
+                        case LocalTime localTime:
+                            return DateTimeFormatter.ISO_LOCAL_TIME.format(localTime);
+                        default:
+                            return x.toString();
+                    }
+                });
+                break;
+
+        }
+    }
+}
diff --git a/sources/relational/debezium-reactivator/src/main/java/com/drasi/models/NodeMapping.java b/sources/relational/debezium-reactivator/src/main/java/com/drasi/models/NodeMapping.java
@@ -14,12 +14,12 @@
  * limitations under the License.
  */
 
-package com.drasi.models;
-
-import java.util.Set;
-
-public class NodeMapping {
-    public String tableName;
-    public String keyField;
-    public Set<String> labels;
-}
+package com.drasi.models;
+
+import java.util.Set;
+
+public class NodeMapping {
+    public String tableName;
+    public String keyField;
+    public Set<String> labels;
+}
diff --git a/sources/relational/sql-proxy/sourceProxy.js b/sources/relational/sql-proxy/sourceProxy.js
@@ -1,19 +1,19 @@
-/**
- * Copyright 2024 The Drasi Authors.
- *
- * Licensed 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.
- */
-
+/**
+ * Copyright 2024 The Drasi Authors.
+ *
+ * Licensed 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.
+ */
+
 require("dotenv").config();
 
 const fs = require('fs');
@@ -108,6 +108,7 @@ function extractConfig(cfg) {
   dbConfig.connection.database = cfg.database;
   dbConfig.connection.ssl = convertConfigValue(cfg.ssl);
   dbConfig.connection.encrypt = convertConfigValue(cfg.encrypt);
+  dbConfig.trustServerCertificate = convertConfigValue(cfg.trustServerCertificate);
 
   console.log("sourceProxy.main - dbConfig:", dbConfig);
   return dbConfig;
@@ -130,4 +131,4 @@ main().catch((error) => {
   console.error("sourceProxy.main - Error:", error.stack);
   fs.writeFileSync("/dev/termination-log", error.message);
   process.exit(1);
-});
+});