From 71084e306eed1415b6f83bd5d05f4d7fc2d69949 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Mon, 7 Dec 2009 22:51:46 +0000 Subject: [PATCH] implement streaming repairs; repair-via-rangecommand TODO patch by Stu Hood; reviewed by jbellis for CASSANDRA-520 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@888171 13f79535-47bb-0310-9956-ffa450edef68 --- .../cassandra/service/AntiEntropyService.java | 144 ++++++++++++------ .../cassandra/service/StreamManager.java | 14 +- .../apache/cassandra/utils/MerkleTree.java | 19 ++- .../org/apache/cassandra/io/SSTableTest.java | 10 +- .../org/apache/cassandra/io/SSTableUtils.java | 77 +++++++++- .../apache/cassandra/io/StreamingTest.java | 2 +- ...ompactEndPointSerializationHelperTest.java | 42 ----- .../service/AntiEntropyServiceTest.java | 17 +-- .../cassandra/utils/MerkleTreeTest.java | 2 +- 9 files changed, 201 insertions(+), 126 deletions(-) delete mode 100644 test/unit/org/apache/cassandra/net/CompactEndPointSerializationHelperTest.java diff --git a/src/java/org/apache/cassandra/service/AntiEntropyService.java b/src/java/org/apache/cassandra/service/AntiEntropyService.java index 2209814ad2..d8feb5b5c7 100644 --- a/src/java/org/apache/cassandra/service/AntiEntropyService.java +++ b/src/java/org/apache/cassandra/service/AntiEntropyService.java @@ -27,9 +27,9 @@ import org.apache.cassandra.concurrent.SingleThreadedStage; import org.apache.cassandra.concurrent.StageManager; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.CompactionManager; +import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.Table; -import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.CompactionIterator.CompactedRow; @@ -37,7 +37,7 @@ import org.apache.cassandra.io.DataInputBuffer; import org.apache.cassandra.io.ICompactSerializer; import org.apache.cassandra.io.SSTable; import org.apache.cassandra.io.SSTableReader; -import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.io.Streaming; import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; @@ -47,7 +47,6 @@ import org.apache.cassandra.utils.LogUtil; import org.apache.cassandra.utils.MerkleTree; import org.apache.log4j.Logger; -import org.apache.commons.lang.StringUtils; import com.google.common.collect.Collections2; import com.google.common.base.Predicate; @@ -158,11 +157,12 @@ public class AntiEntropyService if (etrees == null) { // double check the creation - Cachetable probable = - new Cachetable(TREE_CACHE_LIFETIME); + Cachetable probable = new Cachetable(TREE_CACHE_LIFETIME); if ((etrees = trees.putIfAbsent(endpoint, probable)) == null) + { // created new store for this endpoint etrees = probable; + } } return etrees; } @@ -189,12 +189,16 @@ public class AntiEntropyService for (Map.Entry> entry : trees.entrySet()) { if (LOCAL.equals(entry.getKey())) + { // don't compare to ourself continue; + } MerkleTree remotetree = entry.getValue().remove(cf); if (remotetree == null) + { // no tree stored for this endpoint at the moment continue; + } differencers.add(new Differencer(cf, LOCAL, entry.getKey(), tree, remotetree)); } @@ -206,8 +210,10 @@ public class AntiEntropyService // we stored a remote tree: queue differencing for local tree MerkleTree localtree = cacheForEndpoint(LOCAL).get(cf); if (localtree != null) + { // compare immediately differencers.add(new Differencer(cf, LOCAL, endpoint, localtree, tree)); + } else { // cache for later comparison @@ -317,12 +323,11 @@ public class AntiEntropyService Validator(CFTuple cf, InetAddress initiator) { - this(cf, initiator, + this(cf, + initiator, // TODO: memory usage (maxsize) should either be tunable per // CF, globally, or as shared for all CFs in a cluster - new MerkleTree(DatabaseDescriptor.getPartitioner(), - MerkleTree.RECOMMENDED_DEPTH, - (int)Math.pow(2,15))); + new MerkleTree(DatabaseDescriptor.getPartitioner(), MerkleTree.RECOMMENDED_DEPTH, (int)Math.pow(2, 15))); } Validator(CFTuple cf, InetAddress initiator, MerkleTree tree) @@ -340,17 +345,20 @@ public class AntiEntropyService public void prepare() { - Predicate cfpred = new Predicate(){ + Predicate cfpred = new Predicate() + { public boolean apply(SSTable ss) { return cf.table.equals(ss.getTableName()) && cf.cf.equals(ss.getColumnFamilyName()); } - }; + }; List keys = SSTableReader.getIndexedDecoratedKeysFor(cfpred, DKPRED); if (keys.isEmpty()) + { // use an even tree distribution tree.init(); + } else { int numkeys = keys.size(); @@ -428,8 +436,7 @@ public class AntiEntropyService private MerkleTree.RowHash rowHash(CompactedRow row) { - byte[] rowhash = FBUtilities.hash("MD5", row.key.key.getBytes(), - row.buffer.getData()); + byte[] rowhash = FBUtilities.hash("MD5", row.key.key.getBytes(), row.buffer.getData()); return new MerkleTree.RowHash(row.key.token, rowhash); } @@ -445,8 +452,7 @@ public class AntiEntropyService while (ranges.hasNext()) { MerkleTree.TreeRange range = ranges.next(); - if (!ranges.hasNext() && !minrows.isEmpty() && - range.contains(tree.partitioner().getMinimumToken())) + if (!ranges.hasNext() && !minrows.isEmpty() && range.contains(tree.partitioner().getMinimumToken())) { // append rows with the minimum token into the last range rows.addAll(minrows); @@ -474,16 +480,17 @@ public class AntiEntropyService InetAddress local = FBUtilities.getLocalAddress(); StorageService ss = StorageService.instance(); - Collection neighbors = - Collections2.filter(ss.getNaturalEndpoints(ss.getLocalToken()), - Predicates.not(Predicates.equalTo(local))); + Collection neighbors = Collections2.filter(ss.getNaturalEndpoints(ss.getLocalToken()), + Predicates.not(Predicates.equalTo(local))); // cache the local tree aes.register(cf, local, tree); if (!local.equals(initiator)) + { // one of our neighbors initiated: broadcast the tree to all of them aes.notifyNeighbors(this, local, neighbors); + } // else: we initiated this validation session: wait for responses // return any old object @@ -532,7 +539,7 @@ public class AntiEntropyService public final InetAddress remote; public final MerkleTree ltree; public final MerkleTree rtree; - public final List differences; + public final List differences; public Differencer(CFTuple cf, InetAddress local, InetAddress remote, MerkleTree ltree, MerkleTree rtree) { @@ -541,7 +548,7 @@ public class AntiEntropyService this.remote = remote; this.ltree = ltree; this.rtree = rtree; - differences = new ArrayList(); + differences = new ArrayList(); } /** @@ -563,7 +570,7 @@ public class AntiEntropyService interesting.retainAll(ss.getRangesForEndPoint(remote)); // compare trees, and filter out uninteresting differences - for (Range diff : MerkleTree.difference(ltree, rtree)) + for (MerkleTree.TreeRange diff : MerkleTree.difference(ltree, rtree)) { for (Range localrange: interesting) { @@ -575,13 +582,71 @@ public class AntiEntropyService } } - // TODO: calculating a percentage here would be all kinds of awesome - logger.info("Found " + differences.size() + " differing ranges between local " + - local + " and remote " + remote + " endpoints for " + cf + "."); + // choose a repair method based on the significance of the difference + float difference = differenceFraction(); + try + { + if (difference == 0.0) + { + logger.debug("Endpoints " + local + " and " + remote + " are consistent for " + cf); + return; + } - // FIXME: trigger repairs! + if (difference < 0.05) + performRangeRepair(); + else + performStreamingRepair(); + } + catch(IOException e) + { + throw new RuntimeException(e); + } } + /** + * @return the fraction of the keyspace that is different, as represented by our + * list of different ranges. A range at depth 0 == 1.0, at depth 1 == 0.5, etc. + */ + float differenceFraction() + { + double fraction = 0.0; + for (MerkleTree.TreeRange diff : differences) + fraction += 1.0 / Math.pow(2, diff.depth); + return (float)fraction; + } + + /** + * Sends our list of differences to the remote endpoint using read + * repairs via the query API. + */ + void performRangeRepair() throws IOException + { + logger.info("Performing range read repair of " + differences.size() + " ranges for " + cf); + // FIXME + logger.debug("Finished range read repair for " + cf); + } + + /** + * Sends our list of differences to the remote endpoint using the + * Streaming API. + */ + void performStreamingRepair() throws IOException + { + logger.info("Performing streaming repair of " + differences.size() + " ranges to " + remote + " for " + cf); + ColumnFamilyStore cfstore = Table.open(cf.table).getColumnFamilyStore(cf.cf); + try + { + List ranges = new ArrayList(differences); + List sstables = CompactionManager.instance().submitAnti(cfstore, ranges, remote).get(); + Streaming.transferSSTables(remote, sstables, cf.table); + } + catch(Exception e) + { + throw new IOException("Streaming repair failed.", e); + } + logger.debug("Finished streaming repair to " + remote + " for " + cf); + } + public String toString() { return "#"; @@ -604,8 +669,7 @@ public class AntiEntropyService ByteArrayOutputStream bos = new ByteArrayOutputStream(); DataOutputStream dos = new DataOutputStream(bos); SERIALIZER.serialize(new CFTuple(table, cf), dos); - return new Message(FBUtilities.getLocalAddress(), AE_SERVICE_STAGE, - TREE_REQUEST_VERB, bos.toByteArray()); + return new Message(FBUtilities.getLocalAddress(), AE_SERVICE_STAGE, TREE_REQUEST_VERB, bos.toByteArray()); } catch(IOException e) { @@ -641,31 +705,27 @@ public class AntiEntropyService // check for cached local tree InetAddress local = FBUtilities.getLocalAddress(); - MerkleTree cached = - AntiEntropyService.instance().getCachedTree(request.table, - request.cf, - local); + MerkleTree cached = AntiEntropyService.instance().getCachedTree(request.table, request.cf, local); if (cached != null) { if (local.equals(message.getFrom())) + { // we are the requestor, and we already have a cached tree return; + } // respond immediately with the recently generated tree Validator valid = new Validator(request, message.getFrom(), cached); Message response = TreeResponseVerbHandler.makeVerb(local, valid); MessagingService.instance().sendOneWay(response, message.getFrom()); - logger.debug("Answered request from " + message.getFrom() + - " for " + request + " with cached tree."); + logger.debug("Answered request from " + message.getFrom() + " for " + request + " with cached tree."); return; } // trigger readonly-compaction - logger.debug("Queueing readonly compaction for request from " + - message.getFrom() + " for " + request); + logger.debug("Queueing readonly compaction for request from " + message.getFrom() + " for " + request); Table table = Table.open(request.table); - CompactionManager.instance().submitReadonly(table.getColumnFamilyStore(request.cf), - message.getFrom()); - } + CompactionManager.instance().submitReadonly(table.getColumnFamilyStore(request.cf), message.getFrom()); + } catch (Exception e) { logger.warn(LogUtil.throwableToString(e)); @@ -711,8 +771,7 @@ public class AntiEntropyService ObjectInputStream ois = new ObjectInputStream(dis); try { - Validator v = new Validator(cf, (InetAddress)ois.readObject(), - (MerkleTree)ois.readObject()); + Validator v = new Validator(cf, (InetAddress)ois.readObject(), (MerkleTree)ois.readObject()); return v; } catch(Exception e) @@ -731,9 +790,8 @@ public class AntiEntropyService { // deserialize the remote tree, and register it Validator rvalidator = this.deserialize(buffer); - AntiEntropyService.instance().register(rvalidator.cf, message.getFrom(), - rvalidator.tree); - } + AntiEntropyService.instance().register(rvalidator.cf, message.getFrom(), rvalidator.tree); + } catch (Exception e) { logger.warn(LogUtil.throwableToString(e)); diff --git a/src/java/org/apache/cassandra/service/StreamManager.java b/src/java/org/apache/cassandra/service/StreamManager.java index b9b564b6c4..961224f262 100644 --- a/src/java/org/apache/cassandra/service/StreamManager.java +++ b/src/java/org/apache/cassandra/service/StreamManager.java @@ -20,7 +20,10 @@ package org.apache.cassandra.service; import java.io.File; import java.io.IOException; -import java.util.*; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.net.InetAddress; @@ -40,15 +43,16 @@ public final class StreamManager { private static Logger logger_ = Logger.getLogger( StreamManager.class ); - private static Map streamManagers_ = new HashMap(); + private static ConcurrentMap streamManagers_ = new ConcurrentHashMap(); public static StreamManager instance(InetAddress to) { StreamManager streamManager = streamManagers_.get(to); if ( streamManager == null ) { - streamManager = new StreamManager(to); - streamManagers_.put(to, streamManager); + StreamManager possibleNew = new StreamManager(to); + if ((streamManager = streamManagers_.putIfAbsent(to, possibleNew)) == null) + streamManager = possibleNew; } return streamManager; } @@ -79,7 +83,7 @@ public final class StreamManager { File file = filesToStream_.get(0); if (logger_.isDebugEnabled()) - logger_.debug("Streaming file " + file + " ..."); + logger_.debug("Streaming " + file.length() + " length file " + file + " ..."); MessagingService.instance().stream(file.getAbsolutePath(), 0L, file.length(), FBUtilities.getLocalAddress(), to_); } } diff --git a/src/java/org/apache/cassandra/utils/MerkleTree.java b/src/java/org/apache/cassandra/utils/MerkleTree.java index 3eb0c49b1b..0e264e2cfc 100644 --- a/src/java/org/apache/cassandra/utils/MerkleTree.java +++ b/src/java/org/apache/cassandra/utils/MerkleTree.java @@ -168,11 +168,11 @@ public class MerkleTree implements Serializable * @param active Only ranges that intersect this range will be returned. * @return A list of the largest contiguous ranges where the given trees disagree. */ - public static List difference(MerkleTree ltree, MerkleTree rtree) + public static List difference(MerkleTree ltree, MerkleTree rtree) { - List diff = new ArrayList(); + List diff = new ArrayList(); Token mintoken = ltree.partitioner.getMinimumToken(); - Range active = new Range(mintoken, mintoken); + TreeRange active = new TreeRange(null, mintoken, mintoken, (byte)0, null); byte[] lhash = ltree.hash(active); byte[] rhash = rtree.hash(active); @@ -194,11 +194,11 @@ public class MerkleTree implements Serializable * Takes two trees and a range for which they have hashes, but are inconsistent. * @return FULLY_INCONSISTENT if active is inconsistent, PARTIALLY_INCONSISTENT if only a subrange is inconsistent. */ - static int differenceHelper(MerkleTree ltree, MerkleTree rtree, List diff, Range active) + static int differenceHelper(MerkleTree ltree, MerkleTree rtree, List diff, TreeRange active) { Token midpoint = ltree.partitioner().midpoint(active.left(), active.right()); - Range left = new Range(active.left(), midpoint); - Range right = new Range(midpoint, active.right()); + TreeRange left = new TreeRange(null, active.left(), midpoint, inc(active.depth), null); + TreeRange right = new TreeRange(null, midpoint, active.right(), inc(active.depth), null); byte[] lhash; byte[] rhash; @@ -471,7 +471,8 @@ public class MerkleTree implements Serializable * * NB: A TreeRange should not be returned by a public method unless the * parents of the range it represents are already invalidated, since it - * will allow someone to modify the hash. + * will allow someone to modify the hash. Alternatively, a TreeRange + * may be created with a null tree, indicating that it is read only. */ public static class TreeRange extends Range { @@ -489,6 +490,7 @@ public class MerkleTree implements Serializable public void hash(byte[] hash) { + assert tree != null : "Not intended for modification!"; hashable.hash(hash); } @@ -512,6 +514,7 @@ public class MerkleTree implements Serializable */ public void validate(PeekingIterator entries) { + assert tree != null : "Not intended for modification!"; assert hashable instanceof Leaf; byte[] roothash; try @@ -587,7 +590,7 @@ public class MerkleTree implements Serializable { StringBuilder buff = new StringBuilder("#").toString(); + return buff.append(">").toString(); } } diff --git a/test/unit/org/apache/cassandra/io/SSTableTest.java b/test/unit/org/apache/cassandra/io/SSTableTest.java index 11c058282a..5bb36f696c 100644 --- a/test/unit/org/apache/cassandra/io/SSTableTest.java +++ b/test/unit/org/apache/cassandra/io/SSTableTest.java @@ -43,8 +43,7 @@ public class SSTableTest extends CleanupHelper TreeMap map = new TreeMap(); map.put(key, bytes); - SSTableReader ssTable = SSTableUtils.writeSSTable("singlewrite", map, 1, - new OrderPreservingPartitioner(), 0.01); + SSTableReader ssTable = SSTableUtils.writeRawSSTable("table", "singlewrite", map); // verify verifySingle(ssTable, bytes, key); @@ -72,8 +71,7 @@ public class SSTableTest extends CleanupHelper } // write - SSTableReader ssTable = SSTableUtils.writeSSTable("manywrites", map, 1000, - new OrderPreservingPartitioner(), 0.01); + SSTableReader ssTable = SSTableUtils.writeRawSSTable("table", "manywrites", map); // verify verifyMany(ssTable, map); @@ -109,9 +107,7 @@ public class SSTableTest extends CleanupHelper } // write - SSTableReader ssTable = SSTableUtils.writeSSTable(ssname, map, 1000, - new OrderPreservingPartitioner(), 0.01); - + SSTableReader ssTable = SSTableUtils.writeRawSSTable("table", ssname, map); // verify Predicate cfpred; diff --git a/test/unit/org/apache/cassandra/io/SSTableUtils.java b/test/unit/org/apache/cassandra/io/SSTableUtils.java index f901b47094..d764e53bf0 100644 --- a/test/unit/org/apache/cassandra/io/SSTableUtils.java +++ b/test/unit/org/apache/cassandra/io/SSTableUtils.java @@ -23,24 +23,85 @@ import java.io.IOException; import java.util.Map; import java.util.SortedMap; +import java.util.Set; +import java.util.TreeMap; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.Column; +import org.apache.cassandra.db.ColumnFamily; +import org.apache.cassandra.db.Table; import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.service.StorageService; +/** + * TODO: These methods imitate Memtable.writeSortedKeys to some degree, but + * because it is so monolithic, we can't reuse much. + */ public class SSTableUtils { - public static File tempSSTableFileName(String cfname) throws IOException + // first configured table and cf + public static String TABLENAME; + public static String CFNAME; + static { - return File.createTempFile(cfname + "-", "-" + SSTable.TEMPFILE_MARKER + "-Data.db"); + try + { + TABLENAME = DatabaseDescriptor.getTables().get(0); + CFNAME = Table.open(TABLENAME).getColumnFamilies().iterator().next(); + } + catch(IOException e) + { + throw new RuntimeException(e); + } } - public static SSTableReader writeSSTable(String cfname, SortedMap entries, int expectedKeys, IPartitioner partitioner, double cacheFraction) throws IOException + public static File tempSSTableFile(String tablename, String cfname) throws IOException { - File f = tempSSTableFileName(cfname); - SSTableWriter writer = new SSTableWriter(f.getAbsolutePath(), expectedKeys, partitioner); - for (Map.Entry entry : entries.entrySet()) + File tempdir = File.createTempFile(tablename, cfname); + if(!tempdir.delete() || !tempdir.mkdir()) + throw new IOException("Temporary directory creation failed."); + tempdir.deleteOnExit(); + File tabledir = new File(tempdir, tablename); + tabledir.mkdir(); + tabledir.deleteOnExit(); + return File.createTempFile(cfname + "-", + "-" + SSTable.TEMPFILE_MARKER + "-Data.db", + tabledir); + } + + public static SSTableReader writeSSTable(Set keys) throws IOException + { + TreeMap map = new TreeMap(); + for (String key : keys) { - writer.append(writer.partitioner.decorateKey(entry.getKey()), entry.getValue()); + ColumnFamily cf = ColumnFamily.create(TABLENAME, CFNAME); + cf.addColumn(new Column(key.getBytes(), key.getBytes(), 0)); + map.put(key, cf); } - return writer.closeAndOpenReader(cacheFraction); + return writeSSTable(map); + } + + public static SSTableReader writeSSTable(SortedMap entries) throws IOException + { + TreeMap map = new TreeMap(); + for (Map.Entry entry : entries.entrySet()) + { + DataOutputBuffer buffer = new DataOutputBuffer(); + ColumnFamily.serializer().serializeWithIndexes(entry.getValue(), buffer); + map.put(entry.getKey(), buffer.getData()); + } + return writeRawSSTable(TABLENAME, CFNAME, map); + } + + public static SSTableReader writeRawSSTable(String tablename, String cfname, SortedMap entries) throws IOException + { + File f = tempSSTableFile(tablename, cfname); + SSTableWriter writer = new SSTableWriter(f.getAbsolutePath(), entries.size(), StorageService.getPartitioner()); + for (Map.Entry entry : entries.entrySet()) + writer.append(writer.partitioner.decorateKey(entry.getKey()), + entry.getValue()); + new File(writer.indexFilename()).deleteOnExit(); + new File(writer.filterFilename()).deleteOnExit(); + return writer.closeAndOpenReader(1.0); } } diff --git a/test/unit/org/apache/cassandra/io/StreamingTest.java b/test/unit/org/apache/cassandra/io/StreamingTest.java index e5118a0917..6eb68ceb40 100644 --- a/test/unit/org/apache/cassandra/io/StreamingTest.java +++ b/test/unit/org/apache/cassandra/io/StreamingTest.java @@ -43,7 +43,7 @@ public class StreamingTest extends CleanupHelper @Test public void testTransferTable() throws Exception { - StorageService.instance().start(); + StorageService.instance().initServer(); // write a temporary SSTable, but don't register it Set content = new HashSet(); diff --git a/test/unit/org/apache/cassandra/net/CompactEndPointSerializationHelperTest.java b/test/unit/org/apache/cassandra/net/CompactEndPointSerializationHelperTest.java deleted file mode 100644 index 84d62cdcc2..0000000000 --- a/test/unit/org/apache/cassandra/net/CompactEndPointSerializationHelperTest.java +++ /dev/null @@ -1,42 +0,0 @@ -package org.apache.cassandra.net; -/* - * - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - * - */ - - -import java.net.UnknownHostException; -import java.net.InetAddress; - -import org.junit.Test; - -import org.apache.cassandra.utils.FBUtilities; - -public class CompactEndPointSerializationHelperTest -{ - @Test - public void testSerialize() throws UnknownHostException - { - InetAddress ep = FBUtilities.getLocalAddress(); - byte[] bytes = ep.getAddress(); - System.out.println(bytes.length); - InetAddress ep2 = InetAddress.getByAddress(bytes); - System.out.println(ep2); - } -} diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java index 20a1496c74..89179be033 100644 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java +++ b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java @@ -91,18 +91,13 @@ public class AntiEntropyServiceTest extends CleanupHelper { Validator validator; - // open an SSTable to give us something to sample - TreeMap map = new TreeMap(); - for ( int i = 0; i < 1000; i++ ) - { - map.put(Integer.toString(i), "blah".getBytes()); - } - // write - SSTableReader ssTable = - SSTableUtils.writeSSTable(cfname, map, 1000, - StorageService.instance().getPartitioner(), 0.01); - tablename = ssTable.getTableName(); + List rms = new LinkedList(); + RowMutation rm; + rm = new RowMutation(tablename, "key1"); + rm.add(new QueryPath(cfname, null, "Column1".getBytes()), "asdf".getBytes(), 0); + rms.add(rm); + ColumnFamilyStoreUtils.writeColumnFamily(rms); // sample validator = new Validator(new CFTuple(tablename, cfname), LOCAL); diff --git a/test/unit/org/apache/cassandra/utils/MerkleTreeTest.java b/test/unit/org/apache/cassandra/utils/MerkleTreeTest.java index d727ac8206..4c9aedd552 100644 --- a/test/unit/org/apache/cassandra/utils/MerkleTreeTest.java +++ b/test/unit/org/apache/cassandra/utils/MerkleTreeTest.java @@ -527,7 +527,7 @@ public class MerkleTreeTest range.validate(new HIterator(range.right())); // trees should disagree for leftmost, (middle.left, rightmost.right] - List diffs = MerkleTree.difference(mt, mt2); + List diffs = MerkleTree.difference(mt, mt2); assertEquals(diffs + " contains wrong number of differences:", 2, diffs.size()); assertTrue(diffs.contains(leftmost)); assertTrue(diffs.contains(new Range(middle.left(), rightmost.right())));