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

merge from 1.0 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1222743 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-12-23 16:44:47 +00:00
parent 00b314d152 23e1aa4c04
commit 6c28169e16
13 changed files with 105 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGES.txt
View File

@ -32,6 +32,9 @@
* optimize memtable iteration during range scan (CASSANDRA-3638)
1.0.7
* attempt hint delivery every ten minutes, or when failure detector
notifies us that a node is back up, whichever comes first. hint
handoff throttle delay default changed to 1ms, from 50 (CASSANDRA-3554)
* add nodetool setstreamthroughput (CASSANDRA-3571)
* fix assertion when dropping a columnfamily with no sstables (CASSANDRA-3614)
* more efficient allocation of small bloom filters (CASSANDRA-3618)
@ -40,6 +43,7 @@
* stop thrift service in shutdown hook so we can quiesce MessagingService
(CASSANDRA-3335)
Merged from 0.8:
* avoid logging (harmless) exception when GC takes < 1ms (CASSANDRA-3656)
* prevent new nodes from thinking down nodes are up forever (CASSANDRA-3626)
* Flush non-cfs backed secondary indexes (CASSANDRA-3659)

4
conf/cassandra.yaml
View File

@ -26,8 +26,8 @@ hinted_handoff_enabled: true
# this defines the maximum amount of time a dead host will have hints
# generated. After it has been dead this long, hints will be dropped.
max_hint_window_in_ms: 3600000 # one hour
# Sleep this long after delivering each row or row fragment
hinted_handoff_throttle_delay_in_ms: 50
# Sleep this long after delivering each hint
hinted_handoff_throttle_delay_in_ms: 1
# authentication backend, implementing IAuthenticator; used to identify users
authenticator: org.apache.cassandra.auth.AllowAllAuthenticator

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

