Merge pull request #914 from Altinity/906-connector-does-not-create-r…

…eplicate_schema_history-table Added logic to create schema history table on connector startup.
Altinity · Nov 15, 2024 · 525821d · 525821d
2 parents 30faf6b + 9e4b161
commit 525821d
Showing 1 changed file with 33 additions and 1 deletion.
diff --git a/...c/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCapture.java b/...c/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCapture.java
@@ -370,6 +370,28 @@ private void createDatabaseForDebeziumStorage(ClickHouseSinkConnectorConfig conf
         }
     }
 
+    private void createSchemaHistoryTable(ClickHouseSinkConnectorConfig config, Properties props) {
+        String createSchemaHistoryTable = props.getProperty(JdbcOffsetBackingStoreConfig.OFFSET_STORAGE_PREFIX + JdbcSchemaHistoryConfig.PROP_TABLE_NAME.name());
+        if(createSchemaHistoryTable == null || createSchemaHistoryTable.isEmpty() == true) {
+            log.warn("Skipping creating schema history table as the query was not provided in configuration");
+            return;
+        }
+
+        DBCredentials dbCredentials = parseDBConfiguration(config);
+        String jdbcUrl = BaseDbWriter.getConnectionString(dbCredentials.getHostName(), dbCredentials.getPort(),
+                "system");
+        ClickHouseConnection conn = BaseDbWriter.createConnection(jdbcUrl, "Client_1",dbCredentials.getUserName(), dbCredentials.getPassword(), config);
+        BaseDbWriter writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
+                "system", dbCredentials.getUserName(),
+                dbCredentials.getPassword(), config, conn);
+
+        try {
+            writer.executeQuery(createSchemaHistoryTable);
+        } catch(Exception e) {
+            log.error("Error creating schema history table", e);
+        }
+    }
+
     /**
      * Function to create view for show_replica_status
      * @param config
@@ -743,7 +765,17 @@ public void connectorStarted() {
                                     isReplicationRunning = true;
                                     log.debug("Connector started");
                                     // Create view.
-                                    createViewForShowReplicaStatus(config, props);
+                                    try {
+                                        createViewForShowReplicaStatus(config, props);
+                                    } catch(Exception e) {
+                                        log.error("Error creating view for replica status", e);
+                                    }
+
+                                    try {
+                                        createSchemaHistoryTable(config, props);
+                                    } catch(Exception e) {
+                                        log.error("Error creating schema history table", e);
+                                    }
                                 }
 
                                 @Override