Do not update pool if already created

In some cases We want to only update the pool if previous
do not exist or is shutdown. This commit adds additional
validation to add_or_renew_pool to make sure this condition
is met when needed.

Fixes: scylladb#317

Author: sylwiaszunejko

cassandra/cluster.py

@@ -1991,7 +1991,7 @@ def on_up(self, host):
             futures_results = []
             callback = partial(self._on_up_future_completed, host, futures, futures_results, futures_lock)
             for session in tuple(self.sessions):
-                future = session.add_or_renew_pool(host, is_host_addition=False)
+                future = session.add_or_renew_pool(host, is_host_addition=False, require_previous_shutdown_or_none=False)
                 if future is not None:
                     have_future = True
                     future.add_done_callback(callback)
@@ -2130,7 +2130,7 @@ def future_completed(future):
 
         have_future = False
         for session in tuple(self.sessions):
-            future = session.add_or_renew_pool(host, is_host_addition=True)
+            future = session.add_or_renew_pool(host, is_host_addition=True, require_previous_shutdown_or_none=False)
             if future is not None:
                 have_future = True
                 futures.add(future)
@@ -2661,7 +2661,7 @@ def __init__(self, cluster, hosts, keyspace=None):
         # create connection pools in parallel
         self._initial_connect_futures = set()
         for host in hosts:
-            future = self.add_or_renew_pool(host, is_host_addition=False)
+            future = self.add_or_renew_pool(host, is_host_addition=False, require_previous_shutdown_or_none=False)
             if future:
                 self._initial_connect_futures.add(future)
 
@@ -3284,7 +3284,7 @@ def __del__(self):
             # when cluster.shutdown() is called explicitly.
             pass
 
-    def add_or_renew_pool(self, host, is_host_addition):
+    def add_or_renew_pool(self, host, is_host_addition, require_previous_shutdown_or_none):
         """
         For internal use only.
         """
@@ -3332,7 +3332,11 @@ def callback(pool, errors):
                         self._lock.acquire()
                         return False
                     self._lock.acquire()
-                self._pools[host] = new_pool
+                if require_previous_shutdown_or_none and (previous and not previous.is_shutdown):
+                    new_pool.shutdown()
+                    return True
+                else:
+                    self._pools[host] = new_pool
 
             log.debug("Added pool for host %s to session", host)
             if previous:
@@ -3372,7 +3376,7 @@ def update_created_pools(self):
                 # to allow us to attempt connections to hosts that have gone from ignored to something
                 # else.
                 if distance != HostDistance.IGNORED and host.is_up in (True, None):
-                    future = self.add_or_renew_pool(host, False)
+                    future = self.add_or_renew_pool(host, False, require_previous_shutdown_or_none=True)
             elif distance != pool.host_distance:
                 # the distance has changed
                 if distance == HostDistance.IGNORED: