Add Kafka Connect

Author: hariso

benchmarks/bench-mongo-kafka-cdc/mongo-kafka.yml

@@ -11,7 +11,7 @@ tools:
     docker-compose: ../tools/docker-compose-conduit.yml
 
 metrics:
-  conduit:
+  conduit_metrics:
     collector: conduit # type of the collector, one of: conduit, prometheus, kafka, docker
     tools:
       - conduit
@@ -35,22 +35,44 @@ tests:
             /scripts/init_db.sh
       pre-tool:
       post-tool:
-        - name: "Install curl"
+        - name: "Conduit: Install tools"
+          tools: 
+            - conduit
           container: "benchi-conduit"
           run: |-
-            /app/scripts/install_curl.sh
-        - name: "Set up CDC"
+            /scripts/install_tools.sh
+
+        - name: "Conduit: Set up CDC"
+          tools: 
+            - conduit
           container: "benchi-conduit"
           run: |-
-            /app/scripts/start_pipeline.sh && /app/scripts/stop_pipeline.sh
+            /scripts/start_pipeline.sh && /scripts/stop_pipeline.sh
+        - name: "Kafka Connect: Set up CDC"
+          tools:
+            - kafka-connect
+          container: "benchi-kafka-connect"
+          run: |-
+            /scripts/create_stopped_pipeline.sh
+
         - name: "Insert test data"
           container: "mongo1"
           run: |
             /scripts/insert_test_data.sh
-        - name: "Start pipeline"
+
+        - name: "Conduit: start pipeline"
+          tools: 
+            - conduit
           container: "benchi-conduit"
           run: |-
-            /app/scripts/start_pipeline.sh
+            /scripts/start_pipeline.sh
+
+        - name: "KC: start pipeline"
+          tools: 
+            - kafka-connect
+          container: "benchi-kafka-connect"
+          run: |-
+            /scripts/start_pipeline.sh
       pre-test:
       during:
       post-test:

benchmarks/infra/docker-compose-kafka.yml

@@ -19,6 +19,7 @@ services:
       retries: 5
       start_period: 30s
     ports:
+      - "9092:9092"
       - "7071:7071"
     volumes:
       - ./kafka/jmx:/opt/kafka/jmx  # Mount JMX files

benchmarks/infra/mongo/scripts/insert-test-users.js

@@ -9,7 +9,7 @@ const browsers = ['Chrome', 'Firefox', 'Safari', 'Edge'];
 
 
 // Total number of documents and batch size
-const totalDocs = 2_000_000;
+const totalDocs = 20_000;
 const batchSize = 20_000;
 let batch = [];
 

…ks/tools/conduit-scripts/install_curl.sh …s/tools/conduit-scripts/install_tools.shbenchmarks/tools/conduit-scripts/install_curl.sh renamed to benchmarks/tools/conduit-scripts/install_tools.sh benchmarks/tools/conduit-scripts/install_curl.sh renamed to benchmarks/tools/conduit-scripts/install_tools.sh

@@ -9,7 +9,7 @@ services:
     environment:
       CONDUIT_LOG_FORMAT: json
     volumes:
-      - ./conduit-scripts:/app/scripts
+      - ./conduit-scripts:/scripts
 
 networks:
   benchi:

benchmarks/tools/docker-compose-conduit.yml

@@ -7,7 +7,7 @@ services:
     ports:
       - "8083:8083"
     environment:
-      CONNECT_BOOTSTRAP_SERVERS: "benchi-kafka:29092"
+      CONNECT_BOOTSTRAP_SERVERS: "benchi-kafka:9092"
       CONNECT_REST_ADVERTISED_HOST_NAME: benchi-kafka-connect
       CONNECT_REST_PORT: 8083
       CONNECT_GROUP_ID: connect-cluster-group
@@ -25,7 +25,12 @@ services:
       CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
       CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
       CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR
-
+    volumes:
+      - ./kafka-connect-scripts/:/scripts/
+    user: root  # Switch to root to install packages
+    command:
+      - bash
+      - /scripts/setup.sh
 networks:
   benchi:
     external: true

benchmarks/tools/docker-compose-kafka-connect.yml

@@ -0,0 +1,24 @@
+{
+    "name": "kafka-connect-mongo",
+    "config":
+    {
+        "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
+        "connection.uri": "mongodb://mongo1:30001,mongo2:30002,mongo3:30003/test?replicaSet=my-replica-set",
+        "database": "test",
+        "collection": "users",
+
+        "change.stream.full.document": "updateLookup",
+
+        "output.schema.infer.value": "false",
+        "output.format.value": "json",
+        "output.format.key": "json",
+
+        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
+        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
+
+        "topic.prefix": "kafkaconnect",
+        "poll.await.time.ms": 5000,
+        "poll.max.batch.size": 1000,
+        "batch.size": 130400000
+    }
+}

benchmarks/tools/kafka-connect-scripts/connector.json

@@ -0,0 +1,7 @@
+#!/bin/sh
+
+SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
+
+curl -X POST -H "Content-Type: application/json" -d @"$SCRIPT_DIR/connector.json" localhost:8083/connectors
+
+"$SCRIPT_DIR/stop_pipeline.sh"

benchmarks/tools/kafka-connect-scripts/create_stopped_pipeline.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+# Install required packages
+yum install -y jq curl wget
+
+# Install MongoDB connector
+confluent-hub install --no-prompt mongodb/kafka-connect-mongodb:latest
+
+# Run the original entrypoint script as the confluent user
+su -c "/etc/confluent/docker/run" appuser

benchmarks/tools/kafka-connect-scripts/setup.sh

@@ -0,0 +1,5 @@
+#!/bin/sh
+
+SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
+
+curl -s -X PUT http://localhost:8083/connectors/$(jq -r '.name' "$SCRIPT_DIR/connector.json")/resume