Removed logic of duplicating instances of DebeziumOffsetStorage

sink-connector-client/create_config.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"database/sql"
+	"fmt"
+	"log"
+
+	_ "github.com/go-sql-driver/mysql"
+)
+
+// Function to create a sink connector configuration.
+
+// Add function to connect to MySQL and validate the username/password
+// return true if succeeds, return error if it fails.
+// Validate MySQL credentials.
+// Add function to connect to MySQL and validate the username/password
+// return true if succeeds, return error if it fails.
+// Validate MySQL credentials.
+// Add function to connect to MySQL and validate the username/password
+func validateMySQL(sourceUsername string, sourcePassword string, sourceHost string, sourcePort string) bool {
+	// Connect to MySQL
+	db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/", sourceUsername, sourcePassword, sourceHost, sourcePort))
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer db.Close()
+
+	// Validate MySQL credentials
+	err = db.Ping()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// check if binlogs are enabled.
+	rows, err := db.Query("SHOW VARIABLES LIKE 'log_bin'")
+	// if log_bin is not enabled, then return false
+	// check if rows has response 'OFF'
+	// if it is 'OFF' then return false
+	// check string comparison of rows with "OFF"
+	// get string value from sql.rows
+
+// 	if rows == "OFF" {
+// 		log.Fatal("Binlogs are not enabled")
+// 		return false
+// 	}
+
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer rows.Close()
+
+	return true
+}

sink-connector-client/go.mod

@@ -10,8 +10,10 @@ require (
 )
 
 require (
+	filippo.io/edwards25519 v1.1.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/go-sql-driver/mysql v1.8.1 // indirect
 	github.com/google/go-querystring v1.0.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect

sink-connector-client/go.sum

@@ -1,9 +1,13 @@
+filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
+filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
 github.com/BurntSushi/toml v1.2.1/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
 github.com/cpuguy83/go-md2man/v2 v2.0.2 h1:p1EgwI/C7NhT0JmVkwCD2ZBK8j4aeHQX2pMHHBfMQ6w=
 github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y=
+github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg=
 github.com/google/go-querystring v1.0.0 h1:Xkwi/a1rcvNg1PPYe5vI8GbeBY/jrVuDX5ASuANWTrk=
 github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
 github.com/levigross/grequests v0.0.0-20221222020224-9eee758d18d5 h1:AsF9Q1mQoyLv0HzvHFW7O+19dHilOcKU74k7E5ufI1A=

sink-connector-client/main.go

@@ -33,14 +33,17 @@ const (
 	STATUS_COMMAND            = "show_replica_status"
 	UPDATE_BINLOG_COMMAND     = "change_replication_source"
 	UPDATE_LSN_COMMAND        = "lsn"
+	DELETE_OFFSETS_COMMAND    = "delete_offsets"
 )
+
 const (
 	START_REPLICATION   = "start"
 	STOP_REPLICATION    = "stop"
 	RESTART_REPLICATION = "restart"
 	STATUS              = "status"
 	UPDATE_BINLOG       = "binlog"
 	UPDATE_LSN          = "lsn"
+	DELETE_OFFSETS      = "offsets"
 )
 
 // Fetches the repos for the given Github users
@@ -53,6 +56,14 @@ func getHTTPCall(url string) *grequests.Response {
 	return resp
 }
 
+func getHTTPDeleteCall(url string) *grequests.Response {
+    resp, err := grequests.Delete(url, requestOptions)
+    // you can modify the request by passing an optional RequestOptions struct
+    if err != nil {
+        log.Fatalln("Unable to make request: ", err)
+    }
+    return resp
+}
 /**
 Function to get server url based on the parameters passed
 */
@@ -215,12 +226,44 @@ func main() {
 				return nil
 			},
 		},
-	}
-
+        {
+            Name:  DELETE_OFFSETS_COMMAND,
+        	Usage: "Delete offsets from the sink connector",
+        	Action: func(c *cli.Context) error {
+        		handleDeleteOffsets(c)
+        		return nil
+        	},
+        },
+    }
 	app.Version = "1.0"
 	app.Run(os.Args)
 }
 
