natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

patch by Berenguer Blasi; reviewed by Andres de la Peña for CASSANDRA-18948
This commit is contained in:
Bereng 2023-11-17 09:09:00 +01:00
parent a912085f5e
commit 3ba6de70c7
2 changed files with 37 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
View File

@ -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,

67
test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java
View File

@ -35,7 +35,6 @@ import org.apache.cassandra.db.Keyspace;
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 class CommitLogSegmentManagerCDCTest extends CQLTester
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 class CommitLogSegmentManagerCDCTest extends CQLTester
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 class CommitLogSegmentManagerCDCTest extends CQLTester
}
@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 class CommitLogSegmentManagerCDCTest extends CQLTester
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 class CommitLogSegmentManagerCDCTest extends CQLTester
// 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 class CommitLogSegmentManagerCDCTest extends CQLTester
{
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 class CommitLogSegmentManagerCDCTest extends CQLTester
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 class CommitLogSegmentManagerCDCTest extends CQLTester
{
// 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 class CommitLogSegmentManagerCDCTest extends CQLTester
// 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 class CommitLogSegmentManagerCDCTest extends CQLTester
// 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 class CommitLogSegmentManagerCDCTest extends CQLTester
{
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 class CommitLogSegmentManagerCDCTest extends CQLTester
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 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
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 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
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 class CommitLogSegmentManagerCDCTest extends CQLTester
@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 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
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 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
expectedState, currentState));
}
}
private static File[] getCDCRawFiles()
{
return new File(DatabaseDescriptor.getCDCLogLocation()).listFiles();
}
private static void deleteCDCRawFiles()
{
for (File f : getCDCRawFiles())
{
f.delete();
}
}
}