feature(kafka-conn-sink): add tests for kafka sink connector

The change implements tests for ScyllaDB Sink Connector
(https://github.com/scylladb/kafka-connect-scylladb).

(cherry picked from commit 56a1d5e)

# Conflicts:
#	sdcm/utils/common.py

configurations/kafka/kafka-sink-avro.yaml

@@ -0,0 +1,24 @@
+stress_cmd: >
+  python_thread -thread=KafkaProducerThread -record-format=avro -topic=table1 -num-records=1000
+  -key-schema='{"type":"record","name":"key_schema","fields":[{"name":"id","type":"int"}]}'
+  -value-schema='{"type":"record","name":"value_schema","fields":[{"name":"firstName","type":"string"},{"name":"lastName","type":"string"}]}'
+stress_read_cmd: >
+  python_thread -thread=KafkaValidatorThread -keyspace=keyspace1 -table=table1 -record-format=avro -num-records=1000 -batch-size=500
+  -value-schema='{"type":"record","name":"value_schema","fields":[{"name":"firstName","type":"string"},{"name":"lastName","type":"string"}]}'
+
+kafka_connectors:
+  - source: 'hub:scylladb/kafka-connect-scylladb:1.1.1'
+    name: 'scylladb-sink-connector'
+    config:
+      "connector.class": 'io.connect.scylladb.ScyllaDbSinkConnector'
+      "tasks.max": 1
+      topics: 'table1'
+      "scylladb.keyspace": 'keyspace1'
+      "scylladb.consistency.level": 'ONE'
+      "key.converter": 'io.confluent.connect.avro.AvroConverter'
+      "key.converter.schema.registry.url": 'http://kafka-schema-registry:8081'
+      "value.converter": 'io.confluent.connect.avro.AvroConverter'
+      "value.converter.schema.registry.url": 'http://kafka-schema-registry:8081'
+      "key.converter.schemas.enable": true
+      "value.converter.schemas.enable": true
+      "topic.table1.keyspace1.table1.mapping": 'id=key.id, firstName=value.firstName, lastName=value.lastName'

configurations/kafka/kafka-sink-json.yaml

@@ -0,0 +1,22 @@
+stress_cmd: >
+  python_thread -thread=KafkaProducerThread -record-format=json-with-schema -topic=table1 -num-records=1000
+  -key-schema='{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}]}'
+  -value-schema='{"type":"struct","fields":[{"type":"string","optional":false,"field":"name"},{"type":"string","optional":true,"field":"department"}]}'
+stress_read_cmd: >
+  python_thread -thread=KafkaValidatorThread -keyspace=keyspace1 -table=table1 -record-format=json-with-schema -num-records=1000 -batch-size=500
+  -value-schema='{"type":"struct","fields":[{"type":"string","optional":false,"field":"name"},{"type":"string","optional":true,"field":"department"}]}'
+
+kafka_connectors:
+  - source: 'hub:scylladb/kafka-connect-scylladb:1.1.1'
+    name: 'scylladb-sink-connector'
+    config:
+      "connector.class": 'io.connect.scylladb.ScyllaDbSinkConnector'
+      "tasks.max": 1
+      topics: 'table1'
+      "scylladb.keyspace": 'keyspace1'
+      "scylladb.consistency.level": 'ONE'
+      "key.converter": 'org.apache.kafka.connect.json.JsonConverter'
+      "value.converter": 'org.apache.kafka.connect.json.JsonConverter'
+      "key.converter.schemas.enable": true
+      "value.converter.schemas.enable": true
+      "topic.table1.keyspace1.table1.mapping": 'id=key.id, name=value.name, department=value.department'

docker/env/version

@@ -1 +1 @@
-1.78-argus-0.13.0
+1.79-confluent-kafka-2.5.3

requirements.in

@@ -53,3 +53,5 @@ deepdiff==6.2.3
 PyGithub==2.1.1
 gimme-aws-creds==2.8.0
 kafka-python==2.0.2
+confluent-kafka==2.5.3
+fastavro==1.9.7

requirements.txt

@@ -452,6 +452,42 @@ colorama==0.4.4 \
     --hash=sha256:5941b2b48a20143d2267e95b1c2a7603ce057ee39fd88e7329b0c292aa16869b \
     --hash=sha256:9f47eda37229f68eee03b24b9748937c7dc3868f906e8ba69fbcbdd3bc5dc3e2
     # via awscli
+confluent-kafka==2.5.3 \
+    --hash=sha256:0751302b0fd8090cbca92d7d34d237768923107b30de2611f3db93c2118cf2a8 \
+    --hash=sha256:0b14928cddba963ea7d1c66aa268b6d37976bc91b4cf2100b5b7336d848ced22 \
+    --hash=sha256:0cdb150c12d5ac6e33572cbf16243284c65a178e3719baa610a48d672e9d92bf \
+    --hash=sha256:0e0cb3b18a59d1c6fcae60297ee25b5c65d5c39c8ad8033a8fa1392498a71c9e \
+    --hash=sha256:27d048429b138667c51541adc04bb398afa61a37a7be89f16ff9a318019d02c6 \
+    --hash=sha256:2b8eef8c2f963ca6f5fcc79a0d6edef4e25fba83dfc0ef3f0401e1644f60ff11 \
+    --hash=sha256:3b69c3120e0cac9ca463ca603ddc9d4e811409ef4ec69d2b6bb8bd94d6fce95e \
+    --hash=sha256:3daad72791ae06dec257c9105278e89ae0924e86ef107a1acb443106f002f361 \
+    --hash=sha256:46c6063726fcdae835902961bb6c0e4c148499b87fdd513e6b2a6b406922ae3e \
+    --hash=sha256:490836e9fc3b4489721327e3df987c8667916a97d178e2296913c8d5de6623a9 \
+    --hash=sha256:4cfb18d69e6912fe90cbbcc9c7d805988122c51ab3041e1424ace64bc31b736f \
+    --hash=sha256:4dd5fa74231fc21c3a26eeda1999a27f84768a6291a8b04c3cd61ac1deea4ace \
+    --hash=sha256:505078b402dde98dc06dc66b6356acd19984742ef6b82dd52fb860f2a14b5a57 \
+    --hash=sha256:5122b8e9f94b6160d47e8f0020857376caa21f715b95c4b13c68683b47260c8f \
+    --hash=sha256:61a92637bea8fca454ec711f46e7753647deb7da56132995103acb5eb5041a29 \
+    --hash=sha256:7125c3f86a76136b25aa21c94303b33709e2dd15f777395ea81fbd6872d9147b \
+    --hash=sha256:75e1da68b199ef2472e47785d9a5c2dc75d307ed78827ad929bb733728b18567 \
+    --hash=sha256:806c43fd1524034a9b6c958b4f9395ff5f56ef697218a336eac1da5006184f66 \
+    --hash=sha256:8a1a2a8756b2c1cd2654ea83d1e819a6e2c0a4337eacec50bfd2ab1f0c24a29c \
+    --hash=sha256:8d86de3e2c7bb59fb16faea468e833712912106f32a3a3ec345088c366042734 \
+    --hash=sha256:8ec7a407bcb2eb122ff159d602cedc41d858f4c66a436c778f5d2f9f15fbec4e \
+    --hash=sha256:a2ed265bf3420811efd802fd8ebf5ec0f20a82e9baeff5299a67f6a84dde1b06 \
+    --hash=sha256:a6f152e704b01c6a726233d081921454b7de106a5e4036994d1d5f4b34e7e46f \
+    --hash=sha256:ac8b5fe45ee9c11ce7a516dc7c41441ebb17d9ff63c8646a59b8e52bd791b154 \
+    --hash=sha256:c284eefed1b27133d90afc0fa2fd735864db8501190f3c2e0c8d8b1a20b07759 \
+    --hash=sha256:cfabe291cda68fdc3136f2f367dd4e5a6c3750113785f0c1936ba9cae09f4e9d \
+    --hash=sha256:dae80e9e1e4417462fe61f64da0ab111395719e35c9f7f3eac7c671ff5e868fe \
+    --hash=sha256:db30418bb36723a02ba51e058312056d0403c5f245beb379bff66e4b0c14337b \
+    --hash=sha256:e9ffb298b3ea3477afdaa5da6033d35dc0be01b10537d9b63994411e79b41477 \
+    --hash=sha256:eb80c22a7ca17839f229f299bafca1450c9fe4d5ca222e60e52428df91d42b56 \
+    --hash=sha256:eca625b0a8742d864a954bbe6493d453c07bacedf9e10d71a54dd1047f775778 \
+    --hash=sha256:f034f13c08ba238154d818294ceabb2257e8df8fb6489f891ec7600c7c541553 \
+    --hash=sha256:f626494cd6ad18fa2ed83f80d687bc0194cff6f61b3d4f2aa183efa23ede2e02 \
+    --hash=sha256:fa2318eaa9b2d5f3ebc2022b71e4ebf6242c13963b4faccf46eea49fea0ad91f
+    # via -r requirements.in
 cryptography==43.0.1 \
     --hash=sha256:014f58110f53237ace6a408b5beb6c427b64e084eb451ef25a28308270086494 \
     --hash=sha256:1bbcce1a551e262dfbafb6e6252f1ae36a248e615ca44ba302df077a846a8806 \
@@ -535,6 +571,39 @@ fabric==2.6.0 \
     --hash=sha256:47f184b070272796fd2f9f0436799e18f2ccba4ee8ee587796fca192acd46cd2 \
     --hash=sha256:7a71714b8b8f28cf828eceb155196f43ebac1bd4c849b7161ed5993d1cbcaa40
     # via -r requirements.in
+fastavro==1.9.7 \
+    --hash=sha256:0b2f9bafa167cb4d1c3dd17565cb5bf3d8c0759e42620280d1760f1e778e07fc \
+    --hash=sha256:13e11c6cb28626da85290933027cd419ce3f9ab8e45410ef24ce6b89d20a1f6c \
+    --hash=sha256:17de68aae8c2525f5631d80f2b447a53395cdc49134f51b0329a5497277fc2d2 \
+    --hash=sha256:1d09227d1f48f13281bd5ceac958650805aef9a4ef4f95810128c1f9be1df736 \
+    --hash=sha256:2af559f30383b79cf7d020a6b644c42ffaed3595f775fe8f3d7f80b1c43dfdc5 \
+    --hash=sha256:2db993ae6cdc63e25eadf9f93c9e8036f9b097a3e61d19dca42536dcc5c4d8b3 \
+    --hash=sha256:2fcce036c6aa06269fc6a0428050fcb6255189997f5e1a728fc461e8b9d3e26b \
+    --hash=sha256:3b683693c8a85ede496ebebe115be5d7870c150986e34a0442a20d88d7771224 \
+    --hash=sha256:4e1289b731214a7315884c74b2ec058b6e84380ce9b18b8af5d387e64b18fc44 \
+    --hash=sha256:536f5644737ad21d18af97d909dba099b9e7118c237be7e4bd087c7abde7e4f0 \
+    --hash=sha256:56304401d2f4f69f5b498bdd1552c13ef9a644d522d5de0dc1d789cf82f47f73 \
+    --hash=sha256:58f76a5c9a312fbd37b84e49d08eb23094d36e10d43bc5df5187bc04af463feb \
+    --hash=sha256:6312fa99deecc319820216b5e1b1bd2d7ebb7d6f221373c74acfddaee64e8e60 \
+    --hash=sha256:7313def3aea3dacface0a8b83f6d66e49a311149aa925c89184a06c1ef99785d \
+    --hash=sha256:76d9d96f98052615ab465c63ba8b76ed59baf2e3341b7b169058db104cbe2aa0 \
+    --hash=sha256:7c911366c625d0a997eafe0aa83ffbc6fd00d8fd4543cb39a97c6f3b8120ea87 \
+    --hash=sha256:912283ed48578a103f523817fdf0c19b1755cea9b4a6387b73c79ecb8f8f84fc \
+    --hash=sha256:919f3549e07a8a8645a2146f23905955c35264ac809f6c2ac18142bc5b9b6022 \
+    --hash=sha256:9be089be8c00f68e343bbc64ca6d9a13e5e5b0ba8aa52bcb231a762484fb270e \
+    --hash=sha256:9de1fa832a4d9016724cd6facab8034dc90d820b71a5d57c7e9830ffe90f31e4 \
+    --hash=sha256:b525c363e267ed11810aaad8fbdbd1c3bd8837d05f7360977d72a65ab8c6e1fa \
+    --hash=sha256:b6b2ccdc78f6afc18c52e403ee68c00478da12142815c1bd8a00973138a166d0 \
+    --hash=sha256:cc811fb4f7b5ae95f969cda910241ceacf82e53014c7c7224df6f6e0ca97f52f \
+    --hash=sha256:d576eccfd60a18ffa028259500df67d338b93562c6700e10ef68bbd88e499731 \
+    --hash=sha256:e87d04b235b29f7774d226b120da2ca4e60b9e6fdf6747daef7f13f218b3517a \
+    --hash=sha256:eac69666270a76a3a1d0444f39752061195e79e146271a568777048ffbd91a27 \
+    --hash=sha256:ec2e96bdabd58427fe683329b3d79f42c7b4f4ff6b3644664a345a655ac2c0a1 \
+    --hash=sha256:ec8499dc276c2d2ef0a68c0f1ad11782b2b956a921790a36bf4c18df2b8d4020 \
+    --hash=sha256:edc28ab305e3c424de5ac5eb87b48d1e07eddb6aa08ef5948fcda33cc4d995ce \
+    --hash=sha256:ee9bf23c157bd7dcc91ea2c700fa3bd924d9ec198bb428ff0b47fa37fe160659 \
+    --hash=sha256:fb8749e419a85f251bf1ac87d463311874972554d25d4a0b19f6bdc56036d7cf
+    # via -r requirements.in
 fido2==0.9.3 \
     --hash=sha256:b45e89a6109cfcb7f1bb513776aa2d6408e95c4822f83a253918b944083466ec
     # via