Merge branch 'cassandra-2.1' into cassandra-2.2

# Conflicts:
#	CHANGES.txt
#	src/java/org/apache/cassandra/db/ColumnFamilyStore.java
#	test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java

CHANGES.txt

@@ -1,22 +1,23 @@
-2.2.18
+2.2.17
  * Fix nomenclature of allow and deny lists (CASSANDRA-15862)
  * Remove generated files from source artifact (CASSANDRA-15849)
  * Remove duplicated tools binaries from tarballs (CASSANDRA-15768)
  * Duplicate results with DISTINCT queries in mixed mode (CASSANDRA-15501)
  * Disable JMX rebinding (CASSANDRA-15653)
-Merged from 2.1:
-  * Fix parse error in cqlsh COPY FROM and formatting for map of blobs (CASSANDRA-15679)
-
-2.2.17
  * Fix Commit log replays when static column clustering keys are collections (CASSANDRA-14365)
  * Fix Red Hat init script on newer systemd versions (CASSANDRA-15273)
  * Allow EXTRA_CLASSPATH to work on tar/source installations (CASSANDRA-15567)
+Merged from 2.1:
+ * Fix writing of snapshot manifest when the table has table-backed secondary indexes (CASSANDRA-10968)
+ * Fix parse error in cqlsh COPY FROM and formatting for map of blobs (CASSANDRA-15679)
+
 
 2.2.16
  * Fix SELECT JSON output for empty blobs (CASSANDRA-15435)
  * In-JVM DTest: Set correct internode message version for upgrade test (CASSANDRA-15371)
  * In-JVM DTest: Support NodeTool in dtest
 
+
 2.2.15
  * Catch non-IOException in FileUtils.close to make sure that all resources are closed (CASSANDRA-15225)
  * Handle exceptions during authentication/authorization (CASSANDRA-15041)

src/java/org/apache/cassandra/config/Config.java

@@ -177,7 +177,7 @@ public class Config
     public volatile Integer stream_throughput_outbound_megabits_per_sec = 200;
     public volatile Integer inter_dc_stream_throughput_outbound_megabits_per_sec = 200;
 
