From 32d7cb5066050ef6123f50a25c6e9b4c9e180ea0 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Wed, 8 Jan 2014 17:14:58 +0100 Subject: [PATCH 1/2] Fix execution of LOCAL_QUORUM queries with SimpleStrategy patch by alexliu68; reviewed by slebresne for CASSANDRA-6545 --- CHANGES.txt | 1 + .../apache/cassandra/db/ConsistencyLevel.java | 60 ++++++++++++------- .../locator/AbstractReplicationStrategy.java | 2 +- 3 files changed, 42 insertions(+), 21 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 5a8597718d..cba97d0c28 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -8,6 +8,7 @@ * 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) 1.2.13 diff --git a/src/java/org/apache/cassandra/db/ConsistencyLevel.java b/src/java/org/apache/cassandra/db/ConsistencyLevel.java index 4d72767f9d..3737c730be 100644 --- a/src/java/org/apache/cassandra/db/ConsistencyLevel.java +++ b/src/java/org/apache/cassandra/db/ConsistencyLevel.java @@ -88,9 +88,16 @@ public enum ConsistencyLevel return codeIdx[code]; } + private int quorumFor(Table table) + { + return (table.getReplicationStrategy().getReplicationFactor() / 2) + 1; + } + private int localQuorumFor(Table table, String dc) { - return (((NetworkTopologyStrategy) table.getReplicationStrategy()).getReplicationFactor(dc) / 2) + 1; + return (table.getReplicationStrategy() instanceof NetworkTopologyStrategy) + ? (((NetworkTopologyStrategy) table.getReplicationStrategy()).getReplicationFactor(dc) / 2) + 1 + : quorumFor(table); } public int blockFor(Table table) @@ -107,17 +114,24 @@ public enum ConsistencyLevel case THREE: return 3; case QUORUM: - return (table.getReplicationStrategy().getReplicationFactor() / 2) + 1; + return quorumFor(table); case ALL: return table.getReplicationStrategy().getReplicationFactor(); case LOCAL_QUORUM: return localQuorumFor(table, DatabaseDescriptor.getLocalDataCenter()); case EACH_QUORUM: - NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) table.getReplicationStrategy(); - int n = 0; - for (String dc : strategy.getDatacenters()) - n += localQuorumFor(table, dc); - return n; + if (table.getReplicationStrategy() instanceof NetworkTopologyStrategy) + { + NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) table.getReplicationStrategy(); + int n = 0; + for (String dc : strategy.getDatacenters()) + n += localQuorumFor(table, dc); + return n; + } + else + { + return quorumFor(table); + } default: throw new UnsupportedOperationException("Invalid consistency level: " + toString()); } @@ -208,16 +222,20 @@ public enum ConsistencyLevel // local hint is acceptable, and local node is always live return true; case LOCAL_ONE: - return countLocalEndpoints(liveEndpoints) >= 1; + return countLocalEndpoints(liveEndpoints) >= 1; case LOCAL_QUORUM: return countLocalEndpoints(liveEndpoints) >= blockFor(table); case EACH_QUORUM: - for (Map.Entry entry : countPerDCEndpoints(table, liveEndpoints).entrySet()) + if (table.getReplicationStrategy() instanceof NetworkTopologyStrategy) { - if (entry.getValue() < localQuorumFor(table, entry.getKey())) - return false; + for (Map.Entry entry : countPerDCEndpoints(table, liveEndpoints).entrySet()) + { + if (entry.getValue() < localQuorumFor(table, entry.getKey())) + return false; + } + return true; } - return true; + // Fallthough on purpose for SimpleStrategy default: return Iterables.size(liveEndpoints) >= blockFor(table); } @@ -250,14 +268,18 @@ public enum ConsistencyLevel } break; case EACH_QUORUM: - for (Map.Entry entry : countPerDCEndpoints(table, liveEndpoints).entrySet()) + if (table.getReplicationStrategy() instanceof NetworkTopologyStrategy) { - int dcBlockFor = localQuorumFor(table, entry.getKey()); - int dcLive = entry.getValue(); - if (dcLive < dcBlockFor) - throw new UnavailableException(this, dcBlockFor, dcLive); + for (Map.Entry entry : countPerDCEndpoints(table, liveEndpoints).entrySet()) + { + int dcBlockFor = localQuorumFor(table, 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) @@ -282,8 +304,6 @@ public enum ConsistencyLevel public void validateForWrite(String table) throws InvalidRequestException { - if(this == EACH_QUORUM) - requireNetworkTopologyStrategy(table); } public void validateCounterForWrite(CFMetaData metadata) throws InvalidRequestException diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index a48bec9d56..e4dd422ff8 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, getTable(), 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, getTable(), callback, writeType); } From de19f963aeed2752374d2f84c1b230f6cab253f1 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 8 Jan 2014 20:53:46 -0600 Subject: [PATCH 2/2] Wait for gossip to settle before accepting client connections patch by Chris Burroughs; reviewed by Tyler Hobbs and jbellis for CASSANDRA-4288 --- CHANGES.txt | 1 + .../cassandra/service/CassandraDaemon.java | 50 +++++++++++++++++++ 2 files changed, 51 insertions(+) diff --git a/CHANGES.txt b/CHANGES.txt index d0b63a02f7..e96a8e00c5 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 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: diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index d36b0db573..d497a38f2c 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -34,10 +34,14 @@ 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.apache.log4j.PropertyConfigurator; 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; @@ -365,6 +369,8 @@ public class CassandraDaemon } } + waitForGossipToSettle(); + // Thift InetAddress rpcAddr = DatabaseDescriptor.getRpcAddress(); int rpcPort = DatabaseDescriptor.getRpcPort(); @@ -489,6 +495,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();