diff --git a/CHANGES.txt b/CHANGES.txt index 58a0906ae6..0a8b9b921f 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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 diff --git a/src/java/org/apache/cassandra/db/ConsistencyLevel.java b/src/java/org/apache/cassandra/db/ConsistencyLevel.java index 0f6aba78ec..6d04314a71 100644 --- a/src/java/org/apache/cassandra/db/ConsistencyLevel.java +++ b/src/java/org/apache/cassandra/db/ConsistencyLevel.java @@ -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 entry : countPerDCEndpoints(keyspace, liveEndpoints).entrySet()) + if (keyspace.getReplicationStrategy() instanceof NetworkTopologyStrategy) { - if (entry.getValue() < localQuorumFor(keyspace, entry.getKey())) - return false; + for (Map.Entry 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 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 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"); diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index f4a2662346..df33813701 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -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); } diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index ccabad51d0..260dcb2baa 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -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();