fix: restart BG monitors when B/G switchover is completed

wrapper/src/main/java/software/amazon/jdbc/hostlistprovider/monitoring/MonitoringGlobalAuroraHostListProvider.java

@@ -28,10 +28,8 @@
 import software.amazon.jdbc.hostlistprovider.GlobalAuroraHostListProvider;
 import software.amazon.jdbc.hostlistprovider.GlobalAuroraTopologyUtils;
 import software.amazon.jdbc.util.FullServicesContainer;
-import software.amazon.jdbc.util.LogUtils;
-import software.amazon.jdbc.util.Messages;
 import software.amazon.jdbc.util.RdsUtils;
-import software.amazon.jdbc.util.StringUtils;
+import software.amazon.jdbc.util.events.MonitorStopEvent;
 
 public class MonitoringGlobalAuroraHostListProvider extends MonitoringRdsHostListProvider {
 
@@ -88,4 +86,10 @@ protected ClusterTopologyMonitor initMonitor() throws SQLException {
   protected List<HostSpec> queryForTopology(Connection connection) throws SQLException {
     return this.topologyUtils.queryForTopology(connection, this.initialHostSpec, this.instanceTemplatesByRegion);
   }
+
+  @Override
+  public void stopMonitor() {
+    this.servicesContainer.getEventPublisher().publish(
+        new MonitorStopEvent(GlobalAuroraTopologyMonitor.class, this.clusterId));
+  }
 }

wrapper/src/main/java/software/amazon/jdbc/hostlistprovider/monitoring/MonitoringRdsHostListProvider.java

@@ -31,6 +31,7 @@
 import software.amazon.jdbc.hostlistprovider.RdsHostListProvider;
 import software.amazon.jdbc.hostlistprovider.TopologyUtils;
 import software.amazon.jdbc.util.FullServicesContainer;
+import software.amazon.jdbc.util.events.MonitorStopEvent;
 
 public class MonitoringRdsHostListProvider
     extends RdsHostListProvider implements BlockingHostListProvider, CanReleaseResources {
@@ -106,4 +107,9 @@ public List<HostSpec> forceRefresh(final boolean shouldVerifyWriter, final long
   public void releaseResources() {
     // Do nothing.
   }
+
+  public void stopMonitor() {
+    this.servicesContainer.getEventPublisher().publish(
+        new MonitorStopEvent(ClusterTopologyMonitorImpl.class, this.clusterId));
+  }
 }

wrapper/src/main/java/software/amazon/jdbc/plugin/bluegreen/BlueGreenStatusMonitor.java

@@ -46,10 +46,12 @@
 import software.amazon.jdbc.HostSpec;
 import software.amazon.jdbc.HostSpecBuilder;
 import software.amazon.jdbc.PluginService;
+import software.amazon.jdbc.cleanup.CanReleaseResources;
 import software.amazon.jdbc.dialect.BlueGreenDialect;
 import software.amazon.jdbc.hostavailability.SimpleHostAvailabilityStrategy;
 import software.amazon.jdbc.hostlistprovider.HostListProvider;
 import software.amazon.jdbc.hostlistprovider.RdsHostListProvider;
+import software.amazon.jdbc.hostlistprovider.monitoring.MonitoringRdsHostListProvider;
 import software.amazon.jdbc.plugin.iam.IamAuthConnectionPlugin;
 import software.amazon.jdbc.util.ConnectionUrlParser;
 import software.amazon.jdbc.util.ExecutorFactory;
@@ -201,6 +203,14 @@ protected void runMonitoringLoop() {
       }
     } finally {
       this.closeConnection();
+      if (this.hostListProvider instanceof MonitoringRdsHostListProvider) {
+        ((MonitoringRdsHostListProvider) this.hostListProvider).stopMonitor();
+      }
+      if (this.hostListProvider instanceof CanReleaseResources) {
+        ((CanReleaseResources) this.hostListProvider).releaseResources();
+      }
+      this.hostListProvider = null;
+      this.openConnectionFuture = null;
       LOGGER.finest(() -> Messages.get("bgd.threadCompleted", new Object[] {this.role}));
     }
   }
