diff --git a/CHANGES.txt b/CHANGES.txt index 91fe632a24..22f7157028 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -40,7 +40,10 @@ dev * fix wrapping-range queries on non-minimum token (CASSANDRA-1700) * truncate includes secondary indexes (CASSANDRA-1747) * retain reference to PendingFile sstables (CASSANDRA-1749) - * pluggable seed provider (CASSANDRA-1669) + * fix sstableimport regression (CASSANDRA-1753) + * fix for bootstrap when no non-system tables are defined (CASSANDRA-1732) + * handle replica unavailability in index scan (CASSANDRA-1755) + * fix service initialization order deadlock (CASSANDRA-1756) 0.7.0-beta3 diff --git a/src/java/org/apache/cassandra/avro/CassandraServer.java b/src/java/org/apache/cassandra/avro/CassandraServer.java index 133d9276ff..88ad9c8d51 100644 --- a/src/java/org/apache/cassandra/avro/CassandraServer.java +++ b/src/java/org/apache/cassandra/avro/CassandraServer.java @@ -1106,6 +1106,10 @@ public class CassandraServer implements Cassandra { { throw new TimedOutException(); } + catch (org.apache.cassandra.thrift.UnavailableException e) + { + throw newUnavailableException(); + } return avronateKeySlices(rows, column_parent, column_predicate); } diff --git a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java index 3c9973bd89..db3d15b24c 100644 --- a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java +++ b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java @@ -77,7 +77,7 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); try { - mbs.registerMBean(this, new ObjectName("org.apache.cassandra.db:type=DynamicEndpointSnitch")); + mbs.registerMBean(this, new ObjectName("org.apache.cassandra.db:type=DynamicEndpointSnitch,instance="+hashCode())); } catch (Exception e) { @@ -178,6 +178,8 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa private void updateScores() // this is expensive { + if (!StorageService.instance.isInitialized()) + return; if (!registered) { ILatencyPublisher handler = (ILatencyPublisher)MessagingService.instance.getVerbHandler(StorageService.Verb.REQUEST_RESPONSE); diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 9794b2b61b..a497137e12 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -646,7 +646,7 @@ public class StorageProxy implements StorageProxyMBean } public static List scan(String keyspace, String column_family, IndexClause index_clause, SlicePredicate column_predicate, ConsistencyLevel consistency_level) - throws IOException, TimeoutException + throws IOException, TimeoutException, UnavailableException { IPartitioner p = StorageService.getPartitioner(); @@ -665,7 +665,11 @@ public class StorageProxy implements StorageProxyMBean RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(keyspace, liveEndpoints); AbstractReplicationStrategy rs = Table.open(keyspace).replicationStrategy; QuorumResponseHandler> handler = rs.getQuorumResponseHandler(resolver, consistency_level); - // TODO bail early if live endpoints can't satisfy requested consistency level + + // bail early if live endpoints can't satisfy requested consistency level + if(handler.blockfor > liveEndpoints.size()) + throw new UnavailableException(); + IndexScanCommand command = new IndexScanCommand(keyspace, column_family, index_clause, column_predicate, range); Message message = command.getMessage(); for (InetAddress endpoint : liveEndpoints) diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 2edc5dd3e7..9e8de67bb9 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -196,11 +196,11 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe }}; + public static final RetryingScheduledThreadPoolExecutor scheduledTasks = new RetryingScheduledThreadPoolExecutor("ScheduledTasks"); + private static IPartitioner partitioner_ = DatabaseDescriptor.getPartitioner(); public static VersionedValue.VersionedValueFactory valueFactory = new VersionedValue.VersionedValueFactory(partitioner_); - - public static RetryingScheduledThreadPoolExecutor scheduledTasks = new RetryingScheduledThreadPoolExecutor("ScheduledTasks"); - + public static final StorageService instance = new StorageService(); public static IPartitioner getPartitioner() { @@ -246,11 +246,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe public void finishBootstrapping() { isBootstrapMode = false; - SystemTable.setBootstrapped(true); - setToken(getLocalToken()); - Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS, valueFactory.normal(getLocalToken())); logger_.info("Bootstrap/move completed! Now serving reads."); - setMode("Normal", false); } /** This method updates the local token on disk */ @@ -313,6 +309,11 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe MessagingService.shutdown(); StageManager.shutdownNow(); } + + public boolean isInitialized() + { + return initialized; + } public synchronized void initClient() throws IOException { @@ -393,6 +394,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe && !SystemTable.isBootstrapped()) logger_.info("This node will not auto bootstrap because it is configured to be a seed node."); + Token token; if (DatabaseDescriptor.isAutoBootstrap() && !(DatabaseDescriptor.getSeeds().contains(FBUtilities.getLocalAddress()) || SystemTable.isBootstrapped())) { @@ -406,25 +408,18 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe throw new UnsupportedOperationException(s); } setMode("Joining: getting bootstrap token", true); - Token token = BootStrapper.getBootstrapToken(tokenMetadata_, StorageLoadBalancer.instance.getLoadInfo()); + token = BootStrapper.getBootstrapToken(tokenMetadata_, StorageLoadBalancer.instance.getLoadInfo()); // don't bootstrap if there are no tables defined. if (DatabaseDescriptor.getNonSystemTables().size() > 0) { bootstrap(token); assert !isBootstrapMode; // bootstrap will block until finished } - else - { - isBootstrapMode = false; - SystemTable.setBootstrapped(true); - tokenMetadata_.updateNormalToken(token, FBUtilities.getLocalAddress()); - Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS, valueFactory.normal(token)); - setMode("Normal", false); - } + // else nothing to do, go directly to participating in ring } else { - Token token = SystemTable.getSavedToken(); + token = SystemTable.getSavedToken(); if (token == null) { String initialToken = DatabaseDescriptor.getInitialToken(); @@ -438,18 +433,18 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe token = partitioner_.getTokenFactory().fromString(initialToken); logger_.info("Saved token not found. Using " + token + " from configuration"); } - SystemTable.updateToken(token); } else { logger_.info("Using saved token " + token); } - SystemTable.setBootstrapped(true); - tokenMetadata_.updateNormalToken(token, FBUtilities.getLocalAddress()); - Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS, valueFactory.normal(token)); - setMode("Normal", false); } - + + SystemTable.setBootstrapped(true); // first startup is only chance to bootstrap + setToken(token); + Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS, valueFactory.normal(getLocalToken())); + setMode("Normal", false); + assert tokenMetadata_.sortedTokens().size() > 0; } diff --git a/src/java/org/apache/cassandra/tools/SSTableImport.java b/src/java/org/apache/cassandra/tools/SSTableImport.java index 372b2632b8..15c63f620f 100644 --- a/src/java/org/apache/cassandra/tools/SSTableImport.java +++ b/src/java/org/apache/cassandra/tools/SSTableImport.java @@ -95,9 +95,9 @@ public class SSTableImport JsonColumn col = new JsonColumn(c); QueryPath path = new QueryPath(cfm.cfName, null, ByteBuffer.wrap(hexToBytes(col.name))); if (col.isDeleted) { - cfamily.addColumn(path, ByteBuffer.wrap(hexToBytes(col.value)), col.timestamp); - } else { cfamily.addTombstone(path, ByteBuffer.wrap(hexToBytes(col.value)), col.timestamp); + } else { + cfamily.addColumn(path, ByteBuffer.wrap(hexToBytes(col.value)), col.timestamp); } } } @@ -125,9 +125,9 @@ public class SSTableImport JsonColumn col = new JsonColumn(c); QueryPath path = new QueryPath(cfm.cfName, superName, ByteBuffer.wrap(hexToBytes(col.name))); if (col.isDeleted) { - cfamily.addColumn(path, ByteBuffer.wrap(hexToBytes(col.value)), col.timestamp); - } else { cfamily.addTombstone(path, ByteBuffer.wrap(hexToBytes(col.value)), col.timestamp); + } else { + cfamily.addColumn(path, ByteBuffer.wrap(hexToBytes(col.value)), col.timestamp); } } diff --git a/test/conf/cassandra.yaml b/test/conf/cassandra.yaml index 14698028bd..2d619ec085 100644 --- a/test/conf/cassandra.yaml +++ b/test/conf/cassandra.yaml @@ -23,6 +23,7 @@ seed_provider: parameters: - seeds: "127.0.0.2" endpoint_snitch: org.apache.cassandra.locator.SimpleSnitch +dynamic_snitch: true request_scheduler: org.apache.cassandra.scheduler.RoundRobinScheduler request_scheduler_id: keyspace keyspaces: diff --git a/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java b/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java index c7eb9824c5..daeacf921d 100644 --- a/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java +++ b/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java @@ -19,10 +19,11 @@ package org.apache.cassandra.locator; +import java.io.IOException; import java.net.InetAddress; -import java.net.UnknownHostException; import java.util.ArrayList; +import org.apache.cassandra.service.StorageService; import org.junit.Test; import org.apache.cassandra.utils.FBUtilities; @@ -30,8 +31,10 @@ import org.apache.cassandra.utils.FBUtilities; public class DynamicEndpointSnitchTest { @Test - public void testSnitch() throws UnknownHostException, InterruptedException + public void testSnitch() throws InterruptedException, IOException { + // do this because SS needs to be initialized before DES can work properly. + StorageService.instance.initClient(); int sleeptime = 150; DynamicEndpointSnitch dsnitch = new DynamicEndpointSnitch(new SimpleSnitch()); InetAddress self = FBUtilities.getLocalAddress(); diff --git a/test/unit/org/apache/cassandra/service/InitClientTest.java b/test/unit/org/apache/cassandra/service/InitClientTest.java new file mode 100644 index 0000000000..025a775b02 --- /dev/null +++ b/test/unit/org/apache/cassandra/service/InitClientTest.java @@ -0,0 +1,37 @@ +package org.apache.cassandra.service; + +import org.junit.Test; + +import java.io.IOException; + +/** + * 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. + */ + + +public class InitClientTest // extends CleanupHelper +{ + @Test + public void testInitClientStartup() + { + try { + StorageService.instance.initClient(); + } catch (IOException ex) { + throw new AssertionError(ex.getMessage()); + } + } +} diff --git a/test/unit/org/apache/cassandra/tools/SSTableImportTest.java b/test/unit/org/apache/cassandra/tools/SSTableImportTest.java index 75e48c4313..719b8819d4 100644 --- a/test/unit/org/apache/cassandra/tools/SSTableImportTest.java +++ b/test/unit/org/apache/cassandra/tools/SSTableImportTest.java @@ -26,6 +26,7 @@ import java.util.Arrays; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamily; +import org.apache.cassandra.db.DeletedColumn; import org.apache.cassandra.db.IColumn; import org.apache.cassandra.db.filter.QueryFilter; import org.apache.cassandra.db.filter.QueryPath; @@ -55,6 +56,7 @@ public class SSTableImportTest extends SchemaLoader QueryFilter qf = QueryFilter.getNamesFilter(Util.dk("rowA"), new QueryPath("Standard1", null, null), ByteBufferUtil.bytes("colAA")); ColumnFamily cf = qf.getSSTableColumnIterator(reader).getColumnFamily(); assert cf.getColumn(ByteBufferUtil.bytes("colAA")).value().equals(ByteBuffer.wrap(hexToBytes("76616c4141"))); + assert !(cf.getColumn(ByteBufferUtil.bytes("colAA")) instanceof DeletedColumn); } @Test