Fix ignored streaming encryption settings in sstableloader

Patch by Dmitry Potepalov; reviewed by bereng and brandonwilliams for
CASSANDRA-17367

Encryption settings were silently lost in BulkLoadConnectionFactory.
Forced the bulk loader test to do some actual streaming besides just
handshaking on the native port, the test does not pass without the
change in BulkLoadConnectionFactory.

.build/build-rat.xml

@@ -62,6 +62,7 @@
                  <exclude name="**/test/conf/commitlog_compression_LZ4.yaml"/>
                  <exclude name="**/test/conf/commitlog_compression_Zstd.yaml"/>
                  <exclude name="**/test/conf/system_keyspaces_directory.yaml"/>
+                 <exclude name="**/test/conf/sstableloader_with_encryption.yaml"/>
                  <exclude name="**/test/conf/unit-test-conf/test-native-port.yaml"/>
                  <exclude name="**/test/data/jmxdump/cassandra-3.0-jmx.yaml"/>
                  <exclude name="**/test/data/jmxdump/cassandra-3.11-jmx.yaml"/>

CHANGES.txt

@@ -95,6 +95,7 @@
  * GossiperTest.testHasVersion3Nodes didn't take into account trunk version changes, fixed to rely on latest version (CASSANDRA-16651)
  * Update JNA library to 5.9.0 and snappy-java to version 1.1.8.4 (CASSANDRA-17040)
 Merged from 4.0:
+ * Fix ignored streaming encryption settings in sstableloader (CASSANDRA-17367)
  * Streaming tasks handle empty SSTables correctly (CASSANDRA-16349)
  * Prevent SSTableLoader from doing unnecessary work (CASSANDRA-16349)
 Merged from 3.0:

build.xml

@@ -1518,6 +1518,7 @@
       <delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/commitlog"/>
       <delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/cdc_raw"/>
       <delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/data"/>
+      <delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/ssl_upload_tables"/>
       <delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/system_data"/>
       <delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/saved_caches"/>
       <delete quiet="true" failonerror="false" dir="${build.test.dir}/cassandra/hints"/>

src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java

@@ -50,7 +50,7 @@ public class BulkLoadConnectionFactory extends NettyStreamingConnectionFactory
         // does not know which node is in which dc/rack, connecting to SSL port is always the option.
         OutboundConnectionSettings template = new OutboundConnectionSettings(getByAddress(to));
         if (encryptionOptions != null && encryptionOptions.internode_encryption != EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.none)
-            template = template.withConnectTo(template.to.withPort(secureStoragePort));
+            template = template.withConnectTo(template.to.withPort(secureStoragePort)).withEncryption(encryptionOptions);
 
         return connect(template, messagingVersion, kind);
     }

test/conf/sstableloader_with_encryption.yaml

@@ -0,0 +1,7 @@
+server_encryption_options:
+    internode_encryption: all
+    keystore: test/conf/cassandra_ssl_test.keystore
+    keystore_password: cassandra
+    truststore: test/conf/cassandra_ssl_test.truststore
+    truststore_password: cassandra
+    protocol: TLSv1.2

test/distributed/org/apache/cassandra/distributed/test/SSTableLoaderEncryptionOptionsTest.java

@@ -19,22 +19,30 @@
 package org.apache.cassandra.distributed.test;
 
 import java.io.IOException;
+import java.io.File;
 import java.util.Collections;
+import java.util.List;
 
 import com.google.common.collect.ImmutableMap;
+import org.apache.commons.io.FileUtils;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.distributed.Cluster;
 import org.apache.cassandra.distributed.api.Feature;
 import org.apache.cassandra.tools.BulkLoader;
 import org.apache.cassandra.tools.ToolRunner;
+import org.apache.cassandra.service.StorageService;
 
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertTrue;
 
+import static org.apache.cassandra.distributed.test.ExecUtil.rethrow;
+import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
+import static org.apache.cassandra.distributed.shared.AssertUtils.row;
 
 public class SSTableLoaderEncryptionOptionsTest extends AbstractEncryptionOptionsImpl
 {
@@ -72,19 +80,22 @@ public class SSTableLoaderEncryptionOptionsTest extends AbstractEncryptionOption
             CLUSTER.close();
     }
     @Test
-    public void bulkLoaderSuccessfullyConnectsOverSsl() throws Throwable
+    public void bulkLoaderSuccessfullyStreamsOverSsl() throws Throwable
     {
+        File sstables_to_upload = prepareSstablesForUpload();
         ToolRunner.ToolResult tool = ToolRunner.invokeClass(BulkLoader.class,
                                                             "--nodes", NODES,
                                                             "--port", Integer.toString(NATIVE_PORT),
-                                                            "--storage-port", Integer.toString(STORAGE_PORT),
+                                                            "--ssl-storage-port", Integer.toString(STORAGE_PORT),
                                                             "--keystore", validKeyStorePath,
                                                             "--keystore-password", validKeyStorePassword,
                                                             "--truststore", validTrustStorePath,
                                                             "--truststore-password", validTrustStorePassword,
-                                                            "test/data/legacy-sstables/na/legacy_tables/legacy_na_clust");
+                                                            "--conf-path", "test/conf/sstableloader_with_encryption.yaml",
+                                                            sstables_to_upload.getAbsolutePath());
         tool.assertOnCleanExit();
         assertTrue(tool.getStdout().contains("Summary statistics"));
+        assertRows(CLUSTER.get(1).executeInternal("SELECT count(*) FROM ssl_upload_tables.test"), row(42L));
     }
 
     @Test
@@ -103,4 +114,47 @@ public class SSTableLoaderEncryptionOptionsTest extends AbstractEncryptionOption
         assertNotEquals(0, tool.getExitCode());
         assertTrue(tool.getStdout().contains("SSLHandshakeException"));
     }
+
+    private static File prepareSstablesForUpload() throws IOException
+    {
+        generateSstables();
+        File sstable_dir = copySstablesFromDataDir("test");
+        truncateGeneratedTables();
+        return sstable_dir;
+    }
+
+    private static void generateSstables() throws IOException
+    {
+        CLUSTER.schemaChange("CREATE KEYSPACE ssl_upload_tables WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}");
+        CLUSTER.schemaChange("CREATE TABLE ssl_upload_tables.test (pk int, val text, PRIMARY KEY (pk))");
+        for (int i = 0; i < 42; i++)
+        {
+            CLUSTER.get(1).executeInternal(String.format("INSERT INTO ssl_upload_tables.test (pk, val) VALUES (%s, '%s')",
+                                                         i, Integer.toString(i)));
+        }
+        CLUSTER.get(1).runOnInstance(rethrow(() -> StorageService.instance.forceKeyspaceFlush("ssl_upload_tables")));
+    }
+
+    private static void truncateGeneratedTables() throws IOException
+    {
+        CLUSTER.get(1).executeInternal("TRUNCATE ssl_upload_tables.test");
+    }
+
+    private static File copySstablesFromDataDir(String table) throws IOException
+    {
+        File cfDir = new File("build/test/cassandra/ssl_upload_tables", table);
+        cfDir.mkdirs();
+        List<File> keyspace_dirs = CLUSTER.get(1).callOnInstance(() -> Keyspace.open("ssl_upload_tables").getColumnFamilyStore(table).getDirectories().getCFDirectories());
+        for (File srcDir : keyspace_dirs)
+        {
+            for (File file : srcDir.listFiles())
+            {
+                if (file.isFile())
+                {
+                    FileUtils.copyFileToDirectory(file, cfDir);
+                }
+            }
+        }
+        return cfDir;
+    }
 }