diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 02440df004..c34b44d1a3 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -31,6 +31,9 @@ auto_bootstrap: false # See http://wiki.apache.org/cassandra/HintedHandoff hinted_handoff_enabled: true +# this defines the maximum amount of time a dead host will have hints +# generated. After it has been dead this long, hints will be dropped. +max_hint_window_in_ms: 3600000 # one hour # authentication backend, implementing IAuthenticator; used to identify users authenticator: org.apache.cassandra.auth.AllowAllAuthenticator diff --git a/contrib/bmt_example/CassandraBulkLoader.java b/contrib/bmt_example/CassandraBulkLoader.java index e6c072681a..3470d32e37 100644 --- a/contrib/bmt_example/CassandraBulkLoader.java +++ b/contrib/bmt_example/CassandraBulkLoader.java @@ -62,6 +62,7 @@ import java.util.concurrent.TimeoutException; import com.google.common.base.Charsets; import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.config.ConfigurationException; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.Column; import org.apache.cassandra.db.ColumnFamily; @@ -112,7 +113,7 @@ public class CassandraBulkLoader { { StorageService.instance.initClient(); } - catch (IOException e) + catch (Exception e) { throw new RuntimeException(e); } diff --git a/contrib/stress/README.txt b/contrib/stress/README.txt new file mode 100644 index 0000000000..76f04a4c9f --- /dev/null +++ b/contrib/stress/README.txt @@ -0,0 +1,59 @@ +stress +====== + +Description +----------- +stress is a tool for benchmarking and load testing a Cassandra +cluster. It is significantly faster than the older py_stress tool. + +Setup +----- +Run `ant` from the Cassandra source directory, then Run `ant` from the +contrib/stress directory. + +Usage +----- +There are three different modes of operation: + + * inserting (loading test data) + * reading + * range slicing (only works with the OrderPreservingPartioner) + * indexed range slicing (works with RandomParitioner on indexed ColumnFamilies) + +Important options: + -o or --operation: + Sets the operation mode, one of 'insert', 'read', 'rangeslice', or 'indexedrangeslice' + -n or --num-keys: + the number of rows to insert/read/slice; defaults to one million + -d or --nodes: + the node(s) to perform the test against. For multiple nodes, supply a + comma-separated list without spaces, ex: cassandra1,cassandra2,cassandra3 + -y or --family-type: + Sets the ColumnFamily type. One of 'Standard' or 'Super'. If using super, + you probably want to set the -u option also. + -c or --columns: + the number of columns per row, defaults to 5 + -u or --supercolumns: + use the number of supercolumns specified NOTE: you must set the -y + option appropriately, or this option has no effect. + -g or --get-range-slice-count: + This is only used for the rangeslice operation and will *NOT* work with + the RandomPartioner. You must set the OrderPreservingPartioner in your + storage-conf.xml (note that you will need to wipe all existing data + when switching partioners.) This option sets the number of rows to + slice at a time and defaults to 1000. + -r or --random: + Only used for reads. By default, stress.py will perform reads on rows + with a guassian distribution, which will cause some repeats. Setting + this option makes the reads completely random instead. + -i or --progress-interval: + The interval, in seconds, at which progress will be output. + +Remember that you must perform inserts before performing reads or range slices. + +Examples +-------- + + * contrib/stress/bin/stress -d 192.168.1.101 # 1M inserts to given host + * contrib/stress/bin/stress -d 192.168.1.101 -o read # 1M reads + * contrib/stress/bin/stress -d 192.168.1.101,192.168.1.102 -n 10000000 # 10M inserts spread across two nodes diff --git a/contrib/stress/bin/stress b/contrib/stress/bin/stress index 1284b3c09a..d43e319348 100755 --- a/contrib/stress/bin/stress +++ b/contrib/stress/bin/stress @@ -23,7 +23,7 @@ if [ "x$CLASSPATH" = "x" ]; then exit 1 fi - # Circuit class files. + # Stress class files. if [ ! -d `dirname $0`/../build/classes ]; then echo "Unable to locate stress class files" >&2 exit 1 diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/Session.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/Session.java index 56071f3c93..853d5987fa 100644 --- a/contrib/stress/src/org/apache/cassandra/contrib/stress/Session.java +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/Session.java @@ -45,28 +45,28 @@ public class Session static { - availableOptions.addOption("h", "help", false, "show this help message and exit."); - availableOptions.addOption("n", "num-keys", true, "Number of keys, default:1000000."); - availableOptions.addOption("N", "skip-keys", true, "Fraction of keys to skip initially, default:0."); - availableOptions.addOption("t", "threads", true, "Number of threads to use, default:50."); - availableOptions.addOption("c", "columns", true, "Number of columns per key, default:5."); - availableOptions.addOption("S", "column-size", true, "Size of column values in bytes, default:34."); - availableOptions.addOption("C", "cardinality", true, "Number of unique values stored in columns, default:50."); - availableOptions.addOption("d", "nodes", true, "Host nodes (comma separated), default:locahost."); - availableOptions.addOption("s", "stdev", true, "Standard Deviation Factor, default:0.1."); - availableOptions.addOption("r", "random", false, "Use random key generator (STDEV will have no effect), default:false."); - availableOptions.addOption("f", "file", true, "Write output to file"); - availableOptions.addOption("p", "port", true, "Thrift port, default:9160."); - availableOptions.addOption("m", "unframed", false, "Use unframed transport, default:false."); - availableOptions.addOption("o", "operation", true, "Operation to perform (INSERT, READ, RANGE_SLICE, INDEXED_RANGE_SLICE, MULTI_GET), default:INSERT."); - availableOptions.addOption("u", "supercolumns", true, "Number of super columns per key, default:1."); - availableOptions.addOption("y", "family-type", true, "Column Family Type (Super, Standard), default:Standard."); - availableOptions.addOption("k", "keep-going", false, "Ignore errors inserting or reading, default:false."); - availableOptions.addOption("i", "progress-interval", true, "Progress Report Interval (seconds), default:10."); - availableOptions.addOption("g", "keys-per-call", true, "Amount of keys to get_range_slices or multiget per call, default:1000."); - availableOptions.addOption("l", "replication-factor", true, "Replication Factor to use when creating needed column families, default:1."); - availableOptions.addOption("e", "consistency-level", true, "Consistency Level to use (ONE, QUORUM, LOCAL_QUORUM, EACH_QUORUM, ALL, ANY), default:ONE."); - availableOptions.addOption("x", "create-index", true, "Type of index to create on needed column families (KEYS)."); + availableOptions.addOption("h", "help", false, "Show this help message and exit"); + availableOptions.addOption("n", "num-keys", true, "Number of keys, default:1000000"); + availableOptions.addOption("N", "skip-keys", true, "Fraction of keys to skip initially, default:0"); + availableOptions.addOption("t", "threads", true, "Number of threads to use, default:50"); + availableOptions.addOption("c", "columns", true, "Number of columns per key, default:5"); + availableOptions.addOption("S", "column-size", true, "Size of column values in bytes, default:34"); + availableOptions.addOption("C", "cardinality", true, "Number of unique values stored in columns, default:50"); + availableOptions.addOption("d", "nodes", true, "Host nodes (comma separated), default:locahost"); + availableOptions.addOption("s", "stdev", true, "Standard Deviation Factor, default:0.1"); + availableOptions.addOption("r", "random", false, "Use random key generator (STDEV will have no effect), default:false"); + availableOptions.addOption("f", "file", true, "Write output to given file"); + availableOptions.addOption("p", "port", true, "Thrift port, default:9160"); + availableOptions.addOption("m", "unframed", false, "Use unframed transport, default:false"); + availableOptions.addOption("o", "operation", true, "Operation to perform (INSERT, READ, RANGE_SLICE, INDEXED_RANGE_SLICE, MULTI_GET), default:INSERT"); + availableOptions.addOption("u", "supercolumns", true, "Number of super columns per key, default:1"); + availableOptions.addOption("y", "family-type", true, "Column Family Type (Super, Standard), default:Standard"); + availableOptions.addOption("k", "keep-going", false, "Ignore errors inserting or reading, default:false"); + availableOptions.addOption("i", "progress-interval", true, "Progress Report Interval (seconds), default:10"); + availableOptions.addOption("g", "keys-per-call", true, "Number of keys to get_range_slices or multiget per call, default:1000"); + availableOptions.addOption("l", "replication-factor", true, "Replication Factor to use when creating needed column families, default:1"); + availableOptions.addOption("e", "consistency-level", true, "Consistency Level to use (ONE, QUORUM, LOCAL_QUORUM, EACH_QUORUM, ALL, ANY), default:ONE"); + availableOptions.addOption("x", "create-index", true, "Type of index to create on needed column families (KEYS)"); } private int numKeys = 1000 * 1000; diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/Stress.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/Stress.java index 9ef1e4f077..72e0942322 100644 --- a/contrib/stress/src/org/apache/cassandra/contrib/stress/Stress.java +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/Stress.java @@ -17,7 +17,7 @@ */ package org.apache.cassandra.contrib.stress; -import org.apache.cassandra.contrib.stress.tests.*; +import org.apache.cassandra.contrib.stress.operations.*; import org.apache.cassandra.contrib.stress.util.OperationThread; import org.apache.commons.cli.Option; diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/tests/IndexedRangeSlicer.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/IndexedRangeSlicer.java similarity index 94% rename from contrib/stress/src/org/apache/cassandra/contrib/stress/tests/IndexedRangeSlicer.java rename to contrib/stress/src/org/apache/cassandra/contrib/stress/operations/IndexedRangeSlicer.java index 333946b426..6891921fe9 100644 --- a/contrib/stress/src/org/apache/cassandra/contrib/stress/tests/IndexedRangeSlicer.java +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/IndexedRangeSlicer.java @@ -15,10 +15,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.cassandra.contrib.stress.tests; +package org.apache.cassandra.contrib.stress.operations; import org.apache.cassandra.contrib.stress.util.OperationThread; import org.apache.cassandra.thrift.*; +import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import java.nio.ByteBuffer; @@ -102,11 +103,11 @@ public class IndexedRangeSlicer extends OperationThread private int getMaxKey(List keySlices) { byte[] firstKey = keySlices.get(0).getKey(); - int maxKey = FBUtilities.byteBufferToInt(ByteBuffer.wrap(firstKey)); + int maxKey = ByteBufferUtil.toInt(ByteBuffer.wrap(firstKey)); for (KeySlice k : keySlices) { - int currentKey = FBUtilities.byteBufferToInt(ByteBuffer.wrap(k.getKey())); + int currentKey = ByteBufferUtil.toInt(ByteBuffer.wrap(k.getKey())); if (currentKey > maxKey) { diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/tests/Inserter.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/Inserter.java similarity index 98% rename from contrib/stress/src/org/apache/cassandra/contrib/stress/tests/Inserter.java rename to contrib/stress/src/org/apache/cassandra/contrib/stress/operations/Inserter.java index 6a9e48909f..7065b115c7 100644 --- a/contrib/stress/src/org/apache/cassandra/contrib/stress/tests/Inserter.java +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/Inserter.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.cassandra.contrib.stress.tests; +package org.apache.cassandra.contrib.stress.operations; import org.apache.cassandra.contrib.stress.util.OperationThread; import org.apache.cassandra.db.ColumnFamilyType; diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/tests/MultiGetter.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/MultiGetter.java similarity index 98% rename from contrib/stress/src/org/apache/cassandra/contrib/stress/tests/MultiGetter.java rename to contrib/stress/src/org/apache/cassandra/contrib/stress/operations/MultiGetter.java index dcead126ca..f0bbe76666 100644 --- a/contrib/stress/src/org/apache/cassandra/contrib/stress/tests/MultiGetter.java +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/MultiGetter.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.cassandra.contrib.stress.tests; +package org.apache.cassandra.contrib.stress.operations; import org.apache.cassandra.contrib.stress.util.OperationThread; import org.apache.cassandra.db.ColumnFamilyType; diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/tests/RangeSlicer.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/RangeSlicer.java similarity index 98% rename from contrib/stress/src/org/apache/cassandra/contrib/stress/tests/RangeSlicer.java rename to contrib/stress/src/org/apache/cassandra/contrib/stress/operations/RangeSlicer.java index 54674acbcb..8cdf6d6f8b 100644 --- a/contrib/stress/src/org/apache/cassandra/contrib/stress/tests/RangeSlicer.java +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/RangeSlicer.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.cassandra.contrib.stress.tests; +package org.apache.cassandra.contrib.stress.operations; import org.apache.cassandra.contrib.stress.util.OperationThread; import org.apache.cassandra.db.ColumnFamilyType; diff --git a/contrib/stress/src/org/apache/cassandra/contrib/stress/tests/Reader.java b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/Reader.java similarity index 98% rename from contrib/stress/src/org/apache/cassandra/contrib/stress/tests/Reader.java rename to contrib/stress/src/org/apache/cassandra/contrib/stress/operations/Reader.java index 41e5d91dda..14552568aa 100644 --- a/contrib/stress/src/org/apache/cassandra/contrib/stress/tests/Reader.java +++ b/contrib/stress/src/org/apache/cassandra/contrib/stress/operations/Reader.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.cassandra.contrib.stress.tests; +package org.apache.cassandra.contrib.stress.operations; import org.apache.cassandra.contrib.stress.util.OperationThread; import org.apache.cassandra.db.ColumnFamilyType; diff --git a/debian/init b/debian/init index 1d2961fb51..c1c3707868 100644 --- a/debian/init +++ b/debian/init @@ -119,6 +119,9 @@ do_start() # 2 if daemon could not be started is_running && return 1 + cassandra_home=`getent passwd cassandra | awk -F ':' '{ print $6; }'` + cd / # jsvc doesn't chdir() for us + $JSVC \ -user cassandra \ -home $JAVA_HOME \ @@ -127,6 +130,8 @@ do_start() -outfile /var/log/$NAME/output.log \ -cp `classpath` \ -Dlog4j.configuration=log4j-server.properties \ + -XX:HeapDumpPath="$cassandra_home/java_`date +%s`.hprof" \ + -XX:ErrorFile="$cassandra_home/hs_err_`date +%s`.log" \ $JVM_OPTS \ org.apache.cassandra.thrift.CassandraDaemon diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index da31e9e3ae..8333d1d97e 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -34,6 +34,7 @@ public class Config public Boolean auto_bootstrap = false; public Boolean hinted_handoff_enabled = true; + public Integer max_hint_window_in_ms = Integer.MAX_VALUE; public SeedProviderDef seed_provider; public DiskAccessMode disk_access_mode = DiskAccessMode.auto; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 70a8a566e2..1aa07d8cef 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -101,8 +101,9 @@ public class DatabaseDescriptor try { url = new URL(configUrl); + url.openStream(); // catches well-formed but bogus URLs } - catch (MalformedURLException e) + catch (Exception e) { ClassLoader loader = DatabaseDescriptor.class.getClassLoader(); url = loader.getResource(configUrl); @@ -1092,6 +1093,11 @@ public class DatabaseDescriptor return conf.hinted_handoff_enabled; } + public static int getMaxHintWindow() + { + return conf.max_hint_window_in_ms; + } + public static AbstractType getValueValidator(String keyspace, String cf, ByteBuffer column) { return getCFMetaData(keyspace, cf).getValueValidator(column); diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java index 2ab553ba23..7ebda3b04b 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java @@ -43,6 +43,7 @@ import org.apache.cassandra.db.UnserializableColumnFamilyException; import org.apache.cassandra.io.DeletionService; import org.apache.cassandra.io.util.BufferedRandomAccessFile; import org.apache.cassandra.io.util.FileUtils; +import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.WrappedRunnable; @@ -298,7 +299,7 @@ public class CommitLog if (logger.isDebugEnabled()) logger.debug(String.format("replaying mutation for %s.%s: %s", rm.getTable(), - rm.key(), + ByteBufferUtil.bytesToHex(rm.key()), "{" + StringUtils.join(rm.getColumnFamilies(), ", ") + "}")); final Table table = Table.open(rm.getTable()); tablesRecovered.add(table); diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index de863f831b..312cf85ae5 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -128,7 +128,7 @@ public class Gossiper implements IFailureDetectionEventListener private Set liveEndpoints_ = new ConcurrentSkipListSet(inetcomparator); /* unreachable member set */ - private Set unreachableEndpoints_ = new ConcurrentSkipListSet(inetcomparator); + private Map unreachableEndpoints_ = new ConcurrentHashMap(); /* initial seeds for joining the cluster */ private Set seeds_ = new ConcurrentSkipListSet(inetcomparator); @@ -179,7 +179,16 @@ public class Gossiper implements IFailureDetectionEventListener public Set getUnreachableMembers() { - return new HashSet(unreachableEndpoints_); + return unreachableEndpoints_.keySet(); + } + + public long getEndpointDowntime(InetAddress ep) + { + Long downtime = unreachableEndpoints_.get(ep); + if (downtime != null) + return System.currentTimeMillis() - downtime; + else + return 0L; } /** @@ -353,7 +362,7 @@ public class Gossiper implements IFailureDetectionEventListener double prob = unreachableEndpoints / (liveEndpoints + 1); double randDbl = random_.nextDouble(); if ( randDbl < prob ) - sendGossip(message, unreachableEndpoints_); + sendGossip(message, unreachableEndpoints_.keySet()); } } @@ -735,7 +744,7 @@ public class Gossiper implements IFailureDetectionEventListener else { liveEndpoints_.remove(addr); - unreachableEndpoints_.add(addr); + unreachableEndpoints_.put(addr, System.currentTimeMillis()); for (IEndpointStateChangeSubscriber subscriber : subscribers_) subscriber.onDead(addr, epState); } @@ -871,7 +880,7 @@ public class Gossiper implements IFailureDetectionEventListener epState.isAGossiper(true); epState.setHasToken(true); endpointStateMap_.put(ep, epState); - unreachableEndpoints_.add(ep); + unreachableEndpoints_.put(ep, System.currentTimeMillis()); } } diff --git a/src/java/org/apache/cassandra/io/util/ColumnSortedMap.java b/src/java/org/apache/cassandra/io/util/ColumnSortedMap.java index 98f7abe668..76aae7a386 100644 --- a/src/java/org/apache/cassandra/io/util/ColumnSortedMap.java +++ b/src/java/org/apache/cassandra/io/util/ColumnSortedMap.java @@ -1,4 +1,25 @@ package org.apache.cassandra.io.util; +/* + * + * 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. + * + */ + import java.io.DataInput; import java.io.IOException; @@ -261,4 +282,4 @@ class ColumnIterator implements Iterator> { throw new UnsupportedOperationException(); } -} \ No newline at end of file +} diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index 7c13068ceb..1a560a02de 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -25,6 +25,7 @@ import java.util.*; import com.google.common.collect.HashMultimap; import com.google.common.collect.Multimap; +import org.apache.cassandra.gms.Gossiper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -163,6 +164,12 @@ public abstract class AbstractReplicationStrategy { if (map.containsKey(ep)) continue; + if (!StorageProxy.shouldHint(ep)) + { + if (logger.isDebugEnabled()) + logger.debug("not hinting " + ep + " which has been down " + Gossiper.instance.getEndpointDowntime(ep) + "ms"); + continue; + } InetAddress destination = map.isEmpty() ? localAddress diff --git a/src/java/org/apache/cassandra/service/RepairCallback.java b/src/java/org/apache/cassandra/service/RepairCallback.java index 7c485baba7..8ddd4849c3 100644 --- a/src/java/org/apache/cassandra/service/RepairCallback.java +++ b/src/java/org/apache/cassandra/service/RepairCallback.java @@ -1,4 +1,25 @@ package org.apache.cassandra.service; +/* + * + * 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. + * + */ + import java.io.IOException; import java.net.InetAddress; diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index abb3d8dd3a..395f165370 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -83,6 +83,7 @@ public class StorageProxy implements StorageProxyMBean // consistency > CL.ONE involves a read in the write path private static final LatencyTracker counterWriteStats = new LatencyTracker(); private static boolean hintedHandoffEnabled = DatabaseDescriptor.hintedHandoffEnabled(); + private static int maxHintWindow = DatabaseDescriptor.getMaxHintWindow(); private static final String UNREACHABLE = "UNREACHABLE"; private static final WritePerformer standardWritePerformer; @@ -528,18 +529,17 @@ public class StorageProxy implements StorageProxyMBean ReadCallback handler = getReadCallback(resolver, command.table, consistency_level); handler.assureSufficientLiveNodes(endpoints); - int targets; + // if we're not going to read repair, cut the endpoints list down to the ones required to satisfy ConsistencyLevel if (randomlyReadRepair(command)) { - targets = endpoints.size(); - if (targets > handler.blockfor) + if (endpoints.size() > handler.blockfor) repairs.add(command); } else { - targets = handler.blockfor; + endpoints = endpoints.subList(0, handler.blockfor); } - Message[] messages = new Message[targets]; + Message[] messages = new Message[endpoints.size()]; // data-request message is sent to dataPoint, the node that will actually get // the data for us. The other replicas are only sent a digest query. @@ -1001,6 +1001,21 @@ public class StorageProxy implements StorageProxyMBean return hintedHandoffEnabled; } + public int getMaxHintWindow() + { + return maxHintWindow; + } + + public void setMaxHintWindow(int ms) + { + maxHintWindow = ms; + } + + public static boolean shouldHint(InetAddress ep) + { + return Gossiper.instance.getEndpointDowntime(ep) <= maxHintWindow; + } + /** * Performs the truncate operatoin, which effectively deletes all data from * the column family cfname diff --git a/src/java/org/apache/cassandra/service/StorageProxyMBean.java b/src/java/org/apache/cassandra/service/StorageProxyMBean.java index b3b9b44778..116c5e7fae 100644 --- a/src/java/org/apache/cassandra/service/StorageProxyMBean.java +++ b/src/java/org/apache/cassandra/service/StorageProxyMBean.java @@ -46,4 +46,6 @@ public interface StorageProxyMBean public boolean getHintedHandoffEnabled(); public void setHintedHandoffEnabled(boolean b); + public int getMaxHintWindow(); + public void setMaxHintWindow(int ms); } diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index ea89e22ee9..7a3a11b126 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -627,22 +627,24 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe logger_.info("Node " + endpoint + " state jump to normal"); // we don't want to update if this node is responsible for the token and it has a later startup time than endpoint. - InetAddress currentNode = tokenMetadata_.getEndpoint(token); - if (currentNode == null) + InetAddress currentOwner = tokenMetadata_.getEndpoint(token); + if (currentOwner == null) { logger_.debug("New node " + endpoint + " at token " + token); tokenMetadata_.updateNormalToken(token, endpoint); if (!isClientMode) SystemTable.updateToken(endpoint, token); } - else if (endpoint.equals(currentNode)) + else if (endpoint.equals(currentOwner)) { - // nothing to do + // set state back to normal, since the node may have tried to leave, but failed and is now back up + // no need to persist, token/ip did not change + tokenMetadata_.updateNormalToken(token, endpoint); } - else if (Gossiper.instance.compareEndpointStartup(endpoint, currentNode) > 0) + else if (Gossiper.instance.compareEndpointStartup(endpoint, currentOwner) > 0) { logger_.info(String.format("Nodes %s and %s have the same token %s. %s is the new owner", - endpoint, currentNode, token, endpoint)); + endpoint, currentOwner, token, endpoint)); tokenMetadata_.updateNormalToken(token, endpoint); if (!isClientMode) SystemTable.updateToken(endpoint, token); @@ -650,7 +652,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe else { logger_.info(String.format("Nodes %s and %s have the same token %s. Ignoring %s", - endpoint, currentNode, token, endpoint)); + endpoint, currentOwner, token, endpoint)); } if (pieces.length > 2) diff --git a/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java b/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java index 5acdaf08a0..ad59e7c257 100644 --- a/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java +++ b/src/java/org/apache/cassandra/utils/BloomFilterSerializer.java @@ -1,4 +1,25 @@ package org.apache.cassandra.utils; +/* + * + * 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. + * + */ + import java.io.DataInputStream; import java.io.DataOutputStream; diff --git a/src/java/org/apache/cassandra/utils/LegacyBloomFilterSerializer.java b/src/java/org/apache/cassandra/utils/LegacyBloomFilterSerializer.java index a4add6916d..62b2df62e1 100644 --- a/src/java/org/apache/cassandra/utils/LegacyBloomFilterSerializer.java +++ b/src/java/org/apache/cassandra/utils/LegacyBloomFilterSerializer.java @@ -1,4 +1,25 @@ package org.apache.cassandra.utils; +/* + * + * 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. + * + */ + import java.util.BitSet; import java.io.DataInputStream;