+func handleDeleteOffsets(c *cli.Context) bool {
+    log.Println("***** Delete offsets from the sink connector *****")
+    log.Println("Are you sure you want to continue? (y/n): ")
+    	var userInput string
+    	fmt.Scanln(&userInput)
+    	if userInput != "y" {
+    		log.Println("Exiting...")
+    		return false
+    	} else {
+    		log.Println("Continuing...")
+    	}
+    // Call a REST DELETE API to delete offsets from the sink connector
+    var deleteOffsetsUrl = getServerUrl(DELETE_OFFSETS, c)
+    log.Println("Sending request to URL: " + deleteOffsetsUrl)
+    resp := getHTTPDeleteCall(deleteOffsetsUrl)
+    time.Sleep(5 * time.Second)
+    if resp.StatusCode == 200 {
+        log.Println("Offsets deleted successfully")
+        return true
+    } else {
+        log.Println("Response Status Code:", resp.StatusCode)
+        log.Println("Error deleting offsets")
+        return false
+    }
+}
 func handleUpdateLsn(c *cli.Context) bool {
 	var lsnPosition = c.String("lsn")
 	log.Println("***** lsn position:", lsnPosition+"   *****")

sink-connector-lightweight/pom.xml

@@ -24,6 +24,7 @@
         <quarkus.platform.version>2.14.0.Final</quarkus.platform.version>
         <surefire-plugin.version>3.0.0-M7</surefire-plugin.version>
         <sink-connector-library-version>0.0.9</sink-connector-library-version>
+        <javalin-version>5.5.0</javalin-version>
     </properties>
     <dependencyManagement>
         <dependencies>
@@ -58,14 +59,31 @@
         <dependency>
             <groupId>io.javalin</groupId>
             <artifactId>javalin</artifactId>
-            <version>5.5.0</version>
+            <version>${javalin-version}</version>
             <exclusions>
             <exclusion>
                 <groupId>org.slf4j</groupId>
                 <artifactId>slf4j-api</artifactId>
             </exclusion>
             </exclusions>
         </dependency>
+        <!-- Dependency for kotlin -->
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-stdlib</artifactId>
+            <version>1.9.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-stdlib-common</artifactId>
+            <version>1.9.0</version>
+        </dependency>
+        <dependency>
+            <groupId>io.javalin</groupId>
+            <artifactId>javalin-testtools</artifactId>
+            <version>${javalin-version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>com.googlecode.json-simple</groupId>
             <artifactId>json-simple</artifactId>

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/api/DebeziumEmbeddedRestApi.java

@@ -60,6 +60,23 @@ public static void startRestApi(Properties props, Injector injector,
 
         });
 
+        //Delete offsets
+        app.delete("/offsets", ctx -> {
+            ClickHouseSinkConnectorConfig config = new ClickHouseSinkConnectorConfig(PropertiesHelper.toMap(finalProps1));
+            String response = "";
+
+            try {
+                debeziumChangeEventCapture.deleteOffsets(finalProps1);
+            } catch (Exception e) {
+                log.error("Client - Error deleting offsets", e);
+                ctx.result(e.toString());
+                ctx.status(HttpStatus.INTERNAL_SERVER_ERROR);
+                return;
+            }
+            ctx.result(response);
+
+        });
+
         app.post("/binlog", ctx -> {
             if(debeziumChangeEventCapture.isReplicationRunning()) {
                 ctx.status(HttpStatus.BAD_REQUEST);
@@ -136,4 +153,9 @@ public static void stop() {
         if(app != null)
             app.stop();
     }
+
+    // Return the app instance.
+    public static Javalin app() {
+        return app;
+    }
 }

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCapture.java

@@ -86,11 +86,16 @@ public class DebeziumChangeEventCapture {
 
     ClickHouseBatchWriter singleThreadedWriter;
 
+
+    DebeziumOffsetStorage debeziumOffsetStorage;
+
     public DebeziumChangeEventCapture() {
         singleThreadDebeziumEventExecutor = Executors.newFixedThreadPool(1);
+        this.debeziumOffsetStorage = new DebeziumOffsetStorage();
     }
 
 
+
     /**
      * Function to perform DDL operation on the main thread.
      * @param DDL DDL to be executed.
@@ -382,6 +387,20 @@ private Pair<String, String> getDebeziumStorageDatabaseName(Properties props) {
         return Pair.of(tableName, databaseName);
     }
 
+    /**
+     * Function to delete offsets from Debezium storage.
+     * @param props
+     */
+    public void deleteOffsets(Properties props) throws SQLException {
+        DBCredentials dbCredentials = parseDBConfiguration(new ClickHouseSinkConnectorConfig(PropertiesHelper.toMap(props)));
+        BaseDbWriter writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
+                dbCredentials.getDatabase(), dbCredentials.getUserName(),
+                dbCredentials.getPassword(), new ClickHouseSinkConnectorConfig(PropertiesHelper.toMap(props)), this.conn);
+        String offsetKey = this.debeziumOffsetStorage.getOffsetKey(props);
+
+        this.debeziumOffsetStorage.deleteOffsetStorageRow(offsetKey, props, writer);
+    }
+
     /**
      * Function to get the status of Debezium storage.
      * @param props
@@ -476,7 +495,7 @@ public long getLatestRecordTimestamp(ClickHouseSinkConnectorConfig config, Prope
                 databaseName, dbCredentials.getUserName(),
                 dbCredentials.getPassword(), config, this.conn);
 
-        String latestRecordTs = new DebeziumOffsetStorage().getDebeziumLatestRecordTimestamp(props, writer);
+        String latestRecordTs = this.debeziumOffsetStorage.getDebeziumLatestRecordTimestamp(props, writer);
 
         // Convert date string from  2024-01-26 21:57:47 format to milliseconds.
         SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@@ -515,14 +534,14 @@ public void updateDebeziumStorageStatus(ClickHouseSinkConnectorConfig config, Pr
         BaseDbWriter writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
                 databaseName, dbCredentials.getUserName(),
                 dbCredentials.getPassword(), config, this.conn);
-        String offsetValue = new DebeziumOffsetStorage().getDebeziumStorageStatusQuery(props, writer);
+        String offsetValue = this.debeziumOffsetStorage.getDebeziumStorageStatusQuery(props, writer);
 
-        String offsetKey = new DebeziumOffsetStorage().getOffsetKey(props);
-        String updateOffsetValue = new DebeziumOffsetStorage().updateBinLogInformation(offsetValue,
+        String offsetKey = this.debeziumOffsetStorage.getOffsetKey(props);
+        String updateOffsetValue = this.debeziumOffsetStorage.updateBinLogInformation(offsetValue,
                 binlogFile, binLogPosition, gtid);
 
-        new DebeziumOffsetStorage().deleteOffsetStorageRow(offsetKey, props, writer);
-        new DebeziumOffsetStorage().updateDebeziumStorageRow(writer, tableName, offsetKey, updateOffsetValue,
+        this.debeziumOffsetStorage.deleteOffsetStorageRow(offsetKey, props, writer);
+        this.debeziumOffsetStorage.updateDebeziumStorageRow(writer, tableName, offsetKey, updateOffsetValue,
                 System.currentTimeMillis());
 
     }
@@ -547,14 +566,14 @@ public void updateDebeziumStorageStatus(ClickHouseSinkConnectorConfig config, Pr
         BaseDbWriter writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
                 databaseName, dbCredentials.getUserName(),
                 dbCredentials.getPassword(), config, this.conn);
-        String offsetValue = new DebeziumOffsetStorage().getDebeziumStorageStatusQuery(props, writer);
+        String offsetValue = this.debeziumOffsetStorage.getDebeziumStorageStatusQuery(props, writer);
 
-        String offsetKey = new DebeziumOffsetStorage().getOffsetKey(props);
-        String updateOffsetValue = new DebeziumOffsetStorage().updateLsnInformation(offsetValue,
+        String offsetKey = this.debeziumOffsetStorage.getOffsetKey(props);
+        String updateOffsetValue = this.debeziumOffsetStorage.updateLsnInformation(offsetValue,
                 lsn);
 
-        new DebeziumOffsetStorage().deleteOffsetStorageRow(offsetKey, props, writer);
-        new DebeziumOffsetStorage().updateDebeziumStorageRow(writer, tableName, offsetKey, updateOffsetValue,
+        this.debeziumOffsetStorage.deleteOffsetStorageRow(offsetKey, props, writer);
+        this.debeziumOffsetStorage.updateDebeziumStorageRow(writer, tableName, offsetKey, updateOffsetValue,
                 System.currentTimeMillis());
 
     }