2.4.0

doc/img/lightweight_main.drawio

@@ -0,0 +1 @@
+<mxGraphModel><root><mxCell id="0"/><mxCell id="1" parent="0"/></root></mxGraphModel>

doc/img/single_threaded.drawio

@@ -0,0 +1,40 @@
+<mxfile host="app.diagrams.net" agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36" version="24.7.14">
+  <diagram name="Page-1" id="DBxunoqnh1ms4ItBoYgd">
+    <mxGraphModel dx="3485" dy="1662" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="850" pageHeight="1100" math="0" shadow="0">
+      <root>
+        <mxCell id="0" />
+        <mxCell id="1" parent="0" />
+        <mxCell id="ik4SbChLGUNpw0kpHCm7-1" value="binlogs" style="outlineConnect=0;dashed=0;verticalLabelPosition=bottom;verticalAlign=top;align=center;html=1;shape=mxgraph.aws3.mysql_db_instance_2;fillColor=#2E73B8;gradientColor=none;" vertex="1" parent="1">
+          <mxGeometry x="-1690" y="293.5" width="60" height="63" as="geometry" />
+        </mxCell>
+        <mxCell id="ik4SbChLGUNpw0kpHCm7-2" value="" style="rounded=1;whiteSpace=wrap;html=1;dashed=1;" vertex="1" parent="1">
+          <mxGeometry x="-1600" y="180" width="400" height="360" as="geometry" />
+        </mxCell>
+        <mxCell id="ik4SbChLGUNpw0kpHCm7-3" value="&lt;b&gt;Debezium&lt;/b&gt;&lt;div&gt;&lt;b&gt;Queue&lt;/b&gt;&lt;/div&gt;" style="verticalLabelPosition=bottom;html=1;verticalAlign=top;align=center;strokeColor=none;fillColor=#00BEF2;shape=mxgraph.azure.queue_generic;pointerEvents=1;" vertex="1" parent="1">
+          <mxGeometry x="-1570" y="300" width="180" height="50" as="geometry" />
+        </mxCell>
+        <mxCell id="ik4SbChLGUNpw0kpHCm7-6" style="edgeStyle=orthogonalEdgeStyle;rounded=0;orthogonalLoop=1;jettySize=auto;html=1;entryX=0;entryY=0.5;entryDx=0;entryDy=0;entryPerimeter=0;" edge="1" parent="1" source="ik4SbChLGUNpw0kpHCm7-1" target="ik4SbChLGUNpw0kpHCm7-3">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="ik4SbChLGUNpw0kpHCm7-7" value="ClickHouse" style="shape=cylinder3;whiteSpace=wrap;html=1;boundedLbl=1;backgroundOutline=1;size=15;fillColor=#fff2cc;strokeColor=#d6b656;" vertex="1" parent="1">
+          <mxGeometry x="-1120" y="300" width="60" height="80" as="geometry" />
+        </mxCell>
+        <mxCell id="ik4SbChLGUNpw0kpHCm7-8" value="&lt;b&gt;Main&lt;br&gt;Thread&lt;/b&gt;" style="whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="-1335" y="190" width="60" height="270" as="geometry" />
+        </mxCell>
+        <mxCell id="ik4SbChLGUNpw0kpHCm7-9" style="edgeStyle=orthogonalEdgeStyle;rounded=0;orthogonalLoop=1;jettySize=auto;html=1;entryX=-0.017;entryY=0.53;entryDx=0;entryDy=0;entryPerimeter=0;" edge="1" parent="1" source="ik4SbChLGUNpw0kpHCm7-3" target="ik4SbChLGUNpw0kpHCm7-8">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="ik4SbChLGUNpw0kpHCm7-16" value="&lt;b&gt;Sink Connector Lightweight&lt;/b&gt;" style="text;html=1;align=center;verticalAlign=middle;whiteSpace=wrap;rounded=0;" vertex="1" parent="1">
+          <mxGeometry x="-1450" y="560" width="190" height="30" as="geometry" />
+        </mxCell>
+        <mxCell id="ik4SbChLGUNpw0kpHCm7-29" style="edgeStyle=orthogonalEdgeStyle;rounded=0;orthogonalLoop=1;jettySize=auto;html=1;entryX=0;entryY=0.5;entryDx=0;entryDy=0;entryPerimeter=0;" edge="1" parent="1" source="ik4SbChLGUNpw0kpHCm7-8" target="ik4SbChLGUNpw0kpHCm7-7">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="ik4SbChLGUNpw0kpHCm7-30" value="&lt;b&gt;Single Threaded&lt;/b&gt;: Batch is defined by debezium batch size parameters&amp;nbsp;" style="text;html=1;align=center;verticalAlign=middle;whiteSpace=wrap;rounded=0;" vertex="1" parent="1">
+          <mxGeometry x="-1540" y="480" width="230" height="30" as="geometry" />
+        </mxCell>
+      </root>
+    </mxGraphModel>
+  </diagram>
+</mxfile>

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+"   *****")