Added logic to use mysql go package to verify connection to MySQL

sink-connector-client/config.yml.template

@@ -0,0 +1,48 @@
+name: "company-1"
+topic.prefix: "sink-connector-1"
+database.hostname: "mysql-master"
+database.port: "3306"
+database.user: "root"
+database.password: "root"
+database.server.id: "12345"
+database.server.name: "ER54"
+database.include.list: test
+table.include.list: ""
+clickhouse.server.url: "clickhouse"
+clickhouse.server.user: "root"
+clickhouse.server.password: "root"
+clickhouse.server.port: "8123"
+database.allowPublicKeyRetrieval: "true"
+snapshot.mode: "initial"
+offset.flush.interval.ms: 5000
+connector.class: "io.debezium.connector.mysql.MySqlConnector"
+offset.storage: "io.debezium.storage.jdbc.offset.JdbcOffsetBackingStore"
+offset.storage.jdbc.offset.table.name: "altinity_sink_connector.replica_source_info"
+offset.storage.jdbc.url: "jdbc:clickhouse://clickhouse:8123/altinity_sink_connector"
+offset.storage.jdbc.user: "root"
+offset.storage.jdbc.password: "root"
+offset.storage.jdbc.offset.table.ddl: "CREATE TABLE if not exists %s
+(
+    `id` String,
+    `offset_key` String,
+    `offset_val` String,
+    `record_insert_ts` DateTime,
+    `record_insert_seq` UInt64,
+    `_version` UInt64 MATERIALIZED toUnixTimestamp64Nano(now64(9))
+)
+ENGINE = ReplacingMergeTree(_version) ORDER BY offset_key SETTINGS index_granularity = 8192"
+offset.storage.jdbc.offset.table.delete: "select * from %s"
+offset.storage.jdbc.offset.table.select: "SELECT id, offset_key, offset_val FROM %s FINAL ORDER BY record_insert_ts, record_insert_seq"
+schema.history.internal: "io.debezium.storage.jdbc.history.JdbcSchemaHistory"
+schema.history.internal.jdbc.url: "jdbc:clickhouse://clickhouse:8123/altinity_sink_connector"
+schema.history.internal.jdbc.user: "root"
+schema.history.internal.jdbc.password: "root"
+schema.history.internal.jdbc.schema.history.table.ddl: "CREATE TABLE if not exists %s
+(`id` VARCHAR(36) NOT NULL, `history_data` VARCHAR(65000), `history_data_seq` INTEGER, `record_insert_ts` TIMESTAMP NOT NULL, `record_insert_seq` INTEGER NOT NULL) ENGINE=ReplacingMergeTree(record_insert_seq) order by id"
+schema.history.internal.jdbc.schema.history.table.name: "altinity_sink_connector.replicate_schema_history"
+enable.snapshot.ddl: "true"
+persist.raw.bytes: "false"
+auto.create.tables: "true"
+auto.create.tables.replicated: "true"
+database.connectionTimeZone: "UTC"
+clickhouse.database.override.map: "employees:employees2, products:productsnew

sink-connector-client/create_config.go

