Adapt efm2, limitless, and fastest response plugins to refactoring work

aws-advanced-jdbc-wrapper-bundle/build.gradle.kts

@@ -25,8 +25,8 @@ repositories {
 
 dependencies {
     implementation("org.apache.httpcomponents:httpclient:4.5.14")
-    implementation("software.amazon.awssdk:rds:2.31.78")
-    implementation("software.amazon.awssdk:sts:2.31.78")
+    implementation("software.amazon.awssdk:rds:2.32.21")
+    implementation("software.amazon.awssdk:sts:2.32.21")
     implementation(project(":aws-advanced-jdbc-wrapper"))
 }
 

examples/AWSDriverExample/build.gradle.kts

@@ -18,9 +18,9 @@ dependencies {
     implementation("org.springframework.boot:spring-boot-starter-jdbc:2.7.13") // 2.7.13 is the last version compatible with Java 8
     implementation("org.postgresql:postgresql:42.7.7")
     implementation("com.mysql:mysql-connector-j:9.3.0")
-    implementation("software.amazon.awssdk:rds:2.31.78")
-    implementation("software.amazon.awssdk:secretsmanager:2.31.12")
-    implementation("software.amazon.awssdk:sts:2.31.78")
+    implementation("software.amazon.awssdk:rds:2.32.21")
+    implementation("software.amazon.awssdk:secretsmanager:2.32.21")
+    implementation("software.amazon.awssdk:sts:2.32.21")
     implementation("com.fasterxml.jackson.core:jackson-databind:2.19.0")
     implementation(project(":aws-advanced-jdbc-wrapper"))
     implementation("io.opentelemetry:opentelemetry-api:1.52.0")

examples/DBCPExample/build.gradle.kts

@@ -19,5 +19,5 @@ dependencies {
     implementation("com.mysql:mysql-connector-j:9.3.0")
     implementation(project(":aws-advanced-jdbc-wrapper"))
     implementation("org.apache.commons:commons-dbcp2:2.13.0")
-    implementation("software.amazon.awssdk:rds:2.31.50")
+    implementation("software.amazon.awssdk:rds:2.32.21")
 }

examples/SpringHibernateBalancedReaderOneDataSourceExample/build.gradle.kts

@@ -23,6 +23,6 @@ dependencies {
     implementation("org.springframework.boot:spring-boot-starter-data-jpa")
     implementation("org.springframework.retry:spring-retry")
     implementation("org.postgresql:postgresql:42.7.7")
-    implementation("software.amazon.awssdk:rds:2.31.50")
+    implementation("software.amazon.awssdk:rds:2.32.21")
     implementation(project(":aws-advanced-jdbc-wrapper"))
 }

examples/SpringHibernateBalancedReaderTwoDataSourceExample/build.gradle.kts

@@ -23,6 +23,6 @@ dependencies {
     implementation("org.springframework.boot:spring-boot-starter-data-jpa")
     implementation("org.springframework.retry:spring-retry")
     implementation("org.postgresql:postgresql:42.7.7")
-    implementation("software.amazon.awssdk:rds:2.31.50")
+    implementation("software.amazon.awssdk:rds:2.32.21")
     implementation(project(":aws-advanced-jdbc-wrapper"))
 }

examples/SpringHibernateExample/build.gradle.kts

@@ -23,6 +23,6 @@ dependencies {
     implementation("org.springframework.boot:spring-boot-starter-data-jpa")
     implementation("org.springframework.boot:spring-boot-starter-web")
     implementation("org.postgresql:postgresql:42.7.7")
-    implementation("software.amazon.awssdk:rds:2.31.50")
+    implementation("software.amazon.awssdk:rds:2.32.21")
     implementation(project(":aws-advanced-jdbc-wrapper"))
 }

examples/SpringTxFailoverExample/README.md

@@ -102,7 +102,7 @@ dependencies {
 	implementation("org.springframework.retry:spring-retry:1.3.4")
 	implementation("org.springframework:spring-aspects:5.3.29")
 	implementation("org.postgresql:postgresql:42.5.4")
-	implementation("software.amazon.awssdk:rds:2.29.23")
+	implementation("software.amazon.awssdk:rds:2.32.21")
 	implementation("software.amazon.jdbc:aws-advanced-jdbc-wrapper:latest")
 }
 ```

examples/SpringWildflyExample/spring/build.gradle.kts

@@ -24,6 +24,6 @@ dependencies {
     implementation("org.springframework.boot:spring-boot-starter-web")
     runtimeOnly("org.springframework.boot:spring-boot-devtools")
     implementation("org.postgresql:postgresql:42.7.7")
-    implementation("software.amazon.awssdk:rds:2.31.50")
+    implementation("software.amazon.awssdk:rds:2.32.21")
     implementation(project(":aws-advanced-jdbc-wrapper"))
 }

wrapper/build.gradle.kts

@@ -34,13 +34,13 @@ if (useJacoco) {
 dependencies {
     implementation("org.checkerframework:checker-qual:3.49.5")
     compileOnly("org.apache.httpcomponents:httpclient:4.5.14")
-    compileOnly("software.amazon.awssdk:rds:2.32.15")
-    compileOnly("software.amazon.awssdk:auth:2.31.45") // Required for IAM (light implementation)
-    compileOnly("software.amazon.awssdk:http-client-spi:2.32.11") // Required for IAM (light implementation)
-    compileOnly("software.amazon.awssdk:sts:2.32.15")
+    compileOnly("software.amazon.awssdk:rds:2.32.21")
+    compileOnly("software.amazon.awssdk:auth:2.32.21") // Required for IAM (light implementation)
+    compileOnly("software.amazon.awssdk:http-client-spi:2.32.21") // Required for IAM (light implementation)
+    compileOnly("software.amazon.awssdk:sts:2.32.21")
     compileOnly("com.zaxxer:HikariCP:4.0.3") // Version 4.+ is compatible with Java 8
     compileOnly("com.mchange:c3p0:0.11.0")
-    compileOnly("software.amazon.awssdk:secretsmanager:2.31.12")
+    compileOnly("software.amazon.awssdk:secretsmanager:2.32.21")
     compileOnly("com.fasterxml.jackson.core:jackson-databind:2.19.0")
     compileOnly("com.mysql:mysql-connector-j:9.3.0")
     compileOnly("org.postgresql:postgresql:42.7.7")
@@ -70,12 +70,12 @@ dependencies {
     testImplementation("com.mchange:c3p0:0.11.0")
     testImplementation("org.springframework.boot:spring-boot-starter-jdbc:2.7.13") // 2.7.13 is the last version compatible with Java 8
     testImplementation("org.mockito:mockito-inline:4.11.0") // 4.11.0 is the last version compatible with Java 8
-    testImplementation("software.amazon.awssdk:rds:2.32.15")
-    testImplementation("software.amazon.awssdk:auth:2.31.45") // Required for IAM (light implementation)
-    testImplementation("software.amazon.awssdk:http-client-spi:2.32.11") // Required for IAM (light implementation)
-    testImplementation("software.amazon.awssdk:ec2:2.32.15")
-    testImplementation("software.amazon.awssdk:secretsmanager:2.31.12")
-    testImplementation("software.amazon.awssdk:sts:2.32.15")
+    testImplementation("software.amazon.awssdk:rds:2.32.21")
+    testImplementation("software.amazon.awssdk:auth:2.32.21") // Required for IAM (light implementation)
+    testImplementation("software.amazon.awssdk:http-client-spi:2.32.21") // Required for IAM (light implementation)
+    testImplementation("software.amazon.awssdk:ec2:2.32.21")
+    testImplementation("software.amazon.awssdk:secretsmanager:2.32.21")
+    testImplementation("software.amazon.awssdk:sts:2.32.21")
     // Note: all org.testcontainers dependencies should have the same version
     testImplementation("org.testcontainers:testcontainers:1.21.2")
     testImplementation("org.testcontainers:mysql:1.21.2")

wrapper/src/main/java/software/amazon/jdbc/Driver.java

@@ -44,13 +44,11 @@
 import software.amazon.jdbc.plugin.OpenedConnectionTracker;
 import software.amazon.jdbc.plugin.customendpoint.CustomEndpointMonitorImpl;
 import software.amazon.jdbc.plugin.efm.HostMonitorThreadContainer;
-import software.amazon.jdbc.plugin.efm2.HostMonitorServiceImpl;
 import software.amazon.jdbc.plugin.federatedauth.FederatedAuthCacheHolder;
 import software.amazon.jdbc.plugin.federatedauth.OktaAuthCacheHolder;
 import software.amazon.jdbc.plugin.iam.IamAuthCacheHolder;
 import software.amazon.jdbc.plugin.limitless.LimitlessRouterServiceImpl;
 import software.amazon.jdbc.plugin.strategy.fastestresponse.FastestResponseStrategyPlugin;
-import software.amazon.jdbc.plugin.strategy.fastestresponse.HostResponseTimeServiceImpl;
 import software.amazon.jdbc.profile.ConfigurationProfile;
 import software.amazon.jdbc.profile.DriverConfigurationProfiles;
 import software.amazon.jdbc.states.ResetSessionStateOnCloseCallable;
@@ -440,11 +438,9 @@ public static void clearCaches() {
 
   public static void releaseResources() {
     CoreServicesContainer.getInstance().getMonitorService().stopAndRemoveAll();
-    HostMonitorServiceImpl.closeAllMonitors();
     HostMonitorThreadContainer.releaseInstance();
     ConnectionProviderManager.releaseResources();
     HikariPoolsHolder.closeAllPools();
-    HostResponseTimeServiceImpl.closeAllMonitors();
     clearCaches();
   }
 

wrapper/src/main/java/software/amazon/jdbc/hostlistprovider/RdsMultiAzDbClusterListProvider.java

@@ -28,7 +28,6 @@
 import java.util.List;
 import java.util.Properties;
 import java.util.logging.Logger;
-import software.amazon.jdbc.HostListProviderService;
 import software.amazon.jdbc.HostRole;
 import software.amazon.jdbc.HostSpec;
 import software.amazon.jdbc.hostavailability.HostAvailability;

wrapper/src/main/java/software/amazon/jdbc/plugin/efm/HostMonitorThreadContainer.java

@@ -36,7 +36,6 @@ public class HostMonitorThreadContainer {
 
   private static HostMonitorThreadContainer singleton = null;
   private final Map<HostMonitor, Future<?>> tasksMap = new ConcurrentHashMap<>();
-  // TODO: remove monitorMap and threadPool and submit monitors to MonitorService instead
   private final Map<String, HostMonitor> monitorMap = new ConcurrentHashMap<>();
   private final ExecutorService threadPool;
   private static final ReentrantLock LOCK_OBJECT = new ReentrantLock();

wrapper/src/main/java/software/amazon/jdbc/plugin/efm2/HostMonitorImpl.java

@@ -27,17 +27,16 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import software.amazon.jdbc.HostSpec;
-import software.amazon.jdbc.PluginService;
 import software.amazon.jdbc.util.ExecutorFactory;
 import software.amazon.jdbc.util.Messages;
 import software.amazon.jdbc.util.PropertyUtils;
+import software.amazon.jdbc.util.connection.ConnectionService;
+import software.amazon.jdbc.util.monitoring.AbstractMonitor;
 import software.amazon.jdbc.util.telemetry.TelemetryContext;
 import software.amazon.jdbc.util.telemetry.TelemetryCounter;
 import software.amazon.jdbc.util.telemetry.TelemetryFactory;
@@ -47,10 +46,11 @@
  * This class uses a background thread to monitor a particular server with one or more active {@link
  * Connection}.
  */
-public class HostMonitorImpl implements HostMonitor {
+public class HostMonitorImpl extends AbstractMonitor implements HostMonitor {
 
   private static final Logger LOGGER = Logger.getLogger(HostMonitorImpl.class.getName());
   private static final long THREAD_SLEEP_NANO = TimeUnit.MILLISECONDS.toNanos(100);
+  private static final long TERMINATION_TIMEOUT_SEC = 30;
   private static final String MONITORING_PROPERTY_PREFIX = "monitoring-";
 
   protected static final Executor ABORT_EXECUTOR =
@@ -59,15 +59,11 @@ public class HostMonitorImpl implements HostMonitor {
   private final Queue<WeakReference<HostMonitorConnectionContext>> activeContexts = new ConcurrentLinkedQueue<>();
   private final Map<Long, Queue<WeakReference<HostMonitorConnectionContext>>> newContexts =
       new ConcurrentHashMap<>();
-  private final PluginService pluginService;
+  private final ConnectionService connectionService;
   private final TelemetryFactory telemetryFactory;
   private final Properties properties;
   private final HostSpec hostSpec;
-  private final AtomicBoolean stopped = new AtomicBoolean(false);
   private Connection monitoringConn = null;
-  // TODO: remove and submit monitors to MonitorService instead
-  private final ExecutorService threadPool =
-      ExecutorFactory.newFixedThreadPool(2, "threadPool");
 
   private final long failureDetectionTimeNano;
   private final long failureDetectionIntervalNano;
@@ -82,7 +78,8 @@ public class HostMonitorImpl implements HostMonitor {
   /**
    * Store the monitoring configuration for a connection.
    *
-   * @param pluginService             A service for creating new connections.
+   * @param connectionService         The service to use to create the monitoring connection.
+   * @param telemetryFactory          The telemetry factory to use to create telemetry data.
    * @param hostSpec                  The {@link HostSpec} of the server this {@link HostMonitorImpl}
    *                                  instance is monitoring.
    * @param properties                The {@link Properties} containing additional monitoring
@@ -93,26 +90,24 @@ public class HostMonitorImpl implements HostMonitor {
    * @param abortedConnectionsCounter Aborted connection telemetry counter.
    */
   public HostMonitorImpl(
-      final @NonNull PluginService pluginService,
+      final @NonNull ConnectionService connectionService,
+      final @NonNull TelemetryFactory telemetryFactory,
       final @NonNull HostSpec hostSpec,
       final @NonNull Properties properties,
       final int failureDetectionTimeMillis,
       final int failureDetectionIntervalMillis,
       final int failureDetectionCount,
       final TelemetryCounter abortedConnectionsCounter) {
+    super(TERMINATION_TIMEOUT_SEC, ExecutorFactory.newFixedThreadPool(2, "efm2-monitor"));
 
-    this.pluginService = pluginService;
-    this.telemetryFactory = pluginService.getTelemetryFactory();
+    this.connectionService = connectionService;
+    this.telemetryFactory = telemetryFactory;
     this.hostSpec = hostSpec;
     this.properties = properties;
     this.failureDetectionTimeNano = TimeUnit.MILLISECONDS.toNanos(failureDetectionTimeMillis);
     this.failureDetectionIntervalNano = TimeUnit.MILLISECONDS.toNanos(failureDetectionIntervalMillis);
     this.failureDetectionCount = failureDetectionCount;
     this.abortedConnectionsCounter = abortedConnectionsCounter;
-
-    this.threadPool.submit(this::newContextRun); // task to handle new contexts
-    this.threadPool.submit(this); // task to handle active monitoring contexts
-    this.threadPool.shutdown(); // No more tasks are accepted by pool.
   }
 
   @Override
@@ -121,21 +116,15 @@ public boolean canDispose() {
   }
 
   @Override
-  public void close() throws Exception {
-    this.stopped.set(true);
-
-    // Waiting for 30s gives a thread enough time to exit monitoring loop and close database connection.
-    if (!this.threadPool.awaitTermination(30, TimeUnit.SECONDS)) {
-      this.threadPool.shutdownNow();
-    }
-    LOGGER.finest(() -> Messages.get(
-        "HostMonitorImpl.stopped",
-        new Object[] {this.hostSpec.getHost()}));
+  public void start() {
+    this.monitorExecutor.submit(this::newContextRun); // task to handle new contexts
+    this.monitorExecutor.submit(this); // task to handle active monitoring contexts
+    this.monitorExecutor.shutdown(); // No more tasks are accepted by pool.
   }
 
   @Override
   public void startMonitoring(final HostMonitorConnectionContext context) {
-    if (this.stopped.get()) {
+    if (this.stop.get()) {
       LOGGER.warning(() -> Messages.get("HostMonitorImpl.monitorIsStopped", new Object[] {this.hostSpec.getHost()}));
     }
 
@@ -166,9 +155,9 @@ public void newContextRun() {
         new Object[] {this.hostSpec.getHost()}));
 
     try {
-      while (!this.stopped.get()) {
-
+      while (!this.stop.get()) {
         final long currentTimeNano = this.getCurrentTimeNano();
+        this.lastActivityTimestampNanos.set(currentTimeNano);
 
         final ArrayList<Long> processedKeys = new ArrayList<>();
         this.newContexts.entrySet().stream()
@@ -212,14 +201,14 @@ public void newContextRun() {
   }
 
   @Override
-  public void run() {
+  public void monitor() {
 
     LOGGER.finest(() -> Messages.get(
         "HostMonitorImpl.startMonitoringThread",
         new Object[] {this.hostSpec.getHost()}));
 
     try {
-      while (!this.stopped.get()) {
+      while (!this.stop.get()) {
 
         if (this.activeContexts.isEmpty() && !this.nodeUnhealthy) {
           TimeUnit.NANOSECONDS.sleep(THREAD_SLEEP_NANO);
@@ -236,7 +225,7 @@ public void run() {
         WeakReference<HostMonitorConnectionContext> monitorContextWeakRef;
 
         while ((monitorContextWeakRef = this.activeContexts.poll()) != null) {
-          if (this.stopped.get()) {
+          if (this.stop.get()) {
             break;
           }
 
@@ -284,7 +273,7 @@ public void run() {
             ex); // We want to print full trace stack of the exception.
       }
     } finally {
-      this.stopped.set(true);
+      this.stop.set(true);
       if (this.monitoringConn != null) {
         try {
           this.monitoringConn.close();
@@ -328,8 +317,7 @@ boolean checkConnectionStatus() {
                 });
 
         LOGGER.finest(() -> "Opening a monitoring connection to " + this.hostSpec.getUrl());
-        // TODO: replace with ConnectionService#open
-        this.monitoringConn = this.pluginService.forceConnect(this.hostSpec, monitoringConnProperties);
+        this.monitoringConn = this.connectionService.open(this.hostSpec, monitoringConnProperties);
         LOGGER.finest(() -> "Opened monitoring connection: " + this.monitoringConn);
         return true;
       }
@@ -402,4 +390,14 @@ private void abortConnection(final @NonNull Connection connectionToAbort) {
     }
   }
 
+  @Override
+  public void close() {
+    if (this.monitoringConn != null) {
+      try {
+        this.monitoringConn.close();
+      } catch (SQLException e) {
+        // ignore
+      }
+    }
+  }
 }

wrapper/src/main/java/software/amazon/jdbc/plugin/efm2/HostMonitorService.java

@@ -17,6 +17,7 @@
 package software.amazon.jdbc.plugin.efm2;
 
 import java.sql.Connection;
+import java.sql.SQLException;
 import java.util.Properties;
 import software.amazon.jdbc.HostSpec;
 
@@ -32,7 +33,7 @@ HostMonitorConnectionContext startMonitoring(
       Properties properties,
       int failureDetectionTimeMillis,
       int failureDetectionIntervalMillis,
-      int failureDetectionCount);
+      int failureDetectionCount) throws SQLException;
 
   /**
    * Stop monitoring for a connection represented by the given {@link HostMonitorConnectionContext}.