feat: add tests for adapter_ha functionality in WsSql and WsConsumer, ensuring pooled connector reuse

Author: qevolg

nodejs/test/sql/sql.test.ts

@@ -1,3 +1,4 @@
+import { ClusterRegistry } from "@src/client/clusterRegistry";
 import { WebSocketConnectionPool } from "@src/client/wsConnectorPool";
 import { WSConfig } from "@src/common/config";
 import { WsSql } from "@src/sql/wsSql";
@@ -7,6 +8,11 @@ import { setLevel } from "@src/common/log";
 let dsn = "ws://localhost:6041";
 let password1 = "Ab1!@#$%,.:?<>;~";
 let password2 = "Bc%^&*()-_+=[]{}";
+
+function resetClusterRegistrySingleton(): void {
+    (ClusterRegistry as any)._instance = undefined;
+}
+
 setLevel("debug");
 beforeAll(async () => {
     let conf: WSConfig = new WSConfig(dsn);
@@ -125,6 +131,47 @@ describe("TDWebSocket.WsSql()", () => {
         expect(version).toBeTruthy();
     });
 
+    test("adapter_ha connect reuses pooled connector across seed variants", async () => {
+        const dsnA =
+            `ws://${testUsername()}:${testPassword()}` +
+            "@localhost:6041?adapter_ha=true";
+        const dsnB =
+            `ws://${testUsername()}:${testPassword()}` +
+            "@localhost:6041,127.0.0.1:6041?adapter_ha=true";
+
+        let connA: WsSql | null = null;
+        let connB: WsSql | null = null;
+
+        WebSocketConnectionPool.instance().destroyed();
+        resetClusterRegistrySingleton();
+
+        try {
+            connA = await WsSql.open(new WSConfig(dsnA));
+            const firstResult = await connA.exec("select server_version()");
+            expect(firstResult).toBeTruthy();
+
+            const connectorA = ((connA as any)._wsClient as any)._wsConnector;
+            expect(connectorA).toBeDefined();
+            expect(connectorA.getPoolKey()).toMatch(/^ws:\/\/[0-9a-f-]{36}\/ws#auth=/);
+
+            connB = await WsSql.open(new WSConfig(dsnB));
+            const connectorB = ((connB as any)._wsClient as any)._wsConnector;
+            expect(connectorB).toBe(connectorA);
+
+            const secondResult = await connB.exec("select server_version()");
+            expect(secondResult).toBeTruthy();
+        } finally {
+            if (connB) {
+                await connB.close();
+            }
+            if (connA) {
+                await connA.close();
+            }
+            WebSocketConnectionPool.instance().destroyed();
+            resetClusterRegistrySingleton();
+        }
+    });
+
     test("show databases", async () => {
         let conf: WSConfig = new WSConfig(dsn);
         conf.setUser(testUsername());

nodejs/test/tmq/tmq.test.ts

@@ -1,3 +1,4 @@
+import { ClusterRegistry } from "@src/client/clusterRegistry";
 import { TMQConstants } from "@src/tmq/constant";
 import { WsConsumer } from "@src/tmq/wsTmq";
 import { WSConfig } from "@src/common/config";
@@ -19,6 +20,10 @@ let dropTopic = `DROP TOPIC IF EXISTS ${topics[0]};`;
 let dsn = `ws://${testUsername()}:${testPassword()}@localhost:6041`;
 let tmqDsn = "ws://localhost:6041";
 
+function resetClusterRegistrySingleton(): void {
+    (ClusterRegistry as any)._instance = undefined;
+}
+
 beforeAll(async () => {
     let conf: WSConfig = new WSConfig(dsn);
     const createDB = `create database if not exists ${db} keep 3650`;
@@ -395,6 +400,95 @@ describe("TDWebSocket.Tmq()", () => {
         await consumer.close();
     });
 
+    test("adapter_ha consumes data and reuses pooled connector across seed variants", async () => {
+        const now = Date.now();
+        const dsnA = "ws://localhost:6041?adapter_ha=true";
+        const dsnB = "ws://localhost:6041,127.0.0.1:6041?adapter_ha=true";
+        const buildConfig = (
+            groupId: string,
+            clientId: string,
+            url: string
+        ): Map<string, any> => {
+            const cfg = new Map<string, any>();
+            cfg.set(TMQConstants.GROUP_ID, groupId);
+            cfg.set(TMQConstants.CONNECT_USER, testUsername());
+            cfg.set(TMQConstants.CONNECT_PASS, testPassword());
+            cfg.set(TMQConstants.AUTO_OFFSET_RESET, "earliest");
+            cfg.set(TMQConstants.CLIENT_ID, clientId);
+            cfg.set(TMQConstants.WS_URL, url);
+            cfg.set(TMQConstants.ENABLE_AUTO_COMMIT, false);
+            cfg.set(TMQConstants.AUTO_COMMIT_INTERVAL_MS, 1000);
+            cfg.set("session.timeout.ms", "10000");
+            cfg.set("max.poll.interval.ms", "30000");
+            cfg.set("msg.with.table.name", "true");
+            return cfg;
+        };
+
+        let consumerA: WsConsumer | null = null;
+        let consumerB: WsConsumer | null = null;
+
+        WebSocketConnectionPool.instance().destroyed();
+        resetClusterRegistrySingleton();
+
+        try {
+            const cfgA = buildConfig(
+                `g_adapter_ha_${now}_a`,
+                `c_adapter_ha_${now}_a`,
+                dsnA
+            );
+            consumerA = await WsConsumer.newConsumer(cfgA);
+            await consumerA.subscribe(topics);
+
+            const connectorA = ((consumerA as any)._wsClient as any)._wsConnector;
+            expect(connectorA).toBeDefined();
+            expect(connectorA.getPoolKey()).toMatch(/^ws:\/\/[0-9a-f-]{36}\/rest\/tmq#auth=/);
+
+            let count = 0;
+            for (let i = 0; i < 20 && count < 10; i++) {
+                const res = await consumerA.poll(500);
+                for (const [, value] of res) {
+                    const data = value.getData();
+                    if (!data || data.length === 0) {
+                        continue;
+                    }
+                    count += data.length;
+                }
+            }
+            expect(count).toBeGreaterThanOrEqual(10);
+
+            await consumerA.unsubscribe();
+            await consumerA.close();
+            consumerA = null;
+
+            const cfgB = buildConfig(
+                `g_adapter_ha_${now}_b`,
+                `c_adapter_ha_${now}_b`,
+                dsnB
+            );
+            consumerB = await WsConsumer.newConsumer(cfgB);
+
+            const connectorB = ((consumerB as any)._wsClient as any)._wsConnector;
+            expect(connectorB).toBe(connectorA);
+
+            await consumerB.subscribe(topics);
+            const res = await consumerB.poll(500);
+            expect(res).toBeTruthy();
+
+            await consumerB.unsubscribe();
+            await consumerB.close();
+            consumerB = null;
+        } finally {
+            if (consumerB) {
+                await consumerB.close();
+            }
+            if (consumerA) {
+                await consumerA.close();
+            }
+            WebSocketConnectionPool.instance().destroyed();
+            resetClusterRegistrySingleton();
+        }
+    });
+
     testEnterprise("connect with token", async () => {
         const conf = new WSConfig(dsn);
         const wsSql = await WsSql.open(conf);