fix: ensure B/G monitor instances are set up (#1612)

Author: sergiyvamz

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

@@ -141,7 +141,9 @@ public BlueGreenStatusMonitor(
     this.onBlueGreenStatusChangeFunc = onBlueGreenStatusChangeFunc;
 
     this.blueGreenDialect = (BlueGreenDialect) this.pluginService.getDialect();
+  }
 
+  public void start() {
     executorService.submit(this::runMonitoringLoop);
     executorService.shutdown(); // executor accepts no more tasks
   }

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

@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
@@ -158,7 +159,7 @@ public BlueGreenStatusProvider(
   }
 
   protected void initMonitoring() {
-    monitors[BlueGreenRole.SOURCE.getValue()] =
+    final BlueGreenStatusMonitor sourceMonitor =
         new BlueGreenStatusMonitor(
             BlueGreenRole.SOURCE,
             this.bgdId,
@@ -167,7 +168,10 @@ protected void initMonitoring() {
             this.getMonitoringProperties(),
             statusCheckIntervalMap,
             this::prepareStatus);
-    monitors[BlueGreenRole.TARGET.getValue()] =
+    monitors[BlueGreenRole.SOURCE.getValue()] = sourceMonitor;
+    sourceMonitor.start();
+
+    final BlueGreenStatusMonitor targetMonitor =
         new BlueGreenStatusMonitor(
             BlueGreenRole.TARGET,
             this.bgdId,
@@ -176,6 +180,8 @@ protected void initMonitoring() {
             this.getMonitoringProperties(),
             statusCheckIntervalMap,
             this::prepareStatus);
+    monitors[BlueGreenRole.TARGET.getValue()] = targetMonitor;
+    targetMonitor.start();
   }
 
   protected Properties getMonitoringProperties() {
@@ -448,15 +454,15 @@ protected void updateSummaryStatus(BlueGreenRole role, BlueGreenInterimStatus in
   protected void updateMonitors() {
     switch (this.summaryStatus.getCurrentPhase()) {
       case NOT_CREATED:
-        Arrays.stream(this.monitors).forEach(x -> {
+        Arrays.stream(this.monitors).filter(Objects::nonNull).forEach(x -> {
           x.setIntervalRate(BlueGreenIntervalRate.BASELINE);
           x.setCollectIpAddresses(false);
           x.setCollectTopology(false);
           x.setUseIpAddress(false);
         });
         break;
       case CREATED:
-        Arrays.stream(this.monitors).forEach(x -> {
+        Arrays.stream(this.monitors).filter(Objects::nonNull).forEach(x -> {
           x.setIntervalRate(BlueGreenIntervalRate.INCREASED);
           x.setCollectIpAddresses(true);
           x.setCollectTopology(true);
@@ -469,15 +475,15 @@ protected void updateMonitors() {
       case PREPARATION:
       case IN_PROGRESS:
       case POST:
-        Arrays.stream(this.monitors).forEach(x -> {
+        Arrays.stream(this.monitors).filter(Objects::nonNull).forEach(x -> {
           x.setIntervalRate(BlueGreenIntervalRate.HIGH);
           x.setCollectIpAddresses(false);
           x.setCollectTopology(false);
           x.setUseIpAddress(true);
         });
         break;
       case COMPLETED:
-        Arrays.stream(this.monitors).forEach(x -> {
+        Arrays.stream(this.monitors).filter(Objects::nonNull).forEach(x -> {
           x.setIntervalRate(BlueGreenIntervalRate.BASELINE);
           x.setCollectIpAddresses(false);
           x.setCollectTopology(false);