fix thread leak on connection error

Author: Bulat Shakirzyanov

lib/cassandra.rb

@@ -507,7 +507,7 @@ def self.cluster_async(options = {})
 
     hosts.shuffle!
   rescue => e
-    futures = options.fetch(:futures_factory) { Driver.new.futures_factory }
+    futures = options.fetch(:futures_factory) { return Future::Error.new(e) }
     futures.error(e)
   else
     driver = Driver.new(options)

lib/cassandra/driver.rb

@@ -145,7 +145,11 @@ def connect(addresses)
         if f.resolved?
           promise.fulfill(cluster)
         else
-          f.on_failure {|e| promise.break(e)}
+          f.on_failure do |e|
+            cluster.close_async.on_complete do |_, _|
+              promise.break(e)
+            end
+          end
         end
       end