Repository: cassandra Updated Branches: refs/heads/cassandra-3.0 6a1b1f26b -> 5e57dd14e refs/heads/cassandra-3.11 59d4c2719 -> ed9b04d6a refs/heads/trunk d2dcd7f88 -> 652d9f64f
Potential AssertionError during ReadRepair of range tombstone and partition deletions patch by Sylvain Lebresne; reviewed by Branimir Lambov for CASSANDRA-13719 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/5e57dd14 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/5e57dd14 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/5e57dd14 Branch: refs/heads/cassandra-3.0 Commit: 5e57dd14eb37adf06d2105227e0105d871ea6f76 Parents: 6a1b1f2 Author: Sylvain Lebresne <sylv...@datastax.com> Authored: Fri Jul 21 16:58:53 2017 +0200 Committer: Sylvain Lebresne <sylv...@datastax.com> Committed: Thu Aug 24 11:28:57 2017 +0200 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../org/apache/cassandra/db/ReadResponse.java | 22 +++++ .../db/partitions/AbstractBTreePartition.java | 14 +-- .../db/partitions/PartitionUpdate.java | 9 ++ .../apache/cassandra/service/DataResolver.java | 70 +++++++++++++-- .../cassandra/service/DataResolverTest.java | 91 +++++++++++++++++++- 6 files changed, 192 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/5e57dd14/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 97dda05..2b49bc3 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.0.15 + * Potential AssertionError during ReadRepair of range tombstone and partition deletions (CASSANDRA-13719) * Don't let stress write warmup data if n=0 (CASSANDRA-13773) * Gossip thread slows down when using batch commit log (CASSANDRA-12966) * Randomize batchlog endpoint selection with only 1 or 2 racks (CASSANDRA-12884) http://git-wip-us.apache.org/repos/asf/cassandra/blob/5e57dd14/src/java/org/apache/cassandra/db/ReadResponse.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java index 693b52b..c59d00a 100644 --- a/src/java/org/apache/cassandra/db/ReadResponse.java +++ b/src/java/org/apache/cassandra/db/ReadResponse.java @@ -92,6 +92,28 @@ public abstract class ReadResponse public abstract boolean isDigestResponse(); + /** + * Creates a string of the requested partition in this read response suitable for debugging. + */ + public String toDebugString(ReadCommand command, DecoratedKey key) + { + if (isDigestResponse()) + return "Digest:0x" + ByteBufferUtil.bytesToHex(digest(command)); + + try (UnfilteredPartitionIterator iter = makeIterator(command)) + { + while (iter.hasNext()) + { + try (UnfilteredRowIterator partition = iter.next()) + { + if (partition.partitionKey().equals(key)) + return ImmutableBTreePartition.create(partition).toString(); + } + } + } + return "<key " + key + " not found>"; + } + protected static ByteBuffer makeDigest(UnfilteredPartitionIterator iterator, ReadCommand command) { MessageDigest digest = FBUtilities.threadLocalMD5Digest(); http://git-wip-us.apache.org/repos/asf/cassandra/blob/5e57dd14/src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java b/src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java index 2aa622e..1f3dbd0 100644 --- a/src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java +++ b/src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java @@ -99,7 +99,7 @@ public abstract class AbstractBTreePartition implements Partition, Iterable<Row> public DeletionTime partitionLevelDeletion() { - return holder().deletionInfo.getPartitionDeletion(); + return deletionInfo().getPartitionDeletion(); } public PartitionColumns columns() @@ -372,17 +372,21 @@ public abstract class AbstractBTreePartition implements Partition, Iterable<Row> { StringBuilder sb = new StringBuilder(); - sb.append(String.format("[%s.%s] key=%s columns=%s", + sb.append(String.format("[%s.%s] key=%s partition_deletion=%s columns=%s", metadata.ksName, metadata.cfName, metadata.getKeyValidator().getString(partitionKey().getKey()), + partitionLevelDeletion(), columns())); if (staticRow() != Rows.EMPTY_STATIC_ROW) - sb.append("\n ").append(staticRow().toString(metadata)); + sb.append("\n ").append(staticRow().toString(metadata, true)); - for (Row row : this) - sb.append("\n ").append(row.toString(metadata)); + try (UnfilteredRowIterator iter = unfilteredIterator()) + { + while (iter.hasNext()) + sb.append("\n ").append(iter.next().toString(metadata, true)); + } return sb.toString(); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/5e57dd14/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java index 2a881a3..7bd5345 100644 --- a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java +++ b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java @@ -299,6 +299,15 @@ public class PartitionUpdate extends AbstractBTreePartition return fromIterator(UnfilteredRowIterators.merge(asIterators, nowInSecs)); } + // We override this, because the version in the super-class calls holder(), which build the update preventing + // further updates, but that's not necessary here and being able to check at least the partition deletion without + // "locking" the update is nice (and used in DataResolver.RepairMergeListener.MergeListener). + @Override + public DeletionInfo deletionInfo() + { + return deletionInfo; + } + /** * Modify this update to set every timestamp for live data to {@code newTimestamp} and * every deletion timestamp to {@code newTimestamp - 1}. http://git-wip-us.apache.org/repos/asf/cassandra/blob/5e57dd14/src/java/org/apache/cassandra/service/DataResolver.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/DataResolver.java b/src/java/org/apache/cassandra/service/DataResolver.java index c96a893..26b1b2a 100644 --- a/src/java/org/apache/cassandra/service/DataResolver.java +++ b/src/java/org/apache/cassandra/service/DataResolver.java @@ -22,6 +22,8 @@ import java.util.*; import java.util.concurrent.TimeoutException; import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Joiner; +import com.google.common.collect.Iterables; import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.concurrent.StageManager; @@ -229,6 +231,17 @@ public class DataResolver extends ResponseResolver return repairs[i]; } + /** + * The partition level deletion with with which source {@code i} is currently repaired, or + * {@code DeletionTime.LIVE} if the source is not repaired on the partition level deletion (meaning it was + * up to date on it). The output* of this method is only valid after the call to + * {@link #onMergedPartitionLevelDeletion}. + */ + private DeletionTime partitionLevelRepairDeletion(int i) + { + return repairs[i] == null ? DeletionTime.LIVE : repairs[i].partitionLevelDeletion(); + } + private Row.Builder currentRow(int i, Clustering clustering) { if (currentRows[i] == null) @@ -273,6 +286,37 @@ public class DataResolver extends ResponseResolver public void onMergedRangeTombstoneMarkers(RangeTombstoneMarker merged, RangeTombstoneMarker[] versions) { + try + { + // The code for merging range tombstones is a tad complex and we had the assertions there triggered + // unexpectedly in a few occasions (CASSANDRA-13237, CASSANDRA-13719). It's hard to get insights + // when that happen without more context that what the assertion errors give us however, hence the + // catch here that basically gather as much as context as reasonable. + internalOnMergedRangeTombstoneMarkers(merged, versions); + } + catch (AssertionError e) + { + // The following can be pretty verbose, but it's really only triggered if a bug happen, so we'd + // rather get more info to debug than not. + CFMetaData table = command.metadata(); + String details = String.format("Error merging RTs on %s.%s: merged=%s, versions=%s, sources={%s}, responses:%n %s", + table.ksName, table.cfName, + merged == null ? "null" : merged.toString(table), + '[' + Joiner.on(", ").join(Iterables.transform(Arrays.asList(versions), rt -> rt == null ? "null" : rt.toString(table))) + ']', + Arrays.toString(sources), + makeResponsesDebugString()); + throw new AssertionError(details, e); + } + } + + private String makeResponsesDebugString() + { + return Joiner.on(",\n") + .join(Iterables.transform(getMessages(), m -> m.from + " => " + m.payload.toDebugString(command, partitionKey))); + } + + private void internalOnMergedRangeTombstoneMarkers(RangeTombstoneMarker merged, RangeTombstoneMarker[] versions) + { // The current deletion as of dealing with this marker. DeletionTime currentDeletion = currentDeletion(); @@ -297,21 +341,27 @@ public class DataResolver extends ResponseResolver // active after that point. Further whatever deletion was open or is open by this marker on the // source, that deletion cannot supersedes the current one. // - // But while the marker deletion (before and/or after this point) cannot supersed the current + // But while the marker deletion (before and/or after this point) cannot supersede the current // deletion, we want to know if it's equal to it (both before and after), because in that case // the source is up to date and we don't want to include repair. // // So in practice we have 2 possible case: - // 1) the source was up-to-date on deletion up to that point (markerToRepair[i] == null). Then - // it won't be from that point on unless it's a boundary and the new opened deletion time - // is also equal to the current deletion (note that this implies the boundary has the same - // closing and opening deletion time, which should generally not happen, but can due to legacy - // reading code not avoiding this for a while, see CASSANDRA-13237). - // 2) the source wasn't up-to-date on deletion up to that point (markerToRepair[i] != null), and - // it may now be (if it isn't we just have nothing to do for that marker). + // 1) the source was up-to-date on deletion up to that point: then it won't be from that point + // on unless it's a boundary and the new opened deletion time is also equal to the current + // deletion (note that this implies the boundary has the same closing and opening deletion + // time, which should generally not happen, but can due to legacy reading code not avoiding + // this for a while, see CASSANDRA-13237). + // 2) the source wasn't up-to-date on deletion up to that point and it may now be (if it isn't + // we just have nothing to do for that marker). assert !currentDeletion.isLive() : currentDeletion.toString(); - if (markerToRepair[i] == null) + // Is the source up to date on deletion? It's up to date if it doesn't have an open RT repair + // nor an "active" partition level deletion (where "active" means that it's greater or equal + // to the current deletion: if the source has a repaired partition deletion lower than the + // current deletion, this means the current deletion is due to a previously open range tombstone, + // and if the source isn't currently repaired for that RT, then it means it's up to date on it). + DeletionTime partitionRepairDeletion = partitionLevelRepairDeletion(i); + if (markerToRepair[i] == null && currentDeletion.supersedes(partitionRepairDeletion)) { // Since there is an ongoing merged deletion, the only way we don't have an open repair for // this source is that it had a range open with the same deletion as current and it's @@ -326,6 +376,8 @@ public class DataResolver extends ResponseResolver markerToRepair[i] = marker.closeBound(isReversed).invert(); } // In case 2) above, we only have something to do if the source is up-to-date after that point + // (which, since the source isn't up-to-date before that point, means we're opening a new deletion + // that is equal to the current one). else if (marker.isOpen(isReversed) && currentDeletion.equals(marker.openDeletionTime(isReversed))) { closeOpenMarker(i, marker.openBound(isReversed).invert()); http://git-wip-us.apache.org/repos/asf/cassandra/blob/5e57dd14/test/unit/org/apache/cassandra/service/DataResolverTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/service/DataResolverTest.java b/test/unit/org/apache/cassandra/service/DataResolverTest.java index 2f72093..65e18ce 100644 --- a/test/unit/org/apache/cassandra/service/DataResolverTest.java +++ b/test/unit/org/apache/cassandra/service/DataResolverTest.java @@ -574,7 +574,7 @@ public class DataResolverTest * same deletion on both side (while is useless but could be created by legacy code pre-CASSANDRA-13237 and could * thus still be sent). */ - public void testRepairRangeTombstoneBoundary(int timestamp1, int timestamp2, int timestamp3) throws UnknownHostException + private void testRepairRangeTombstoneBoundary(int timestamp1, int timestamp2, int timestamp3) throws UnknownHostException { DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2); InetAddress peer1 = peer(); @@ -621,6 +621,95 @@ public class DataResolverTest assertRepairContainsDeletions(msg, null, expected); } + /** + * Test for CASSANDRA-13719: tests that having a partition deletion shadow a range tombstone on another source + * doesn't trigger an assertion error. + */ + @Test + public void testRepairRangeTombstoneWithPartitionDeletion() + { + DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2); + InetAddress peer1 = peer(); + InetAddress peer2 = peer(); + + // 1st "stream": just a partition deletion + UnfilteredPartitionIterator iter1 = iter(PartitionUpdate.fullPartitionDelete(cfm, dk, 10, nowInSec)); + + // 2nd "stream": a range tombstone that is covered by the 1st stream + RangeTombstone rt = tombstone("0", true , "10", true, 5, nowInSec); + UnfilteredPartitionIterator iter2 = iter(new RowUpdateBuilder(cfm, nowInSec, 1L, dk) + .addRangeTombstone(rt) + .buildUpdate()); + + resolver.preprocess(readResponseMessage(peer1, iter1)); + resolver.preprocess(readResponseMessage(peer2, iter2)); + + // No results, we've only reconciled tombstones. + try (PartitionIterator data = resolver.resolve()) + { + assertFalse(data.hasNext()); + // 2nd stream should get repaired + assertRepairFuture(resolver, 1); + } + + assertEquals(1, messageRecorder.sent.size()); + + MessageOut msg = getSentMessage(peer2); + assertRepairMetadata(msg); + assertRepairContainsNoColumns(msg); + + assertRepairContainsDeletions(msg, new DeletionTime(10, nowInSec)); + } + + /** + * Additional test for CASSANDRA-13719: tests the case where a partition deletion doesn't shadow a range tombstone. + */ + @Test + public void testRepairRangeTombstoneWithPartitionDeletion2() + { + DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2); + InetAddress peer1 = peer(); + InetAddress peer2 = peer(); + + // 1st "stream": a partition deletion and a range tombstone + RangeTombstone rt1 = tombstone("0", true , "9", true, 11, nowInSec); + PartitionUpdate upd1 = new RowUpdateBuilder(cfm, nowInSec, 1L, dk) + .addRangeTombstone(rt1) + .buildUpdate(); + ((MutableDeletionInfo)upd1.deletionInfo()).add(new DeletionTime(10, nowInSec)); + UnfilteredPartitionIterator iter1 = iter(upd1); + + // 2nd "stream": a range tombstone that is covered by the other stream rt + RangeTombstone rt2 = tombstone("2", true , "3", true, 11, nowInSec); + RangeTombstone rt3 = tombstone("4", true , "5", true, 10, nowInSec); + UnfilteredPartitionIterator iter2 = iter(new RowUpdateBuilder(cfm, nowInSec, 1L, dk) + .addRangeTombstone(rt2) + .addRangeTombstone(rt3) + .buildUpdate()); + + resolver.preprocess(readResponseMessage(peer1, iter1)); + resolver.preprocess(readResponseMessage(peer2, iter2)); + + // No results, we've only reconciled tombstones. + try (PartitionIterator data = resolver.resolve()) + { + assertFalse(data.hasNext()); + // 2nd stream should get repaired + assertRepairFuture(resolver, 1); + } + + assertEquals(1, messageRecorder.sent.size()); + + MessageOut msg = getSentMessage(peer2); + assertRepairMetadata(msg); + assertRepairContainsNoColumns(msg); + + // 2nd stream should get both the partition deletion, as well as the part of the 1st stream RT that it misses + assertRepairContainsDeletions(msg, new DeletionTime(10, nowInSec), + tombstone("0", true, "2", false, 11, nowInSec), + tombstone("3", false, "9", true, 11, nowInSec)); + } + // Forces the start to be exclusive if the condition holds private static RangeTombstone withExclusiveStartIf(RangeTombstone rt, boolean condition) { --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org For additional commands, e-mail: commits-h...@cassandra.apache.org