@@ -244,6 +254,13 @@ public void setUseIpAddress(boolean useIpAddress) {
   public void setStop(boolean stop) {
     this.stop.set(stop);
     this.notifyChanges();
+    try {
+      if (!this.executorService.awaitTermination(5, TimeUnit.SECONDS)) {
+        this.executorService.shutdownNow();
+      }
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
   }
 
   public void resetCollectedData() {

wrapper/src/main/java/software/amazon/jdbc/plugin/bluegreen/BlueGreenStatusProvider.java

@@ -490,11 +490,6 @@ protected void updateMonitors() {
           x.setUseIpAddress(false);
           x.resetCollectedData();
         });
-
-        // Stop monitoring old1 cluster/instance.
-        if (!this.rollback && monitors[BlueGreenRole.SOURCE.getValue()] != null) {
-          monitors[BlueGreenRole.SOURCE.getValue()].setStop(true);
-        }
         break;
       default:
         throw new UnsupportedOperationException(Messages.get("bgd.unknownPhase",
@@ -1118,6 +1113,16 @@ protected void resetContextWhenCompleted() {
       this.greenNodeChangeNameTimes.clear();
       this.monitorResetOnInProgressCompleted.set(false);
       this.monitorResetOnTopologyCompleted.set(false);
+
+      if (this.monitors[BlueGreenRole.SOURCE.getValue()] != null) {
+        this.monitors[BlueGreenRole.SOURCE.getValue()].setStop(true);
+        this.monitors[BlueGreenRole.SOURCE.getValue()] = null;
+      }
+      if (this.monitors[BlueGreenRole.TARGET.getValue()] != null) {
+        this.monitors[BlueGreenRole.TARGET.getValue()].setStop(true);
+        this.monitors[BlueGreenRole.TARGET.getValue()] = null;
+      }
+      this.initMonitoring();
     }
   }
 

wrapper/src/main/java/software/amazon/jdbc/util/events/MonitorStopEvent.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package software.amazon.jdbc.util.events;
+
+import java.util.Objects;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import software.amazon.jdbc.util.monitoring.Monitor;
+
+public class MonitorStopEvent implements Event {
+
+  protected final @NonNull Class<? extends Monitor> monitorClass;
+  protected final @NonNull Object key;
+
+  public MonitorStopEvent(@NonNull Class<? extends Monitor> monitorClass, @NonNull Object key) {
+    this.monitorClass = monitorClass;
+    this.key = key;
+  }
+
+  public @NonNull Class<? extends Monitor> getMonitorClass() {
+    return monitorClass;
+  }
+
+  public @NonNull Object getKey() {
+    return key;
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) {
+      return true;
+    }
+
+    if (obj == null) {
+      return false;
+    }
+
+    if (getClass() != obj.getClass()) {
+      return false;
+    }
+
+    MonitorStopEvent event = (MonitorStopEvent) obj;
+    return Objects.equals(this.monitorClass, event.monitorClass)
+        && Objects.equals(this.key, event.key);
+  }
+
+  @Override
+  public int hashCode() {
+    final int prime = 31;
+    int result = 1;
+    result = prime * result + this.monitorClass.hashCode();
+    result = prime * result + this.key.hashCode();
+    return result;
+  }
+
+  @Override
+  public boolean isImmediateDelivery() {
+    return true;
+  }
+}

wrapper/src/main/java/software/amazon/jdbc/util/monitoring/MonitorServiceImpl.java

@@ -47,6 +47,7 @@
 import software.amazon.jdbc.util.events.Event;
 import software.amazon.jdbc.util.events.EventPublisher;
 import software.amazon.jdbc.util.events.EventSubscriber;
+import software.amazon.jdbc.util.events.MonitorStopEvent;
 import software.amazon.jdbc.util.storage.ExternallyManagedCache;
 import software.amazon.jdbc.util.storage.StorageService;
 import software.amazon.jdbc.util.telemetry.TelemetryFactory;
@@ -383,6 +384,12 @@ public void processEvent(Event event) {
       return;
     }
 
+    if (event instanceof MonitorStopEvent) {
+      MonitorStopEvent stopEvent = (MonitorStopEvent) event;
+      this.stopAndRemove(stopEvent.getMonitorClass(), stopEvent.getKey());
+      return;
+    }
+
     // Other event types should be propagated to monitors
     for (CacheContainer container : this.monitorCaches.values()) {
       for (MonitorItem item : container.getCache().getEntries().values()) {