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

merge from 2.0

This commit is contained in:
Jonathan Ellis 2014-01-08 20:54:51 -06:00
parent c2294aa21e de19f963ae
commit db07b20edb
4 changed files with 92 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -26,12 +26,14 @@
2.0.5
* Wait for gossip to settle before accepting client connections (CASSANDRA-4288)
* Delete unfinished compaction incrementally (CASSANDRA-6086)
* Allow specifying custom secondary index options in CQL3 (CASSANDRA-6480)
Merged from 1.2:
* fsync compression metadata (CASSANDRA-6531)
* Validate CF existence on execution for prepared statement (CASSANDRA-6535)
* Add ability to throttle batchlog replay (CASSANDRA-6550)
* Fix executing LOCAL_QUORUM with SimpleStrategy (CASSANDRA-6545)
2.0.4

59
src/java/org/apache/cassandra/db/ConsistencyLevel.java
View File

@ -89,9 +89,16 @@ public enum ConsistencyLevel
return codeIdx[code];
}
private int quorumFor(Keyspace keyspace)
{
return (keyspace.getReplicationStrategy().getReplicationFactor() / 2) + 1;
}
private int localQuorumFor(Keyspace keyspace, String dc)
{
return (((NetworkTopologyStrategy) keyspace.getReplicationStrategy()).getReplicationFactor(dc) / 2) + 1;
return (keyspace.getReplicationStrategy() instanceof NetworkTopologyStrategy)
? (((NetworkTopologyStrategy) keyspace.getReplicationStrategy()).getReplicationFactor(dc) / 2) + 1
: quorumFor(keyspace);
}
public int blockFor(Keyspace keyspace)
@ -108,17 +115,24 @@ public enum ConsistencyLevel
case THREE:
return 3;
case QUORUM:
return (keyspace.getReplicationStrategy().getReplicationFactor() / 2) + 1;
return quorumFor(keyspace);
case ALL:
return keyspace.getReplicationStrategy().getReplicationFactor();
case LOCAL_QUORUM:
return localQuorumFor(keyspace, DatabaseDescriptor.getLocalDataCenter());
case EACH_QUORUM:
NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) keyspace.getReplicationStrategy();
int n = 0;
for (String dc : strategy.getDatacenters())
n += localQuorumFor(keyspace, dc);
return n;
if (keyspace.getReplicationStrategy() instanceof NetworkTopologyStrategy)
{
NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) keyspace.getReplicationStrategy();
int n = 0;
for (String dc : strategy.getDatacenters())
n += localQuorumFor(keyspace, dc);
return n;
}
else
{
return quorumFor(keyspace);
}
default:
throw new UnsupportedOperationException("Invalid consistency level: " + toString());
}
@ -213,12 +227,16 @@ public enum ConsistencyLevel
case LOCAL_QUORUM:
return countLocalEndpoints(liveEndpoints) >= blockFor(keyspace);
case EACH_QUORUM:
for (Map.Entry<String, Integer> entry : countPerDCEndpoints(keyspace, liveEndpoints).entrySet())
if (keyspace.getReplicationStrategy() instanceof NetworkTopologyStrategy)
{
if (entry.getValue() < localQuorumFor(keyspace, entry.getKey()))
return false;
for (Map.Entry<String, Integer> entry : countPerDCEndpoints(keyspace, liveEndpoints).entrySet())
{
if (entry.getValue() < localQuorumFor(keyspace, entry.getKey()))
return false;
}
return true;
}
return true;
// Fallthough on purpose for SimpleStrategy
default:
return Iterables.size(liveEndpoints) >= blockFor(keyspace);
}
@ -251,14 +269,18 @@ public enum ConsistencyLevel
}
break;
case EACH_QUORUM:
for (Map.Entry<String, Integer> entry : countPerDCEndpoints(keyspace, liveEndpoints).entrySet())
if (keyspace.getReplicationStrategy() instanceof NetworkTopologyStrategy)
{
int dcBlockFor = localQuorumFor(keyspace, entry.getKey());
int dcLive = entry.getValue();
if (dcLive < dcBlockFor)
throw new UnavailableException(this, dcBlockFor, dcLive);
for (Map.Entry<String, Integer> entry : countPerDCEndpoints(keyspace, liveEndpoints).entrySet())
{
int dcBlockFor = localQuorumFor(keyspace, entry.getKey());
int dcLive = entry.getValue();
if (dcLive < dcBlockFor)
throw new UnavailableException(this, dcBlockFor, dcLive);
}
break;
}
break;
// Fallthough on purpose for SimpleStrategy
default:
int live = Iterables.size(liveEndpoints);
if (live < blockFor)
@ -285,9 +307,6 @@ public enum ConsistencyLevel
{
switch (this)
{
case EACH_QUORUM:
requireNetworkTopologyStrategy(keyspaceName);
break;
case SERIAL:
case LOCAL_SERIAL:
throw new InvalidRequestException("You must use conditional updates for serializable writes");

2
src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
View File

@ -134,7 +134,7 @@ public abstract class AbstractReplicationStrategy
// block for in this context will be localnodes block.
return new DatacenterWriteResponseHandler(naturalEndpoints, pendingEndpoints, consistency_level, getKeyspace(), callback, writeType);
}
else if (consistency_level == ConsistencyLevel.EACH_QUORUM)
else if (consistency_level == ConsistencyLevel.EACH_QUORUM && (this instanceof NetworkTopologyStrategy))
{
return new DatacenterSyncWriteResponseHandler(naturalEndpoints, pendingEndpoints, consistency_level, getKeyspace(), callback, writeType);
}

51
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -30,11 +30,14 @@ import javax.management.ObjectName;
import javax.management.StandardMBean;
import com.addthis.metrics.reporter.config.ReporterConfig;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.Uninterruptibles;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
@ -321,6 +324,8 @@ public class CassandraDaemon
}
}
waitForGossipToSettle();
// Thift
InetAddress rpcAddr = DatabaseDescriptor.getRpcAddress();
int rpcPort = DatabaseDescriptor.getRpcPort();
@ -446,6 +451,50 @@ public class CassandraDaemon
destroy();
}
private void waitForGossipToSettle()
{
int forceAfter = Integer.getInteger("cassandra.skip_wait_for_gossip_to_settle", -1);
if (forceAfter == 0)
{
return;
}
final int GOSSIP_SETTLE_MIN_WAIT_MS = 5000;
final int GOSSIP_SETTLE_POLL_INTERVAL_MS = 1000;
final int GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED = 3;
logger.info("waiting for gossip to settle before accepting client requests...");
Uninterruptibles.sleepUninterruptibly(GOSSIP_SETTLE_MIN_WAIT_MS, TimeUnit.MILLISECONDS);
int totalPolls = 0;
int numOkay = 0;
JMXEnabledThreadPoolExecutor gossipStage = (JMXEnabledThreadPoolExecutor)StageManager.getStage(Stage.GOSSIP);
while (numOkay < GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED)
{
Uninterruptibles.sleepUninterruptibly(GOSSIP_SETTLE_POLL_INTERVAL_MS, TimeUnit.MILLISECONDS);
long completed = gossipStage.getCompletedTasks();
long active = gossipStage.getActiveCount();
long pending = gossipStage.getPendingTasks();
totalPolls++;
if (active == 0 && pending == 0)
{
logger.debug("gossip looks settled. CompletedTasks: {}", completed);
numOkay++;
}
else
{
logger.info("gossip not settled after {} polls. Gossip Stage active/pending/completed: {}/{}/{}", totalPolls, active, pending, completed);
numOkay = 0;
}
if (forceAfter > 0 && totalPolls > forceAfter)
{
logger.warn("Gossip not settled but startup forced by cassandra.skip_wait_for_gossip_to_settle. Gossip Stage active/pending/completed: {}/{}/{}",
totalPolls, active, pending, completed);
break;
}
}
logger.info("gossip settled after {} extra polls; proceeding", totalPolls - GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED);
}
public static void stop(String[] args)
{
instance.deactivate();