natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

merge w/ 0.7 branch 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1038864 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Eric Evans 2010-11-24 23:04:38 +00:00
parent 84cd72b34d 5b22841e53
commit 930b4658b3
16 changed files with 142 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGES.txt
View File

@ -8,6 +8,9 @@ dev
* require index_type to be present when specifying index_name
on ColumnDef (CASSANDRA-1759)
* fix add/remove index bugs in CFMetadata (CASSANDRA-1768)
* rebuild Strategy during system_update_keyspace (CASSANDRA-1762)
* cli updates prompt to ... in continuation lines (CASSANDRA-1770)
* improvements to Debian init script (CASSANDRA-1772)
0.7.0-rc1

100
bin/json2sstable
View File

@ -1,50 +1,50 @@
#!/bin/sh
# 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.
if [ "x$CASSANDRA_INCLUDE" = "x" ]; then
for include in /usr/share/cassandra/cassandra.in.sh \
/usr/local/share/cassandra/cassandra.in.sh \
/opt/cassandra/cassandra.in.sh \
~/.cassandra.in.sh \
`dirname $0`/cassandra.in.sh; do
if [ -r $include ]; then
. $include
break
fi
done
elif [ -r $CASSANDRA_INCLUDE ]; then
. $CASSANDRA_INCLUDE
fi
# Use JAVA_HOME if set, otherwise look for java in PATH
if [ -x $JAVA_HOME/bin/java ]; then
JAVA=$JAVA_HOME/bin/java
else
JAVA=`which java`
fi
if [ -z $CLASSPATH ]; then
echo "You must set the CLASSPATH var" >&2
exit 1
fi
$JAVA -cp $CLASSPATH -Dstorage-config=$CASSANDRA_CONF \
-Dlog4j.configuration=log4j-tools.properties \
org.apache.cassandra.tools.SSTableImport "$@"
# vi:ai sw=4 ts=4 tw=0 et
#!/bin/sh
# 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.
if [ "x$CASSANDRA_INCLUDE" = "x" ]; then
for include in /usr/share/cassandra/cassandra.in.sh \
/usr/local/share/cassandra/cassandra.in.sh \
/opt/cassandra/cassandra.in.sh \
~/.cassandra.in.sh \
`dirname $0`/cassandra.in.sh; do
if [ -r $include ]; then
. $include
break
fi
done
elif [ -r $CASSANDRA_INCLUDE ]; then
. $CASSANDRA_INCLUDE
fi
# Use JAVA_HOME if set, otherwise look for java in PATH
if [ -x $JAVA_HOME/bin/java ]; then
JAVA=$JAVA_HOME/bin/java
else
JAVA=`which java`
fi
if [ -z $CLASSPATH ]; then
echo "You must set the CLASSPATH var" >&2
exit 1
fi
$JAVA -cp $CLASSPATH -Dstorage-config=$CASSANDRA_CONF \
-Dlog4j.configuration=log4j-tools.properties \
org.apache.cassandra.tools.SSTableImport "$@"
# vi:ai sw=4 ts=4 tw=0 et

35
debian/init vendored
View File