@ -169,6 +169,17 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
public void setCompactionStrategyClass(String compactionStrategyClass) throws ConfigurationException
{
metadata.compactionStrategyClass = CFMetaData.createCompactionStrategy(compactionStrategyClass);
maybeReloadCompactionStrategy();
}
public String getCompactionStrategyClass()
{
return metadata.compactionStrategyClass.getName();
}
private ColumnFamilyStore(Table table, String columnFamilyName, IPartitioner partitioner, int generation, CFMetaData metadata)
{
assert metadata != null : "null metadata for " + table + ":" + columnFamilyName;
@ -1267,7 +1278,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
* @return true if we found all keys we were looking for, otherwise false
*/
public List<Row> getRangeSlice(ByteBuffer superColumn, final AbstractBounds<RowPosition> range, int maxResults, IFilter columnFilter)
throws ExecutionException, InterruptedException
{
assert range instanceof Bounds
|| !((Range)range).isWrapAround() || range.right.isMinimum()

13
src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java
View File

@ -22,6 +22,8 @@ import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutionException;
import org.apache.cassandra.config.ConfigurationException;
/**
* The MBean interface for ColumnFamilyStore
*/
@ -193,6 +195,17 @@ public interface ColumnFamilyStoreMBean
*/
public void setMaximumCompactionThreshold(int threshold);
/**
* Sets the compaction strategy by class name
* @param className the name of the compaction strategy class
*/
public void setCompactionStrategyClass(String className) throws ConfigurationException;
/**
* Gets the compaction strategy class name
*/
public String getCompactionStrategyClass();
/**
* Disable automatic compaction.
*/

81
src/java/org/apache/cassandra/db/HintedHandOffManager.java
View File

@ -20,23 +20,26 @@ package org.apache.cassandra.db;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.collect.ImmutableSortedSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.filter.IFilter;
import org.apache.cassandra.db.filter.NamesQueryFilter;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.dht.IPartitioner;
@ -61,7 +64,7 @@ import org.cliffc.high_scale_lib.NonBlockingHashSet;
* (We have to use String keys for compatibility with OPP.)
* SuperColumns in these rows are the mutations to replay, with uuid names:
*
* <dest ip>: { // key
* <dest token>: { // key
* <uuid>: { // supercolumn
* mutation: <mutation> // subcolumn
* version: <mutation serialization version>
@ -96,7 +99,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
private final ExecutorService executor_ = new JMXEnabledThreadPoolExecutor("HintedHandoff", Thread.MIN_PRIORITY);
public HintedHandOffManager()
public void start()
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
@ -107,25 +110,23 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
{
throw new RuntimeException(e);
}
}
public void registerMBean()
{
logger_.debug("Created HHOM instance, registered MBean.");
Runnable runnable = new Runnable()
{
public void run()
{
scheduleAllDeliveries();
}
};
StorageService.optionalTasks.scheduleWithFixedDelay(runnable, 10, 10, TimeUnit.MINUTES);
}
private static boolean sendMutation(InetAddress endpoint, RowMutation mutation) throws IOException
private static void sendMutation(InetAddress endpoint, RowMutation mutation) throws TimeoutException
{
IWriteResponseHandler responseHandler = WriteResponseHandler.create(endpoint);
MessagingService.instance().sendRR(mutation, endpoint, responseHandler);
try
{
responseHandler.get();
}
catch (TimeoutException e)
{
return false;
}
responseHandler.get();
try
{
@ -135,8 +136,6 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
{
throw new AssertionError(e);
}
return true;
}
private static void deleteHint(ByteBuffer tokenBytes, ByteBuffer hintId, long timestamp) throws IOException
@ -226,7 +225,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
logger_.debug("schema for {} matches local schema", endpoint);
return waited;
}
private void deliverHintsToEndpoint(InetAddress endpoint) throws IOException, DigestMismatchException, InvalidRequestException, TimeoutException, InterruptedException
{
ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(HINTS_CF);
@ -275,12 +274,12 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
while (true)
{
QueryFilter filter = QueryFilter.getSliceFilter(epkey, new QueryPath(HINTS_CF), startColumn, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, PAGE_SIZE);
ColumnFamily hintColumnFamily = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(filter), Integer.MAX_VALUE);
if (pagingFinished(hintColumnFamily, startColumn))
ColumnFamily hintsPage = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(filter), Integer.MAX_VALUE);
if (pagingFinished(hintsPage, startColumn))
break;
page:
for (IColumn hint : hintColumnFamily.getSortedColumns())
for (IColumn hint : hintsPage.getSortedColumns())
{
startColumn = hint.name();
for (IColumn subColumn : hint.getSubColumns())
@ -305,14 +304,15 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(mutationColumn.value()));
RowMutation rm = RowMutation.serializer().deserialize(in, ByteBufferUtil.toInt(versionColumn.value()));
if (sendMutation(endpoint, rm))
try
{
sendMutation(endpoint, rm);
deleteHint(tokenBytes, hint.name(), hint.maxTimestamp());
rowsReplayed++;
}
else
catch (TimeoutException e)
{
logger_.info("Could not complete hinted handoff to " + endpoint);
logger_.info(String.format("Timed out replaying hints to %s; aborting further deliveries", endpoint));
break delivery;
}
}
@ -335,12 +335,37 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
rowsReplayed, endpoint));
}
/**
* Attempt delivery to any node for which we have hints. Necessary since we can generate hints even for
* nodes which are never officially down/failed.
*/
private void scheduleAllDeliveries()
{
if (logger_.isDebugEnabled())
logger_.debug("Started scheduleAllDeliveries");
ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(HINTS_CF);
IPartitioner p = StorageService.getPartitioner();
Range range = new Range(p.getMinimumToken(), p.getMinimumToken(), p);
IFilter filter = new NamesQueryFilter(ImmutableSortedSet.<ByteBuffer>of());
List<Row> rows = hintStore.getRangeSlice(null, range, Integer.MAX_VALUE, filter);
for (Row row : rows)
{
Token<?> token = StorageService.getPartitioner().getTokenFactory().fromByteArray(row.key.key);
InetAddress target = StorageService.instance.getTokenMetadata().getEndpoint(token);
scheduleHintDelivery(target);
}
if (logger_.isDebugEnabled())
logger_.debug("Finished scheduleAllDeliveries");
}
/*
* This method is used to deliver hints to a particular endpoint.
* When we learn that some endpoint is back up we deliver the data
* to him via an event driven mechanism.
*/
public void deliverHints(final InetAddress to)
public void scheduleHintDelivery(final InetAddress to)
{
logger_.debug("deliverHints to {}", to);
if (!queuedDeliveries.add(to))
@ -356,9 +381,9 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
executor_.execute(r);
}
public void deliverHints(String to) throws UnknownHostException
public void scheduleHintDelivery(String to) throws UnknownHostException
{
deliverHints(InetAddress.getByName(to));
scheduleHintDelivery(InetAddress.getByName(to));
}
public List<String> listEndpointsPendingHints()

8
src/java/org/apache/cassandra/db/HintedHandOffManagerMBean.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.db;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
@ -25,9 +26,9 @@ public interface HintedHandOffManagerMBean
{
/**
* Nuke all hints from this node to `ep`.
* @param epaddr String rep. of endpoint address to delete hints for, either ip address ("127.0.0.1") or hostname
* @param host String rep. of endpoint address to delete hints for, either ip address ("127.0.0.1") or hostname
*/
public void deleteHintsForEndpoint(final String epaddr);
public void deleteHintsForEndpoint(final String host);
/**
* List all the endpoints that this node has hints for.
@ -42,5 +43,8 @@ public interface HintedHandOffManagerMBean
* @return map of endpoint -> hint count
*/
public Map<String, Integer> countPendingHints();
/** force hint delivery to an endpoint **/
public void scheduleHintDelivery(String host) throws UnknownHostException;
}

