feat(cdc): app-level Cassandra+Neo4j change events to Kafka (->S3) on generate

infra/airflow/scripts/cdc_emit.py

@@ -0,0 +1,66 @@
+"""Lightweight change-event emitter to Kafka for sources without a native
+Debezium connector (Cassandra 4.1 with cdc disabled, Neo4j 4.4 Community).
+
+This performs an application-level dual-write: as the generators insert data,
+each new record is also published to Kafka as a Debezium-style change event.
+The platform's all-topics S3 consumer then archives these to object storage,
+so Cassandra/Neo4j changes flow DB -> Kafka -> S3 just like the CDC sources.
+
+Best-effort: never raises, so data generation cannot be broken by Kafka issues.
+"""
+import json
+import os
+import time
+
+KAFKA_BOOTSTRAP = os.getenv("CDC_KAFKA_BOOTSTRAP", "10.0.21.36:9092")
+
+_producer = None
+
+
+def _default(o):
+    # make datetimes / uuids / Decimals JSON-serializable
+    try:
+        import datetime
+        if isinstance(o, (datetime.datetime, datetime.date)):
+            return o.isoformat()
+    except Exception:
+        pass
+    return str(o)
+
+
+def _get_producer():
+    global _producer
+    if _producer is None:
+        from kafka import KafkaProducer
+        _producer = KafkaProducer(
+            bootstrap_servers=KAFKA_BOOTSTRAP,
+            value_serializer=lambda v: json.dumps(v, default=_default).encode("utf-8"),
+            linger_ms=50,
+            acks=1,
+            retries=3,
+            request_timeout_ms=20000,
+        )
+    return _producer
+
+
+def emit_changes(topic, db, table, records):
+    """Publish a list of dict records as change events. Returns count sent."""
+    if not records:
+        return 0
+    try:
+        p = _get_producer()
+        now = int(time.time() * 1000)
+        for r in records:
+            evt = {
+                "op": "c",
+                "ts_ms": now,
+                "source": {"connector": "app-cdc", "db": db, "table": table},
+                "after": r,
+            }
+            p.send(topic, evt)
+        p.flush(timeout=30)
+        print(f"[cdc] emitted {len(records)} change events -> {topic}")
+        return len(records)
+    except Exception as e:
+        print(f"[cdc] emit warning ({topic}): {str(e)[:200]}")
+        return 0

infra/airflow/scripts/generate_cassandra_telemetry_data.py

@@ -9,6 +9,13 @@ import random
 from datetime import datetime, timedelta
 import uuid
 import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+try:
+    from cdc_emit import emit_changes
+except Exception:
+    def emit_changes(*a, **k):
+        return 0
 
 # Database connection details
 DB_HOST = "10.0.21.51"
@@ -57,12 +64,25 @@ def main():
     )
 
     total_generated = 0
+    cdc_buffer = []
+    CDC_TOPIC = "cassandra_telemetry.cdc.device_metrics"
     for i in range(TARGET_ROWS):
-        session.execute(insert_stmt, generate_fake_device_metric())
+        row = generate_fake_device_metric()
+        session.execute(insert_stmt, row)
         total_generated += 1
+        cdc_buffer.append({
+            "device_id": row[0], "metric_ts": row[1], "metric_type": row[2],
+            "metric_value": row[3], "payload": row[4],
+        })
+        if len(cdc_buffer) >= 1000:
+            emit_changes(CDC_TOPIC, "telemetry", "device_metrics", cdc_buffer)
+            cdc_buffer = []
         if total_generated % 1000 == 0:
             print(f"Generated {total_generated} rows...")
 
+    if cdc_buffer:
+        emit_changes(CDC_TOPIC, "telemetry", "device_metrics", cdc_buffer)
+
     session.shutdown()
     cluster.shutdown()
     

infra/airflow/scripts/generate_neo4j_graph_data.py

@@ -8,6 +8,16 @@ from neo4j import GraphDatabase
 import random
 import uuid
 import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+try:
+    from cdc_emit import emit_changes
+except Exception:
+    def emit_changes(*a, **k):
+        return 0
+
+NODES_TOPIC = "neo4j_graph.cdc.nodes"
+RELS_TOPIC = "neo4j_graph.cdc.relationships"
 
 # Database connection details
 DB_HOST = "10.0.21.51"
@@ -121,6 +131,7 @@ def main():
                     batch=batch
                 )
                 total_products += len(batch)
+                emit_changes(NODES_TOPIC, "graph", "Product", batch)
                 batch = []
                 
                 if total_products % 50000 == 0:
@@ -143,6 +154,7 @@ def main():
                 batch=batch
             )
             total_products += len(batch)
+            emit_changes(NODES_TOPIC, "graph", "Product", batch)
         
         print(f"Generated {total_products} product nodes.")
         
@@ -172,6 +184,7 @@ def main():
                     batch=batch
                 )
                 total_suppliers += len(batch)
+                emit_changes(NODES_TOPIC, "graph", "Supplier", batch)
                 batch = []
         
         if batch:
@@ -189,6 +202,7 @@ def main():
                 batch=batch
             )
             total_suppliers += len(batch)
+            emit_changes(NODES_TOPIC, "graph", "Supplier", batch)
         
         print(f"Generated {total_suppliers} supplier nodes.")
         
@@ -213,6 +227,7 @@ def main():
                 if len(batch) >= BATCH_SIZE:
                     _flush_rels(session, batch)
                     total_relationships += len(batch)
+                    emit_changes(RELS_TOPIC, "graph", "RELATIONSHIP", batch)
                     batch = []
                     
                     if total_relationships % 50000 == 0:
@@ -221,6 +236,7 @@ def main():
         if batch:
             _flush_rels(session, batch)
             total_relationships += len(batch)
+            emit_changes(RELS_TOPIC, "graph", "RELATIONSHIP", batch)
         
         print(f"Created {total_relationships} relationships.")