mirror of https://github.com/apache/cassandra
merge from 0.8
git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1153198 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis
parent
8274a9a351
commit
2816bd471f
18
CHANGES.txt
18
CHANGES.txt
|
|
@ -22,8 +22,6 @@
|
|||
* check column family validity in nodetool repair (CASSANDRA-2933)
|
||||
* use lazy initialization instead of class initialization in NodeId
|
||||
(CASSANDRA-2953)
|
||||
* fix potential use of free'd native memory in SerializingCache
|
||||
(CASSANDRA-2951)
|
||||
* add paging to get_count (CASSANDRA-2894)
|
||||
* fix "short reads" in [multi]get (CASSANDRA-2643)
|
||||
|
||||
|
|
@ -31,7 +29,7 @@
|
|||
0.8.3
|
||||
* add ability to drop local reads/writes that are going to timeout
|
||||
(CASSANDRA-2943)
|
||||
* revamp token removal process, keep gossip states for 3 days (CASSANDRA-2946)
|
||||
* revamp token removal process, keep gossip states for 3 days (CASSANDRA-2496)
|
||||
* don't accept extra args for 0-arg nodetool commands (CASSANDRA-2740)
|
||||
* log unavailableexception details at debug level (CASSANDRA-2856)
|
||||
* expose data_dir though jmx (CASSANDRA-2770)
|
||||
|
|
@ -39,13 +37,25 @@
|
|||
* log Java classpath on startup (CASSANDRA-2895)
|
||||
* keep gossipped version in sync with actual on migration coordinator
|
||||
(CASSANDRA-2946)
|
||||
* use lazy initialization instead of class initialization in NodeId
|
||||
(CASSANDRA-2953)
|
||||
* check column family validity in nodetool repair (CASSANDRA-2933)
|
||||
* speedup bytes to hex conversions dramatically (CASSANDRA-2850)
|
||||
* fix bug where dirty commit logs were removed (and avoid keeping segment
|
||||
* Flush memtables on shutdown when durable writes are disabled
|
||||
(CASSANDRA-2958)
|
||||
* improved POSIX compatibility of start scripts (CASsANDRA-2965)
|
||||
* add counter support to Hadoop InputFormat (CASSANDRA-2981)
|
||||
* fix bug where dirty commit logs were removed (and avoid keeping segments
|
||||
with no post-flush activity permanently dirty) (CASSANDRA-2829)
|
||||
* fix throwing exception with batch mutation of counter super columns
|
||||
(CASSANDRA-2949)
|
||||
* ignore system tables during repair (CASSANDRA-2979)
|
||||
* throw exception when NTS is given replication_factor as an option
|
||||
(CASSANDRA-2960)
|
||||
* fix assertion error during compaction of counter CFs (CASSANDRA-2968)
|
||||
* avoid trying to create index names, when no index exists (CASSANDRA-2867)
|
||||
* fix potential use of free'd native memory in SerializingCache
|
||||
(CASSANDRA-2951)
|
||||
|
||||
|
||||
0.8.2
|
||||
|
|
|
|||
|
|
@ -61,32 +61,35 @@
|
|||
# Be aware that you will be entirely responsible for populating the needed
|
||||
# environment variables.
|
||||
|
||||
# NB: Developers should be aware that this script should remain compatible with
|
||||
# POSIX sh and Solaris sh. This means, in particular, no $(( )) and no $( ).
|
||||
|
||||
# If an include wasn't specified in the environment, then search for one...
|
||||
if [ "x$CASSANDRA_INCLUDE" = "x" ]; then
|
||||
# Locations (in order) to use when searching for an include file.
|
||||
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
|
||||
"$HOME/.cassandra.in.sh" \
|
||||
"`dirname $0`/cassandra.in.sh"; do
|
||||
if [ -r "$include" ]; then
|
||||
. "$include"
|
||||
break
|
||||
fi
|
||||
done
|
||||
# ...otherwise, source the specified include.
|
||||
elif [ -r $CASSANDRA_INCLUDE ]; then
|
||||
. $CASSANDRA_INCLUDE
|
||||
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
|
||||
if [ -n "$JAVA_HOME" ]; then
|
||||
JAVA="$JAVA_HOME/bin/java"
|
||||
else
|
||||
JAVA=`which java`
|
||||
JAVA=java
|
||||
fi
|
||||
|
||||
if [ -z $CASSANDRA_CONF -o -z $CLASSPATH ]; then
|
||||
if [ -z "$CASSANDRA_CONF" -o -z "$CLASSPATH" ]; then
|
||||
echo "You must set the CASSANDRA_CONF and CLASSPATH vars" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
|
@ -119,11 +122,11 @@ launch_service()
|
|||
# to close stdout/stderr, but it's up to us not to background.
|
||||
if [ "x$foreground" != "x" ]; then
|
||||
cassandra_parms="$cassandra_parms -Dcassandra-foreground=yes"
|
||||
exec $JAVA $JVM_OPTS $cassandra_parms -cp $CLASSPATH $props $class
|
||||
exec "$JAVA" $JVM_OPTS $cassandra_parms -cp "$CLASSPATH" $props "$class"
|
||||
# Startup CassandraDaemon, background it, and write the pid.
|
||||
else
|
||||
exec $JAVA $JVM_OPTS $cassandra_parms -cp $CLASSPATH $props $class <&- &
|
||||
[ ! -z $pidpath ] && printf "%d" $! > $pidpath
|
||||
exec "$JAVA" $JVM_OPTS $cassandra_parms -cp "$CLASSPATH" $props "$class" <&- &
|
||||
[ ! -z "$pidpath" ] && printf "%d" $! > "$pidpath"
|
||||
fi
|
||||
|
||||
return $?
|
||||
|
|
@ -150,7 +153,7 @@ while true; do
|
|||
exit 0
|
||||
;;
|
||||
-v)
|
||||
$JAVA -cp $CLASSPATH org.apache.cassandra.tools.GetVersion
|
||||
"$JAVA" -cp "$CLASSPATH" org.apache.cassandra.tools.GetVersion
|
||||
exit 0
|
||||
;;
|
||||
-D)
|
||||
|
|
|
|||
|
|
@ -24,10 +24,15 @@ calculate_heap_sizes()
|
|||
;;
|
||||
FreeBSD)
|
||||
system_memory_in_bytes=`sysctl hw.physmem | awk '{print $2}'`
|
||||
system_memory_in_mb=$((system_memory_in_bytes / 1024 / 1024))
|
||||
system_memory_in_mb=`expr $system_memory_in_bytes / 1024 / 1024`
|
||||
system_cpu_cores=`sysctl hw.ncpu | awk '{print $2}'`
|
||||
break
|
||||
;;
|
||||
SunOS)
|
||||
system_memory_in_mb=`prtconf | awk '/Memory size:/ {print $3}'`
|
||||
system_cpu_cores=`psrinfo | wc -l`
|
||||
break
|
||||
;;
|
||||
*)
|
||||
# assume reasonable defaults for e.g. a modern desktop or
|
||||
# cheap server
|
||||
|
|
@ -35,14 +40,14 @@ calculate_heap_sizes()
|
|||
system_cpu_cores="2"
|
||||
;;
|
||||
esac
|
||||
max_heap_size_in_mb=$((system_memory_in_mb / 2))
|
||||
max_heap_size_in_mb=`expr $system_memory_in_mb / 2`
|
||||
MAX_HEAP_SIZE="${max_heap_size_in_mb}M"
|
||||
|
||||
# Young gen: min(max_sensible_per_modern_cpu_core * num_cores, 1/4 * heap size)
|
||||
max_sensible_yg_per_core_in_mb="100"
|
||||
max_sensible_yg_in_mb=$((max_sensible_yg_per_core_in_mb * system_cpu_cores))
|
||||
max_sensible_yg_in_mb=`expr $max_sensible_yg_per_core_in_mb "*" $system_cpu_cores`
|
||||
|
||||
desired_yg_in_mb=$((max_heap_size_in_mb / 4))
|
||||
desired_yg_in_mb=`expr $max_heap_size_in_mb / 4`
|
||||
|
||||
if [ "$desired_yg_in_mb" -gt "$max_sensible_yg_in_mb" ]
|
||||
then
|
||||
|
|
@ -92,7 +97,7 @@ JMX_PORT="7199"
|
|||
JVM_OPTS="$JVM_OPTS -ea"
|
||||
|
||||
# add the jamm javaagent
|
||||
check_openjdk=$(java -version 2>&1 | awk '{if (NR == 2) {print $1}}')
|
||||
check_openjdk=`"${JAVA:-java}" -version 2>&1 | awk '{if (NR == 2) {print $1}}'`
|
||||
if [ "$check_openjdk" != "OpenJDK" ]
|
||||
then
|
||||
JVM_OPTS="$JVM_OPTS -javaagent:$CASSANDRA_HOME/lib/jamm-0.2.2.jar"
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ case "$1" in
|
|||
cassandra
|
||||
fi
|
||||
|
||||
if [ -n $2 ] && dpkg --compare-versions "$2" le 0.6.4-2; then
|
||||
if [ -n "$2" ] && dpkg --compare-versions "$2" le 0.6.4-2; then
|
||||
chown -R cassandra: /var/lib/cassandra
|
||||
chown -R cassandra: /var/log/cassandra
|
||||
fi
|
||||
|
|
|
|||
|
|
@ -30,23 +30,15 @@ JVM_SEARCH_DIRS="/usr/lib/jvm/java-6-openjdk /usr/lib/jvm/java-6-sun"
|
|||
[ -e /etc/cassandra/cassandra.yaml ] || exit 0
|
||||
[ -e /etc/cassandra/cassandra-env.sh ] || exit 0
|
||||
|
||||
# Read Cassandra environment file.
|
||||
. /etc/cassandra/cassandra-env.sh
|
||||
|
||||
# Read configuration variable file if it is present
|
||||
[ -r /etc/default/$NAME ] && . /etc/default/$NAME
|
||||
|
||||
if [ -z "$JVM_OPTS" ]; then
|
||||
echo "Initialization failed; \$JVM_OPTS not set!" >&2
|
||||
exit 3
|
||||
fi
|
||||
|
||||
# If JAVA_HOME has not been set, try to determine it.
|
||||
if [ -z "$JAVA_HOME" ]; then
|
||||
# If java is in PATH, use a JAVA_HOME that corresponds to that. This is
|
||||
# both consistent with how the upstream startup script works, and how
|
||||
# Debian works (read: the use of alternatives to set a system JVM).
|
||||
if [ -n `which java` ]; then
|
||||
if [ -n "`which java`" ]; then
|
||||
java=`which java`
|
||||
# Dereference symlink(s)
|
||||
while true; do
|
||||
|
|
@ -67,6 +59,15 @@ if [ -z "$JAVA_HOME" ]; then
|
|||
done
|
||||
fi
|
||||
fi
|
||||
JAVA="$JAVA_HOME/bin/java"
|
||||
|
||||
# Read Cassandra environment file.
|
||||
. /etc/cassandra/cassandra-env.sh
|
||||
|
||||
if [ -z "$JVM_OPTS" ]; then
|
||||
echo "Initialization failed; \$JVM_OPTS not set!" >&2
|
||||
exit 3
|
||||
fi
|
||||
|
||||
# Load the VERBOSE setting and other rcS variables
|
||||
. /lib/init/vars.sh
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
WordCount hadoop example: Inserts a bunch of words across multiple rows,
|
||||
and counts them, with RandomPartitioner.
|
||||
and counts them, with RandomPartitioner. The word_count_counters example sums
|
||||
the value of counter columns for a key.
|
||||
|
||||
The scripts in bin/ assume you are running with cwd of contrib/word_count.
|
||||
|
||||
|
|
@ -9,6 +10,7 @@ then run
|
|||
contrib/word_count$ ant
|
||||
contrib/word_count$ bin/word_count_setup
|
||||
contrib/word_count$ bin/word_count
|
||||
contrib/word_count$ bin/word_count_counters
|
||||
|
||||
In order to view the results in Cassandra, one can use bin/cassandra-cli and
|
||||
perform the following operations:
|
||||
|
|
@ -25,5 +27,8 @@ in the 'wordcount' keyspace. 'cassandra' is the default.
|
|||
|
||||
Read the code in src/ for more details.
|
||||
|
||||
The word_count_counters example sums the counter columns for a row. The output
|
||||
is written to a text file in /tmp/word_count_counters.
|
||||
|
||||
*If you want to point wordcount at a real cluster, modify the seed
|
||||
and listenaddress settings accordingly.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,58 @@
|
|||
#!/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.
|
||||
|
||||
cwd=`dirname $0`
|
||||
|
||||
# Cassandra class files.
|
||||
if [ ! -d $cwd/../../../build/classes/main ]; then
|
||||
echo "Unable to locate cassandra class files" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# word_count Jar.
|
||||
if [ ! -e $cwd/../build/word_count.jar ]; then
|
||||
echo "Unable to locate word_count jar" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
CLASSPATH=$CLASSPATH:$cwd/../build/word_count.jar
|
||||
CLASSPATH=$CLASSPATH:$cwd/../../../build/classes/main
|
||||
CLASSPATH=$CLASSPATH:$cwd/../../../build/classes/thrift
|
||||
for jar in $cwd/../build/lib/jars/*.jar; do
|
||||
CLASSPATH=$CLASSPATH:$jar
|
||||
done
|
||||
for jar in $cwd/../../../lib/*.jar; do
|
||||
CLASSPATH=$CLASSPATH:$jar
|
||||
done
|
||||
for jar in $cwd/../../../build/lib/jars/*.jar; do
|
||||
CLASSPATH=$CLASSPATH:$jar
|
||||
done
|
||||
|
||||
if [ -x $JAVA_HOME/bin/java ]; then
|
||||
JAVA=$JAVA_HOME/bin/java
|
||||
else
|
||||
JAVA=`which java`
|
||||
fi
|
||||
|
||||
if [ "x$JAVA" = "x" ]; then
|
||||
echo "Java executable not found (hint: set JAVA_HOME)" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
#echo $CLASSPATH
|
||||
$JAVA -Xmx1G -ea -cp $CLASSPATH WordCountCounters
|
||||
|
|
@ -0,0 +1,105 @@
|
|||
/**
|
||||
* 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.nio.ByteBuffer;
|
||||
import java.util.*;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.conf.Configured;
|
||||
import org.apache.hadoop.fs.Path;
|
||||
import org.apache.hadoop.io.Text;
|
||||
import org.apache.hadoop.io.LongWritable;
|
||||
import org.apache.hadoop.mapreduce.Job;
|
||||
import org.apache.hadoop.mapreduce.Mapper;
|
||||
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
|
||||
import org.apache.hadoop.util.Tool;
|
||||
import org.apache.hadoop.util.ToolRunner;
|
||||
|
||||
import org.apache.cassandra.db.IColumn;
|
||||
import org.apache.cassandra.hadoop.ColumnFamilyInputFormat;
|
||||
import org.apache.cassandra.hadoop.ConfigHelper;
|
||||
import org.apache.cassandra.thrift.*;
|
||||
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||
|
||||
/**
|
||||
* This sums the word count stored in the input_words_count ColumnFamily for the key "key-if-verse1".
|
||||
*
|
||||
* Output is written to a text file.
|
||||
*/
|
||||
public class WordCountCounters extends Configured implements Tool
|
||||
{
|
||||
private static final Logger logger = LoggerFactory.getLogger(WordCountCounters.class);
|
||||
|
||||
static final String COUNTER_COLUMN_FAMILY = "input_words_count";
|
||||
private static final String OUTPUT_PATH_PREFIX = "/tmp/word_count_counters";
|
||||
|
||||
|
||||
public static void main(String[] args) throws Exception
|
||||
{
|
||||
// Let ToolRunner handle generic command-line options
|
||||
ToolRunner.run(new Configuration(), new WordCountCounters(), args);
|
||||
System.exit(0);
|
||||
}
|
||||
|
||||
public static class SumMapper extends Mapper<ByteBuffer, SortedMap<ByteBuffer, IColumn>, Text, LongWritable>
|
||||
{
|
||||
public void map(ByteBuffer key, SortedMap<ByteBuffer, IColumn> columns, Context context) throws IOException, InterruptedException
|
||||
{
|
||||
long sum = 0;
|
||||
for (IColumn column : columns.values())
|
||||
{
|
||||
logger.debug("read " + key + ":" + column.name() + " from " + context.getInputSplit());
|
||||
sum += ByteBufferUtil.toLong(column.value());
|
||||
}
|
||||
context.write(new Text(ByteBufferUtil.string(key)), new LongWritable(sum));
|
||||
}
|
||||
}
|
||||
|
||||
public int run(String[] args) throws Exception
|
||||
{
|
||||
Job job = new Job(getConf(), "wordcountcounters");
|
||||
job.setJarByClass(WordCountCounters.class);
|
||||
job.setMapperClass(SumMapper.class);
|
||||
|
||||
job.setOutputKeyClass(Text.class);
|
||||
job.setOutputValueClass(LongWritable.class);
|
||||
FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH_PREFIX));
|
||||
|
||||
|
||||
job.setInputFormatClass(ColumnFamilyInputFormat.class);
|
||||
|
||||
|
||||
ConfigHelper.setRpcPort(job.getConfiguration(), "9160");
|
||||
ConfigHelper.setInitialAddress(job.getConfiguration(), "localhost");
|
||||
ConfigHelper.setPartitioner(job.getConfiguration(), "org.apache.cassandra.dht.RandomPartitioner");
|
||||
ConfigHelper.setInputColumnFamily(job.getConfiguration(), WordCount.KEYSPACE, WordCountCounters.COUNTER_COLUMN_FAMILY);
|
||||
SlicePredicate predicate = new SlicePredicate().setSlice_range(
|
||||
new SliceRange().
|
||||
setStart(ByteBufferUtil.EMPTY_BYTE_BUFFER).
|
||||
setFinish(ByteBufferUtil.EMPTY_BYTE_BUFFER).
|
||||
setCount(100));
|
||||
ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);
|
||||
|
||||
job.waitForCompletion(true);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
|
@ -81,6 +81,19 @@ public class WordCountSetup
|
|||
client.batch_mutate(mutationMap, ConsistencyLevel.ONE);
|
||||
logger.info("added text3");
|
||||
|
||||
// sentence data for the counters
|
||||
final ByteBuffer key = ByteBufferUtil.bytes("key-if-verse1");
|
||||
final ColumnParent colParent = new ColumnParent(WordCountCounters.COUNTER_COLUMN_FAMILY);
|
||||
for (String sentence : sentenceData())
|
||||
{
|
||||
client.add(key,
|
||||
colParent,
|
||||
new CounterColumn(ByteBufferUtil.bytes(sentence),
|
||||
(long)sentence.split("\\s").length),
|
||||
ConsistencyLevel.ONE );
|
||||
}
|
||||
logger.info("added key-if-verse1");
|
||||
|
||||
System.exit(0);
|
||||
}
|
||||
|
||||
|
|
@ -115,6 +128,10 @@ public class WordCountSetup
|
|||
output.setComparator_type("AsciiType");
|
||||
output.setDefault_validation_class("AsciiType");
|
||||
cfDefList.add(output);
|
||||
CfDef counterInput = new CfDef(WordCount.KEYSPACE, WordCountCounters.COUNTER_COLUMN_FAMILY);
|
||||
counterInput.setComparator_type("UTF8Type");
|
||||
counterInput.setDefault_validation_class("CounterColumnType");
|
||||
cfDefList.add(counterInput);
|
||||
|
||||
KsDef ksDef = new KsDef(WordCount.KEYSPACE, "org.apache.cassandra.locator.SimpleStrategy", cfDefList);
|
||||
ksDef.putToStrategy_options("replication_factor", "1");
|
||||
|
|
@ -150,4 +167,18 @@ public class WordCountSetup
|
|||
|
||||
return new Cassandra.Client(protocol);
|
||||
}
|
||||
|
||||
private static String[] sentenceData()
|
||||
{ // Public domain context, source http://en.wikisource.org/wiki/If%E2%80%94
|
||||
return new String[]{
|
||||
"If you can keep your head when all about you",
|
||||
"Are losing theirs and blaming it on you",
|
||||
"If you can trust yourself when all men doubt you,",
|
||||
"But make allowance for their doubting too:",
|
||||
"If you can wait and not be tired by waiting,",
|
||||
"Or being lied about, don’t deal in lies,",
|
||||
"Or being hated, don’t give way to hating,",
|
||||
"And yet don’t look too good, nor talk too wise;"
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,18 +25,14 @@ import java.io.DataOutput;
|
|||
import java.io.IOError;
|
||||
import java.io.IOException;
|
||||
import java.security.MessageDigest;
|
||||
import java.util.*;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
|
||||
import com.google.common.base.Predicates;
|
||||
import com.google.common.collect.Iterators;
|
||||
|
||||
import org.apache.cassandra.db.Column;
|
||||
import org.apache.cassandra.db.ColumnFamily;
|
||||
import org.apache.cassandra.db.ColumnFamilyStore;
|
||||
import org.apache.cassandra.db.ColumnIndexer;
|
||||
import org.apache.cassandra.db.CounterColumn;
|
||||
import org.apache.cassandra.db.IColumn;
|
||||
import org.apache.cassandra.db.SuperColumn;
|
||||
import org.apache.cassandra.db.*;
|
||||
import org.apache.cassandra.db.marshal.AbstractType;
|
||||
import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
|
||||
import org.apache.cassandra.io.util.DataOutputBuffer;
|
||||
|
|
|
|||
|
|
@ -31,11 +31,15 @@ import com.google.common.collect.AbstractIterator;
|
|||
|
||||
import org.apache.cassandra.auth.SimpleAuthenticator;
|
||||
import org.apache.cassandra.config.ConfigurationException;
|
||||
import org.apache.cassandra.db.IColumn;
|
||||
import org.apache.cassandra.db.*;
|
||||
import org.apache.cassandra.db.marshal.AbstractType;
|
||||
import org.apache.cassandra.db.marshal.TypeParser;
|
||||
import org.apache.cassandra.dht.IPartitioner;
|
||||
import org.apache.cassandra.thrift.*;
|
||||
import org.apache.cassandra.thrift.Column;
|
||||
import org.apache.cassandra.thrift.CounterColumn;
|
||||
import org.apache.cassandra.thrift.SuperColumn;
|
||||
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
import org.apache.cassandra.utils.Pair;
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
|
|
@ -293,8 +297,13 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
|
|||
|
||||
private IColumn unthriftify(ColumnOrSuperColumn cosc)
|
||||
{
|
||||
if (cosc.column == null)
|
||||
if (cosc.counter_column != null)
|
||||
return unthriftifyCounter(cosc.counter_column);
|
||||
if (cosc.counter_super_column != null)
|
||||
return unthriftifySuperCounter(cosc.counter_super_column);
|
||||
if (cosc.super_column != null)
|
||||
return unthriftifySuper(cosc.super_column);
|
||||
assert cosc.column != null;
|
||||
return unthriftifySimple(cosc.column);
|
||||
}
|
||||
|
||||
|
|
@ -312,5 +321,20 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
|
|||
{
|
||||
return new org.apache.cassandra.db.Column(column.name, column.value, column.timestamp);
|
||||
}
|
||||
|
||||
private IColumn unthriftifyCounter(CounterColumn column)
|
||||
{
|
||||
//CounterColumns read the nodeID from the System table, so need the StorageService running and access
|
||||
//to cassandra.yaml. To avoid a Hadoop needing access to yaml return a regular Column.
|
||||
return new org.apache.cassandra.db.Column(column.name, ByteBufferUtil.bytes(column.value), 0);
|
||||
}
|
||||
|
||||
private IColumn unthriftifySuperCounter(CounterSuperColumn superColumn)
|
||||
{
|
||||
org.apache.cassandra.db.SuperColumn sc = new org.apache.cassandra.db.SuperColumn(superColumn.name, subComparator);
|
||||
for (CounterColumn column : superColumn.columns)
|
||||
sc.addColumn(unthriftifyCounter(column));
|
||||
return sc;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -223,7 +223,7 @@ public final class MessagingService implements MessagingServiceMBean
|
|||
else
|
||||
throw e;
|
||||
}
|
||||
logger_.info("Starting Messaging Service on port {}", DatabaseDescriptor.getStoragePort());
|
||||
logger_.info("Starting Messaging Service on {}", address);
|
||||
}
|
||||
return ss;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -44,6 +44,7 @@ import org.apache.cassandra.concurrent.StageManager;
|
|||
import org.apache.cassandra.config.CFMetaData;
|
||||
import org.apache.cassandra.config.ConfigurationException;
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.config.KSMetaData;
|
||||
import org.apache.cassandra.db.*;
|
||||
import org.apache.cassandra.db.commitlog.CommitLog;
|
||||
import org.apache.cassandra.dht.*;
|
||||
|
|
@ -408,6 +409,18 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
|
|||
mutationStage.awaitTermination(1, TimeUnit.SECONDS);
|
||||
CommitLog.instance.shutdownBlocking();
|
||||
}
|
||||
|
||||
List<Future<?>> flushes = new ArrayList<Future<?>>();
|
||||
for (Table table : Table.all())
|
||||
{
|
||||
KSMetaData ksm = DatabaseDescriptor.getKSMetaData(table.name);
|
||||
if (!ksm.isDurableWrites())
|
||||
{
|
||||
for (ColumnFamilyStore cfs : table.getColumnFamilyStores())
|
||||
flushes.add(cfs.forceFlush());
|
||||
}
|
||||
}
|
||||
FBUtilities.waitOnFutures(flushes);
|
||||
}
|
||||
});
|
||||
Runtime.getRuntime().addShutdownHook(drainOnShutdown);
|
||||
|
|
@ -1356,7 +1369,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
|
|||
|
||||
public void forceTableCleanup(String tableName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException
|
||||
{
|
||||
if (tableName.equals("system"))
|
||||
if (tableName.equals(Table.SYSTEM_TABLE))
|
||||
throw new RuntimeException("Cleanup of the system table is neither necessary nor wise");
|
||||
|
||||
NodeId.OneShotRenewer nodeIdRenewer = new NodeId.OneShotRenewer();
|
||||
|
|
|
|||
Loading…
Reference in New Issue