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

revert incomplete BMT-excision 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1127179 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-05-24 17:43:32 +00:00
parent c6aa109459
commit a2665b174c
8 changed files with 589 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

293
examples/bmt/CassandraBulkLoader.java Normal file
View File

@ -0,0 +1,293 @@
/*
* 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.
*/
/**
* Cassandra has a back door called the Binary Memtable. The purpose of this backdoor is to
* mass import large amounts of data, without using the Thrift interface.
*
* Inserting data through the binary memtable, allows you to skip the commit log overhead, and an ack
* from Thrift on every insert. The example below utilizes Hadoop to generate all the data necessary
* to send to Cassandra, and sends it using the Binary Memtable interface. What Hadoop ends up doing is
* creating the actual data that gets put into an SSTable as if you were using Thrift. With enough Hadoop nodes
* inserting the data, the bottleneck at this point should become the network.
*
* We recommend adjusting the compaction threshold to 0, while the import is running. After the import, you need
* to run `nodeprobe -host <IP> flush_binary <Keyspace>` on every node, as this will flush the remaining data still left
* in memory to disk. Then it's recommended to adjust the compaction threshold to it's original value.
*
* The example below is a sample Hadoop job, and it inserts SuperColumns. It can be tweaked to work with normal Columns.
*
* You should construct your data you want to import as rows delimited by a new line. You end up grouping by <Key>
* in the mapper, so that the end result generates the data set into a column oriented subset. Once you get to the
* reduce aspect, you can generate the ColumnFamilies you want inserted, and send it to your nodes.
*
* For Cassandra 0.6.4, we modified this example to wait for acks from all Cassandra nodes for each row
* before proceeding to the next. This means to keep Cassandra similarly busy you can either
* 1) add more reducer tasks,
* 2) remove the "wait for acks" block of code,
* 3) parallelize the writing of rows to Cassandra, e.g. with an Executor.
*
* THIS CANNOT RUN ON THE SAME IP ADDRESS AS A CASSANDRA INSTANCE.
*/
package org.apache.cassandra.bulkloader;
import java.io.IOException;
import java.net.InetAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import com.google.common.base.Charsets;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Column;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.db.RowMutation;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.net.IAsyncResult;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;
public class CassandraBulkLoader {
public static class Map extends MapReduceBase implements Mapper<Text, Text, Text, Text> {
public void map(Text key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
// This is a simple key/value mapper.
output.collect(key, value);
}
}
public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
private Path[] localFiles;
private JobConf jobconf;
public void configure(JobConf job) {
this.jobconf = job;
String cassConfig;
// Get the cached files
try
{
localFiles = DistributedCache.getLocalCacheFiles(job);
}
catch (IOException e)
{
throw new RuntimeException(e);
}
cassConfig = localFiles[0].getParent().toString();
System.setProperty("storage-config",cassConfig);
try
{
StorageService.instance.initClient();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
try
{
Thread.sleep(10*1000);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
}
public void close()
{
try
{
// release the cache
DistributedCache.releaseCache(new URI("/cassandra/storage-conf.xml#storage-conf.xml"), this.jobconf);
}
catch (IOException e)
{
throw new RuntimeException(e);
}
catch (URISyntaxException e)
{
throw new RuntimeException(e);
}
try
{
// Sleep just in case the number of keys we send over is small
Thread.sleep(3*1000);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
StorageService.instance.stopClient();
}
public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException
{
ColumnFamily columnFamily;
String keyspace = "Keyspace1";
String cfName = "Super1";
Message message;
List<ColumnFamily> columnFamilies;
columnFamilies = new LinkedList<ColumnFamily>();
String line;
/* Create a column family */
columnFamily = ColumnFamily.create(keyspace, cfName);
while (values.hasNext()) {
// Split the value (line based on your own delimiter)
line = values.next().toString();
String[] fields = line.split("\1");
String SuperColumnName = fields[1];
String ColumnName = fields[2];
String ColumnValue = fields[3];
int timestamp = 0;
columnFamily.addColumn(new QueryPath(cfName,
ByteBufferUtil.bytes(SuperColumnName),
ByteBufferUtil.bytes(ColumnName)),
ByteBufferUtil.bytes(ColumnValue),
timestamp);
}
columnFamilies.add(columnFamily);
/* Get serialized message to send to cluster */
message = createMessage(keyspace, key.getBytes(), cfName, columnFamilies);
List<IAsyncResult> results = new ArrayList<IAsyncResult>();
for (InetAddress endpoint: StorageService.instance.getNaturalEndpoints(keyspace, ByteBufferUtil.bytes(key)))
{
/* Send message to end point */
results.add(MessagingService.instance().sendRR(message, endpoint));
}
/* wait for acks */
for (IAsyncResult result : results)
{
try
{
result.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
}
catch (TimeoutException e)
{
// you should probably add retry logic here
throw new RuntimeException(e);
}
}
output.collect(key, new Text(" inserted into Cassandra node(s)"));
}
}
public static void runJob(String[] args)
{
JobConf conf = new JobConf(CassandraBulkLoader.class);
if(args.length >= 4)
{
conf.setNumReduceTasks(new Integer(args[3]));
}
try
{
// We store the cassandra storage-conf.xml on the HDFS cluster
DistributedCache.addCacheFile(new URI("/cassandra/storage-conf.xml#storage-conf.xml"), conf);
}
catch (URISyntaxException e)
{
throw new RuntimeException(e);
}
conf.setInputFormat(KeyValueTextInputFormat.class);
conf.setJobName("CassandraBulkLoader_v2");
conf.setMapperClass(Map.class);
conf.setReducerClass(Reduce.class);
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(Text.class);
FileInputFormat.setInputPaths(conf, new Path(args[1]));
FileOutputFormat.setOutputPath(conf, new Path(args[2]));
try
{
JobClient.runJob(conf);
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
public static Message createMessage(String keyspace, byte[] key, String columnFamily, List<ColumnFamily> columnFamilies)
{
ColumnFamily baseColumnFamily;
DataOutputBuffer bufOut = new DataOutputBuffer();
RowMutation rm;
Message message;
Column column;
/* Get the first column family from list, this is just to get past validation */
baseColumnFamily = new ColumnFamily(ColumnFamilyType.Standard,
DatabaseDescriptor.getComparator(keyspace, columnFamily),
DatabaseDescriptor.getSubComparator(keyspace, columnFamily),
CFMetaData.getId(keyspace, columnFamily));
for(ColumnFamily cf : columnFamilies) {
bufOut.reset();
ColumnFamily.serializer().serializeWithIndexes(cf, bufOut);
byte[] data = new byte[bufOut.getLength()];
System.arraycopy(bufOut.getData(), 0, data, 0, bufOut.getLength());
column = new Column(FBUtilities.toByteBuffer(cf.id()), ByteBuffer.wrap(data), 0);
baseColumnFamily.addColumn(column);
}
rm = new RowMutation(keyspace, ByteBuffer.wrap(key));
rm.add(baseColumnFamily);
try
{
/* Make message */
message = rm.makeRowMutationMessage(StorageService.Verb.BINARY, MessagingService.version_);
}
catch (IOException e)
{
throw new RuntimeException(e);
}
return message;
}
public static void main(String[] args) throws Exception
{
runJob(args);
}
}

34
examples/bmt/README.txt Normal file
View File

@ -0,0 +1,34 @@
This is an example for the deprecated BinaryMemtable bulk-load interface.
Inserting data through the binary memtable, allows you to skip the
commit log overhead, and an ack from Thrift on every insert. The
example below utilizes Hadoop to generate all the data necessary to
send to Cassandra, and sends it using the Binary Memtable
interface. What Hadoop ends up doing is creating the actual data that
gets put into an SSTable as if you were using Thrift. With enough
Hadoop nodes inserting the data, the bottleneck at this point should
become the network.
We recommend adjusting the compaction threshold to 0 while the import
is running. After the import, you need to run `nodeprobe -host <IP>
flush_binary <Keyspace>` on every node, as this will flush the
remaining data still left in memory to disk. Then it's recommended to
adjust the compaction threshold to it's original value.
The example in CassandraBulkLoader.java is a sample Hadoop job that
inserts SuperColumns. It can be tweaked to work with normal Columns.
You should construct your data you want to import as rows delimited by
a new line. You end up grouping by <Key> in the mapper, so that
the end result generates the data set into a column oriented
subset. Once you get to the reduce aspect, you can generate the
ColumnFamilies you want inserted, and send it to your nodes.
For Cassandra 0.6.4, we modified this example to wait for acks from
all Cassandra nodes for each row before proceeding to the next. This
means to keep Cassandra similarly busy you can either
1) add more reducer tasks,
2) remove the "wait for acks" block of code,
3) parallelize the writing of rows to Cassandra, e.g. with an Executor.
THIS CANNOT RUN ON THE SAME IP ADDRESS AS A CASSANDRA INSTANCE.

2
src/java/org/apache/cassandra/config/Config.java
View File

@ -77,6 +77,8 @@ public class Config
public Boolean snapshot_before_compaction = false;
public Integer compaction_thread_priority = Thread.MIN_PRIORITY;
public Integer binary_memtable_throughput_in_mb = 256;
/* if the size of columns or super-columns are more than this, indexing will kick in */
public Integer column_index_size_in_kb = 64;
public Integer in_memory_compaction_limit_in_mb = 256;

5
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -937,6 +937,11 @@ public class DatabaseDescriptor
return conf.sliced_buffer_size_in_kb;
}
public static int getBMTThreshold()
{
return conf.binary_memtable_throughput_in_mb;
}
public static int getCompactionThreadPriority()
{
return conf.compaction_thread_priority;

160
src/java/org/apache/cassandra/db/BinaryMemtable.java Normal file
View File

@ -0,0 +1,160 @@
/**
* 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.
*/
package org.apache.cassandra.db;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.sstable.SSTableWriter;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.WrappedRunnable;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
public class BinaryMemtable implements IFlushable
{
private static final Logger logger = LoggerFactory.getLogger(BinaryMemtable.class);
private final int threshold = DatabaseDescriptor.getBMTThreshold() * 1024 * 1024;
private final AtomicInteger currentSize = new AtomicInteger(0);
/* Table and ColumnFamily name are used to determine the ColumnFamilyStore */
private boolean isFrozen = false;
private final Map<DecoratedKey, ByteBuffer> columnFamilies = new NonBlockingHashMap<DecoratedKey, ByteBuffer>();
/* Lock and Condition for notifying new clients about Memtable switches */
private final Lock lock = new ReentrantLock();
Condition condition;
private final IPartitioner partitioner = StorageService.getPartitioner();
private final ColumnFamilyStore cfs;
public BinaryMemtable(ColumnFamilyStore cfs)
{
this.cfs = cfs;
condition = lock.newCondition();
}
boolean isThresholdViolated()
{
return currentSize.get() >= threshold;
}
/*
* This version is used by the external clients to put data into
* the memtable. This version will respect the threshold and flush
* the memtable to disk when the size exceeds the threshold.
*/
void put(DecoratedKey key, ByteBuffer buffer)
{
if (isThresholdViolated())
{
lock.lock();
try
{
if (!isFrozen)
{
isFrozen = true;
cfs.submitFlush(this, new CountDownLatch(1), null);
cfs.switchBinaryMemtable(key, buffer);
}
else
{
cfs.applyBinary(key, buffer);
}
}
finally
{
lock.unlock();
}
}
else
{
resolve(key, buffer);
}
}
public boolean isClean()
{
return columnFamilies.isEmpty();
}
private void resolve(DecoratedKey key, ByteBuffer buffer)
{
columnFamilies.put(key, buffer);
currentSize.addAndGet(buffer.remaining() + key.key.remaining());
}
private List<DecoratedKey> getSortedKeys()
{
assert !columnFamilies.isEmpty();
logger.info("Sorting " + this);
List<DecoratedKey> keys = new ArrayList<DecoratedKey>(columnFamilies.keySet());
Collections.sort(keys);
return keys;
}
private SSTableReader writeSortedContents(List<DecoratedKey> sortedKeys, ReplayPosition context) throws IOException
{
logger.info("Writing " + this);
SSTableWriter writer = cfs.createFlushWriter(sortedKeys.size(), DatabaseDescriptor.getBMTThreshold(), context);
for (DecoratedKey key : sortedKeys)
{
ByteBuffer bytes = columnFamilies.get(key);
assert bytes.remaining() > 0;
writer.append(key, bytes);
}
SSTableReader sstable = writer.closeAndOpenReader();
logger.info("Completed flushing " + writer.getFilename());
return sstable;
}
public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer, final ReplayPosition context)
{
sorter.execute(new Runnable()
{
public void run()
{
final List<DecoratedKey> sortedKeys = getSortedKeys();
writer.execute(new WrappedRunnable()
{
public void runMayThrow() throws IOException
{
cfs.addSSTable(writeSortedContents(sortedKeys, context));
latch.countDown();
}
});
}
});
}
}