@ -18,6 +18,7 @@ PIDFILE=/var/run/$NAME.pid
SCRIPTNAME=/etc/init.d/$NAME
CONFDIR=/etc/cassandra
JSVC=/usr/bin/jsvc
WAIT_FOR_START=10
# The first existing directory is used for JAVA_HOME if needed.
JVM_SEARCH_DIRS="/usr/lib/jvm/java-6-openjdk /usr/lib/jvm/java-6-sun"
@ -87,16 +88,21 @@ classpath()
}
#
# Function that returns 0 if process is running, 1 if not
# Function that returns 0 if process is running, or nonzero if not.
#
# The nonzero value is 3 if the process is simply not running, and 1 if the
# process is not running but the pidfile exists (to match the exit codes for
# the "status" command; see LSB core spec 3.1, section 20.2)
#
CMD_PATT="-user.cassandra.+CassandraDaemon"
is_running()
{
if [ -f $PIDFILE ]; then
if ps -p `cat $PIDFILE` &> /dev/null; then
return 0
fi
pid=`cat $PIDFILE`
grep -Eq "$CMD_PATT" "/proc/$pid/cmdline" 2>/dev/null && return 0
return 1
fi
return 1
return 3
}
#
@ -121,7 +127,12 @@ do_start()
$JVM_OPTS \
org.apache.cassandra.thrift.CassandraDaemon
if ! is_running; then return 2; fi
is_running && return 0
for tries in `seq $WAIT_FOR_START`; do
sleep 1
is_running && return 0
done
return 2
}
#
@ -175,8 +186,18 @@ case "$1" in
;;
esac
;;
status)
is_running
stat=$?
case "$stat" in
0) log_success_msg "$DESC is running" ;;
1) log_failure_msg "could not access pidfile for $DESC" ;;
*) log_success_msg "$DESC is not running" ;;
esac
exit "$stat"
;;
*)
echo "Usage: $SCRIPTNAME {start|stop|restart|force-reload}" >&2
echo "Usage: $SCRIPTNAME {start|stop|restart|force-reload|status}" >&2
exit 3
;;
esac

6
src/java/org/apache/cassandra/cli/CliMain.java
View File

