sliding expiration cache fix

Author: joyc-bq

aws_advanced_python_wrapper/fastest_response_strategy_plugin.py

@@ -105,16 +105,25 @@ def get_host_info_by_strategy(self, role: HostRole, strategy: str) -> HostInfo:
 
         fastest_response_host: Optional[HostInfo] = self._cached_fastest_response_host_by_role.get(role.name)
         if fastest_response_host is not None:
+
             # Found a fastest host. Let's find it in the the latest topology.
             for host in self._plugin_service.hosts:
-                if host is not None:
+                if host == fastest_response_host:
                     # found the fastest host in the topology
                     return host
                 # It seems that the fastest cached host isn't in the latest topology.
                 # Let's ignore cached results and find the fastest host.
 
             # Cached result isn't available. Need to find the fastest response time host.
-            calculated_fastest_response_host = None
+            host_dict = {}
+            for host in self._plugin_service.hosts:
+                if role == host.role and self._host_response_time_service is not None:
+                    response_time_tuple = FastestResponseStrategyPlugin.ResponseTimeTuple(host,
+                                                                                          self._host_response_time_service.get_response_time(host))
+                    host_dict[response_time_tuple.host_info] = response_time_tuple.response_time
+            # sort by response time then retrieve the first host
+            sorted_host_dict = dict(sorted(host_dict.items(), key=lambda item: item[1]))
+            calculated_fastest_response_host: HostInfo = next(iter(sorted_host_dict.keys()))
             if calculated_fastest_response_host is None:
                 return self._random_host_selector.get_host(self._plugin_service.hosts, role, self._properties)
 
@@ -219,7 +228,7 @@ def run(self):
                     logger.debug("NodeResponseTimeMonitor.InterruptedExceptionDuringMonitoring", self._host_info.host)
         except Exception as e:
             # this should not be reached; log and exit thread
-            logger.debug("NodeResponseTimeMonitor.RxceptionDuringMonitoringStop",
+            logger.debug("NodeResponseTimeMonitor.ExceptionDuringMonitoringStop",
                          self._host_info.host,
                          e)  # print full trace stack of the exception.
         finally:
@@ -243,7 +252,7 @@ def _open_connection(self):
 
                 logger.debug("NodeResponseTimeMonitor.OpeningConnection", self._host_info.url)
                 self._monitoring_conn = self._plugin_service.force_connect(self._host_info, monitoring_conn_properties, None)
-                logger.debug("NodeResponseTimeMonitor.openedConnection", self._host_info.url)
+                logger.debug("NodeResponseTimeMonitor.OpenedConnection", self._host_info.url)
 
         except Exception:
             if self._monitoring_conn is not None:
@@ -258,10 +267,11 @@ def _open_connection(self):
 class HostResponseTimeService:
     """
     Return a response time in milliseconds to the host.
-    Return _MAX_VALUE if response time is not available.
+    Return HostResponseTimeService._MAX_VALUE if response time is not available.
 
     @param hostSpec the host details
-    @return response time in milliseconds for a desired host. It should return _MAX_VALUE if response time couldn't be measured.
+    @return response time in milliseconds for a desired host. It should return HostResponseTimeService._MAX_VALUE
+    if response time couldn't be measured.
    """
 
     @abstractmethod
@@ -280,8 +290,8 @@ class HostResponseTimeServiceImpl(HostResponseTimeService):
     _lock: Lock = Lock()
     _monitoring_nodes: ClassVar[SlidingExpirationCacheWithCleanupThread[str, NodeResponseTimeMonitor]] = \
         SlidingExpirationCacheWithCleanupThread(_CACHE_CLEANUP_NANO,
-                                                should_dispose_func=lambda queue_pool: queue_pool.checkedout() == 0,
-                                                item_disposal_func=lambda queue_pool: queue_pool.dispose())
+                                                should_dispose_func=lambda monitor: True,
+                                                item_disposal_func=lambda monitor: monitor.dispose())
 
     def __init__(self, plugin_service: PluginService, props: Properties, interval_ms: int):
         self._plugin_service = plugin_service
@@ -309,4 +319,3 @@ def set_hosts(self, hosts: Tuple[HostInfo, ...]) -> None:
                                                                  new_host,
                                                                  self._properties,
                                                                  self._interval_ms), self._CACHE_EXPIRATION_NANO)
-

aws_advanced_python_wrapper/resources/aws_advanced_python_wrapper_messages.properties

@@ -250,6 +250,8 @@ RoundRobinHostSelector.ClusterInfoNone=[RoundRobinHostSelector] The round robin
 RoundRobinHostSelector.RoundRobinInvalidDefaultWeight=[RoundRobinHostSelector] The provided default weight value is not valid. Weight values must be an integer greater than or equal to 1.
 RoundRobinHostSelector.RoundRobinInvalidHostWeightPairs= [RoundRobinHostSelector] The provided host weight pairs have not been configured correctly. Please ensure the provided host weight pairs is a comma separated list of pairs, each pair in the format of <host>:<weight>. Weight values must be an integer greater than or equal to the default weight value of 1.
 
+SlidingExpirationCache.CleaningUp=[SlidingExpirationCache] Cleaning up...
+
 SqlAlchemyPooledConnectionProvider.PoolNone=[SqlAlchemyPooledConnectionProvider] Attempted to find or create a pool for '{}' but the result of the attempt evaluated to None.
 SqlAlchemyPooledConnectionProvider.UnableToCreateDefaultKey=[SqlAlchemyPooledConnectionProvider] Unable to create a default key for internal connection pools. By default, the user parameter is used, but the given user evaluated to None or the empty string (""). Please ensure you have passed a valid user in the connection properties.
 

aws_advanced_python_wrapper/utils/sliding_expiration_cache.py

@@ -100,7 +100,7 @@ def _cleanup(self):
             self._remove_if_expired(key)
 
 
-class SlidingExpirationCacheWithCleanupThread(SlidingExpirationCache):
+class SlidingExpirationCacheWithCleanupThread(SlidingExpirationCache, Generic[K, V]):
 
     _executor: ClassVar[Executor] = ThreadPoolExecutor(thread_name_prefix="SlidingExpirationCacheWithCleanupThreadExecutor")