diff --git a/.rat-excludes b/.rat-excludes index 953d3c06c4..1ab279b4d7 100644 --- a/.rat-excludes +++ b/.rat-excludes @@ -29,3 +29,4 @@ drivers/txpy/txcql/cassandra/* drivers/py/cql/cassandra/* doc/cql/CQL* build.properties.default +test/data/legacy-sstables/** diff --git a/CHANGES.txt b/CHANGES.txt index 925d8881c3..ee149c8e22 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -27,7 +27,12 @@ * more efficient allocation of small bloom filters (CASSANDRA-3618) +1.0.7 + * fix assertion when dropping a columnfamily with no sstables (CASSANDRA-3614) + + 1.0.6 + * (CQL) fix cqlsh support for replicate_on_write (CASSANDRA-3596) * fix adding to leveled manifest after streaming (CASSANDRA-3536) * filter out unavailable cipher suites when using encryption (CASSANDRA-3178) * (HADOOP) add old-style api support for CFIF and CFRR (CASSANDRA-2799) @@ -49,13 +54,20 @@ * add back partitioner to sstable metadata (CASSANDRA-3540) * fix NPE in get_count for counters (CASSANDRA-3601) Merged from 0.8: + * remove invalid assertion that table was opened before dropping it + (CASSANDRA-3580) + * range and index scans now only send requests to enough replicas to + satisfy requested CL + RR (CASSANDRA-3598) * use cannonical host for local node in nodetool info (CASSANDRA-3556) + * remove nonlocal DC write optimization since it only worked with + CL.ONE or CL.LOCAL_QUORUM (CASSANDRA-3577, 3585) * detect misuses of CounterColumnType (CASSANDRA-3422) * turn off string interning in json2sstable, take 2 (CASSANDRA-2189) * validate compression parameters on add/update of the ColumnFamily (CASSANDRA-3573) * Check for 0.0.0.0 is incorrect in CFIF (CASSANDRA-3584) * Increase vm.max_map_count in debian packaging (CASSANDRA-3563) + * gossiper will never add itself to saved endpoints (CASSANDRA-3485) 1.0.5 diff --git a/NEWS.txt b/NEWS.txt index e29ee88a0b..a21dbb95e5 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -33,15 +33,28 @@ Upgrading want to use such large batches. +1.0.6 +===== + +Upgrading +--------- + - This release fixes an issue related to the chunk_length_kb option for + compressed sstables. If you use compression on some column families, it + is recommended after the upgrade to check the value for this option on + these column families (the default value is 64). In case the option would + not be set correctly, you should update the column family definition, + setting the right value and then run scrub on the column family. + - Please report to instruction for 1.0.5 if coming from an older version. + + 1.0.5 ===== -JMX ---- - - A command has been added to stop running compaction. It is available - through JMX and through nodetool stop (see the nodetool help for - details). Please note that stopped compaction are terminated and cannot - be restarted afterwards. +Upgrading +--------- + - 1.0.5 comes to fix two important regression of 1.0.4. So all information + concerning 1.0.4 are valid for this release, but please avoids upgrading + to 1.0.4. 1.0.4 diff --git a/build.xml b/build.xml index d17f66d259..def3b33031 100644 --- a/build.xml +++ b/build.xml @@ -59,7 +59,6 @@ - @@ -285,9 +284,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} - - - @@ -969,22 +965,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} - - - - - - - - - - - - - @@ -1036,6 +1016,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} + @@ -1044,6 +1025,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} + @@ -1070,14 +1052,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} - - - - - - - - + diff --git a/debian/cassandra-sysctl.conf b/debian/cassandra-sysctl.conf new file mode 100644 index 0000000000..2173765615 --- /dev/null +++ b/debian/cassandra-sysctl.conf @@ -0,0 +1 @@ +vm.max_map_count = 1048575 diff --git a/debian/changelog b/debian/changelog index 4fc4714c4d..9047f19df5 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,15 @@ +cassandra (1.0.6) unstable; urgency=low + + * New release + + -- Sylvain Lebresne Sat, 10 Dec 2011 18:21:50 -0600 + +cassandra (1.0.5) unstable; urgency=low + + * New release + + -- Sylvain Lebresne Tue, 29 Nov 2011 19:36:09 +0100 + cassandra (1.0.4) unstable; urgency=low * New release diff --git a/pylib/cqlshlib/cqlhandling.py b/pylib/cqlshlib/cqlhandling.py index 2df80610c6..30c22fb9d9 100644 --- a/pylib/cqlshlib/cqlhandling.py +++ b/pylib/cqlshlib/cqlhandling.py @@ -37,7 +37,7 @@ columnfamily_options = ( ('max_compaction_threshold', None), ('row_cache_save_period_in_seconds', None), ('key_cache_save_period_in_seconds', None), - ('replication_on_write', 'replicate_on_write') + ('replicate_on_write', None) ) cql_type_to_apache_class = { diff --git a/src/java/org/apache/cassandra/db/DataTracker.java b/src/java/org/apache/cassandra/db/DataTracker.java index 8a541658c4..a710b272f5 100644 --- a/src/java/org/apache/cassandra/db/DataTracker.java +++ b/src/java/org/apache/cassandra/db/DataTracker.java @@ -292,6 +292,11 @@ public class DataTracker } while (!view.compareAndSet(currentView, newView)); + if (notCompacting.isEmpty()) + { + // notifySSTablesChanged -> LeveledManifest.promote doesn't like a no-op "promotion" + return; + } notifySSTablesChanged(notCompacting, Collections.emptySet()); postReplace(notCompacting, Collections.emptySet()); } diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java index b70f50fad0..57704773b8 100644 --- a/src/java/org/apache/cassandra/db/SystemTable.java +++ b/src/java/org/apache/cassandra/db/SystemTable.java @@ -138,6 +138,8 @@ public class SystemTable */ public static synchronized void updateToken(InetAddress ep, Token token) { + if (ep == FBUtilities.getLocalAddress()) + return; IPartitioner p = StorageService.getPartitioner(); ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, STATUS_CF); cf.addColumn(new Column(p.getTokenFactory().toByteArray(token), ByteBuffer.wrap(ep.getAddress()), System.currentTimeMillis())); diff --git a/src/java/org/apache/cassandra/db/migration/DropKeyspace.java b/src/java/org/apache/cassandra/db/migration/DropKeyspace.java index b8ea1238c3..00d9b220e4 100644 --- a/src/java/org/apache/cassandra/db/migration/DropKeyspace.java +++ b/src/java/org/apache/cassandra/db/migration/DropKeyspace.java @@ -63,8 +63,7 @@ public class DropKeyspace extends Migration } // remove the table from the static instances. - Table table = Table.clear(ksm.name, schema); - assert table != null; + Table.clear(ksm.name, schema); // reset defs. schema.clearTableDefinition(ksm, newVersion); } diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 92f61ff273..9e88a7867b 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -377,9 +377,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean * @param endpoint - the endpoint being removed * @param token - the token being removed * @param mytoken - my own token for replication coordination - * @param delay */ - public void advertiseRemoving(InetAddress endpoint, Token token, Token mytoken, int delay) + public void advertiseRemoving(InetAddress endpoint, Token token, Token mytoken) { EndpointState epState = endpointStateMap.get(endpoint); // remember this node's generation @@ -388,7 +387,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean logger.info("Sleeping for " + StorageService.RING_DELAY + "ms to ensure " + endpoint + " does not change"); try { - Thread.sleep(delay); + Thread.sleep(StorageService.RING_DELAY); } catch (InterruptedException e) { @@ -432,6 +431,66 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean } } + /** + * Do not call this method unless you know what you are doing. + * It will try extremely hard to obliterate any endpoint from the ring, + * even if it does not know about it. + * This should only ever be called by human via JMX. + * @param address + * @throws UnknownHostException + */ + public void unsafeAssassinateEndpoint(String address) throws UnknownHostException + { + InetAddress endpoint = InetAddress.getByName(address); + EndpointState epState = endpointStateMap.get(endpoint); + Token token = null; + logger.warn("Assassinating {} via gossip", endpoint); + if (epState == null) + { + epState = new EndpointState(new HeartBeatState((int)((System.currentTimeMillis() + 60000) / 1000), 9999)); + } + else + { + try + { + token = StorageService.instance.getTokenMetadata().getToken(endpoint); + } + catch (AssertionError e) + { + } + int generation = epState.getHeartBeatState().getGeneration(); + logger.info("Sleeping for " + StorageService.RING_DELAY + "ms to ensure " + endpoint + " does not change"); + try + { + Thread.sleep(StorageService.RING_DELAY); + } + catch (InterruptedException e) + { + throw new AssertionError(e); + } + // make sure it did not change + epState = endpointStateMap.get(endpoint); + if (epState.getHeartBeatState().getGeneration() != generation) + throw new RuntimeException("Endpoint " + endpoint + " generation changed while trying to remove it"); + epState.updateTimestamp(); // make sure we don't evict it too soon + epState.getHeartBeatState().forceNewerGenerationUnsafe(); + } + if (token == null) + token = StorageService.instance.getBootstrapToken(); + // do not pass go, do not collect 200 dollars, just gtfo + epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.left(token, computeExpireTime())); + handleMajorStateChange(endpoint, epState); + try + { + Thread.sleep(intervalInMillis * 4); + } + catch (InterruptedException e) + { + throw new AssertionError(e); + } + logger.warn("Finished killing {}", endpoint); + } + public boolean isKnownEndpoint(InetAddress endpoint) { return endpointStateMap.containsKey(endpoint); @@ -1035,6 +1094,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean */ public void addSavedEndpoint(InetAddress ep) { + if (ep == FBUtilities.getLocalAddress()) + { + logger.debug("Attempt to add self as saved endpoint"); + return; + } EndpointState epState = new EndpointState(new HeartBeatState(0)); epState.markDead(); epState.setHasToken(true); diff --git a/src/java/org/apache/cassandra/gms/GossiperMBean.java b/src/java/org/apache/cassandra/gms/GossiperMBean.java index 62e77eadad..574f4b83eb 100644 --- a/src/java/org/apache/cassandra/gms/GossiperMBean.java +++ b/src/java/org/apache/cassandra/gms/GossiperMBean.java @@ -28,4 +28,6 @@ public interface GossiperMBean public int getCurrentGenerationNumber(String address) throws UnknownHostException; + public void unsafeAssassinateEndpoint(String address) throws UnknownHostException; + } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 1ce96cb49f..1f29c1cbdc 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -862,7 +862,7 @@ public class StorageProxy implements StorageProxyMBean RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints); ReadCallback> handler = getReadCallback(resolver, command, consistency_level, liveEndpoints); handler.assureSufficientLiveNodes(); - for (InetAddress endpoint : liveEndpoints) + for (InetAddress endpoint : handler.endpoints) { MessagingService.instance().sendRR(c2, endpoint, handler); if (logger.isDebugEnabled()) @@ -1140,7 +1140,7 @@ public class StorageProxy implements StorageProxyMBean IndexScanCommand command = new IndexScanCommand(keyspace, column_family, index_clause, column_predicate, range); MessageProducer producer = new CachingMessageProducer(command); - for (InetAddress endpoint : liveEndpoints) + for (InetAddress endpoint : handler.endpoints) { MessagingService.instance().sendRR(producer, endpoint, handler); if (logger.isDebugEnabled()) diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 348e3c0871..ec4a7687c9 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -79,7 +79,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe { private static Logger logger_ = LoggerFactory.getLogger(StorageService.class); - public static final int RING_DELAY = 30 * 1000; // delay after which we assume ring has stablized + public static final int RING_DELAY = getRingDelay(); // delay after which we assume ring has stablized /* All verb handler identifiers */ public enum Verb @@ -150,6 +150,17 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe put(Verb.UNUSED_3, Stage.INTERNAL_RESPONSE); }}; + private static int getRingDelay() + { + String newdelay = System.getProperty("cassandra.ring_delay_ms"); + if (newdelay != null) + { + logger_.warn("Overriding RING_DELAY to {}ms", newdelay); + return Integer.parseInt(newdelay); + } + else + return 30 * 1000; + } /** * This pool is used for periodic short (sub-second) tasks. @@ -404,8 +415,16 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe logger_.info("Loading persisted ring state"); for (Map.Entry entry : SystemTable.loadTokens().entrySet()) { - tokenMetadata_.updateNormalToken(entry.getKey(), entry.getValue()); - Gossiper.instance.addSavedEndpoint(entry.getValue()); + if (entry.getValue() == FBUtilities.getLocalAddress()) + { + // entry has been mistakenly added, delete it + SystemTable.removeToken(entry.getKey()); + } + else + { + tokenMetadata_.updateNormalToken(entry.getKey(), entry.getValue()); + Gossiper.instance.addSavedEndpoint(entry.getValue()); + } } } @@ -2330,11 +2349,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe * @param tokenString token for the node */ public void removeToken(String tokenString) - { - removeToken(tokenString, RING_DELAY); - } - - public void removeToken(String tokenString, int delay) { InetAddress myAddress = FBUtilities.getBroadcastAddress(); Token localToken = tokenMetadata_.getToken(myAddress); @@ -2382,7 +2396,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe calculatePendingRanges(); // the gossiper will handle spoofing this node's state to REMOVING_TOKEN for us // we add our own token so other nodes to let us know when they're done - Gossiper.instance.advertiseRemoving(endpoint, token, localToken, delay); + Gossiper.instance.advertiseRemoving(endpoint, token, localToken); // kick off streaming commands restoreReplicaCount(endpoint, myAddress); diff --git a/test/cassandra.in.sh b/test/cassandra.in.sh index 47514f28b6..682327f7d4 100644 --- a/test/cassandra.in.sh +++ b/test/cassandra.in.sh @@ -50,4 +50,5 @@ JVM_OPTS=" \ -XX:+HeapDumpOnOutOfMemoryError \ -Dcom.sun.management.jmxremote.port=8090 \ -Dcom.sun.management.jmxremote.ssl=false \ - -Dcom.sun.management.jmxremote.authenticate=false" + -Dcom.sun.management.jmxremote.authenticate=false \ + -Dcassandra.ring_delay_ms=1000" diff --git a/test/distributed/README.txt b/test/distributed/README.txt deleted file mode 100644 index 651038b77c..0000000000 --- a/test/distributed/README.txt +++ /dev/null @@ -1,57 +0,0 @@ -Distributed Test Harness - - -Sub-project description ------------------------ - -A distributed test harness that deploys a cluster to a cloud provider, -via Apache Whirr, runs tests against that cluster, then tears down -the deployed cluster. - -Requirements ------------- - * A cloud provider account. [see: http://incubator.apache.org/whirr/] - - -Getting started ---------------- - -First, setup an account w/ a supported cloud provider. Then, refer to -the Whirr documentation for configuration instructions. Refer to: - * http://incubator.apache.org/whirr/quick-start-guide.html - -Setup your personal whirr configuration properties. The shared whirr -configuration is located at: - * test/resources/whirr-default.properties - -An example EC2/S3 whirr configuration would be: -############################################### -whirr.cluster-user=[username] -whirr.provider=aws-ec2 -whirr.location-id=us-west-1 -whirr.image-id=us-west-1/ami-16f3a253 -whirr.hardware-id=m1.large -whirr.identity=[EC2 Access Key ID] -whirr.credential=[EC2 Secret Access Key] -whirr.private-key-file=${sys:user.home}/.ssh/id_rsa -whirr.public-key-file=${sys:user.home}/.ssh/id_rsa.pub -whirr.run-url-base=http://hoodidge.net/scripts/ -whirr.blobstore.provider=aws-s3 -whirr.blobstore.container=cassandratests -############################################### - -The distributed tests are located in: - * test/distributed - -Run the tests via ant: - * ant distributed-test -Dwhirr.config=my-whirr.properties - -The ant target will: - * download extra dependencies via Apache Ivy - * compile the distributed tests - * push the local working copy to a blobstore to fetch from the test nodes - * deploy a cluster via Apache Whirr - * run the distributed tests against the cluster - * tear down the deployed cluster - - diff --git a/test/distributed/org/apache/cassandra/CassandraServiceController.java b/test/distributed/org/apache/cassandra/CassandraServiceController.java deleted file mode 100644 index 10fdef5884..0000000000 --- a/test/distributed/org/apache/cassandra/CassandraServiceController.java +++ /dev/null @@ -1,336 +0,0 @@ -/** - * 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. - */ - -package org.apache.cassandra; - -import java.net.InetAddress; -import java.net.URI; -import java.util.*; - -import com.google.common.base.Predicate; - -import org.apache.cassandra.thrift.Cassandra; -import org.apache.cassandra.utils.BlobUtils; -import org.apache.cassandra.utils.KeyPair; -import org.apache.cassandra.utils.Pair; -import org.apache.commons.configuration.CompositeConfiguration; -import org.apache.commons.configuration.PropertiesConfiguration; -import org.apache.thrift.TException; -import org.apache.thrift.protocol.TBinaryProtocol; -import org.apache.thrift.protocol.TProtocol; -import org.apache.thrift.transport.TFramedTransport; -import org.apache.thrift.transport.TSocket; -import org.apache.thrift.transport.TTransport; -import org.apache.whirr.service.*; -import org.apache.whirr.service.Cluster.Instance; -import org.apache.whirr.service.cassandra.CassandraClusterActionHandler; -import org.apache.whirr.service.jclouds.StatementBuilder; - -import org.jclouds.blobstore.domain.BlobMetadata; -import org.jclouds.compute.ComputeService; -import org.jclouds.compute.domain.ExecResponse; -import org.jclouds.compute.domain.NodeMetadata; -import org.jclouds.compute.options.RunScriptOptions; -import org.jclouds.compute.RunScriptOnNodesException; -import org.jclouds.domain.Credentials; -import org.jclouds.scriptbuilder.domain.OsFamily; -import org.jclouds.scriptbuilder.domain.Statements; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public class CassandraServiceController -{ - private static final Logger LOG = - LoggerFactory.getLogger(CassandraServiceController.class); - - protected static int CLIENT_PORT = 9160; - protected static int JMX_PORT = 7199; - - private static final CassandraServiceController INSTANCE = - new CassandraServiceController(); - - public static CassandraServiceController getInstance() - { - return INSTANCE; - } - - private boolean running; - - private ClusterSpec clusterSpec; - private Service service; - private Cluster cluster; - private ComputeService computeService; - private CompositeConfiguration config; - private BlobMetadata tarball; - private List hosts; - - private CassandraServiceController() - { - } - - public Cassandra.Client createClient(InetAddress addr) throws TException - { - TTransport transport = new TSocket( - addr.getHostAddress(), - CLIENT_PORT, - 200000); - transport = new TFramedTransport(transport); - TProtocol protocol = new TBinaryProtocol(transport); - - Cassandra.Client client = new Cassandra.Client(protocol); - transport.open(); - - return client; - } - - private void waitForClusterInitialization() - { - for (InetAddress host : hosts) - waitForNodeInitialization(host); - } - - private void waitForNodeInitialization(InetAddress addr) - { - while (true) - { - try - { - Cassandra.Client client = createClient(addr); - client.describe_cluster_name(); - break; - } - catch (TException e) - { - LOG.debug(e.toString()); - try - { - Thread.sleep(1000); - } - catch (InterruptedException ie) - { - break; - } - } - } - } - - public synchronized void startup() throws Exception - { - LOG.info("Starting up cluster..."); - - config = new CompositeConfiguration(); - if (System.getProperty("whirr.config") != null) - { - config.addConfiguration( - new PropertiesConfiguration(System.getProperty("whirr.config"))); - } - config.addConfiguration(new PropertiesConfiguration("whirr-default.properties")); - - clusterSpec = new ClusterSpec(config); - if (clusterSpec.getPrivateKey() == null) - { - Map pair = KeyPair.generate(); - clusterSpec.setPublicKey(pair.get("public")); - clusterSpec.setPrivateKey(pair.get("private")); - } - - // if a local tarball is available deploy it to the blobstore where it will be available to cassandra - if (System.getProperty("whirr.cassandra_tarball") != null) - { - Pair blob = BlobUtils.storeBlob(config, clusterSpec, System.getProperty("whirr.cassandra_tarball")); - tarball = blob.left; - config.setProperty(CassandraClusterActionHandler.BIN_TARBALL, blob.right.toURL().toString()); - // TODO: parse the CassandraVersion property file instead - config.setProperty(CassandraClusterActionHandler.MAJOR_VERSION, "0.8"); - } - - service = new ServiceFactory().create(clusterSpec.getServiceName()); - cluster = service.launchCluster(clusterSpec); - computeService = ComputeServiceContextBuilder.build(clusterSpec).getComputeService(); - hosts = new ArrayList(); - for (Instance instance : cluster.getInstances()) - { - hosts.add(instance.getPublicAddress()); - } - - ShutdownHook shutdownHook = new ShutdownHook(this); - Runtime.getRuntime().addShutdownHook(shutdownHook); - - waitForClusterInitialization(); - - running = true; - } - - public synchronized void shutdown() - { - // catch and log errors, we're in a runtime shutdown hook - try - { - LOG.info("Shutting down cluster..."); - if (tarball != null) - BlobUtils.deleteBlob(config, clusterSpec, tarball); - if (service != null) - service.destroyCluster(clusterSpec); - running = false; - } - catch (Exception e) - { - LOG.error("Error shutting down cluster.", e); - } - } - - public class ShutdownHook extends Thread - { - private CassandraServiceController controller; - - public ShutdownHook(CassandraServiceController controller) - { - this.controller = controller; - } - - public void run() - { - controller.shutdown(); - } - } - - public synchronized boolean ensureClusterRunning() throws Exception - { - if (running) - { - LOG.info("Cluster already running."); - return false; - } - else - { - startup(); - return true; - } - } - - /** - * Execute nodetool with args against localhost from the given host. - */ - public void nodetool(String args, InetAddress... hosts) - { - callOnHosts(Arrays.asList(hosts), "nodetool_cassandra", args); - } - - /** - * Wipes all persisted state for the given node, leaving it as if it had just started. - */ - public void wipeHosts(InetAddress... hosts) - { - callOnHosts(Arrays.asList(hosts), "wipe_cassandra"); - } - - public Failure failHosts(List hosts) - { - return new Failure(hosts).trigger(); - } - - public Failure failHosts(InetAddress... hosts) - { - return new Failure(Arrays.asList(hosts)).trigger(); - } - - /** TODO: Move to CassandraService? */ - protected void callOnHosts(List hosts, String functionName, String... functionArgs) - { - final Set hostset = new HashSet(); - - for (InetAddress host : hosts) - hostset.add(host.getHostAddress()); - - StatementBuilder statementBuilder = new StatementBuilder(); - statementBuilder.addStatement(Statements.call(functionName, functionArgs)); - Credentials credentials = new Credentials(clusterSpec.getClusterUser(), clusterSpec.getPrivateKey()); - - Map results; - try - { - results = computeService.runScriptOnNodesMatching(new Predicate() - { - public boolean apply(NodeMetadata node) - { - Set intersection = new HashSet(hostset); - intersection.retainAll(node.getPublicAddresses()); - return !intersection.isEmpty(); - } - }, - statementBuilder, - RunScriptOptions.Builder.overrideCredentialsWith(credentials).wrapInInitScript(false).runAsRoot(false)); - } - catch (RunScriptOnNodesException e) - { - throw new RuntimeException(e); - } - - if (results.size() != hostset.size()) - { - throw new RuntimeException(results.size() + " hosts matched " + hostset + ": " + results); - } - - for (ExecResponse response : results.values()) - { - if (response.getExitCode() != 0) - { - throw new RuntimeException("Call " + functionName + " failed on at least one of " + hostset + ": " + results.values()); - } - } - } - - public List getHosts() - { - return hosts; - } - - class Failure - { - private List hosts; - - public Failure(List hosts) - { - this.hosts = hosts; - } - - public Failure trigger() - { - callOnHosts(hosts, "stop_cassandra"); - return this; - } - - public void resolve() - { - callOnHosts(hosts, "start_cassandra"); - for (InetAddress host : hosts) - { - waitForNodeInitialization(host); - } - } - } - - public InetAddress getPublicHost(InetAddress privateHost) - { - for (Instance instance : cluster.getInstances()) - if (privateHost.equals(instance.getPrivateAddress())) - return instance.getPublicAddress(); - throw new RuntimeException("No public host for private host " + privateHost); - } -} diff --git a/test/distributed/org/apache/cassandra/CountersTest.java b/test/distributed/org/apache/cassandra/CountersTest.java deleted file mode 100644 index 71e635d367..0000000000 --- a/test/distributed/org/apache/cassandra/CountersTest.java +++ /dev/null @@ -1,88 +0,0 @@ -/** - * 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. - */ - -package org.apache.cassandra; - -import java.io.*; -import java.net.InetAddress; -import java.nio.ByteBuffer; -import java.util.List; - -import org.apache.thrift.TException; - -import org.apache.cassandra.thrift.*; - -import org.apache.cassandra.CassandraServiceController.Failure; - -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.Test; - -import static junit.framework.Assert.assertEquals; -import static junit.framework.Assert.assertNull; - -public class CountersTest extends TestBase -{ - @Test - public void testWriteOneReadAll() throws Exception - { - List hosts = controller.getHosts(); - // create a keyspace that performs counter validation - final String keyspace = "TestOneNodeWrite"; - keyspace(keyspace).rf(3).validator("CounterColumnType").create(); - - for (InetAddress host : hosts) - { - Cassandra.Client client = controller.createClient(host); - client.set_keyspace(keyspace); - - ByteBuffer key = newKey(); - - add(client, key, "Standard1", "c1", 1, ConsistencyLevel.ONE); - add(client, key, "Standard1", "c2", 2, ConsistencyLevel.ONE); - - new CounterGet(client, "Standard1", key).name("c1").value(1L).perform(ConsistencyLevel.ALL); - new CounterGet(client, "Standard1", key).name("c2").value(2L).perform(ConsistencyLevel.ALL); - } - } - - protected class CounterGet extends RetryingAction - { - public CounterGet(Cassandra.Client client, String cf, ByteBuffer key) - { - super(client, cf, key); - } - - public void tryPerformAction(ConsistencyLevel cl) throws Exception - { - ByteBuffer bname = ByteBuffer.wrap(name.getBytes()); - ColumnPath cpath = new ColumnPath(cf).setColumn(bname); - CounterColumn col = client.get(key, cpath, cl).counter_column; - assertEquals(bname, col.name); - assertEquals(value.longValue(), col.value); - } - } - - /** NB: Counter increments are unfortunately not idempotent, so we don't provide a RetyingAction to perform them. */ - protected void add(Cassandra.Client client, ByteBuffer key, String cf, String name, long value, ConsistencyLevel cl) - throws InvalidRequestException, UnavailableException, TimedOutException, TException - { - CounterColumn col = new CounterColumn(ByteBuffer.wrap(name.getBytes()), value); - client.add(key, new ColumnParent(cf), col, cl); - } -} diff --git a/test/distributed/org/apache/cassandra/MovementTest.java b/test/distributed/org/apache/cassandra/MovementTest.java deleted file mode 100644 index 68d9086318..0000000000 --- a/test/distributed/org/apache/cassandra/MovementTest.java +++ /dev/null @@ -1,125 +0,0 @@ -/** - * 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. - */ - -package org.apache.cassandra; - -import java.io.BufferedReader; -import java.io.InputStreamReader; -import java.io.OutputStream; -import java.io.OutputStreamWriter; -import java.io.Writer; -import java.net.InetAddress; -import java.nio.ByteBuffer; -import java.util.*; - -import org.apache.cassandra.thrift.*; -import org.apache.cassandra.tools.NodeProbe; -import org.apache.cassandra.utils.ByteBufferUtil; -import org.apache.cassandra.utils.WrappedRunnable; - -import org.apache.cassandra.CassandraServiceController.Failure; - -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.Test; - -import static junit.framework.Assert.assertEquals; -import static junit.framework.Assert.assertNull; - -public class MovementTest extends TestBase -{ - private static final String STANDARD_CF = "Standard1"; - private static final ColumnParent STANDARD = new ColumnParent(STANDARD_CF); - - /** Inserts 1000 keys with names such that at least 1 key ends up on each host. */ - private static Map> insertBatch(Cassandra.Client client) throws Exception - { - final int N = 1000; - Column col1 = new Column(ByteBufferUtil.bytes("c1")) - .setValue(ByteBufferUtil.bytes("v1")) - .setTimestamp(0); - Column col2 = new Column(ByteBufferUtil.bytes("c2")) - .setValue(ByteBufferUtil.bytes("v2")) - .setTimestamp(0); - - // build N rows - Map> rows = new HashMap>(); - Map>> batch = new HashMap>>(); - for (int i = 0; i < N; i++) - { - String rawKey = String.format("test.key.%d", i); - ByteBuffer key = ByteBufferUtil.bytes(rawKey); - Mutation m1 = (new Mutation()).setColumn_or_supercolumn((new ColumnOrSuperColumn()).setColumn(col1)); - Mutation m2 = (new Mutation()).setColumn_or_supercolumn((new ColumnOrSuperColumn()).setColumn(col2)); - rows.put(key, Arrays.asList(m1.getColumn_or_supercolumn(), - m2.getColumn_or_supercolumn())); - - // add row to batch - Map> rowmap = new HashMap>(); - rowmap.put(STANDARD_CF, Arrays.asList(m1, m2)); - batch.put(key, rowmap); - } - // insert the batch - client.batch_mutate(batch, ConsistencyLevel.ONE); - return rows; - } - - private static void verifyBatch(Cassandra.Client client, Map> batch) throws Exception - { - for (Map.Entry> entry : batch.entrySet()) - { - // verify slice - SlicePredicate sp = new SlicePredicate(); - sp.setSlice_range( - new SliceRange( - ByteBuffer.wrap(new byte[0]), - ByteBuffer.wrap(new byte[0]), - false, - 1000 - ) - ); - assertEquals(client.get_slice(entry.getKey(), STANDARD, sp, ConsistencyLevel.ONE), - entry.getValue()); - } - } - - @Test - public void testLoadbalance() throws Exception - { - final String keyspace = "TestLoadbalance"; - addKeyspace(keyspace, 1); - List hosts = controller.getHosts(); - Cassandra.Client client = controller.createClient(hosts.get(0)); - client.set_keyspace(keyspace); - - // add keys to each node - Map> rows = insertBatch(client); - - Thread.sleep(100); - - // ask a node to move to a new location - controller.nodetool("loadbalance", hosts.get(0)); - - // trigger cleanup on all nodes - for (InetAddress host : hosts) - controller.nodetool("cleanup", host); - - // check that all keys still exist - verifyBatch(client, rows); - } -} diff --git a/test/distributed/org/apache/cassandra/MutationTest.java b/test/distributed/org/apache/cassandra/MutationTest.java deleted file mode 100644 index ab93fd3d4c..0000000000 --- a/test/distributed/org/apache/cassandra/MutationTest.java +++ /dev/null @@ -1,188 +0,0 @@ -/** - * 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. - */ - -package org.apache.cassandra; - -import java.io.IOException; -import java.net.InetAddress; -import java.nio.ByteBuffer; -import java.util.*; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.cassandra.client.RingCache; -import org.apache.cassandra.dht.RandomPartitioner; -import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.thrift.*; -import org.apache.cassandra.utils.ByteBufferUtil; -import org.apache.cassandra.utils.WrappedRunnable; -import org.apache.thrift.TException; - -import org.apache.cassandra.CassandraServiceController.Failure; - -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.Test; - -import static junit.framework.Assert.assertEquals; -import static junit.framework.Assert.assertNull; - -public class MutationTest extends TestBase -{ - private static final Logger logger = LoggerFactory.getLogger(MutationTest.class); - - @Test - public void testInsert() throws Exception - { - List hosts = controller.getHosts(); - final String keyspace = "TestInsert"; - addKeyspace(keyspace, 3); - Cassandra.Client client = controller.createClient(hosts.get(0)); - client.set_keyspace(keyspace); - - ByteBuffer key = newKey(); - - insert(client, key, "Standard1", "c1", "v1", 0, ConsistencyLevel.ONE); - insert(client, key, "Standard1", "c2", "v2", 0, ConsistencyLevel.ONE); - - // block until the column is available - new Get(client, "Standard1", key).name("c1").value("v1").perform(ConsistencyLevel.ONE); - new Get(client, "Standard1", key).name("c2").value("v2").perform(ConsistencyLevel.ONE); - - List coscs = get_slice(client, key, "Standard1", ConsistencyLevel.ONE); - assertColumnEqual("c1", "v1", 0, coscs.get(0).column); - assertColumnEqual("c2", "v2", 0, coscs.get(1).column); - } - - @Test - public void testWriteAllReadOne() throws Exception - { - List hosts = controller.getHosts(); - Cassandra.Client client = controller.createClient(hosts.get(0)); - - final String keyspace = "TestWriteAllReadOne"; - addKeyspace(keyspace, 3); - client.set_keyspace(keyspace); - - ByteBuffer key = newKey(); - - insert(client, key, "Standard1", "c1", "v1", 0, ConsistencyLevel.ALL); - // should be instantly available - assertColumnEqual("c1", "v1", 0, getColumn(client, key, "Standard1", "c1", ConsistencyLevel.ONE)); - - List endpoints = endpointsForKey(hosts.get(0), key, keyspace); - InetAddress coordinator = nonEndpointForKey(hosts.get(0), key, keyspace); - Failure failure = controller.failHosts(endpoints.subList(1, endpoints.size())); - - try { - client = controller.createClient(coordinator); - client.set_keyspace(keyspace); - - new Get(client, "Standard1", key).name("c1").value("v1") - .perform(ConsistencyLevel.ONE); - - new Insert(client, "Standard1", key).name("c3").value("v3") - .expecting(UnavailableException.class).perform(ConsistencyLevel.ALL); - } finally { - failure.resolve(); - Thread.sleep(10000); - } - } - - @Test - public void testWriteQuorumReadQuorum() throws Exception - { - List hosts = controller.getHosts(); - Cassandra.Client client = controller.createClient(hosts.get(0)); - - final String keyspace = "TestWriteQuorumReadQuorum"; - addKeyspace(keyspace, 3); - client.set_keyspace(keyspace); - - ByteBuffer key = newKey(); - - // with quorum-1 nodes up - List endpoints = endpointsForKey(hosts.get(0), key, keyspace); - InetAddress coordinator = nonEndpointForKey(hosts.get(0), key, keyspace); - Failure failure = controller.failHosts(endpoints.subList(1, endpoints.size())); //kill all but one nodes - - client = controller.createClient(coordinator); - client.set_keyspace(keyspace); - try { - new Insert(client, "Standard1", key).name("c1").value("v1") - .expecting(UnavailableException.class).perform(ConsistencyLevel.QUORUM); - } finally { - failure.resolve(); - Thread.sleep(10000); - } - - // with all nodes up - new Insert(client, "Standard1", key).name("c2").value("v2").perform(ConsistencyLevel.QUORUM); - - failure = controller.failHosts(endpoints.get(0)); - try { - new Get(client, "Standard1", key).name("c2").value("v2").perform(ConsistencyLevel.QUORUM); - } finally { - failure.resolve(); - Thread.sleep(10000); - } - } - - @Test - public void testWriteOneReadAll() throws Exception - { - List hosts = controller.getHosts(); - Cassandra.Client client = controller.createClient(hosts.get(0)); - - final String keyspace = "TestWriteOneReadAll"; - addKeyspace(keyspace, 3); - client.set_keyspace(keyspace); - - ByteBuffer key = newKey(); - - List endpoints = endpointsForKey(hosts.get(0), key, keyspace); - InetAddress coordinator = nonEndpointForKey(hosts.get(0), key, keyspace); - client = controller.createClient(coordinator); - client.set_keyspace(keyspace); - - insert(client, key, "Standard1", "c1", "v1", 0, ConsistencyLevel.ONE); - assertColumnEqual("c1", "v1", 0, getColumn(client, key, "Standard1", "c1", ConsistencyLevel.ALL)); - - // with each of HH, read repair and proactive repair: - // with one node up - // write with one (success) - // read with all (failure) - // bring nodes up - // repair - // read with all (success) - - Failure failure = controller.failHosts(endpoints); - try { - new Insert(client, "Standard1", key).name("c2").value("v2") - .expecting(UnavailableException.class).perform(ConsistencyLevel.ONE); - } finally { - failure.resolve(); - } - } - - protected ByteBuffer newKey() - { - return ByteBufferUtil.bytes(String.format("test.key.%d", System.currentTimeMillis())); - } -} diff --git a/test/distributed/org/apache/cassandra/TestBase.java b/test/distributed/org/apache/cassandra/TestBase.java deleted file mode 100644 index e182270391..0000000000 --- a/test/distributed/org/apache/cassandra/TestBase.java +++ /dev/null @@ -1,329 +0,0 @@ -/** - * 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. - */ - -package org.apache.cassandra; - -import java.io.*; -import java.net.InetAddress; -import java.nio.ByteBuffer; -import java.util.*; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.thrift.TException; - -import org.apache.cassandra.client.*; -import org.apache.cassandra.dht.RandomPartitioner; -import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.thrift.*; - -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.Test; - -import static junit.framework.Assert.assertEquals; -import static junit.framework.Assert.assertNull; - -public abstract class TestBase -{ - private static final Logger logger = LoggerFactory.getLogger(TestBase.class); - - protected static CassandraServiceController controller = - CassandraServiceController.getInstance(); - - static class KeyspaceCreation - { - private String name; - private int rf; - private CfDef cfdef; - public KeyspaceCreation(String name) - { - this.name = name; - cfdef = new CfDef(name, "Standard1"); - cfdef.setComparator_type("BytesType"); - cfdef.setKey_cache_size(10000); - cfdef.setRow_cache_size(1000); - cfdef.setRow_cache_save_period_in_seconds(0); - cfdef.setKey_cache_save_period_in_seconds(3600); - cfdef.setMemtable_throughput_in_mb(255); - cfdef.setMemtable_operations_in_millions(0.29); - } - - public KeyspaceCreation validator(String validator) - { - cfdef.setDefault_validation_class(validator); - return this; - } - - public KeyspaceCreation rf(int rf) - { - this.rf = rf; - return this; - } - - public void create() throws Exception - { - List hosts = controller.getHosts(); - Cassandra.Client client = controller.createClient(hosts.get(0)); - Map stratOptions = new HashMap(); - stratOptions.put("replication_factor", "" + rf); - client.system_add_keyspace(new KsDef(name, - "org.apache.cassandra.locator.SimpleStrategy", - Arrays.asList(cfdef)) - .setStrategy_options(stratOptions)); - - // poll, until KS added - for (InetAddress host : hosts) - { - try - { - client = controller.createClient(host); - poll: - while (true) - { - List ksDefList = client.describe_keyspaces(); - for (KsDef ks : ksDefList) - { - if (ks.name.equals(name)) - break poll; - } - - try - { - Thread.sleep(1000); - } - catch (InterruptedException e) - { - break poll; - } - } - } - catch (TException te) - { - continue; - } - } - } - } - - protected static KeyspaceCreation keyspace(String name) - { - return new KeyspaceCreation(name); - } - - protected static void addKeyspace(String name, int rf) throws Exception - { - keyspace(name).rf(rf).create(); - } - - @BeforeClass - public static void setUp() throws Exception - { - controller.ensureClusterRunning(); - } - - protected ByteBuffer newKey() - { - return ByteBuffer.wrap(String.format("test.key.%d", System.currentTimeMillis()).getBytes()); - } - - protected void insert(Cassandra.Client client, ByteBuffer key, String cf, String name, String value, long timestamp, ConsistencyLevel cl) - throws InvalidRequestException, UnavailableException, TimedOutException, TException - { - Column col = new Column(ByteBuffer.wrap(name.getBytes())) - .setValue(ByteBuffer.wrap(value.getBytes())) - .setTimestamp(timestamp); - client.insert(key, new ColumnParent(cf), col, cl); - } - - protected Column getColumn(Cassandra.Client client, ByteBuffer key, String cf, String col, ConsistencyLevel cl) - throws InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException - { - ColumnPath cpath = new ColumnPath(cf); - cpath.setColumn(col.getBytes()); - return client.get(key, cpath, cl).column; - } - - protected class Get extends RetryingAction - { - public Get(Cassandra.Client client, String cf, ByteBuffer key) - { - super(client, cf, key); - } - - public void tryPerformAction(ConsistencyLevel cl) throws Exception - { - assertColumnEqual(name, value, timestamp, getColumn(client, key, cf, name, cl)); - } - } - - protected class Insert extends RetryingAction - { - public Insert(Cassandra.Client client, String cf, ByteBuffer key) - { - super(client, cf, key); - } - - public void tryPerformAction(ConsistencyLevel cl) throws Exception - { - insert(client, key, cf, name, value, timestamp, cl); - } - } - - /** Performs an action repeatedly until timeout, success or failure. */ - protected abstract class RetryingAction - { - protected Cassandra.Client client; - protected String cf; - protected ByteBuffer key; - protected String name; - protected T value; - protected long timestamp; - - private Set> expected = new HashSet>(); - private long timeout = StorageService.RING_DELAY * 2; - - public RetryingAction(Cassandra.Client client, String cf, ByteBuffer key) - { - this.client = client; - this.cf = cf; - this.key = key; - this.timestamp = 0; - } - - public RetryingAction name(String name) - { - this.name = name; return this; - } - - /** A parameterized value for the action. */ - public RetryingAction value(T value) - { - this.value = value; return this; - } - - /** The total time to allow before failing. */ - public RetryingAction timeout(long timeout) - { - this.timeout = timeout; return this; - } - - /** The expected timestamp of the returned column. */ - public RetryingAction timestamp(long timestamp) - { - this.timestamp = timestamp; return this; - } - - /** The exception classes that indicate success. */ - public RetryingAction expecting(Class... tempExceptions) - { - this.expected.clear(); - for (Class exclass : tempExceptions) - expected.add((Class)exclass); - return this; - } - - public void perform(ConsistencyLevel cl) throws AssertionError - { - long deadline = System.currentTimeMillis() + timeout; - int attempts = 0; - String template = "%s for " + this + " after %d attempt(s) with %d ms to spare."; - Exception e = null; - while(deadline > System.currentTimeMillis()) - { - try - { - attempts++; - tryPerformAction(cl); - logger.info(String.format(template, "Succeeded", attempts, deadline - System.currentTimeMillis())); - return; - } - catch (Exception ex) - { - e = ex; - if (!expected.contains(ex.getClass())) - continue; - logger.info(String.format(template, "Caught expected exception: " + e, attempts, deadline - System.currentTimeMillis())); - return; - } - } - String err = String.format(template, "Caught unexpected: " + e, attempts, deadline - System.currentTimeMillis()); - logger.error(err, e); - throw new AssertionError(err); - } - - public String toString() - { - return this.getClass().getSimpleName() + "(" + key + "," + name + ")"; - } - - protected abstract void tryPerformAction(ConsistencyLevel cl) throws Exception; - } - - protected List get_slice(Cassandra.Client client, ByteBuffer key, String cf, ConsistencyLevel cl) - throws InvalidRequestException, UnavailableException, TimedOutException, TException - { - SlicePredicate sp = new SlicePredicate(); - sp.setSlice_range( - new SliceRange( - ByteBuffer.wrap(new byte[0]), - ByteBuffer.wrap(new byte[0]), - false, - 1000 - ) - ); - return client.get_slice(key, new ColumnParent(cf), sp, cl); - } - - protected void assertColumnEqual(String name, String value, long timestamp, Column col) - { - assertEquals(ByteBuffer.wrap(name.getBytes()), col.name); - assertEquals(ByteBuffer.wrap(value.getBytes()), col.value); - assertEquals(timestamp, col.timestamp); - } - - protected List endpointsForKey(InetAddress seed, ByteBuffer key, String keyspace) - throws IOException - { - Configuration conf = new Configuration(); - - RingCache ring = new RingCache(keyspace, new RandomPartitioner(), seed.getHostAddress(), 9160); - List privateendpoints = ring.getEndpoint(key); - List endpoints = new ArrayList(); - for (InetAddress endpoint : privateendpoints) - { - endpoints.add(controller.getPublicHost(endpoint)); - } - return endpoints; - } - - protected InetAddress nonEndpointForKey(InetAddress seed, ByteBuffer key, String keyspace) - throws IOException - { - List endpoints = endpointsForKey(seed, key, keyspace); - for (InetAddress host : controller.getHosts()) - { - if (!endpoints.contains(host)) - { - return host; - } - } - return null; - } -} diff --git a/test/distributed/org/apache/cassandra/utils/BlobUtils.java b/test/distributed/org/apache/cassandra/utils/BlobUtils.java deleted file mode 100644 index ddcbfa5fc1..0000000000 --- a/test/distributed/org/apache/cassandra/utils/BlobUtils.java +++ /dev/null @@ -1,163 +0,0 @@ -/** - * 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. - */ - -package org.apache.cassandra.utils; - -import java.io.File; -import java.io.FileWriter; -import java.io.IOException; -import java.net.URI; -import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; - -import com.google.common.io.Files; -import org.apache.commons.configuration.Configuration; - -import org.apache.whirr.service.ClusterSpec; - -import org.jclouds.blobstore.BlobStoreContext; -import org.jclouds.blobstore.BlobStoreContextFactory; -import org.jclouds.blobstore.InputStreamMap; -import org.jclouds.blobstore.domain.BlobMetadata; - -import org.jclouds.s3.S3AsyncClient; -import org.jclouds.s3.S3Client; -import org.jclouds.s3.domain.AccessControlList; -import org.jclouds.s3.domain.CannedAccessPolicy; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public final class BlobUtils -{ - private static final Logger LOG = LoggerFactory.getLogger(BlobUtils.class); - - public static final String BLOB_PROVIDER = "whirr.blobstore.provider"; - public static final String BLOB_CONTAINER = "whirr.blobstore.container"; - - private static BlobStoreContext getContext(Configuration config, ClusterSpec spec) - { - return new BlobStoreContextFactory().createContext(getProvider(config), spec.getIdentity(), spec.getCredential()); - } - - private static String getProvider(Configuration config) - { - String provider = config.getString(BLOB_PROVIDER, null); - if (provider == null) - throw new RuntimeException("Please set " + BLOB_PROVIDER + " to a jclouds supported provider."); - return provider; - } - - private static String getContainer(Configuration config) - { - String container = config.getString(BLOB_CONTAINER, null); - if (container == null) - throw new RuntimeException("Please set " + BLOB_CONTAINER + " to an existing container for your chosen provider."); - return container; - } - - /** - * Stores the given local file as a public blob, and returns metadata for the blob. - */ - public static Pair storeBlob(Configuration config, ClusterSpec spec, String filename) - { - File file = new File(filename); - String container = getContainer(config); - String provider = getProvider(config); - - // blob name and checksum of the file - String blobName = System.nanoTime() + "/" + file.getName(); - String blobNameChecksum = blobName + ".md5"; - - BlobStoreContext context = getContext(config, spec); - - File checksumFile; - - try - { - checksumFile = File.createTempFile("dtchecksum", "md5"); - checksumFile.deleteOnExit(); - - FileWriter checksumWriter = new FileWriter(checksumFile); - - String checksum = FBUtilities.bytesToHex(Files.getDigest(file, MessageDigest.getInstance("MD5"))); - - checksumWriter.write(String.format("%s %s", checksum, file.getName())); - checksumWriter.close(); - } - catch (IOException e) - { - throw new RuntimeException("Can't create a checksum of the file: " + filename); - } - catch (NoSuchAlgorithmException e) - { - throw new RuntimeException(e.getMessage()); - } - - try - { - InputStreamMap map = context.createInputStreamMap(container); - - map.putFile(blobName, file); - map.putFile(blobNameChecksum, checksumFile); - - // TODO: magic! in order to expose the blob as public, we need to dive into provider specific APIs - // the hope is that permissions are encapsulated in jclouds in the future - if (provider.contains("s3")) - { - S3Client sss = context.getProviderSpecificContext().getApi(); - String ownerId = sss.getObjectACL(container, blobName).getOwner().getId(); - - sss.putObjectACL(container, - blobName, - AccessControlList.fromCannedAccessPolicy(CannedAccessPolicy.PUBLIC_READ, ownerId)); - - sss.putObjectACL(container, - blobNameChecksum, - AccessControlList.fromCannedAccessPolicy(CannedAccessPolicy.PUBLIC_READ, ownerId)); - } - else - { - LOG.warn(provider + " may not be properly supported for tarball transfer."); - } - - // resolve the full URI of the blob (see http://code.google.com/p/jclouds/issues/detail?id=431) - BlobMetadata blob = context.getBlobStore().blobMetadata(container, blobName); - URI uri = context.getProviderSpecificContext().getEndpoint().resolve("/" + container + "/" + blob.getName()); - return new Pair(blob, uri); - } - finally - { - context.close(); - } - } - - public static void deleteBlob(Configuration config, ClusterSpec spec, BlobMetadata blob) - { - String container = getContainer(config); - BlobStoreContext context = getContext(config, spec); - try - { - context.getBlobStore().removeBlob(container, blob.getName()); - } - finally - { - context.close(); - } - } -} diff --git a/test/distributed/org/apache/cassandra/utils/KeyPair.java b/test/distributed/org/apache/cassandra/utils/KeyPair.java deleted file mode 100644 index a8d9049027..0000000000 --- a/test/distributed/org/apache/cassandra/utils/KeyPair.java +++ /dev/null @@ -1,49 +0,0 @@ -/** - * 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. - */ - -package org.apache.cassandra.utils; - -import java.io.ByteArrayOutputStream; -import java.util.Map; - -import com.google.common.collect.ImmutableMap; -import com.jcraft.jsch.JSch; -import com.jcraft.jsch.JSchException; - -/** - * A convenience class for generating an RSA key pair. - */ -public class KeyPair { - - /** - * return a "public" -> rsa public key, "private" -> its corresponding - * private key - */ - public static Map generate() throws JSchException { - com.jcraft.jsch.KeyPair pair = com.jcraft.jsch.KeyPair.genKeyPair( - new JSch(), com.jcraft.jsch.KeyPair.RSA); - ByteArrayOutputStream publicKeyOut = new ByteArrayOutputStream(); - ByteArrayOutputStream privateKeyOut = new ByteArrayOutputStream(); - pair.writePublicKey(publicKeyOut, "whirr"); - pair.writePrivateKey(privateKeyOut); - String publicKey = new String(publicKeyOut.toByteArray()); - String privateKey = new String(privateKeyOut.toByteArray()); - return ImmutableMap. of("public", publicKey, - "private", privateKey); - } -} diff --git a/test/unit/org/apache/cassandra/service/RemoveTest.java b/test/unit/org/apache/cassandra/service/RemoveTest.java index 7bfa385cf9..c394e8f033 100644 --- a/test/unit/org/apache/cassandra/service/RemoveTest.java +++ b/test/unit/org/apache/cassandra/service/RemoveTest.java @@ -125,7 +125,7 @@ public class RemoveTest extends CleanupHelper { try { - ss.removeToken(token, 0); + ss.removeToken(token); } catch (Exception e) {