-    public String[] data_file_directories;
+    public String[] data_file_directories = new String[0];
 
     public String saved_caches_directory;
 

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -2426,9 +2426,9 @@ public void snapshotWithoutFlush(String snapshotName)
     public Set<SSTableReader> snapshotWithoutFlush(String snapshotName, Predicate<SSTableReader> predicate, boolean ephemeral)
     {
         Set<SSTableReader> snapshottedSSTables = new HashSet<>();
+        final JSONArray filesJSONArr = new JSONArray();
         for (ColumnFamilyStore cfs : concatWithIndexes())
         {
-            final JSONArray filesJSONArr = new JSONArray();
             try (RefViewFragment currentView = cfs.selectAndReference(CANONICAL_SSTABLES))
             {
                 for (SSTableReader ssTable : currentView.sstables)
@@ -2444,10 +2444,9 @@ public Set<SSTableReader> snapshotWithoutFlush(String snapshotName, Predicate<SS
                         logger.trace("Snapshot for {} keyspace data file {} created in {}", keyspace, ssTable.getFilename(), snapshotDirectory);
                     snapshottedSSTables.add(ssTable);
                 }
-
-                writeSnapshotManifest(filesJSONArr, snapshotName);
             }
         }
+        writeSnapshotManifest(filesJSONArr, snapshotName);
         if (ephemeral)
             createEphemeralSnapshotMarkerFile(snapshotName);
         return snapshottedSSTables;

src/java/org/apache/cassandra/db/Directories.java

@@ -422,7 +422,7 @@ public static File getSnapshotDirectory(Descriptor desc, String snapshotName)
      */
     public static File getSnapshotDirectory(File location, String snapshotName)
     {
-        if (location.getName().startsWith(SECONDARY_INDEX_NAME_SEPARATOR))
+        if (isSecondaryIndexFolder(location))
         {
             return getOrCreate(location.getParentFile(), SNAPSHOT_SUBDIR, snapshotName, location.getName());
         }
@@ -456,7 +456,7 @@ public static File getBackupsDirectory(Descriptor desc)
 
     public static File getBackupsDirectory(File location)
     {
-        if (location.getName().startsWith(SECONDARY_INDEX_NAME_SEPARATOR))
+        if (isSecondaryIndexFolder(location))
         {
             return getOrCreate(location.getParentFile(), BACKUPS_SUBDIR, location.getName());
         }
@@ -680,9 +680,9 @@ private List<File> listSnapshots()
         final List<File> snapshots = new LinkedList<>();
         for (final File dir : dataPaths)
         {
-            File snapshotDir = dir.getName().startsWith(SECONDARY_INDEX_NAME_SEPARATOR) ?
-                                       new File(dir.getParent(), SNAPSHOT_SUBDIR) :
-                                       new File(dir, SNAPSHOT_SUBDIR);
+            File snapshotDir = isSecondaryIndexFolder(dir)
+                               ? new File(dir.getParent(), SNAPSHOT_SUBDIR)
+                               : new File(dir, SNAPSHOT_SUBDIR);
             if (snapshotDir.exists() && snapshotDir.isDirectory())
             {
                 final File[] snapshotDirs  = snapshotDir.listFiles();
@@ -705,7 +705,7 @@ public boolean snapshotExists(String snapshotName)
         for (File dir : dataPaths)
         {
             File snapshotDir;
-            if (dir.getName().startsWith(SECONDARY_INDEX_NAME_SEPARATOR))
+            if (isSecondaryIndexFolder(dir))
             {
                 snapshotDir = new File(dir.getParentFile(), join(SNAPSHOT_SUBDIR, snapshotName, dir.getName()));
             }
@@ -764,9 +764,9 @@ public long trueSnapshotsSize()
         long result = 0L;
         for (File dir : dataPaths)
         {
-            File snapshotDir = dir.getName().startsWith(SECONDARY_INDEX_NAME_SEPARATOR) ?
-                                       new File(dir.getParent(), SNAPSHOT_SUBDIR) :
-                                       new File(dir, SNAPSHOT_SUBDIR);
+            File snapshotDir = isSecondaryIndexFolder(dir)
+                               ? new File(dir.getParent(), SNAPSHOT_SUBDIR)
+                               : new File(dir, SNAPSHOT_SUBDIR);
             result += getTrueAllocatedSizeIn(snapshotDir);
         }
         return result;
@@ -809,6 +809,11 @@ public static List<File> getKSChildDirectories(String ksName)
         return result;
     }
 
+    public static boolean isSecondaryIndexFolder(File dir)
+    {
+        return dir.getName().startsWith(SECONDARY_INDEX_NAME_SEPARATOR);
+    }
+
     public List<File> getCFDirectories()
     {
         List<File> result = new ArrayList<>();

src/java/org/apache/cassandra/io/sstable/Descriptor.java

@@ -151,6 +151,11 @@ private void appendFileName(StringBuilder buff)
     public String relativeFilenameFor(Component component)
     {
         final StringBuilder buff = new StringBuilder();
+        if (Directories.isSecondaryIndexFolder(directory))
+        {
+            buff.append(directory.getName()).append(File.separator);
+        }
+
         appendFileName(buff);
         buff.append(separator).append(component.name());
         return buff.toString();
@@ -271,7 +276,7 @@ else if (Descriptor.Type.TEMPLINK.marker.equals(nexttok))
             File cfDirectory = parentDirectory;
             // check if this is secondary index
             String indexName = "";
-            if (cfDirectory.getName().startsWith(Directories.SECONDARY_INDEX_NAME_SEPARATOR))
+            if (Directories.isSecondaryIndexFolder(cfDirectory))
             {
                 indexName = cfDirectory.getName();
                 cfDirectory = cfDirectory.getParentFile();

test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java

@@ -19,6 +19,7 @@
 package org.apache.cassandra.db;
 
 import java.io.File;
+import java.io.FileReader;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.CharacterCodingException;
@@ -95,6 +96,9 @@
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.WrappedRunnable;
+import org.json.simple.JSONArray;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
 
 import static org.apache.cassandra.Util.cellname;
 import static org.apache.cassandra.Util.column;
@@ -2356,4 +2360,39 @@ public void testRebuildSecondaryIndex() throws IOException
 
         PerRowSecondaryIndexTest.TestIndex.reset();
     }
+
+    @Test
+    public void testSnapshotWithoutFlushWithSecondaryIndexes() throws Exception
+    {
+        Keyspace keyspace = Keyspace.open(KEYSPACE1);
+        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_INDEX1);
+        cfs.truncateBlocking();
+
+        List<Mutation> rms = new LinkedList<>();
+        Mutation rm;
+
+        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("k1"));
+        rm.add(CF_INDEX1, cellname("birthdate"), ByteBufferUtil.bytes(1L), 0);
+        rm.add(CF_INDEX1, cellname("nobirthdate"), ByteBufferUtil.bytes(1L), 0);
+        rms.add(rm);
+        Util.writeColumnFamily(rms);
+
+        String snapshotName = "newSnapshot";
+        cfs.snapshotWithoutFlush(snapshotName);
+
+        File snapshotManifestFile = cfs.directories.getSnapshotManifestFile(snapshotName);
+        JSONParser parser = new JSONParser();
+        JSONObject manifest = (JSONObject) parser.parse(new FileReader(snapshotManifestFile));
+        JSONArray files = (JSONArray) manifest.get("files");
+
+        // Keyspace1-Indexed1 and the corresponding index
+        assert files.size() == 2;
+
+        // Snapshot of the secondary index is stored in the subfolder with the same file name
+        String baseTableFile = (String) files.get(0);
+        String indexTableFile = (String) files.get(1);
+        assert !baseTableFile.equals(indexTableFile);
+        assert Directories.isSecondaryIndexFolder(new File(indexTableFile).getParentFile());
+        assert indexTableFile.endsWith(baseTableFile);
+    }
 }