rotor: migrate to confluent kafka library (#1222)

* rotor: migrate to confluent kafka library

Author: absorbb

pnpm-lock.yaml

@@ -3,7 +3,7 @@
   "version": "0.0.0",
   "dependencies": {
     "isolated-vm": "6.0.0",
-    "@confluentinc/kafka-javascript": "^1.4.0",
+    "@confluentinc/kafka-javascript": "^1.4.1",
     "@mongodb-js/zstd": "^2.0.1"
   }
 }

services/profiles/dist_package.json

@@ -29,7 +29,7 @@
     "pg-cursor": "^2.11.0",
     "prom-client": "^15.1.3",
     "tslib": "^2.6.3",
-    "@confluentinc/kafka-javascript": "^1.4.0",
+    "@confluentinc/kafka-javascript": "^1.4.1",
     "@clickhouse/client": "^1.10.1"
   },
   "devDependencies": {

services/profiles/package.json

@@ -1,21 +1,24 @@
 import { getLog } from "juava";
 import { connectToKafka } from "../src/lib/kafka-config";
-
-import "@sensejs/kafkajs-zstd-support";
 import { test } from "@jest/globals";
 
 const log = getLog("kafka-test");
 test.skip("Kafka Test", async () => {
-  const kafka = connectToKafka({ defaultAppId: "test", brokers: "localhost:9092" });
+  const kafka = connectToKafka({ defaultAppId: "test", brokers: ["localhost:9092"] });
   const consumer = kafka.consumer({
-    groupId: "test",
-    allowAutoTopicCreation: true,
-    sessionTimeout: 10000,
+    kafkaJS: {
+      groupId: "test",
+      allowAutoTopicCreation: true,
+      sessionTimeout: 10000,
+      autoCommitInterval: 10000,
+      autoCommit: true,
+      fromBeginning: true,
+    },
   });
   await consumer.connect();
-  await consumer.subscribe({ topics: ["autocommit-test"], fromBeginning: true });
+  await consumer.subscribe({ topics: ["autocommit-test"] });
 
-  const producer = kafka.producer({ allowAutoTopicCreation: false });
+  const producer = kafka.producer({ kafkaJS: { allowAutoTopicCreation: false } });
   await producer.connect();
 
   for (let i = 0; i < 200; i++) {
@@ -30,8 +33,6 @@ test.skip("Kafka Test", async () => {
   }
 
   await consumer.run({
-    autoCommitInterval: 10000,
-    autoCommit: true,
     partitionsConsumedConcurrently: 8,
     eachMessage: async ({ topic, partition, message }) => {
       log.atInfo().log(`${topic}:${partition}: ${message.offset} => ${message.value?.toString()}`);

services/rotor/__tests__/kafka.test.ts

@@ -4,12 +4,7 @@
   "dependencies": {
     "undici": "^7.11.0",
     "isolated-vm": "6.0.0",
-    "@sensejs/kafkajs-zstd-support": "^0.11.0",
+    "@confluentinc/kafka-javascript": "^1.4.1",
     "@mongodb-js/zstd": "^2.0.1"
-  },
-  "overrides": {
-    "@sensejs/kafkajs-zstd-support": {
-      "zstd-napi": "^0.0.10"
-    }
   }
 }

services/rotor/dist_package.json

@@ -25,16 +25,14 @@
     "@jitsu/core-functions": "workspace:*",
     "@jitsu/functions-lib": "workspace:*",
     "@maxmind/geoip2-node": "^5.0.0",
-    "@sensejs/kafkajs-zstd-support": "^0.11.0",
     "@types/pg-cursor": "^2.7.2",
     "dayjs": "^1.11.12",
     "express": "^4.21.2",
     "ioredis": "^5.4.1",
     "json5": "^2.2.3",
     "jsondiffpatch": "workspace:*",
     "juava": "workspace:*",
-    "kafkajs": "^2.2.4",
-    "kafkajs-snappy": "^1.1.0",
+    "@confluentinc/kafka-javascript": "^1.4.1",
     "node-cache": "^5.1.2",
     "node-fetch-commonjs": "^3.3.2",
     "object-hash": "^3.0.0",