test: include rw integration tests

aws · Jan 27, 2025 · b1ff9e6 · b1ff9e6
1 parent bb7d4c5
commit b1ff9e6
Show file tree

Hide file tree

Showing 2 changed files with 140 additions and 2 deletions.
diff --git a/common/lib/host_list_provider/monitoring/cluster_topology_monitor.ts b/common/lib/host_list_provider/monitoring/cluster_topology_monitor.ts
@@ -115,6 +115,7 @@ export class ClusterTopologyMonitorImpl implements ClusterTopologyMonitor {
     this.hostMonitorsStop = true;
     this.requestToUpdateTopology = true;
     await Promise.all(this.untrackedPromises);
+    await this.closeConnection(this.monitoringClient);
     await this.closeConnection(this.hostMonitorsWriterClient);
     await this.closeConnection(this.hostMonitorsReaderClient);
     this.untrackedPromises = [];
@@ -213,7 +214,7 @@ export class ClusterTopologyMonitorImpl implements ClusterTopologyMonitor {
             this.writerHostInfo = this.initialHostInfo;
             writerVerifiedByThisThread = true;
           }
-        } catch {
+        } catch (error) {
           // Do nothing.
         }
       } else {
@@ -250,8 +251,9 @@ export class ClusterTopologyMonitorImpl implements ClusterTopologyMonitor {
   }
 
   async closeConnection(client: ClientWrapper) {
-    if (client) {
+    if (client !== null) {
       await client.abort();
+      client = null;
     }
   }
 

diff --git a/tests/integration/container/tests/aurora_failover2.test.ts b/tests/integration/container/tests/aurora_failover2.test.ts
@@ -36,6 +36,7 @@ const itIf =
     ? it
     : it.skip;
 const itIfTwoInstance = instanceCount == 2 ? itIf : it.skip;
+const itIfMinThreeInstance = instanceCount >= 3 ? itIf : it.skip;
 
 let env: TestEnvironment;
 let driver;
@@ -65,6 +66,27 @@ async function initDefaultConfig(host: string, port: number, connectToProxy: boo
   return config;
 }
 
+async function initConfigWithRWSplitting(host: string, port: number, connectToProxy: boolean): Promise<any> {
+  let config: any = {
+    user: env.databaseInfo.username,
+    host: host,
+    database: env.databaseInfo.defaultDbName,
+    password: env.databaseInfo.password,
+    port: port,
+    plugins: "readWriteSplitting,failover2",
+    failoverTimeoutMs: 400000,
+    enableTelemetry: true,
+    telemetryTracesBackend: "OTLP",
+    telemetryMetricsBackend: "OTLP"
+  };
+
+  if (connectToProxy) {
+    config["clusterInstanceHostPattern"] = "?." + env.proxyDatabaseInfo.instanceEndpointSuffix;
+  }
+  config = DriverHelper.addDriverSpecificConfiguration(config, env.engine);
+  return config;
+}
+
 describe("aurora failover2", () => {
   beforeEach(async () => {
     logger.info(`Test started: ${expect.getState().currentTestName}`);
@@ -259,4 +281,118 @@ describe("aurora failover2", () => {
     },
     1320000
   );
+
+  itIfMinThreeInstance(
+    "test failover to new writer set read only true false",
+    async () => {
+      // Connect to writer instance
+      const writerConfig = await initConfigWithRWSplitting(
+        env.proxyDatabaseInfo.writerInstanceEndpoint,
+        env.proxyDatabaseInfo.instanceEndpointPort,
+        true
+      );
+      client = initClientFunc(writerConfig);
+      await client.connect();
+
+      const initialWriterId = await auroraTestUtility.queryInstanceId(client);
+      expect(await auroraTestUtility.isDbInstanceWriter(initialWriterId)).toStrictEqual(true);
+
+      // Kill all reader instances
+      for (const host of env.proxyDatabaseInfo.instances) {
+        if (host.instanceId && host.instanceId !== initialWriterId) {
+          await ProxyHelper.disableConnectivity(env.engine, host.instanceId);
+        }
+      }
+
+      // Force internal reader connection to the writer instance
+      await client.setReadOnly(true);
+      const currentId0 = await auroraTestUtility.queryInstanceId(client);
+
+      expect(currentId0).toStrictEqual(initialWriterId);
+      await client.setReadOnly(false);
+
+      await ProxyHelper.enableAllConnectivity();
+      // Crash instance 1 and nominate a new writer
+      await auroraTestUtility.failoverClusterAndWaitUntilWriterChanged();
+      await TestEnvironment.verifyClusterStatus();
+
+      await expect(async () => {
+        await auroraTestUtility.queryInstanceId(client);
+      }).rejects.toThrow(FailoverSuccessError);
+      const newWriterId = await auroraTestUtility.queryInstanceId(client);
+      expect(await auroraTestUtility.isDbInstanceWriter(newWriterId)).toStrictEqual(true);
+      expect(newWriterId).not.toBe(initialWriterId);
+
+      await client.setReadOnly(true);
+      const currentReaderId = await auroraTestUtility.queryInstanceId(client);
+      expect(currentReaderId).not.toBe(newWriterId);
+
+      await client.setReadOnly(false);
+      const currentId = await auroraTestUtility.queryInstanceId(client);
+      expect(currentId).toStrictEqual(newWriterId);
+    },
+    1320000
+  );
+
+  itIfMinThreeInstance(
+    "test failover to new reader set read only false true",
+    async () => {
+      // Connect to writer instance
+      const writerConfig = await initConfigWithRWSplitting(
+        env.proxyDatabaseInfo.writerInstanceEndpoint,
+        env.proxyDatabaseInfo.instanceEndpointPort,
+        true
+      );
+      writerConfig["failoverMode"] = "reader-or-writer";
+      client = initClientFunc(writerConfig);
+
+      await client.connect();
+      const initialWriterId = await auroraTestUtility.queryInstanceId(client);
+      expect(await auroraTestUtility.isDbInstanceWriter(initialWriterId)).toStrictEqual(true);
+      await client.setReadOnly(true);
+
+      const readerConnectionId = await auroraTestUtility.queryInstanceId(client);
+      expect(readerConnectionId).not.toBe(initialWriterId);
+
+      // Get a reader instance
+      let otherReaderId;
+      for (const host of env.proxyDatabaseInfo.instances) {
+        if (host.instanceId && host.instanceId !== readerConnectionId && host.instanceId !== initialWriterId) {
+          otherReaderId = host.instanceId;
+          break;
+        }
+      }
+
+      if (!otherReaderId) {
+        throw new Error("Could not find a reader instance");
+      }
+      // Kill all instances except one other reader
+      for (const host of env.proxyDatabaseInfo.instances) {
+        if (host.instanceId && host.instanceId !== otherReaderId) {
+          await ProxyHelper.disableConnectivity(env.engine, host.instanceId);
+        }
+      }
+
+      await expect(async () => {
+        await auroraTestUtility.queryInstanceId(client);
+      }).rejects.toThrow(FailoverSuccessError);
+
+      const currentReaderId0 = await auroraTestUtility.queryInstanceId(client);
+
+      expect(currentReaderId0).toStrictEqual(otherReaderId);
+      expect(currentReaderId0).not.toBe(readerConnectionId);
+
+      await ProxyHelper.enableAllConnectivity();
+      await client.setReadOnly(false);
+
+      const currentId = await auroraTestUtility.queryInstanceId(client);
+      expect(currentId).toStrictEqual(initialWriterId);
+
+      await client.setReadOnly(true);
+
+      const currentReaderId2 = await auroraTestUtility.queryInstanceId(client);
+      expect(currentReaderId2).toStrictEqual(otherReaderId);
+    },
+    1320000
+  );
 });