Merge branch 'cassandra-2.1' into cassandra-2.2

CHANGES.txt

@@ -25,6 +25,7 @@
  * Always close cluster with connection in CqlRecordWriter (CASSANDRA-11553)
  * Fix slice queries on ordered COMPACT tables (CASSANDRA-10988)
 Merged from 2.1:
+ * cqlsh COPY FROM: shutdown parent cluster after forking, to avoid corrupting SSL connections (CASSANDRA-11749)
  * cqlsh: apply current keyspace to source command (CASSANDRA-11152)
  * Backport CASSANDRA-11578 (CASSANDRA-11750)
  * Clear out parent repair session if repair coordinator dies (CASSANDRA-11824)

pylib/cqlshlib/copyutil.py

@@ -59,6 +59,7 @@ PROFILE_ON = False
 STRACE_ON = False
 DEBUG = False  # This may be set to True when initializing the task
 IS_LINUX = platform.system() == 'Linux'
+IS_WINDOWS = platform.system() == 'Windows'
 
 CopyOptions = namedtuple('CopyOptions', 'copy dialect unrecognized')
 
@@ -425,9 +426,13 @@ class CopyTask(object):
     def make_params(self):
         """
         Return a dictionary of parameters to be used by the worker processes.
-        On Windows this dictionary must be pickle-able.
+        On Windows this dictionary must be pickle-able, therefore we do not pass the
+        parent connection since it may not be pickle-able. Also, on Windows child
+        processes are spawned and not forked, and therefore we don't need to shutdown
+        the parent connection anyway, see CASSANDRA-11749 for more details.
         """
         shell = self.shell
+
         return dict(ks=self.ks,
                     table=self.table,
                     local_dc=self.local_dc,
@@ -438,6 +443,7 @@ class CopyTask(object):
                     port=shell.port,
                     ssl=shell.ssl,
                     auth_provider=shell.auth_provider,
+                    parent_cluster=shell.conn if not IS_WINDOWS else None,
                     cql_version=shell.conn.cql_version,
                     config_file=self.config_file,
                     protocol_version=self.protocol_version,
@@ -1100,7 +1106,8 @@ class ImportTask(CopyTask):
                 self.processes.append(ImportProcess(self.update_params(params, i)))
 
             feeder = FeedingProcess(self.outmsg.channels[-1], self.inmsg.channels[-1],
-                                    self.outmsg.channels[:-1], self.fname, self.options)
+                                    self.outmsg.channels[:-1], self.fname, self.options,
+                                    self.shell.conn if not IS_WINDOWS else None)
             self.processes.append(feeder)
 
             self.start_processes()
@@ -1207,7 +1214,7 @@ class FeedingProcess(mp.Process):
     """
     A process that reads from import sources and sends chunks to worker processes.
     """
-    def __init__(self, inmsg, outmsg, worker_channels, fname, options):
+    def __init__(self, inmsg, outmsg, worker_channels, fname, options, parent_cluster):
         mp.Process.__init__(self, target=self.run)
         self.inmsg = inmsg
         self.outmsg = outmsg
@@ -1217,6 +1224,15 @@ class FeedingProcess(mp.Process):
         self.ingest_rate = options.copy['ingestrate']
         self.num_worker_processes = options.copy['numprocesses']
         self.chunk_id = 0
+        self.parent_cluster = parent_cluster
+
+    def on_fork(self):
+        """
+        Release any parent connections after forking, see CASSANDRA-11749 for details.
+        """
+        if self.parent_cluster:
+            printdebugmsg("Closing parent cluster sockets")
+            self.parent_cluster.shutdown()
 
     def run(self):
         pr = profile_on() if PROFILE_ON else None
@@ -1233,6 +1249,9 @@ class FeedingProcess(mp.Process):
         here we throttle using the ingest rate in the feeding process because of memory usage concerns.
         When finished we send back to the parent process the total number of rows sent.
         """
+
+        self.on_fork()
+
         reader = self.reader
         reader.start()
         channels = self.worker_channels
@@ -1296,6 +1315,7 @@ class ChildProcess(mp.Process):
         self.connect_timeout = params['connect_timeout']
         self.cql_version = params['cql_version']
         self.auth_provider = params['auth_provider']
+        self.parent_cluster = params['parent_cluster']
         self.ssl = params['ssl']
         self.protocol_version = params['protocol_version']
         self.config_file = params['config_file']
@@ -1313,6 +1333,14 @@ class ChildProcess(mp.Process):
         else:
             self.test_failures = None
 
+    def on_fork(self):
+        """
+        Release any parent connections after forking, see CASSANDRA-11749 for details.
+        """
+        if self.parent_cluster:
+            printdebugmsg("Closing parent cluster sockets")
+            self.parent_cluster.shutdown()
+
     def close(self):
         printdebugmsg("Closing queues...")
         self.inmsg.close()
@@ -1439,6 +1467,9 @@ class ExportProcess(ChildProcess):
         we can signal a global error by sending (None, error).
         We terminate when the inbound queue is closed.
         """
+
+        self.on_fork()
+
         while True:
             if self.num_requests() > self.max_requests:
                 time.sleep(0.001)  # 1 millisecond
@@ -2083,6 +2114,7 @@ class ImportProcess(ChildProcess):
         try:
             pr = profile_on() if PROFILE_ON else None
 
+            self.on_fork()
             self.inner_run(*self.make_params())
 
             if pr: