Author: slebresne
Date: Mon Aug 1 14:22:50 2011
New Revision: 1152793
URL: http://svn.apache.org/viewvc?rev=1152793&view=rev
Log:
fix bug where dirty commit logs were removed (and avoid keeping segment with no
post-flush activity permanently dirty)
patch by slebresne; reviewed by jbellis for CASSANDRA-2829
Modified:
cassandra/branches/cassandra-0.8/CHANGES.txt
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CommitLogTest.java
Modified: cassandra/branches/cassandra-0.8/CHANGES.txt
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/CHANGES.txt?rev=1152793&r1=1152792&r2=1152793&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.8/CHANGES.txt Mon Aug 1 14:22:50 2011
@@ -17,6 +17,8 @@
(CASSANDRA-2958)
* improved POSIX compatibility of start scripts (CASsANDRA-2965)
* add counter support to Hadoop InputFormat (CASSANDRA-2981)
+ * fix bug where dirty commit logs were removed (and avoid keeping segment
+ with no post-flush activity permanently dirty) (CASSANDRA-2829)
0.8.2
Modified:
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLog.java?rev=1152793&r1=1152792&r2=1152793&view=diff
==============================================================================
---
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
(original)
+++
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
Mon Aug 1 14:22:50 2011
@@ -385,6 +385,12 @@ public class CommitLog
}
}
+ // for tests mainly
+ public int segmentsCount()
+ {
+ return segments.size();
+ }
+
/*
* Adds the specified row to the commit log. This method will reset the
* file offset to what it is before the start of the operation in case
@@ -450,30 +456,36 @@ public class CommitLog
CommitLogSegment segment = iter.next();
if (segment.id == context.segment)
{
- // we can't just mark the segment where the flush happened
clean,
- // since there may have been writes to it between when the
flush
- // started and when it finished.
- segment.turnOn(id);
+ // Only unmark this segment if there were not write since the
+ // ReplayPosition was grabbed.
+ segment.turnOffIfNotWritten(id, context.position);
+ maybeDiscardSegment(segment, iter);
break;
}
segment.turnOff(id);
- if (segment.isSafeToDelete() && iter.hasNext())
- {
- logger.info("Discarding obsolete commit log:" + segment);
- segment.close();
- DeletionService.executeDelete(segment.getPath());
- // usually this will be the first (remaining) segment, but not
always, if segment A contains
- // writes to a CF that is unflushed but is followed by segment
B whose CFs are all flushed.
- iter.remove();
- }
- else
- {
- if (logger.isDebugEnabled())
- logger.debug("Not safe to delete commit log " + segment +
"; dirty is " + segment.dirtyString() + "; hasNext: " + iter.hasNext());
- }
+ maybeDiscardSegment(segment, iter);
}
}
+
+ private void maybeDiscardSegment(CommitLogSegment segment,
Iterator<CommitLogSegment> iter)
+ {
+ if (segment.isSafeToDelete() && iter.hasNext())
+ {
+ logger.info("Discarding obsolete commit log:" + segment);
+ segment.close();
+ DeletionService.executeDelete(segment.getPath());
+ // usually this will be the first (remaining) segment, but not
always, if segment A contains
+ // writes to a CF that is unflushed but is followed by segment B
whose CFs are all flushed.
+ iter.remove();
+ }
+ else
+ {
+ if (logger.isDebugEnabled())
+ logger.debug("Not safe to delete commit log " + segment + ";
dirty is " + segment.dirtyString() + "; hasNext: " + iter.hasNext());
+ }
+ }
+
void sync() throws IOException
{
Modified:
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java?rev=1152793&r1=1152792&r2=1152793&view=diff
==============================================================================
---
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
(original)
+++
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
Mon Aug 1 14:22:50 2011
@@ -23,8 +23,8 @@ package org.apache.cassandra.db.commitlo
import java.io.File;
import java.io.IOError;
import java.io.IOException;
-import java.util.HashSet;
-import java.util.Set;
+import java.util.HashMap;
+import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.zip.CRC32;
@@ -37,6 +37,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.RowMutation;
+import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
public class CommitLogSegment
@@ -48,7 +49,7 @@ public class CommitLogSegment
private final BufferedRandomAccessFile logWriter;
// cache which cf is dirty in this segment to avoid having to lookup all
ReplayPositions to decide if we could delete this segment
- private Set<Integer> cfDirty = new HashSet<Integer>();
+ private Map<Integer, Integer> cfLastWrite = new HashMap<Integer,
Integer>();
public CommitLogSegment()
{
@@ -102,6 +103,21 @@ public class CommitLogSegment
assert currentPosition <= Integer.MAX_VALUE;
ReplayPosition cLogCtx = new ReplayPosition(id, (int)
currentPosition);
+ for (ColumnFamily columnFamily : rowMutation.getColumnFamilies())
+ {
+ // check for null cfm in case a cl write goes through after
the cf is
+ // defined but before a new segment is created.
+ CFMetaData cfm =
DatabaseDescriptor.getCFMetaData(columnFamily.id());
+ if (cfm == null)
+ {
+ logger.error("Attempted to write commit log entry for
unrecognized column family: " + columnFamily.id());
+ }
+ else
+ {
+ turnOn(cfm.cfId, (int) currentPosition);
+ }
+ }
+
// write mutation, w/ checksum on the size and data
Checksum checksum = new CRC32();
byte[] serializedRow =
rowMutation.getSerializedBuffer(MessagingService.version_);
@@ -168,21 +184,32 @@ public class CommitLogSegment
}
}
- void turnOn(Integer cfId)
+ void turnOn(Integer cfId, Integer position)
+ {
+ cfLastWrite.put(cfId, position);
+ }
+
+ /**
+ * Turn the dirty bit off only if there has been no write since the flush
+ * position was grabbed.
+ */
+ void turnOffIfNotWritten(Integer cfId, Integer flushPosition)
{
- cfDirty.add(cfId);
+ Integer lastWritten = cfLastWrite.get(cfId);
+ if (lastWritten == null || lastWritten < flushPosition)
+ cfLastWrite.remove(cfId);
}
void turnOff(Integer cfId)
{
- cfDirty.remove(cfId);
+ cfLastWrite.remove(cfId);
}
// For debugging, not fast
String dirtyString()
{
StringBuilder sb = new StringBuilder();
- for (Integer cfId : cfDirty)
+ for (Integer cfId : cfLastWrite.keySet())
{
CFMetaData m = DatabaseDescriptor.getCFMetaData(cfId);
sb.append(m == null ? m.cfName : "<deleted>").append("
(").append(cfId).append("), ");
@@ -192,7 +219,7 @@ public class CommitLogSegment
boolean isSafeToDelete()
{
- return cfDirty.isEmpty();
+ return cfLastWrite.isEmpty();
}
@Override
Modified:
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CommitLogTest.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CommitLogTest.java?rev=1152793&r1=1152792&r2=1152793&view=diff
==============================================================================
---
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CommitLogTest.java
(original)
+++
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CommitLogTest.java
Mon Aug 1 14:22:50 2011
@@ -20,6 +20,7 @@
package org.apache.cassandra.db;
import java.io.*;
+import java.nio.ByteBuffer;
import java.util.zip.CRC32;
import java.util.zip.Checksum;
@@ -29,6 +30,7 @@ import org.apache.cassandra.CleanupHelpe
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.utils.Pair;
+import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
public class CommitLogTest extends CleanupHelper
{
@@ -87,6 +89,74 @@ public class CommitLogTest extends Clean
testRecoveryWithBadSizeArgument(-10, 10); // negative size, but no EOF
}
+ @Test
+ public void testDontDeleteIfDirty() throws Exception
+ {
+ CommitLog.instance.resetUnsafe();
+ // Roughly 32 MB mutation
+ RowMutation rm = new RowMutation("Keyspace1", bytes("k"));
+ rm.add(new QueryPath("Standard1", null, bytes("c1")),
ByteBuffer.allocate(32 * 1024 * 1024), 0);
+
+ // Adding it 5 times
+ CommitLog.instance.add(rm);
+ CommitLog.instance.add(rm);
+ CommitLog.instance.add(rm);
+ CommitLog.instance.add(rm);
+ CommitLog.instance.add(rm);
+
+ // Adding new mutation on another CF
+ RowMutation rm2 = new RowMutation("Keyspace1", bytes("k"));
+ rm2.add(new QueryPath("Standard2", null, bytes("c1")),
ByteBuffer.allocate(4), 0);
+ CommitLog.instance.add(rm2);
+
+ assert CommitLog.instance.segmentsCount() == 2 : "Expecting 2
segments, got " + CommitLog.instance.segmentsCount();
+
+ int cfid2 = rm2.getColumnFamilyIds().iterator().next();
+ CommitLog.instance.discardCompletedSegments(cfid2,
CommitLog.instance.getContext());
+
+ // Assert we still have both our segment
+ assert CommitLog.instance.segmentsCount() == 2 : "Expecting 2
segments, got " + CommitLog.instance.segmentsCount();
+ }
+
+ @Test
+ public void testDeleteIfNotDirty() throws Exception
+ {
+ CommitLog.instance.resetUnsafe();
+ // Roughly 32 MB mutation
+ RowMutation rm = new RowMutation("Keyspace1", bytes("k"));
+ rm.add(new QueryPath("Standard1", null, bytes("c1")),
ByteBuffer.allocate(32 * 1024 * 1024), 0);
+
+ // Adding it twice (won't change segment)
+ CommitLog.instance.add(rm);
+ CommitLog.instance.add(rm);
+
+ assert CommitLog.instance.segmentsCount() == 1 : "Expecting 1 segment,
got " + CommitLog.instance.segmentsCount();
+
+ // "Flush": this won't delete anything
+ int cfid1 = rm.getColumnFamilyIds().iterator().next();
+ CommitLog.instance.discardCompletedSegments(cfid1,
CommitLog.instance.getContext());
+
+ assert CommitLog.instance.segmentsCount() == 1 : "Expecting 1 segment,
got " + CommitLog.instance.segmentsCount();
+
+ // Adding new mutation on another CF so that a new segment is created
+ RowMutation rm2 = new RowMutation("Keyspace1", bytes("k"));
+ rm2.add(new QueryPath("Standard2", null, bytes("c1")),
ByteBuffer.allocate(64 * 1024 * 1024), 0);
+ CommitLog.instance.add(rm2);
+ CommitLog.instance.add(rm2);
+
+ assert CommitLog.instance.segmentsCount() == 2 : "Expecting 2
segments, got " + CommitLog.instance.segmentsCount();
+
+
+ // "Flush" second cf: The first segment should be deleted since we
+ // didn't write anything on cf1 since last flush (and we flush cf2)
+
+ int cfid2 = rm2.getColumnFamilyIds().iterator().next();
+ CommitLog.instance.discardCompletedSegments(cfid2,
CommitLog.instance.getContext());
+
+ // Assert we still have both our segment
+ assert CommitLog.instance.segmentsCount() == 1 : "Expecting 1 segment,
got " + CommitLog.instance.segmentsCount();
+ }
+
protected void testRecoveryWithBadSizeArgument(int size, int dataSize)
throws Exception
{
Checksum checksum = new CRC32();