@ -183,7 +183,7 @@ public class CliMain
private static void printBanner()
{
sessionState.out.println("Welcome to cassandra CLI.\n");
sessionState.out.println("Type 'help' or '?' for help. Type 'quit' or 'exit' to quit.");
sessionState.out.println("Type 'help;' or '?' for help. Type 'quit;' or 'exit;' to quit.");
}
/**
@ -323,7 +323,7 @@ public class CliMain
while (line != null)
{
prompt = (inCompoundStatement) ? "\t" : getPrompt(cliClient);
prompt = (inCompoundStatement) ? "...\t" : getPrompt(cliClient);
line = reader.readLine(prompt).trim();
@ -332,7 +332,7 @@ public class CliMain
currentStatement += line;
if (line.endsWith(";"))
if (line.endsWith(";") || line.equals("?"))
{
processStatement(currentStatement);
currentStatement = "";

27
src/java/org/apache/cassandra/db/Table.java
View File

@ -96,7 +96,7 @@ public class Table
public final Map<Integer, ColumnFamilyStore> columnFamilyStores = new HashMap<Integer, ColumnFamilyStore>(); // TODO make private again
private final Object[] indexLocks;
private ScheduledFuture<?> flushTask;
public final AbstractReplicationStrategy replicationStrategy;
private volatile AbstractReplicationStrategy replicationStrategy;
public static Table open(String table)
{
@ -244,11 +244,7 @@ public class Table
KSMetaData ksm = DatabaseDescriptor.getKSMetaData(table);
try
{
replicationStrategy = AbstractReplicationStrategy.createReplicationStrategy(table,
ksm.strategyClass,
StorageService.instance.getTokenMetadata(),
DatabaseDescriptor.getEndpointSnitch(),
ksm.strategyOptions);
createReplicationStrategy(ksm);
}
catch (ConfigurationException e)
{
@ -302,7 +298,19 @@ public class Table
};
flushTask = StorageService.scheduledTasks.scheduleWithFixedDelay(runnable, minCheckMs, minCheckMs, TimeUnit.MILLISECONDS);
}
public void createReplicationStrategy(KSMetaData ksm) throws ConfigurationException
{
if (replicationStrategy != null)
StorageService.instance.getTokenMetadata().unregister(replicationStrategy);
replicationStrategy = AbstractReplicationStrategy.createReplicationStrategy(ksm.name,
ksm.strategyClass,
StorageService.instance.getTokenMetadata(),
DatabaseDescriptor.getEndpointSnitch(),
ksm.strategyOptions);
}
// best invoked on the compaction mananger.
public void dropCf(Integer cfId) throws IOException
{
@ -557,6 +565,11 @@ public class Table
return new IndexBuilder(cfs, columns, iter);
}
public AbstractReplicationStrategy getReplicationStrategy()
{
return replicationStrategy;
}
public class IndexBuilder implements ICompactionInfo
{
private final ColumnFamilyStore cfs;

14
src/java/org/apache/cassandra/db/migration/UpdateKeyspace.java
View File

@ -5,6 +5,7 @@ import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
@ -59,7 +60,18 @@ public class UpdateKeyspace extends Migration
{
DatabaseDescriptor.clearTableDefinition(oldKsm, newVersion);
DatabaseDescriptor.setTableDefinition(newKsm, newVersion);
Table.open(newKsm.name).replicationStrategy.clearEndpointCache();
Table table = Table.open(newKsm.name);
try
{
table.createReplicationStrategy(newKsm);
}
catch (ConfigurationException e)
{
throw new IOException(e);
}
logger.info("Keyspace updated. Please perform any manual operations.");
}

2
src/java/org/apache/cassandra/dht/BootStrapper.java
View File

@ -191,7 +191,7 @@ public class BootStrapper
Multimap<Range, InetAddress> getRangesWithSources(String table)
{
assert tokenMetadata.sortedTokens().size() > 0;
final AbstractReplicationStrategy strat = Table.open(table).replicationStrategy;
final AbstractReplicationStrategy strat = Table.open(table).getReplicationStrategy();
Collection<Range> myRanges = strat.getPendingAddressRanges(tokenMetadata, token, address);
Multimap<Range, InetAddress> myRangeAddresses = ArrayListMultimap.create();

5
src/java/org/apache/cassandra/locator/TokenMetadata.java
View File

@ -540,6 +540,11 @@ public class TokenMetadata
subscribers.add(subscriber);
}
public void unregister(AbstractReplicationStrategy subscriber)
{
subscribers.remove(subscriber);
}
/**
* write endpoints may be different from read endpoints, because read endpoints only need care about the
* "natural" nodes for a token, but write endpoints also need to account for nodes that are bootstrapping

2
src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java
View File

@ -65,7 +65,7 @@ public class DatacenterQuorumResponseHandler<T> extends QuorumResponseHandler<T>
@Override
public int determineBlockFor(ConsistencyLevel consistency_level, String table)
{
NetworkTopologyStrategy stategy = (NetworkTopologyStrategy) Table.open(table).replicationStrategy;
NetworkTopologyStrategy stategy = (NetworkTopologyStrategy) Table.open(table).getReplicationStrategy();
return (stategy.getReplicationFactor(localdc) / 2) + 1;
}
}

2
src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java
View File

@ -63,7 +63,7 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHan
super(writeEndpoints, hintedEndpoints, consistencyLevel);
assert consistencyLevel == ConsistencyLevel.LOCAL_QUORUM;
strategy = (NetworkTopologyStrategy) Table.open(table).replicationStrategy;
strategy = (NetworkTopologyStrategy) Table.open(table).getReplicationStrategy();
for (String dc : strategy.getDatacenters())
{

2
src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java
View File

@ -65,7 +65,7 @@ public class DatacenterWriteResponseHandler extends WriteResponseHandler
@Override
protected int determineBlockFor(String table)
{
NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) Table.open(table).replicationStrategy;
NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) Table.open(table).getReplicationStrategy();
return (strategy.getReplicationFactor(localdc) / 2) + 1;
}

10
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -105,7 +105,7 @@ public class StorageProxy implements StorageProxyMBean
{
mostRecentRowMutation = rm;
String table = rm.getTable();
AbstractReplicationStrategy rs = Table.open(table).replicationStrategy;
AbstractReplicationStrategy rs = Table.open(table).getReplicationStrategy();
List<InetAddress> naturalEndpoints = ss.getNaturalEndpoints(table, rm.key());
Collection<InetAddress> writeEndpoints = ss.getTokenMetadata().getWriteEndpoints(StorageService.getPartitioner().getToken(rm.key()), table, naturalEndpoints);
@ -343,7 +343,7 @@ public class StorageProxy implements StorageProxyMBean
if (logger.isDebugEnabled())
logger.debug("strongread reading " + (m == message ? "data" : "digest") + " for " + command + " from " + m.getMessageId() + "@" + endpoint);
}
AbstractReplicationStrategy rs = Table.open(command.table).replicationStrategy;
AbstractReplicationStrategy rs = Table.open(command.table).getReplicationStrategy();
QuorumResponseHandler<Row> quorumResponseHandler = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), consistency_level);
MessagingService.instance.sendRR(messages, endpoints, quorumResponseHandler);
quorumResponseHandlers.add(quorumResponseHandler);
@ -369,7 +369,7 @@ public class StorageProxy implements StorageProxyMBean
}
catch (DigestMismatchException ex)
{
AbstractReplicationStrategy rs = Table.open(command.table).replicationStrategy;
AbstractReplicationStrategy rs = Table.open(command.table).getReplicationStrategy();
QuorumResponseHandler<Row> qrhRepair = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), ConsistencyLevel.QUORUM);
if (logger.isDebugEnabled())
logger.debug("Digest mismatch:", ex);
@ -449,7 +449,7 @@ public class StorageProxy implements StorageProxyMBean
// collect replies and resolve according to consistency level
RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints);
AbstractReplicationStrategy rs = Table.open(command.keyspace).replicationStrategy;
AbstractReplicationStrategy rs = Table.open(command.keyspace).getReplicationStrategy();
QuorumResponseHandler<List<Row>> handler = rs.getQuorumResponseHandler(resolver, consistency_level);
// TODO bail early if live endpoints can't satisfy requested consistency level
for (InetAddress endpoint : liveEndpoints)
@ -663,7 +663,7 @@ public class StorageProxy implements StorageProxyMBean
// collect replies and resolve according to consistency level
RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(keyspace, liveEndpoints);
AbstractReplicationStrategy rs = Table.open(keyspace).replicationStrategy;
AbstractReplicationStrategy rs = Table.open(keyspace).getReplicationStrategy();
QuorumResponseHandler<List<Row>> handler = rs.getQuorumResponseHandler(resolver, consistency_level);
// bail early if live endpoints can't satisfy requested consistency level

16
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -558,7 +558,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
Map<Range, List<InetAddress>> rangeToEndpointMap = new HashMap<Range, List<InetAddress>>();
for (Range range : ranges)
{
rangeToEndpointMap.put(range, Table.open(keyspace).replicationStrategy.getNaturalEndpoints(range.right));
rangeToEndpointMap.put(range, Table.open(keyspace).getReplicationStrategy().getNaturalEndpoints(range.right));
}
return rangeToEndpointMap;
}
@ -824,7 +824,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
private void calculatePendingRanges()
{
for (String table : DatabaseDescriptor.getNonSystemTables())
calculatePendingRanges(Table.open(table).replicationStrategy, table);
calculatePendingRanges(Table.open(table).getReplicationStrategy(), table);
}
// public & static for testing purposes
@ -894,7 +894,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
private Multimap<InetAddress, Range> getNewSourceRanges(String table, Set<Range> ranges)
{
InetAddress myAddress = FBUtilities.getLocalAddress();
Multimap<Range, InetAddress> rangeAddresses = Table.open(table).replicationStrategy.getRangeAddresses(tokenMetadata_);
Multimap<Range, InetAddress> rangeAddresses = Table.open(table).getReplicationStrategy().getRangeAddresses(tokenMetadata_);
Multimap<InetAddress, Range> sourceRanges = HashMultimap.create();
IFailureDetector failureDetector = FailureDetector.instance;
@ -1017,7 +1017,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
// Find (for each range) all nodes that store replicas for these ranges as well
for (Range range : ranges)
currentReplicaEndpoints.put(range, Table.open(table).replicationStrategy.calculateNaturalEndpoints(range.right, tokenMetadata_));
currentReplicaEndpoints.put(range, Table.open(table).getReplicationStrategy().calculateNaturalEndpoints(range.right, tokenMetadata_));
TokenMetadata temp = tokenMetadata_.cloneAfterAllLeft();
@ -1035,7 +1035,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
// range.
for (Range range : ranges)
{
Collection<InetAddress> newReplicaEndpoints = Table.open(table).replicationStrategy.calculateNaturalEndpoints(range.right, temp);
Collection<InetAddress> newReplicaEndpoints = Table.open(table).getReplicationStrategy().calculateNaturalEndpoints(range.right, temp);
newReplicaEndpoints.removeAll(currentReplicaEndpoints.get(range));
if (logger_.isDebugEnabled())
if (newReplicaEndpoints.isEmpty())
@ -1359,7 +1359,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
*/
Collection<Range> getRangesForEndpoint(String table, InetAddress ep)
{
return Table.open(table).replicationStrategy.getAddressRanges().get(ep);
return Table.open(table).getReplicationStrategy().getAddressRanges().get(ep);
}
/**
@ -1409,7 +1409,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
*/
public List<InetAddress> getNaturalEndpoints(String table, Token token)
{
return Table.open(table).replicationStrategy.getNaturalEndpoints(token);
return Table.open(table).getReplicationStrategy().getNaturalEndpoints(token);
}
/**
@ -1427,7 +1427,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
public List<InetAddress> getLiveNaturalEndpoints(String table, Token token)
{
List<InetAddress> liveEps = new ArrayList<InetAddress>();
List<InetAddress> endpoints = Table.open(table).replicationStrategy.getNaturalEndpoints(token);
List<InetAddress> endpoints = Table.open(table).getReplicationStrategy().getNaturalEndpoints(token);
for (InetAddress endpoint : endpoints)
{

2
test/unit/org/apache/cassandra/locator/ReplicationStrategyEndpointCacheTest.java
View File

@ -43,7 +43,7 @@ public class ReplicationStrategyEndpointCacheTest extends SchemaLoader
tmd = new TokenMetadata();
searchToken = new BigIntegerToken(String.valueOf(15));
strategy = getStrategyWithNewTokenMetadata(Table.open("Keyspace3").replicationStrategy, tmd);
strategy = getStrategyWithNewTokenMetadata(Table.open("Keyspace3").getReplicationStrategy(), tmd);
tmd.updateNormalToken(new BigIntegerToken(String.valueOf(10)), InetAddress.getByName("127.0.0.1"));
tmd.updateNormalToken(new BigIntegerToken(String.valueOf(20)), InetAddress.getByName("127.0.0.2"));

2
test/unit/org/apache/cassandra/locator/SimpleStrategyTest.java
View File

@ -43,7 +43,7 @@ public class SimpleStrategyTest extends SchemaLoader
@Test
public void tryValidTable()
{
assert Table.open("Keyspace1").replicationStrategy != null;
assert Table.open("Keyspace1").getReplicationStrategy() != null;
}
@Test

2
test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java
View File

@ -183,7 +183,7 @@ public class AntiEntropyServiceTest extends CleanupHelper
// generate rf*2 nodes, and ensure that only neighbors specified by the ARS are returned
addTokens(2 * DatabaseDescriptor.getReplicationFactor(tablename));
AbstractReplicationStrategy ars = Table.open(tablename).replicationStrategy;
AbstractReplicationStrategy ars = Table.open(tablename).getReplicationStrategy();
Set<InetAddress> expected = new HashSet<InetAddress>();
for (Range replicaRange : ars.getAddressRanges().get(FBUtilities.getLocalAddress()))
{