add step to start sink connector

sink-connector-lightweight/tests/integration/helpers/cluster.py

@@ -527,6 +527,8 @@ def node(self, name):
             return ClickHouseNode(self, name, data_base_type="clickhouse")
         if name.startswith("mysql"):
             return MySQLNode(self, name, data_base_type="mysql")
+        if name.startswith("clickhouse-sink"):
+            return SinkConnector(self, name)
         return Node(self, name)
 
     def down(self, timeout=300):
@@ -925,6 +927,45 @@ def query(
         )
 
 
+class SinkConnector(DatabaseNode):
+    """ClickHouse Sink Connector node."""
+
+    sink_connector_cli = "./sink-connector-client "
+
+    def start_sink_connector(self, timeout=300):
+        with Given("I start ClickHouse Sink Connector"):
+            self.command(
+                node=self.name,
+                command=f"java -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005 -jar /app.jar "
+                f"/config.yml com.altinity.clickhouse.debezium.embedded.ClickHouseDebeziumEmbeddedApplication",
+                timeout=timeout,
+            )
+
+    def stop_replication(self, timeout=300):
+        with Given("I stop ClickHouse Sink Connector replication"):
+            self.command(
+                node=self.name,
+                command=f"{self.sink_connector_cli} stop_replica",
+                timeout=timeout,
+            )
+
+    def start_replication(self):
+        with Given("I start ClickHouse Sink Connector replication"):
+            self.command(
+                node=self.name,
+                command=f"{self.sink_connector_cli} start_replica",
+                timeout=300,
+            )
+
+    def change_replica_source(self):
+        with Given("I change ClickHouse Sink Connector replica source"):
+            self.command(
+                node=self.name,
+                command=f"{self.sink_connector_cli} change_replica_source",
+                timeout=300,
+            )
+
+
 class ClickHouseNode(DatabaseNode):
     """Node with ClickHouse server."""
 

sink-connector-lightweight/tests/integration/regression_auto.py

@@ -6,7 +6,6 @@
 
 from testflows.core import *
 
-
 append_path(sys.path, "..")
 
 from integration.helpers.argparser import argparser
@@ -15,7 +14,6 @@
 from integration.requirements.requirements import *
 from integration.tests.steps.steps_global import *
 
-
 ffails = {
     "schema changes/table recreation with different datatypes": (
         Skip,
@@ -122,7 +120,6 @@
     ),
 }
 
-
 xflags = {}
 
 
@@ -182,6 +179,11 @@ def regression(
             caller_dir=os.path.join(current_dir()),
         )
 
+    with And("I start ClickHouse sink connector"):
+        sink_node = self.context.cluster.node("clickhouse-sink-connector-lt")
+
+        sink_node.start_sink_connector()
+
     self.context.cluster = cluster
 
     self.context.env = env