@@ -36,13 +36,10 @@ func validateMySQL(sourceUsername string, sourcePassword string, sourceHost stri
 	// 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 rows == 'OFF' {
+	// 	log.fatal("Binlogs are not enabled")
+	// 	return false
+	// }
 
 	if err != nil {
 		log.Fatal(err)

sink-connector-client/create_config_test.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/testcontainers/testcontainers-go"
+)
+
+// Use testcontainers to spin up a mysql container
+func TestValidateMySQL(t *testing.T) {
+	// Spin up a MySQL container
+	req := testcontainers.ContainerRequest{
+		Image:        "mysql:5.7",
+		ExposedPorts: []string{"3306/tcp"},
+		Env: map[string]string{
+			"MYSQL_ROOT_PASSWORD": "password",
+		},
+	}
+	mysqlContainer, err := testcontainers.GenericContainer(context.Background(), testcontainers.GenericContainerRequest{
+		ContainerRequest: req,
+		Started:          true,
+	})
+	if err != nil {
+		t.Fatalf("Could not start MySQL container: %s", err)
+	}
+	defer mysqlContainer.Terminate(context.Background())
+
+	// Wait for container to start
+	// sleep till container has started
+	time.Sleep(10 * time.Second)
+
+	// Get the MySQL container's IP address
+	ip, err := mysqlContainer.Host(context.Background())
+	if err != nil {
+		t.Fatalf("Could not get MySQL container IP: %s, error %s", ip, err)
+	}
+	port, err := mysqlContainer.MappedPort(context.Background(), "3306")
+	if err != nil {
+		t.Fatalf("Could not get MySQL container port: %s error %s", port, err)
+	}
+
+	// Set the MySQL container's IP address and port
+
+	// Validate MySQL credentials
+	if !validateMySQL("root", "password", ip, port.Port()) {
+		t.Fatalf("Could not validate MySQL credentials")
+	}
+}

sink-connector-client/go.mod

@@ -1,22 +1,77 @@
 module sink-connector-client
 
-go 1.18
+go 1.21
+
+toolchain go1.21.4
 
 require (
 	github.com/levigross/grequests v0.0.0-20221222020224-9eee758d18d5
-	github.com/stretchr/testify v1.8.2
+	github.com/stretchr/testify v1.9.0
+	github.com/testcontainers/testcontainers-go v0.32.0
 	github.com/tidwall/pretty v1.2.1
 	github.com/urfave/cli v1.22.13
+	gopkg.in/yaml.v2 v2.4.0
+//testcontainers
 )
 
+require filippo.io/edwards25519 v1.1.0 // indirect
+
 require (
-	filippo.io/edwards25519 v1.1.0 // indirect
+	dario.cat/mergo v1.0.0 // indirect
+	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
+	github.com/Microsoft/go-winio v0.6.2 // indirect
+	github.com/Microsoft/hcsshim v0.11.5 // indirect
+	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
+	github.com/containerd/containerd v1.7.18 // indirect
+	github.com/containerd/errdefs v0.1.0 // indirect
+	github.com/containerd/log v0.1.0 // indirect
+	github.com/cpuguy83/dockercfg v0.3.1 // 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/distribution/reference v0.6.0 // indirect
+	github.com/docker/docker v27.0.3+incompatible // indirect
+	github.com/docker/go-connections v0.5.0 // indirect
+	github.com/docker/go-units v0.5.0 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
+	github.com/go-logr/logr v1.4.1 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-ole/go-ole v1.2.6 // indirect
+	github.com/go-sql-driver/mysql v1.8.1
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/go-querystring v1.0.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/klauspost/compress v1.17.4 // indirect
+	github.com/kr/text v0.2.0 // indirect
+	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
+	github.com/magiconair/properties v1.8.7 // indirect
+	github.com/moby/docker-image-spec v1.3.1 // indirect
+	github.com/moby/patternmatcher v0.6.0 // indirect
+	github.com/moby/sys/sequential v0.5.0 // indirect
+	github.com/moby/sys/user v0.1.0 // indirect
+	github.com/moby/term v0.5.0 // indirect
+	github.com/morikuni/aec v1.0.0 // indirect
+	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/opencontainers/image-spec v1.1.0 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
-	golang.org/x/net v0.0.0-20181011144130-49bb7cea24b1 // indirect
+	github.com/shirou/gopsutil/v3 v3.23.12 // indirect
+	github.com/shoenig/go-m1cpu v0.1.6 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
+	github.com/tklauser/go-sysconf v0.3.12 // indirect
+	github.com/tklauser/numcpus v0.6.1 // indirect
+	github.com/yusufpapurcu/wmi v1.2.3 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect
+	go.opentelemetry.io/otel v1.24.0 // indirect
+	go.opentelemetry.io/otel/metric v1.24.0 // indirect
+	go.opentelemetry.io/otel/trace v1.24.0 // indirect
+	golang.org/x/crypto v0.22.0 // indirect
+	golang.org/x/net v0.23.0 // indirect
+	golang.org/x/sys v0.19.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b // indirect
+	google.golang.org/grpc v1.59.0 // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )