Remove the isMultiWriterCluster flag from the AuroraTopologyService (#405)

* refactor: remove isMultiWriterCluster flag

* account for multiple stale records

* remove unnecessary return

* fix comment

* simplify logic to take the latest updated writer in the case of stale records

* simplify further as writers and resultset are already in order

* order hosts by last update timestamp

* fix tests

* compare timestamps as strings

* switch last update timestamp property to iso string

* do not convert to iso string twice

Author: crystall-bitquill

src/main/core-api/java/com/mysql/cj/conf/HostInfo.java

@@ -31,6 +31,7 @@
 
 import static com.mysql.cj.util.StringUtils.isNullOrEmpty;
 
+import com.mysql.cj.jdbc.ha.plugins.failover.TopologyServicePropertyKeys;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -175,6 +176,15 @@ public String getProperty(String key) {
         return this.hostProperties.get(key);
     }
 
+    /**
+     * Returns the last updated time as a String.
+     *
+     * @return the last updated time.
+     */
+    public String getLastUpdatedTime() {
+        return this.getProperty(TopologyServicePropertyKeys.LAST_UPDATED);
+    }
+
     /**
      * Shortcut to the database connection argument.
      * 

src/main/resources/com/mysql/cj/LocalizedErrorMessages.properties

@@ -1055,7 +1055,7 @@ ConnectionProperties.useAwsIam=Set to true to use AWS IAM database authenticatio
 
 AuroraTopologyService.1=[AuroraTopologyService] clusterId=''{0}''
 AuroraTopologyService.2=[AuroraTopologyService] clusterInstance host=''{0}'', port={1,number,#}, database=''{2}''
-AuroraTopologyService.3=[AuroraTopologyService] The topology query returned an invalid topology - no writer instance detected
+AuroraTopologyService.3=[AuroraTopologyService] The topology query returned an invalid topology - no writer instance detected.
 
 ClusterAwareConnectionProxy.1=Transaction resolution unknown. Please re-configure session state if required and try restarting transaction.
 ClusterAwareConnectionProxy.2=Unable to establish SQL connection to writer node.

src/main/user-impl/java/com/mysql/cj/jdbc/ha/plugins/failover/AuroraTopologyService.java

@@ -49,7 +49,9 @@
 import java.sql.Statement;
 import java.sql.Timestamp;
 import java.time.Instant;
+import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -59,6 +61,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 /**
  * An implementation of topology service for Aurora RDS. It uses
@@ -89,7 +92,6 @@ public class AuroraTopologyService implements ITopologyService {
   public static final CacheMap<String, List<HostInfo>> topologyCache = new CacheMap<>();
   public static final CacheMap<String, Set<String>> downHostCache = new CacheMap<>();
   public static final CacheMap<String, HostInfo> lastUsedReaderCache = new CacheMap<>();
-  public static final CacheMap<String, Boolean> multiWriterClusterCache = new CacheMap<>();
 
   protected String clusterId;
   protected HostInfo clusterInstanceTemplate;
@@ -190,9 +192,6 @@ public List<HostInfo> getTopology(JdbcConnection conn, boolean forceUpdate)
       ClusterTopologyInfo latestTopologyInfo = queryForTopology(conn);
 
       if (latestTopologyInfo != null) {
-        multiWriterClusterCache.put(
-            this.clusterId, latestTopologyInfo.getMultiWriterCluster(), this.refreshRateNanos);
-
         downHostCache.get(this.clusterId, ConcurrentHashMap.newKeySet(), this.refreshRateNanos).clear();
 
         if (!Util.isNullOrEmpty(latestTopologyInfo.getHosts())) {
@@ -233,7 +232,7 @@ protected ClusterTopologyInfo queryForTopology(JdbcConnection conn) throws SQLEx
 
     return topologyInfo != null
         ? topologyInfo
-        : new ClusterTopologyInfo(new ArrayList<>(),false);
+        : new ClusterTopologyInfo(new ArrayList<>());
   }
 
   /**
@@ -257,33 +256,35 @@ private boolean gatherPerfMetrics(PropertySet props) {
    */
   private ClusterTopologyInfo processQueryResults(ResultSet resultSet)
       throws SQLException {
-    int writerCount = 0;
 
     List<HostInfo> hosts = new ArrayList<>();
+    List<HostInfo> writers = new ArrayList<>();
     while (resultSet.next()) {
+      HostInfo currentHost = createHost(resultSet);
+
       if (!WRITER_SESSION_ID.equalsIgnoreCase(resultSet.getString(FIELD_SESSION_ID))) {
-        hosts.add(createHost(resultSet));
+        hosts.add(currentHost);
         continue;
       }
 
-      if (writerCount == 0) {
-        // store the first writer to its expected position [0]
-        hosts.add(
-            FailoverConnectionPlugin.WRITER_CONNECTION_INDEX,
-            createHost(resultSet));
-      } else {
-        // append other writers, if any, to the end of the host list
-        hosts.add(createHost(resultSet));
-      }
-      writerCount++;
+      writers.add(currentHost);
     }
 
-    if (writerCount == 0) {
+    int writersCount = writers.size();
+
+    if (writersCount == 0) {
       this.log.logError(Messages.getString("AuroraTopologyService.3"));
       hosts.clear();
+    } else if (writersCount == 1) {
+      hosts.add(FailoverConnectionPlugin.WRITER_CONNECTION_INDEX, writers.get(0));
+    } else {
+      // Store the first writer to its expected position [0]. If there are other writers or stale records, ignore them.
+      List<HostInfo> sortedWriters = writers.stream()
+          .sorted(Comparator.comparing(HostInfo::getLastUpdatedTime).reversed()).collect(Collectors.toList());
+      hosts.add(FailoverConnectionPlugin.WRITER_CONNECTION_INDEX, sortedWriters.get(0));
     }
 
-    return new ClusterTopologyInfo(hosts, writerCount > 1);
+    return new ClusterTopologyInfo(hosts);
   }
 
   private HostInfo createHost(ResultSet resultSet) throws SQLException {
@@ -351,7 +352,8 @@ private Map<String, String> getPropertiesFromTopology(ResultSet resultSet)
   }
 
   private String convertTimestampToString(Timestamp timestamp) {
-    return timestamp == null ? null : timestamp.toString();
+    DateTimeFormatter formatter = DateTimeFormatter.ISO_LOCAL_DATE_TIME;
+    return timestamp == null ? null : formatter.format(timestamp.toLocalDateTime());
   }
 
   /**
@@ -464,16 +466,6 @@ public void removeFromDownHostList(HostInfo host) {
       .remove(host.getHostPortPair());
   }
 
-  /**
-   * Check if cached topology belongs to multi-writer cluster.
-   *
-   * @return True, if it's multi-writer cluster.
-   */
-  @Override
-  public boolean isMultiWriterCluster() {
-    return multiWriterClusterCache.get(this.clusterId, false, this.refreshRateNanos);
-  }
-
   /**
    * Set new topology refresh rate. Different service instances may have different topology refresh
    * rate while sharing the same topology cache.
@@ -490,7 +482,6 @@ public void setRefreshRate(int refreshRateMillis) {
   public void clearAll() {
     topologyCache.clear();
     downHostCache.clear();
-    multiWriterClusterCache.clear();
     lastUsedReaderCache.clear();
   }
 
@@ -499,20 +490,16 @@ public void clearAll() {
   public void clear() {
     topologyCache.remove(this.clusterId);
     downHostCache.remove(this.clusterId);
-    multiWriterClusterCache.remove(this.clusterId);
     lastUsedReaderCache.remove(this.clusterId);
   }
 
   private static class ClusterTopologyInfo {
     private List<HostInfo> hosts;
-    private boolean isMultiWriterCluster;
 
-    ClusterTopologyInfo(List<HostInfo> hosts, boolean isMultiWriterCluster) {
+    ClusterTopologyInfo(List<HostInfo> hosts) {
       this.hosts = hosts;
-      this.isMultiWriterCluster = isMultiWriterCluster;
     }
 
     List<HostInfo> getHosts() { return this.hosts; }
-    boolean getMultiWriterCluster() { return this.isMultiWriterCluster; }
   }
 }

src/main/user-impl/java/com/mysql/cj/jdbc/ha/plugins/failover/FailoverConnectionPlugin.java

@@ -147,7 +147,6 @@ public class FailoverConnectionPlugin implements IConnectionPlugin {
   protected boolean inTransaction = false;
   protected boolean explicitlyAutoCommit = true;
   protected boolean isClusterTopologyAvailable = false;
-  protected boolean isMultiWriterCluster = false;
   protected boolean isRdsProxy = false;
   protected boolean isRds = false;
   protected ITopologyService topologyService;
@@ -307,7 +306,6 @@ public boolean isFailoverEnabled() {
     return this.enableFailoverSetting
         && !this.isRdsProxy
         && this.isClusterTopologyAvailable
-        && !this.isMultiWriterCluster
         && (this.hosts == null || this.hosts.size() > 1);
   }
 
@@ -652,8 +650,7 @@ protected boolean shouldExceptionTriggerConnectionSwitch(Throwable t) {
 
     if (!this.enableFailoverSetting
         || this.isRdsProxy
-        || !this.isClusterTopologyAvailable
-        || this.isMultiWriterCluster) {
+        || !this.isClusterTopologyAvailable) {
       if (this.logger.isDebugEnabled()) {
         this.logger.logDebug(Messages.getString("ClusterAwareConnectionProxy.13"));
       }
@@ -1057,7 +1054,6 @@ private void fetchTopology() throws SQLException {
                       new Object[]{"isClusterTopologyAvailable",
                               this.isClusterTopologyAvailable}));
     }
-    this.isMultiWriterCluster = this.topologyService.isMultiWriterCluster();
     this.currentHostIndex =
             getHostIndex(topologyService.getHostByName(this.currentConnectionProvider.getCurrentConnection()));
 

src/main/user-impl/java/com/mysql/cj/jdbc/ha/plugins/failover/ITopologyService.java

@@ -140,13 +140,6 @@ List<HostInfo> getTopology(JdbcConnection conn, boolean forceUpdate)
    */
   void removeFromDownHostList(HostInfo host);
 
-  /**
-   * Check if topology belongs to multi-writer cluster.
-   *
-   * @return True, if it's multi-writer cluster.
-   */
-  boolean isMultiWriterCluster();
-
   /**
    * Set new topology refresh rate. Topology is considered valid (up to date) within provided
    * duration of time.