Make sure pools keep trying to connect now

aholmberg · aholmberg · commit 5b7627a84866 · 2016-04-12T16:17:52.000-05:00
This follows from changing the default conviction (we now keep pools up,
and failures should keep trying to connect)

PYTHON-286
diff --git a/cassandra/pool.py b/cassandra/pool.py
@@ -335,12 +335,17 @@ def return_connection(self, connection):
 
     def _replace(self, connection):
         log.debug("Replacing connection (%s) to %s", id(connection), self.host)
-        conn = self._session.cluster.connection_factory(self.host.address)
-        if self._session.keyspace:
-            conn.set_keyspace_blocking(self._session.keyspace)
-        self._connection = conn
-        with self._lock:
-            self._is_replacing = False
+        try:
+            conn = self._session.cluster.connection_factory(self.host.address)
+            if self._session.keyspace:
+                conn.set_keyspace_blocking(self._session.keyspace)
+            self._connection = conn
+        except Exception:
+            log.warning("Failed reconnecting %s. Retrying." % (self.host.address,))
+            self._session.submit(self._replace, connection)
+        else:
+            with self._lock:
+                self._is_replacing = False
 
     def shutdown(self):
         with self._lock:
@@ -499,10 +504,10 @@ def _add_conn_if_under_max(self):
         max_conns = self._session.cluster.get_max_connections_per_host(self.host_distance)
         with self._lock:
             if self.is_shutdown:
-                return False
+                return True
 
             if self.open_count >= max_conns:
-                return False
+                return True
 
             self.open_count += 1
 
@@ -646,17 +651,22 @@ def _replace(self, connection):
 
         if should_replace:
             log.debug("Replacing connection (%s) to %s", id(connection), self.host)
-
-            def close_and_replace():
-                connection.close()
-                self._add_conn_if_under_max()
-
-            self._session.submit(close_and_replace)
+            connection.close()
+            self._session.submit(self._retrying_replace)
         else:
-            # just close it
             log.debug("Closing connection (%s) to %s", id(connection), self.host)
             connection.close()
 
+    def _retrying_replace(self):
+        replaced = False
+        try:
+            replaced = self._add_conn_if_under_max()
+        except Exception:
+            log.exception("Failed replacing connection to %s", self.host)
+        if not replaced:
+            log.debug("Failed replacing connection to %s. Retrying.", self.host)
+            self._session.submit(self._retrying_replace)
+
     def shutdown(self):
         with self._lock:
             if self.is_shutdown:
