pool: ignore failures from replaced pools

Author: dkropachev

cassandra/pool.py

@@ -613,23 +613,8 @@ def return_connection(self, connection, stream_was_orphaned=False):
             if not connection.signaled_error:
                 log.debug("Defunct or closed connection (%s) returned to pool, potentially "
                           "marking host %s as down", id(connection), self.host)
-                with self.host.lock:
-                    endpoint_matches = _endpoints_match(
-                        self._session.cluster, self.host.endpoint, self.endpoint)
-                    host_is_current = (endpoint_matches and
-                                       _host_is_current_for_endpoint(
-                                           self._session.cluster, self.host,
-                                           self.endpoint))
-                    if not endpoint_matches:
-                        log.debug("Ignoring stale connection failure for host %s; endpoint changed from %s",
-                                  self.host, self.endpoint)
-                        stale_endpoint_failure = True
-                    elif not host_is_current:
-                        log.debug("Ignoring stale connection failure for host %s; endpoint reassigned from %s",
-                                  self.host, self.endpoint)
-                        stale_endpoint_failure = True
-                    else:
-                        is_down = self.host.signal_connection_failure(connection.last_error)
+                is_down, stale_endpoint_failure = \
+                    self._signal_connection_failure_if_current(connection)
                 connection.signaled_error = True
 
             if stale_endpoint_failure:
@@ -683,6 +668,52 @@ def on_orphaned_stream_released(self):
         with self._stream_available_condition:
             self._stream_available_condition.notify()
 
+    def _is_current_pool_for_endpoint_locked(self, expected_endpoint):
+        pools = getattr(self._session, "_pools", None)
+        if not isinstance(pools, dict):
+            return True
+
+        for pool_host, host_pool in pools.items():
+            if pool_host is not self.host or host_pool is not self:
+                continue
+            pool_endpoint = getattr(host_pool, 'endpoint', None)
+            if pool_endpoint is None:
+                pool_endpoint = host_pool.host.endpoint
+            return _endpoints_match(
+                self._session.cluster, pool_endpoint, expected_endpoint)
+        return False
+
+    def _signal_connection_failure_if_current(self, connection):
+        session_lock = getattr(self._session, "_lock", None)
+        if session_lock is None:
+            with self.host.lock:
+                return self._signal_connection_failure_locked(connection)
+
+        with session_lock:
+            with self.host.lock:
+                if not self._is_current_pool_for_endpoint_locked(self.endpoint):
+                    log.debug("Ignoring stale connection failure for host %s; pool was replaced for %s",
+                              self.host, self.endpoint)
+                    return False, True
+                return self._signal_connection_failure_locked(connection)
+
+    def _signal_connection_failure_locked(self, connection):
+        endpoint_matches = _endpoints_match(
+            self._session.cluster, self.host.endpoint, self.endpoint)
+        host_is_current = (endpoint_matches and
+                           _host_is_current_for_endpoint(
+                               self._session.cluster, self.host,
+                               self.endpoint))
+        if not endpoint_matches:
+            log.debug("Ignoring stale connection failure for host %s; endpoint changed from %s",
+                      self.host, self.endpoint)
+            return False, True
+        if not host_is_current:
+            log.debug("Ignoring stale connection failure for host %s; endpoint reassigned from %s",
+                      self.host, self.endpoint)
+            return False, True
+        return self.host.signal_connection_failure(connection.last_error), False
+
     def _remove_stale_pool(self, expected_endpoint):
         future = self._session.remove_pool(
             self.host, expected_host=self.host,

tests/unit/test_host_connection_pool.py

@@ -296,6 +296,45 @@ def test_return_defunct_connection_after_endpoint_reassignment_is_ignored(self):
         conn.close.assert_called_once_with()
         assert not pool.is_shutdown
 
+    def test_return_defunct_connection_from_removed_pool_after_endpoint_flip_back_is_ignored(self):
+        endpoint = DefaultEndPoint('127.0.0.1')
+        host = Host(endpoint, SimpleConvictionPolicy, host_id=uuid.uuid4())
+        host.signal_connection_failure = Mock(return_value=True)
+
+        session = self.make_session()
+        session._lock = Lock()
+        session._pools = {}
+        session.remove_pool.return_value = None
+        session.cluster.metadata = Metadata()
+        session.cluster.metadata.add_or_return_host(host)
+        session.cluster._endpoints_match.side_effect = Cluster._endpoints_match
+        conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False,
+                            is_closed=False, max_request_id=100,
+                            signaled_error=False,
+                            orphaned_threshold_reached=False)
+        session.cluster.connection_factory.return_value = conn
+
+        stale_pool = self.PoolImpl(host, HostDistance.LOCAL, session)
+        stale_pool.borrow_connection(timeout=0.01)
+
+        # The old pool was already removed, then the host endpoint flipped back
+        # to the same endpoint and a replacement pool became current.
+        current_pool = Mock()
+        current_pool.host = host
+        current_pool.endpoint = endpoint
+        session._pools[host] = current_pool
+
+        conn.is_defunct = True
+        stale_pool.return_connection(conn)
+
+        host.signal_connection_failure.assert_not_called()
+        session.cluster.on_down.assert_not_called()
+        session.remove_pool.assert_called_once_with(
+            host, expected_host=host, expected_endpoint=endpoint,
+            expected_pool=stale_pool)
+        conn.close.assert_called_once_with()
+        assert not stale_pool.is_shutdown
+
     def test_host_instantiations(self):
         """
         Ensure Host fails if not initialized properly