2
src/java/org/apache/cassandra/db/RowMutation.java
View File

@ -112,7 +112,7 @@ public class RowMutation implements IMutation, MessageProducer
* The format is the following:
*
* HintsColumnFamily: { // cf
* <dest ip>: { // key
* <dest token>: { // key
* <uuid>: { // super-column
* table: <table> // columns
* key: <key>

5
src/java/org/apache/cassandra/io/compress/CompressionParameters.java
View File

@ -150,7 +150,10 @@ public class CompressionParameters
try
{
return 1024 * Integer.parseInt(chLengthKB);
int parsed = Integer.parseInt(chLengthKB);
if (parsed > Integer.MAX_VALUE / 1024)
throw new ConfigurationException("Value of " + CHUNK_LENGTH_KB + " is too large (" + parsed + ")");
return 1024 * parsed;
}
catch (NumberFormatException e)
{

5
src/java/org/apache/cassandra/net/IncomingTcpConnection.java
View File

@ -82,8 +82,9 @@ public class IncomingTcpConnection extends Thread
}
else
{
// streaming connections are per-session and have a fixed version. we can't do anything with a new-version stream connection, so drop it.
logger.error("Received untranslated stream from newer protocol version. Terminating connection!");
// streaming connections are per-session and have a fixed version. we can't do anything with a wrong-version stream connection, so drop it.
logger.error("Received stream using protocol version {} (my version {}). Terminating connection",
version, MessagingService.version_);
}
// We are done with this connection....
return;

3
src/java/org/apache/cassandra/service/GCInspector.java
View File

@ -99,8 +99,7 @@ public class GCInspector
if (previousTotal.equals(total))
continue;
gctimes.put(gc.getName(), total);
Long duration = total - previousTotal;
assert duration > 0;
Long duration = total - previousTotal; // may be zero for a really fast collection
Long previousCount = gccounts.get(gc.getName());
Long count = gc.getCollectionCount();

19
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -838,21 +838,10 @@ public class StorageProxy implements StorageProxyMBean
if (logger.isDebugEnabled())
logger.debug("local range slice");
ColumnFamilyStore cfs = Table.open(command.keyspace).getColumnFamilyStore(command.column_family);
try
{
rows.addAll(cfs.getRangeSlice(command.super_column,
range,
command.max_keys,
QueryFilter.getFilter(command.predicate, cfs.getComparator())));
}
catch (ExecutionException e)
{
throw new RuntimeException(e.getCause());
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
rows.addAll(cfs.getRangeSlice(command.super_column,
range,
command.max_keys,
QueryFilter.getFilter(command.predicate, cfs.getComparator())));
}
else
{

15
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -514,7 +514,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
MigrationManager.passiveAnnounce(Schema.instance.getVersion());
Gossiper.instance.addLocalApplicationState(ApplicationState.RELEASE_VERSION, valueFactory.releaseVersion());
HintedHandOffManager.instance.registerMBean();
HintedHandOffManager.instance.start();
if (DatabaseDescriptor.isAutoBootstrap()
&& DatabaseDescriptor.getSeeds().contains(FBUtilities.getBroadcastAddress())
@ -1494,7 +1494,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
public void onAlive(InetAddress endpoint, EndpointState state)
{
if (!isClientMode && getTokenMetadata().isMember(endpoint))
deliverHints(endpoint);
HintedHandOffManager.instance.scheduleHintDelivery(endpoint);
}
public void onRemove(InetAddress endpoint)
@ -1545,18 +1545,9 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
return map;
}
/**
* Deliver hints to the specified node when it has crashed
* and come back up/ marked as alive after a network partition
*/
public final void deliverHints(InetAddress endpoint)
{
HintedHandOffManager.instance.deliverHints(endpoint);
}
public final void deliverHints(String host) throws UnknownHostException
{
HintedHandOffManager.instance.deliverHints(host);
HintedHandOffManager.instance.scheduleHintDelivery(host);
}
public Token getLocalToken()

3
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -300,9 +300,6 @@ public interface StorageServiceMBean
*/
public void truncate(String keyspace, String columnFamily) throws UnavailableException, TimeoutException, IOException;
/** force hint delivery to an endpoint **/
public void deliverHints(String host) throws UnknownHostException;
/**
* given a list of tokens (representing the nodes in the cluster), returns
* a mapping from "token -> %age of cluster owned by that token"