Address review comments

Author: kibertoad

packages/kafka/load-tests/package.json

@@ -24,6 +24,9 @@
         "load:direct:batch:medium": "node src/run-direct-batch.ts --rate 1000 --duration 60",
         "load:direct:batch:heavy": "node src/run-direct-batch.ts --rate 5000 --duration 120"
     },
+    "engines": {
+        "node": ">=22.18.0"
+    },
     "dependencies": {
         "@message-queue-toolkit/kafka": "file:../../kafka",
         "@message-queue-toolkit/core": "file:../../core",
@@ -36,8 +39,5 @@
     "devDependencies": {
         "@types/pg": "^8.16.0",
         "typescript": "^5.9.3"
-    },
-    "overrides": {
-        "zod": "$zod"
     }
 }

packages/kafka/load-tests/scripts/init-crdb.sh

@@ -33,14 +33,27 @@ CREATE TABLE IF NOT EXISTS orders (
 
 SET CLUSTER SETTING kv.rangefeed.enabled = true;
 
+SQL
+
+# Check for existing running changefeeds before creating a new one
+EXISTING=$(/cockroach/cockroach sql --insecure --host="$CRDB_HOST" --port="$CRDB_PORT" \
+  --format=csv -e "SELECT count(*) FROM [SHOW CHANGEFEED JOBS] WHERE status = 'running'" 2>/dev/null | tail -1)
+
+if [ "${EXISTING:-0}" -gt 0 ]; then
+  echo "Changefeed already exists, skipping creation."
+else
+  /cockroach/cockroach sql --insecure --host="$CRDB_HOST" --port="$CRDB_PORT" <<'SQL'
+USE loadtest;
+
 CREATE CHANGEFEED FOR events, orders
   INTO 'kafka://kafka:9093'
   WITH format = json,
        updated,
        resolved = '5s',
        diff,
        kafka_sink_config = '{"Flush":{"MaxMessages":100,"Frequency":"500ms"}}';
-
 SQL
+  echo "Changefeed created successfully."
+fi
 
-echo "Database, tables, and changefeed created successfully."
+echo "Database and tables initialized successfully."