68
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -25,6 +25,7 @@ import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Pattern;
@ -71,10 +72,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
private static Logger logger = LoggerFactory.getLogger(ColumnFamilyStore.class);
/*
* maybeSwitchMemtable puts Memtable.getSortedContents on the writer executor. When the write is complete,
* submitFlush first puts [Binary]Memtable.getSortedContents on the flushSorter executor,
* which then puts the sorted results on the writer executor. This is because sorting is CPU-bound,
* and writing is disk-bound; we want to be able to do both at once. When the write is complete,
* we turn the writer into an SSTableReader and add it to ssTables_ where it is available for reads.
*
* There are two other things that maybeSwitchMemtable does.
* For BinaryMemtable that's about all that happens. For live Memtables there are two other things
* that switchMemtable does (which should be the only caller of submitFlush in this case).
* First, it puts the Memtable into memtablesPendingFlush, where it stays until the flush is complete
* and it's been added as an SSTableReader to ssTables_. Second, it adds an entry to commitLogUpdater
* that waits for the flush to complete, then calls onMemtableFlush. This allows multiple flushes
@ -82,6 +86,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
* which is necessary for replay in case of a restart since CommitLog assumes that when onMF is
* called, all data up to the given context has been persisted to SSTables.
*/
private static final ExecutorService flushSorter
= new JMXEnabledThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
StageManager.KEEPALIVE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(Runtime.getRuntime().availableProcessors()),
new NamedThreadFactory("FlushSorter"),
"internal");
private static final ExecutorService flushWriter
= new JMXEnabledThreadPoolExecutor(DatabaseDescriptor.getFlushWriters(),
StageManager.KEEPALIVE,
@ -123,6 +134,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
private final ConcurrentSkipListMap<ByteBuffer, ColumnFamilyStore> indexedColumns;
// TODO binarymemtable ops are not threadsafe (do they need to be?)
private AtomicReference<BinaryMemtable> binaryMemtable;
private LatencyTracker readStats = new LatencyTracker();
private LatencyTracker writeStats = new LatencyTracker();
@ -243,6 +257,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
this.keyCacheSaveInSeconds = new DefaultInteger(metadata.getKeyCacheSavePeriodInSeconds());
this.partitioner = partitioner;
fileIndexGenerator.set(generation);
binaryMemtable = new AtomicReference<BinaryMemtable>(new BinaryMemtable(this));
if (logger.isDebugEnabled())
logger.debug("Starting CFS {}", columnFamily);
@ -643,11 +658,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
final CountDownLatch latch = new CountDownLatch(icc.size());
for (ColumnFamilyStore cfs : icc)
{
Memtable memtable = cfs.data.switchMemtable();
logger.info("Enqueuing flush of {}", memtable);
memtable.flushAndSignal(latch, flushWriter, ctx);
}
submitFlush(cfs.data.switchMemtable(), latch, ctx);
// we marked our memtable as frozen as part of the concurrency control,
// so even if there was nothing to flush we need to switch it out
@ -688,6 +699,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
: DatabaseDescriptor.getCFMetaData(metadata.cfId) == null;
}
void switchBinaryMemtable(DecoratedKey key, ByteBuffer buffer)
{
binaryMemtable.set(new BinaryMemtable(this));
binaryMemtable.get().put(key, buffer);
}
public void forceFlushIfExpired()
{
if (getMemtableThreadSafe().isExpired())
@ -718,6 +735,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
future.get();
}
public void forceFlushBinary()
{
if (binaryMemtable.get().isClean())
return;
submitFlush(binaryMemtable.get(), new CountDownLatch(1), null);
}
public void updateRowCache(DecoratedKey key, ColumnFamily columnFamily)
{
if (rowCache.isPutCopying())
@ -768,6 +793,20 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return flushRequested ? mt : null;
}
/*
* Insert/Update the column family for this key. param @ lock - lock that
* Caller is responsible for acquiring Table.flusherLock!
* param @ lock - lock that needs to be used.
* needs to be used. param @ key - key for update/insert param @
* columnFamily - columnFamily changes
*/
void applyBinary(DecoratedKey key, ByteBuffer buffer)
{
long start = System.nanoTime();
binaryMemtable.get().put(key, buffer);
writeStats.addNano(System.nanoTime() - start);
}
public static ColumnFamily removeDeletedCF(ColumnFamily cf, int gcBefore)
{
// in case of a timestamp tie, tombstones get priority over non-tombstones.
@ -959,6 +998,21 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
/**
* submits flush sort on the flushSorter executor, which will in turn submit to flushWriter when sorted.
* TODO because our executors use CallerRunsPolicy, when flushSorter fills up, no writes will proceed
* because the next flush will start executing on the caller, mutation-stage thread that has the
* flush write lock held. (writes aquire this as a read lock before proceeding.)
* This is good, because it backpressures flushes, but bad, because we can't write until that last
* flushing thread finishes sorting, which will almost always be longer than any of the flushSorter threads proper
* (since, by definition, it started last).
*/
void submitFlush(IFlushable flushable, CountDownLatch latch, ReplayPosition context)
{
logger.info("Enqueuing flush of {}", flushable);
flushable.flushAndSignal(latch, flushSorter, flushWriter, context);
}
public long getMemtableColumnsCount()
{
return getMemtableThreadSafe().getOperations();

32
src/java/org/apache/cassandra/db/IFlushable.java Normal file
View File

@ -0,0 +1,32 @@
package org.apache.cassandra.db;
/*
*
* 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.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import org.apache.cassandra.db.commitlog.ReplayPosition;
public interface IFlushable
{
public void flushAndSignal(CountDownLatch condition, ExecutorService sorter, ExecutorService writer, ReplayPosition context);
}

4
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -46,7 +46,7 @@ import org.apache.cassandra.io.sstable.SSTableWriter;
import org.apache.cassandra.utils.WrappedRunnable;
import org.github.jamm.MemoryMeter;
public class Memtable implements Comparable<Memtable>
public class Memtable implements Comparable<Memtable>, IFlushable
{
private static final Logger logger = LoggerFactory.getLogger(Memtable.class);
@ -256,7 +256,7 @@ public class Memtable implements Comparable<Memtable>
return ssTable;
}
public void flushAndSignal(final CountDownLatch latch, ExecutorService writer, final ReplayPosition context)
public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer, final ReplayPosition context)
{
writer.execute(new WrappedRunnable()
{