feat: add kafka helper methods

Co-authored-by: Luca Schmitz <schmitz@sipgate.de>

Author: flying7eleven

pyproject.toml

@@ -31,6 +31,9 @@ db = [
 metrics = [
     "prometheus-client~=0.23.1"
 ]
+kafka = [
+    "confluent-kafka==2.10.0"
+]
 dev = [
     "pre-commit",
 ]

requirements.txt

@@ -1,5 +1,5 @@
 #
-# This file is autogenerated by pip-compile with Python 3.13
+# This file is autogenerated by pip-compile with Python 3.14
 # by the following command:
 #
 #    pip-compile --all-extras --allow-unsafe --generate-hashes --output-file=requirements.txt pyproject.toml
@@ -142,6 +142,43 @@ cfgv==3.5.0 \
     --hash=sha256:a8dc6b26ad22ff227d2634a65cb388215ce6cc96bbcc5cfde7641ae87e8dacc0 \
     --hash=sha256:d5b1034354820651caa73ede66a6294d6e95c1b00acc5e9b098e917404669132
     # via pre-commit
+confluent-kafka==2.10.0 \
+    --hash=sha256:0fb9ea56dc8d1773678f7ea3c1a1c0f59001502c8ff44df79142ab471915e81e \
+    --hash=sha256:10230c3da4cd047cf488cef405f6e209da4e8403c0e52e45b89b1d878420082b \
+    --hash=sha256:13f49fe760ff33c43c852c3e09ef7a914efea9a8a7e553080e13d611cd3eedad \
+    --hash=sha256:1448d6a5e8afd99c81221db2fd7cf33bd2079ce04df8ff5882b208911b0112a7 \
+    --hash=sha256:2268a607d374325e3a8fc779d8639de4344ce4d1ad2b88e8f9d13d97baae4e71 \
+    --hash=sha256:30a346908f3ad49c4bc1cb5557e7a8ce484190f8633aa18f9b87b2620809ac13 \
+    --hash=sha256:30d40b0b43f0e15d830b610640eeb625ebd49cc792254fcdcd5ae7396d91004d \
+    --hash=sha256:32f5ea8ecee98d1c1afd523115cbdb719a6f87b852c19b8898fa76db95ec5f4a \
+    --hash=sha256:3e1915c428c044a5764fdd79cd12e94c961d9e74863ae1440287141173bf3138 \
+    --hash=sha256:413d7a3d4caca6b2c28dbe86595ef17af83d679edea403bcb75fd5af3bd5afa7 \
+    --hash=sha256:46734d24b0e0c8fd84b05174a7338a71eb3a7d34a494e4c458838f89c537508f \
+    --hash=sha256:4e4d4350f66eb3c4cd987f603f08974e461ce732dc63741744e09e0b490a3b1d \
+    --hash=sha256:514c4a4dcb522aacdc17fa3e2751ae8670553850c9e22844ac997ec6459b3a48 \
+    --hash=sha256:5784cf3301470ce85a41b5dbcf1b7d4fe0ac31eee120f3c184b18ed79f32dc9f \
+    --hash=sha256:5a7a16592ce6c5fffdfec4f587ab887f478e68f70df14484184e4e8ec3a86c90 \
+    --hash=sha256:635c62a51ec31ff82864043120dbc7c61eb62abf4e0682cc6ee8526e9b04e16d \
+    --hash=sha256:649ccde18b4f2ff509a16093a06ca13629e4f4b3146d4ef5a82805c869cf8cbd \
+    --hash=sha256:6b692f7e1af2c2085a342ffb0d1c2d0fd6664b3bff43a3c092c0c4d2c6c8f77b \
+    --hash=sha256:6b8b057d436d82c1b2d819915d03e2b446bfa55403acd440f9222509ba2147de \
+    --hash=sha256:7252fee8d7287bae71f1a506967b3782e67ba5b699c4d095d16fabce92a63cfd \
+    --hash=sha256:735c560ecb076e33eb5e14741a1cd55023f0cbc156e4d09f8591dcf9e55efc0b \
+    --hash=sha256:7eae0c1e49fb42e5085555c4387102034c0fe3194e2572049769ef4c30f733cc \
+    --hash=sha256:7eead1134e3aae9380bf5c923d93e25052cfca5b6a55bf99d11ebadf3ec6f21e \
+    --hash=sha256:82800e5cf27e21a7bbccb5aa6f2d9114ab253364526b08b0b97614369bf95612 \
+    --hash=sha256:84c4e50aaf0a51402810bd82cc3b9601a17c866fb44a1aa310b7686009d06b56 \
+    --hash=sha256:965743d6832d3947c29970eeeedcb3728dfe69b5296fd5578fdb18ce39792318 \
+    --hash=sha256:9a80084018177238dbe0dd6562c46259e1883e2eaf9eef8945bf74f13fc69b5c \
+    --hash=sha256:a89e9ba5c8e18891b3bbf333001a9e12af43bd5bc3c3353065109582f3f078fc \
+    --hash=sha256:b81d70ba4fbb81350daccea9de7d99f190bb02bc4d0dfbb151ce9ddee501e245 \
+    --hash=sha256:bba76d0965b31e61c9d7a45831e8f801f089af98a781508b74daeac54268b883 \
+    --hash=sha256:bf706b1ebb540276791b5c42ac624c93195c34782848b24b2da1467caeef479b \
+    --hash=sha256:d5e841e2e2a75fab254acf806449ce013d4e72470bf0aa7c96469fa9b3829693 \
+    --hash=sha256:ddb9400786eedd8a6c0cbcceaddb8329a2597f4c02dc72389e83859da47d22d6 \
+    --hash=sha256:e7dd3a0607139ed3de4d3c511861ff8a84af0f04c47acb6c899f033540027c58 \
+    --hash=sha256:ef5aad8c04c54fe6ef89dc08a4ba24a80be90b05bbbc3348370c1c51222e0a5e
+    # via sipgate_e2e_test_utils (pyproject.toml)
 distlib==0.4.0 \
     --hash=sha256:9659f7d87e46584a30b5780e43ac7a2143098441670ff0a49d5f9034c54a6c16 \
     --hash=sha256:feec40075be03a04501a973d81f633735b4b69f98b05450592310c0f401a4e0d
@@ -286,87 +323,6 @@ frozenlist==1.8.0 \
     # via
     #   aiohttp
     #   aiosignal
-greenlet==3.5.1 \
-    --hash=sha256:001775efe7b8e758861294c7a27c28af87f3f3f1c20468a2bc618c45b346c061 \
-    --hash=sha256:00929c98ec525fd9bf075875d8c5f6a983a90906cdf78a66e6de2d8e466c2a19 \
-    --hash=sha256:017a544f0385d441e88714160d089d6900ef46c9eff9d99b6715a5ef2d127747 \
-    --hash=sha256:089fff7a6ce8d9316d1f65ebc00273a56be258c1725b32b94de90a3a979557e1 \
-    --hash=sha256:1072b4f9edcc1e192d9283a66a3e68d6b84c561de33a83d7858beb9ba1effe10 \
-    --hash=sha256:10a9a1c0bfbc93d41156ffcb90c75fbc05544054faf15dcc1fdf9765f8b607f0 \
-    --hash=sha256:110a1ca7b49b014b097f6078272c3f4ed31af45b254de5228b79adba879f6af9 \
-    --hash=sha256:111e2390ffffc47d5840b01711dd7fac07d4c09283d0283e7f3264b14e284c64 \
-    --hash=sha256:17d86354f0ae6b61bf9be5148d0dd34e06c3cb7c602c671f79f29ac3b150e659 \
-    --hash=sha256:1ffdb3c0bb002c99cd8f298957e046c3dbf6006b5b7cdf11a4e19194624a0a0a \
-    --hash=sha256:2baee5ca02031757ffe8cc3d69f0cc0aec7065ce362622da74f32d3bcab1c541 \
-    --hash=sha256:2c18ef16bf6d4dd410e4dd52996888ea1497be26892fe5bbc73580aba4287b8e \
-    --hash=sha256:2f82b3597e9d83b63408affed0b48fd0f54935edac4302237b9a837be0dae33c \
-    --hash=sha256:3bfbd69cc349e43bf3a8ae1c85548ff0718efc887615c2db16c3833d7b0b072d \
-    --hash=sha256:3c8bb982ad117d29478ef8f5533e97df21f1e2befd17a299257b0c96d1371c0b \
-    --hash=sha256:3d955c89b75eeca4723d7cc14135f393cd47c32e2a6cb4a8e4c6e760a26b0986 \
-    --hash=sha256:4378720dd888136c27215a0214d32a4d37c3852765d45bc37aad0623423cfd78 \
-    --hash=sha256:45718441607f9325d948db98cbc691276059316d0358c188c246da4e1d4d23d2 \
-    --hash=sha256:5028648bf2253ec4745add746129d3904121fa7fe871a76bed23c5720573ce0a \
-    --hash=sha256:50ae25a67bea74ea41fb14b960bc532df73eb713417b2d61892dced82fe8d3bc \
-    --hash=sha256:51518ff74664078fc51bffcc6fc529b0df5ae58da192691cee765d45ce944a2b \
-    --hash=sha256:540dae7b956209af4d70a3be35927b4055f617763771e5e84a5255bea934d2f5 \
-    --hash=sha256:5a56aeb7d5d9cc4b3a735efb5095bd4b4f6f0e4f93e5ca876d0e2315137b7829 \
-    --hash=sha256:5e300185139abc337ade480c327183adf42a875ac7181bfe66d7d4efea31fbea \
-    --hash=sha256:67821bb03e4e98664490edb787ff6af501194c29bbee0f5c1dfdcf1dc3d9d436 \
-    --hash=sha256:6c09df69dc1712d131332054a858a3e5cca400967fa3a672e2324fbb0971448c \
-    --hash=sha256:6ebeb75c81211f5c702576cf81f315e77e23cfdb2c7c6fcb9dd143e6de35c360 \
-    --hash=sha256:73f78f9b9f0a5c06e5c946ba1e8e36f5114923b6be109ee618c54f079c3ea14f \
-    --hash=sha256:7546556f0d649f99f6a361098a55f761181bb2ea12ff150bb16d26092ad88244 \
-    --hash=sha256:7715a5a2c3378ba602c3a440558261e13a820bb53a82693aacd7b7f6d964e283 \
-    --hash=sha256:7b5f5fae05b8ac6d176a61b60c394a8cbdc2b5b91b81793066e68745cf165e54 \
-    --hash=sha256:7eacb17a9d41538a2bc4912eba5ef13823c83cb69e4d141d0813debe7163187f \
-    --hash=sha256:7ffdb990dcaa0234cf9845aead5df2e3c3a8b6507d409274dd87e0d5ab05ffc2 \
-    --hash=sha256:80eb4b04dadc4e67df3fae179a32c4706a3f495bc7f22fc8a81115d5f5512188 \
-    --hash=sha256:88e300d136eac057b2397aa1cfd7328b4c87c7eb66a09c7bc6a1292234db474e \
-    --hash=sha256:89101bfd5011e069be974903cb3a4e4523845e4ece2d62dcd8d358933c0ef249 \
-    --hash=sha256:8a17c42330e261299766b75ac1ea32caa437a9453c8f65d16a13140db378ecd3 \
-    --hash=sha256:8a271fcd66c74615cda6a964fda3f304267a12e50a084472218a39bb0376f563 \
-    --hash=sha256:8d8a23250ea3ec7b36de8fa4b541e9e2db3ee82915cc060ab0631609ad8b28de \
-    --hash=sha256:92fd6d44ac5e5a887c8a5dc4a8ba0ba908527c31c12f78c6bc7dcfe8aab279f6 \
-    --hash=sha256:975eac34b44a7077ca4d421348455b94f0f518246a7f14bc6d2fdcfe5b584368 \
-    --hash=sha256:9ab3c3a0b2ae6198e67c898dad5215a49f9ae0d0081b3c3ec59f333e39eeca26 \
-    --hash=sha256:9b1ec3274918a81d3ea778b9e75b56b72b33f300edb6cf7f3a7fe1dae56683de \
-    --hash=sha256:9d59e840387076a51016777a9328b3f2c427c6f9208a6e958bad251be50a648d \
-    --hash=sha256:a0cbed8bb44e23c5b199f888f4e4ce096b45ad9f25ff74a7ad0213875e936bb2 \
-    --hash=sha256:a19570c52a21420dcbc94e661994bc325c0b5b11304540fed514586da5dc8f2e \
-    --hash=sha256:a203a8bd0acb0701653d3bbb26e404854a68674139ed5cbb778830f42b09bb33 \
-    --hash=sha256:a4764e0bfc6a4d114c865b32520805c16a990ef5f286a514413b05d5ecd6a23d \
-    --hash=sha256:a57b0d05a0448eed231d59c0ceb287dde984551e54cbc51ac2d4865712838e9c \
-    --hash=sha256:a5c81f74d204d3edd136ebfd50dce53acbb776995d721a0fe801626cfc93b8cd \
-    --hash=sha256:a5ea42a752d47a145eae922b605cd1634665ac3d5ec1e72402d5048e8d60d207 \
-    --hash=sha256:a6fdf2433a5441ef9a95464f7c3e674775da1c8c1177fff311cee1acad4626ed \
-    --hash=sha256:add5217d68b31130f0beca584d7fef4878327d2e31642b66618a14eef312b63b \
-    --hash=sha256:b0703c2cef53e01baec47f7a3868009913ad71ec678bbecb42a6f40895e4ce62 \
-    --hash=sha256:b9152fca4a6466e114aaec745ae61cba739903a109754a9d4e1262f01e9259b1 \
-    --hash=sha256:c0141e37414c10164e702b8fb1473304221ad98f71600850c6ef7ff4880feba0 \
-    --hash=sha256:c3d35f87c7253b715d13d679e0783d845910144f282cb939fe1ba4ac8616269c \
-    --hash=sha256:c5551170cf4f5ff5623e9af81323751979fee2c731e2287b61f73cd27257b823 \
-    --hash=sha256:cbfc69be86e10dcfef5b1e6269d1d6926552aa89ee39e1de3353360c1b6989ab \
-    --hash=sha256:cc6ab7e555c8a112ad3a76e368e86e12a2754bcae1652a5602e133ec7b635523 \
-    --hash=sha256:cd443683db272ebaaca03af98c0b063ab30db70ea8a31a1559f35e3f7b744ccd \
-    --hash=sha256:d0932b81d72f552ded9d810d00021b64d89f2195a91ce115b893f943b7a4ab3c \
-    --hash=sha256:d40a890035c0058cadbdc4af7569800fd28a0e527a0fdbb7b5f9418f176846ce \
-    --hash=sha256:d5ee3ea898009fa898f85f9982255d35278c477bebe185beca249cab42d4526c \
-    --hash=sha256:d8ab31c9de8651a2facdd5c5bb0011f2380dd1a7af78ce2adf4b56095294fc07 \
-    --hash=sha256:dc71ff466927a201b08305acac451ebe1aedfcea002f62f1f2f2ac2ac1e6a135 \
-    --hash=sha256:de2daaaebd1a5aa88c49045b6baf9310b3263796bd88db713edf37cf53e7bb4e \
-    --hash=sha256:ded7b068c7c31c1a8657d4fd42d886b3e051ae29f88b80c5ff9d502257b0f071 \
-    --hash=sha256:e5cc9606aa5f4e0bde0d3bd502b44f743864c3ffa5cfa1011b1e30f5aa02366f \
-    --hash=sha256:e630136e905fe5ff43e86945ae41220b6d1470956a39220e708110ac48d01ea5 \
-    --hash=sha256:e6cd99ea59dd5d89f0c956606571d79bfe6f68c9eb7f4a4083a41a7f1587edee \
-    --hash=sha256:e7516cf6ae6b8a582c2770a0caed47b8a48373ed732c33d69a72913ae6ac923e \
-    --hash=sha256:ea37d5a157eb9493820d3792ac4ece28619a394391d2b9f2f78057d396ff0f0f \
-    --hash=sha256:ea8da1e900d758d078810d4255d8c6aa572181896a31ec79d779eb79c3adc9ad \
-    --hash=sha256:ed8cdb691169715a9a492844a83246f090182247d1a5031dc78a403f68ba1e97 \
-    --hash=sha256:ef08c1567c78074b22d1a200183d52d04a14df447bf70bcbb6a3507a48e776fc \
-    --hash=sha256:f16ba1efc0715b680a18b8123d90dad887c6112ae3555b4b5c32c149540c6b4e \
-    --hash=sha256:fa4f98af3a528f0c3fd592a26df7f376f93329c8f4d987f6bb979057af8bf5e2 \
-    --hash=sha256:ffea73584b216150eab159b6d12348fb253e68757974de1e2c40d8a318ac89ed
-    # via sqlalchemy
 http-request-recorder @ https://github.com/sipgate/http-request-recorder/archive/bf99fe64fd67db9df845be0e8da7e2601d9c8eef.zip \
     --hash=sha256:4b51faa761dd8ce21d5752030ff810c1b202c14ebfdcad4cb1e623779c213bab
     # via sipgate_e2e_test_utils (pyproject.toml)

sipgate_e2e_test_utils/kafka.py

@@ -0,0 +1,88 @@
+import asyncio
+from datetime import date, datetime
+
+from confluent_kafka import Producer, Consumer, TopicPartition, cimpl
+from confluent_kafka.schema_registry import SchemaRegistryClient
+from confluent_kafka.schema_registry.avro import AvroSerializer
+from confluent_kafka.serialization import SerializationContext, MessageField
+
+SHARED_KAFKA_CLIENT_PROPS = {
+    "bootstrap.servers": "kafka.local.sipgate.com:29092",
+    "security.protocol": "SASL_SSL",
+    "sasl.mechanism": 'SCRAM-SHA-256',
+    "sasl.username": "python-runner",
+    "sasl.password": "crq6HPocZQSo",
+    "ssl.ca.location": "/certs/ca-cert.pem",
+}
+
+
+def epoch_day(d: date) -> int:
+    epoch_sec = datetime(year=d.year, month=d.month, day=d.day).timestamp()
+    epoch_day = epoch_sec / (24 * 3600)
+    return int(epoch_day)
+
+
+def epoch_milli(d: datetime) -> int:
+    return int(d.timestamp()) * 1000
+
+
+def generic_source_part(db: str, table: str, created: datetime) -> dict:
+    return {
+        "version": "3.1.2.Final",
+        "connector": "mysql",
+        "name": "numbering.internal.cdc.dbnms",
+        "snapshot": "false",
+        "db": db,
+        "table": table,
+        "ts_ms": epoch_milli(created),
+        "ts_us": epoch_milli(created) * 1000,
+        "ts_ns": epoch_milli(created) * 1000 * 1000,
+        "server_id": 0,
+        "file": "mysql-bin.277229",
+        "pos": 16668676,  # seems as irrelevant as the other meta info... is it?
+        "row": 0,
+        "thread": 1,
+    }
+
+
+def publish_avro_record(schema_registry_client: SchemaRegistryClient, producer: Producer, topic: str, key_schema_filename: str, key: dict[str, any], value_schema_filename: str, value: dict[str, any]) -> None:
+    with open("/python-runner/e2e_tests/avro-schemata/" + key_schema_filename) as f:
+        key_schema_str = f.read()
+    with open("/python-runner/e2e_tests/avro-schemata/" + value_schema_filename) as f:
+        value_schema_str = f.read()
+
+    avro_key_serializer = AvroSerializer(schema_registry_client, key_schema_str, conf={"auto.register.schemas": True})
+    avro_value_serializer = AvroSerializer(schema_registry_client, value_schema_str, conf={"auto.register.schemas": True})
+
+    producer.produce(
+        topic=topic,
+        key=avro_key_serializer(key, SerializationContext(topic, MessageField.KEY)),
+        value=avro_value_serializer(value, SerializationContext(topic, MessageField.VALUE)),
+    )
+
+    producer.flush()
+
+
+async def likely_latest_msg_offset(consumer: Consumer, topic: str) -> int:
+    offset = -1
+    unchanged_for_iterations = 0
+    while unchanged_for_iterations < 20:
+        last_offset = offset
+        _, high_watermark_offset = consumer.get_watermark_offsets(TopicPartition(topic, 0), timeout=1.0)
+        # wierdly the high watermark offset is offset + 1
+        offset = high_watermark_offset - 1
+
+        if last_offset == offset:
+            unchanged_for_iterations += 1
+
+        await asyncio.sleep(0.500)
+
+    return offset
+
+
+def consume(consumer: Consumer, msg_count: int) -> list[cimpl.Message]:
+    msgs: list[cimpl.Message] = []
+    while len(msgs) < msg_count:
+        msgs += consumer.consume(1, timeout=1.0)
+
+    return msgs