Test failure: org.apache.cassandra.db.commitlog.CommitLogSegmentManagerCDCTest

patch by Berenguer Blasi; reviewed by Andres de la Peña for CASSANDRA-18948

Author: bereng

src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java

@@ -59,7 +59,7 @@ public abstract class CommitLogSegment
 {
     private final static long idBase;
 
-    private CDCState cdcState = CDCState.PERMITTED;
+    private volatile CDCState cdcState = CDCState.PERMITTED;
     public enum CDCState
     {
         PERMITTED,

test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java

@@ -35,7 +35,6 @@
 import org.apache.cassandra.db.RowUpdateBuilder;
 import org.apache.cassandra.db.commitlog.CommitLogSegment.CDCState;
 import org.apache.cassandra.exceptions.CDCWriteException;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.schema.TableMetadata;
 
 public class CommitLogSegmentManagerCDCTest extends CQLTester
@@ -68,7 +67,7 @@ public void testCDCWriteFailure() throws Throwable
         TableMetadata cfm = currentTableMetadata();
 
         // Confirm that logic to check for whether or not we can allocate new CDC segments works
-        Integer originalCDCSize = DatabaseDescriptor.getCDCSpaceInMB();
+        int originalCDCSize = DatabaseDescriptor.getCDCSpaceInMB();
         try
         {
             DatabaseDescriptor.setCDCSpaceInMB(32);
@@ -98,11 +97,10 @@ public void testCDCWriteFailure() throws Throwable
             Keyspace.open(keyspace()).getColumnFamilyStore(currentTable()).forceBlockingFlush();
             CommitLog.instance.forceRecycleAllSegments();
             cdcMgr.awaitManagementTasksCompletion();
-            Assert.assertTrue("Expected files to be moved to overflow.", getCDCRawCount() > 0);
+            Assert.assertTrue("Expected files to be moved to overflow.", getCDCRawFiles().length > 0);
 
             // Simulate a CDC consumer reading files then deleting them
-            for (File f : new File(DatabaseDescriptor.getCDCLogLocation()).listFiles())
-                FileUtils.deleteWithConfirm(f);
+            deleteCDCRawFiles();
 
             // Update size tracker to reflect deleted files. Should flip flag on current allocatingFrom to allow.
             cdcMgr.updateCDCTotalSize();
@@ -115,7 +113,7 @@ public void testCDCWriteFailure() throws Throwable
     }
 
     @Test
-    public void testSegmentFlaggingOnCreation() throws Throwable
+    public void testSegmentFlaggingOnCreation()
     {
         CommitLogSegmentManagerCDC cdcMgr = (CommitLogSegmentManagerCDC)CommitLog.instance.segmentManager;
         String ct = createTable("CREATE TABLE %s (idx int, data text, primary key(idx)) WITH cdc=true;");
@@ -143,16 +141,13 @@ public void testSegmentFlaggingOnCreation() throws Throwable
 
             cdcMgr.awaitManagementTasksCompletion();
             // Delete all files in cdc_raw
-            for (File f : new File(DatabaseDescriptor.getCDCLogLocation()).listFiles())
-                f.delete();
+            deleteCDCRawFiles();
             cdcMgr.updateCDCTotalSize();
             // Confirm cdc update process changes flag on active segment
             expectCurrentCDCState(CDCState.PERMITTED);
 
             // Clear out archived CDC files
-            for (File f : new File(DatabaseDescriptor.getCDCLogLocation()).listFiles()) {
-                FileUtils.deleteWithConfirm(f);
-            }
+            deleteCDCRawFiles();
         }
         finally
         {
@@ -179,8 +174,8 @@ public void testCDCIndexFileWriteOnSync() throws IOException
         // Read index value and confirm it's == end from last sync
         BufferedReader in = new BufferedReader(new FileReader(cdcIndexFile));
         String input = in.readLine();
-        Integer offset = Integer.parseInt(input);
-        Assert.assertEquals(syncOffset, (long)offset);
+        int offset = Integer.parseInt(input);
+        Assert.assertEquals(syncOffset, offset);
         in.close();
     }
 
@@ -189,7 +184,7 @@ public void testCompletedFlag() throws IOException
     {
         createTable("CREATE TABLE %s (idx int, data text, primary key(idx)) WITH cdc=true;");
         CommitLogSegment initialSegment = CommitLog.instance.segmentManager.allocatingFrom();
-        Integer originalCDCSize = DatabaseDescriptor.getCDCSpaceInMB();
+        int originalCDCSize = DatabaseDescriptor.getCDCSpaceInMB();
 
         DatabaseDescriptor.setCDCSpaceInMB(8);
         try
@@ -220,7 +215,7 @@ public void testCompletedFlag() throws IOException
         BufferedReader in = new BufferedReader(new FileReader(cdcIndexFile));
         String input = in.readLine();
         input = in.readLine();
-        Assert.assertTrue("Expected COMPLETED in index file, got: " + input, input.equals("COMPLETED"));
+        Assert.assertEquals("Expected COMPLETED in index file, got: " + input, "COMPLETED", input);
         in.close();
     }
 
@@ -282,7 +277,7 @@ public void testReplayLogic() throws IOException
     {
         // Assert.assertEquals(0, new File(DatabaseDescriptor.getCDCLogLocation()).listFiles().length);
         String table_name = createTable("CREATE TABLE %s (idx int, data text, primary key(idx)) WITH cdc=true;");
-        Integer originalCDCSize = DatabaseDescriptor.getCDCSpaceInMB();
+        int originalCDCSize = DatabaseDescriptor.getCDCSpaceInMB();
 
         DatabaseDescriptor.setCDCSpaceInMB(8);
         TableMetadata ccfm = Keyspace.open(keyspace()).getColumnFamilyStore(table_name).metadata();
@@ -310,13 +305,12 @@ public void testReplayLogic() throws IOException
 
         // Build up a list of expected index files after replay and then clear out cdc_raw
         List<CDCIndexData> oldData = parseCDCIndexData();
-        for (File f : new File(DatabaseDescriptor.getCDCLogLocation()).listFiles())
-            FileUtils.deleteWithConfirm(f.getAbsolutePath());
+        deleteCDCRawFiles();
 
         try
         {
             Assert.assertEquals("Expected 0 files in CDC folder after deletion. ",
-                                0, new File(DatabaseDescriptor.getCDCLogLocation()).listFiles().length);
+                                0, getCDCRawFiles().length);
         }
         finally
         {
@@ -331,7 +325,7 @@ public void testReplayLogic() throws IOException
 
         // Rough sanity check -> should be files there now.
         Assert.assertTrue("Expected non-zero number of files in CDC folder after restart.",
-                          new File(DatabaseDescriptor.getCDCLogLocation()).listFiles().length > 0);
+                          getCDCRawFiles().length > 0);
 
         // Confirm all the old indexes in old are present and >= the original offset, as we flag the entire segment
         // as cdc written on a replay.
@@ -345,6 +339,7 @@ public void testReplayLogic() throws IOException
                 {
                     Assert.assertTrue("New CDC index file expected to have >= offset in old.", ncid.offset >= cid.offset);
                     found = true;
+                    break;
                 }
             }
             if (!found)
@@ -365,7 +360,10 @@ public void testReplayLogic() throws IOException
             for (CDCIndexData cid : oldData)
             {
                 if (cid.fileName.equals(ncid.fileName))
+                {
                     found = true;
+                    break;
+                }
             }
             if (!found)
                 Assert.fail(String.format("Unexpected new CDCIndexData found after replay: %s\n", ncid));
@@ -377,7 +375,7 @@ private List<CDCIndexData> parseCDCIndexData()
         List<CDCIndexData> results = new ArrayList<>();
         try
         {
-            for (File f : new File(DatabaseDescriptor.getCDCLogLocation()).listFiles())
+            for (File f : getCDCRawFiles())
             {
                 if (f.getName().contains("_cdc.idx"))
                     results.add(new CDCIndexData(f));
@@ -397,15 +395,12 @@ private static class CDCIndexData
 
         CDCIndexData(File f) throws IOException
         {
-            String line = "";
+            String line;
             try (BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(f))))
             {
                 line = br.readLine();
             }
-            catch (Exception e)
-            {
-                throw e;
-            }
+
             fileName = f.getName();
             offset = Integer.parseInt(line);
         }
@@ -419,6 +414,8 @@ public String toString()
         @Override
         public boolean equals(Object other)
         {
+            if (!(other instanceof CDCIndexData))
+                return false;
             CDCIndexData cid = (CDCIndexData)other;
             return fileName.equals(cid.fileName) && offset == cid.offset;
         }
@@ -431,11 +428,6 @@ private ByteBuffer randomizeBuffer(int size)
         return ByteBuffer.wrap(toWrap);
     }
 
-    private int getCDCRawCount()
-    {
-        return new File(DatabaseDescriptor.getCDCLogLocation()).listFiles().length;
-    }
-
     private void expectCurrentCDCState(CDCState expectedState)
     {
         CDCState currentState = CommitLog.instance.segmentManager.allocatingFrom().getCDCState();
@@ -447,4 +439,17 @@ private void expectCurrentCDCState(CDCState expectedState)
                         expectedState, currentState));
         }
     }
+
+    private static File[] getCDCRawFiles()
+    {
+        return new File(DatabaseDescriptor.getCDCLogLocation()).listFiles();
+    }
+
+    private static void deleteCDCRawFiles()
+    {
+        for (File f : getCDCRawFiles())
+        {
+            f.delete();
+        }
+    }
 }