diff --git a/CHANGES.txt b/CHANGES.txt index 39b95b41d0..d34759892c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -18,6 +18,8 @@ * Assert ranges are not overlapping in AB.normalize (CASSANDRA-2641) * Don't write CounterUpdateColumn to disk in tests (CASSANDRA-2650) * Add sstable bulk loading utility (CASSANDRA-1278) + * avoid replaying hints to dropped columnfamilies (CASSANDRA-2685) + * add placeholders for missing rows in range query pseudo-RR (CASSANDRA-2680) 0.8.0-final @@ -26,7 +28,8 @@ * update CQL consistency levels (CASSANDRA-2566) * debian packaging fixes (CASSANDRA-2481, 2647) * fix UUIDType, IntegerType for direct buffers (CASSANDRA-2682, 2684) - + * switch to native Thrift for Hadoop map/reduce (CASSANDRA-2667) + * fix StackOverflowError when building from eclipse (CASSANDRA-2687) 0.8.0-rc1 * faster flushes and compaction from fixing excessively pessimistic diff --git a/NEWS.txt b/NEWS.txt index 48c20a32cd..4eb8fbe897 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -13,8 +13,10 @@ Upgrading - 0.8 is fully API-compatible with 0.7. You can continue to use your 0.7 clients. - Avro record classes used in map/reduce and Hadoop streaming code have - moved from org.apache.cassandra.avro to org.apache.cassandra.hadoop.avro, - applications using these classes will need to be updated accordingly. + been removed. Map/reduce can be switched to Thrift by changing + org.apache.cassandra.avro in import statements to + org.apache.cassandra.thrift (no class names change). Streaming support + has been removed for the time being. - The loadbalance command has been removed from nodetool. For similar behavior, decommission then rebootstrap with empty initial_token. - Thrift unframed mode has been removed. @@ -57,6 +59,9 @@ Other to compact that anyway (which will free up space if there are a lot of expired tombstones), use the new forceUserDefinedCompaction JMX method on CompactionManager. + - most of contrib/ (which was not part of the binary releases) + has been moved either to examples/ or tools/. We plan to move the + rest for 0.8.1. JMX --- diff --git a/build.xml b/build.xml index 99b1da3e55..737e7ea691 100644 --- a/build.xml +++ b/build.xml @@ -1252,10 +1252,10 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} - - - - + + + + diff --git a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java index 5fcd1ce358..0b5e7f4ab8 100644 --- a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java +++ b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java @@ -22,6 +22,7 @@ import java.util.*; import org.apache.cassandra.config.ConfigurationException; import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.db.marshal.TypeParser; import org.apache.cassandra.thrift.*; import org.apache.cassandra.utils.FBUtilities; import org.apache.commons.logging.Log; @@ -32,9 +33,9 @@ import org.apache.cassandra.db.IColumn; import org.apache.cassandra.db.SuperColumn; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.hadoop.*; -import org.apache.cassandra.hadoop.avro.Mutation; -import org.apache.cassandra.hadoop.avro.Deletion; -import org.apache.cassandra.hadoop.avro.ColumnOrSuperColumn; +import org.apache.cassandra.thrift.Mutation; +import org.apache.cassandra.thrift.Deletion; +import org.apache.cassandra.thrift.ColumnOrSuperColumn; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.hadoop.conf.Configuration; @@ -158,7 +159,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface // super ArrayList subcols = new ArrayList(); - for (IColumn subcol : ((SuperColumn)col).getSubColumns()) + for (IColumn subcol : col.getSubColumns()) subcols.add(columnToTuple(subcol.name(), subcol, cfDef)); pair.set(1, new DefaultDataBag(subcols)); @@ -179,8 +180,8 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface AbstractType default_validator = null; try { - comparator = FBUtilities.getComparator(cfDef.comparator_type); - default_validator = FBUtilities.getComparator(cfDef.default_validation_class); + comparator = TypeParser.parse(cfDef.comparator_type); + default_validator = TypeParser.parse(cfDef.default_validation_class); } catch (ConfigurationException e) { @@ -202,7 +203,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface AbstractType validator = null; try { - validator = FBUtilities.getComparator(cd.getValidation_class()); + validator = TypeParser.parse(cd.getValidation_class()); validators.put(cd.name, validator); } catch (ConfigurationException e) @@ -385,13 +386,13 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface if (pair.get(1) == null) { mutation.deletion = new Deletion(); - mutation.deletion.predicate = new org.apache.cassandra.hadoop.avro.SlicePredicate(); + mutation.deletion.predicate = new org.apache.cassandra.thrift.SlicePredicate(); mutation.deletion.predicate.column_names = Arrays.asList(objToBB(pair.get(0))); mutation.deletion.timestamp = System.currentTimeMillis() * 1000; } else { - org.apache.cassandra.hadoop.avro.Column column = new org.apache.cassandra.hadoop.avro.Column(); + org.apache.cassandra.thrift.Column column = new org.apache.cassandra.thrift.Column(); column.name = marshallers.get(0).decompose((pair.get(0))); if (validators.get(column.name) == null) // Have to special case BytesType to convert DataByteArray into ByteBuffer diff --git a/examples/hadoop_streaming_output/README.txt b/examples/hadoop_streaming_output/README.txt deleted file mode 100644 index 8acdcd8ccc..0000000000 --- a/examples/hadoop_streaming_output/README.txt +++ /dev/null @@ -1,24 +0,0 @@ -Hadoop Streaming WordCount example: runs a simple Python wordcount mapper script, and a slightly -more complex Cassandra specific reducer that outputs the results of the wordcount as Avro -records to Cassandra. - -Dependencies for the example: -* Hadoop 0.20.x installed, such that the 'bin/hadoop' launcher is available on your path -* Avro's Python library installed - -Unfortunately, due to the way Hadoop builds its CLASSPATH, it is also necessary to -upgrade a library that conflicts between Hadoop and Avro. Within your Hadoop distribution, -you'll need to ensure that the following jars are of a sufficiently high version, or else -you will see a classloader error at runtime: -* jackson-core-asl >= 1.4.0 -* jackson-mapper-asl >= 1.4.0 - -To run the example, edit bin/streaming to point to a valid Cassandra cluster and -then execute it over a text input, located on the local filesystem or in Hadoop: -$ bin/streaming -input - -Hadoop streaming will execute a simple wordcount over your input files, and write the generated -counts to Keyspace1,Standard1. bin/reducer.py gives an example of how to format the output of -a script as Avro records which can be consumed by Cassandra's AvroOutputReader, and bin/streaming -shows the necessary incantations to execute a Hadoop Streaming job with Cassandra as output. - diff --git a/examples/hadoop_streaming_output/bin/mapper.py b/examples/hadoop_streaming_output/bin/mapper.py deleted file mode 100644 index 62d4629c26..0000000000 --- a/examples/hadoop_streaming_output/bin/mapper.py +++ /dev/null @@ -1,43 +0,0 @@ -#!/usr/bin/env python - -# 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. - - -# Simple reducer for writing to Cassandra using Hadoop streaming, Python and Avro, -# based on http://www.michael-noll.com/wiki/Writing_An_Hadoop_MapReduce_Program_In_Python - - -# Wordcount mapper -# based on http://www.michael-noll.com/wiki/Writing_An_Hadoop_MapReduce_Program_In_Python - -import sys - -# input comes from STDIN (standard input) -for line in sys.stdin: - # remove leading and trailing whitespace - line = line.strip() - # split the line into words - words = line.split() - # increase counters - for word in words: - # write the results to STDOUT (standard output); - # what we output here will be the input for the - # Reduce step, i.e. the input for reducer.py - # - # tab-delimited; the trivial word count is 1 - print '%s\t%s' % (word, 1) - diff --git a/examples/hadoop_streaming_output/bin/reducer.py b/examples/hadoop_streaming_output/bin/reducer.py deleted file mode 100644 index cbf447561c..0000000000 --- a/examples/hadoop_streaming_output/bin/reducer.py +++ /dev/null @@ -1,72 +0,0 @@ -#!/usr/bin/env python - -# 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. - - -# Wordcount reducer for writing to Cassandra using Hadoop streaming and Avro, -# based on http://www.michael-noll.com/wiki/Writing_An_Hadoop_MapReduce_Program_In_Python - -from avro.io import BinaryEncoder, DatumWriter -import avro.protocol -import sys,time - -# input comes from STDIN (standard input) -word2count = {} -for line in sys.stdin: - # remove leading and trailing whitespace - line = line.strip() - - # parse the input we got from mapper.py - word, count = line.split('\t', 1) - # convert count (currently a string) to int - try: - count = int(count) - word2count[word] = word2count.get(word, 0) + count - except ValueError: - # count was not a number, so silently - # ignore/discard this line - pass - -# -# NB: the AvroOutputReader specific portion begins here -# - -def new_column(name, value): - column = dict() - column['name'] = '%s' % name - column['value'] = '%s' % value - column['timestamp'] = long(time.time() * 1e6) - column['ttl'] = 0 - return column - -# parse the current avro schema -proto = avro.protocol.parse(open('cassandra.avpr').read()) -schema = proto.types_dict['StreamingMutation'] -# open an avro encoder and writer for stdout -enc = BinaryEncoder(sys.stdout) -writer = DatumWriter(schema) - -# output a series of objects matching 'StreamingMutation' in the Avro interface -smutation = dict() -try: - for word, count in word2count.iteritems(): - smutation['key'] = word - smutation['mutation'] = {'column_or_supercolumn': {'column': new_column('count', count)}} - writer.write(smutation, enc) -finally: - sys.stdout.flush() - diff --git a/examples/hadoop_streaming_output/bin/streaming b/examples/hadoop_streaming_output/bin/streaming deleted file mode 100644 index db71723606..0000000000 --- a/examples/hadoop_streaming_output/bin/streaming +++ /dev/null @@ -1,40 +0,0 @@ -#!/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` - -ARCHIVES=`ls -1 $cwd/../../../build/apache-cassandra*.jar` -for jar in `ls -1 $cwd/../../../build/lib/jars/*.jar $cwd/../../../lib/*.jar`; do - ARCHIVES=$ARCHIVES,$jar -done - -hadoop jar $cwd/../../../build/lib/jars/hadoop-streaming*.jar \ - -D stream.reduce.output=cassandra_avro_output \ - -D stream.io.identifier.resolver.class=org.apache.cassandra.hadoop.streaming.AvroResolver \ - -D cassandra.output.keyspace=Keyspace1 \ - -D cassandra.output.columnfamily=Standard1 \ - -D cassandra.partitioner.class=org.apache.cassandra.dht.RandomPartitioner \ - -D cassandra.thrift.address=127.0.0.1 \ - -D cassandra.thrift.port=9160 \ - -libjars $ARCHIVES \ - -file $cwd/../../../src/gen-java/org/apache/cassandra/hadoop/hadoop.avpr \ - -outputformat org.apache.cassandra.hadoop.ColumnFamilyOutputFormat \ - -output /tmp/ignored \ - -mapper $cwd/mapper.py -reducer $cwd/reducer.py \ - $* - diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index 3a5b13f728..db1a34a97c 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -30,6 +30,7 @@ import java.util.concurrent.TimeoutException; import static com.google.common.base.Charsets.UTF_8; +import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.utils.FBUtilities; import org.apache.commons.lang.ArrayUtils; import org.slf4j.Logger; @@ -127,8 +128,12 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean return false; } + if (CFMetaData.getId(tableName, cfName) == null) + { + logger_.debug("Discarding hints for dropped keyspace or columnfamily {}/{}", tableName, cfName); + return true; + } Table table = Table.open(tableName); - DecoratedKey dkey = StorageService.getPartitioner().decorateKey(key); ColumnFamilyStore cfs = table.getColumnFamilyStore(cfName); int pageSize = PAGE_SIZE; @@ -141,6 +146,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean logger_.debug("average hinted-row column size is {}; using pageSize of {}", averageColumnSize, pageSize); } + DecoratedKey dkey = StorageService.getPartitioner().decorateKey(key); ByteBuffer startColumn = ByteBufferUtil.EMPTY_BYTE_BUFFER; while (true) { diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java index 1d50f10e51..2d8e3362ee 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java @@ -31,7 +31,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.auth.SimpleAuthenticator; -import org.apache.cassandra.hadoop.avro.Mutation; import org.apache.cassandra.thrift.*; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.mapreduce.*; diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java index d1f245bee1..a40d637b0f 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java @@ -59,8 +59,8 @@ import org.apache.cassandra.utils.ByteBufferUtil; * @see OutputFormat * */ -final class ColumnFamilyRecordWriter extends RecordWriter> -implements org.apache.hadoop.mapred.RecordWriter> +final class ColumnFamilyRecordWriter extends RecordWriter> +implements org.apache.hadoop.mapred.RecordWriter> { // The configuration this writer is associated with. private final Configuration conf; @@ -122,7 +122,7 @@ implements org.apache.hadoop.mapred.RecordWriter value) throws IOException + public void write(ByteBuffer keybuff, List value) throws IOException { Range range = ringCache.getRange(keybuff); @@ -136,77 +136,8 @@ implements org.apache.hadoop.mapred.RecordWriter(keybuff, avroToThrift(amut))); - } - - /** - * Deep copies the given Avro mutation into a new Thrift mutation. - */ - private Mutation avroToThrift(org.apache.cassandra.hadoop.avro.Mutation amut) - { - Mutation mutation = new Mutation(); - org.apache.cassandra.hadoop.avro.ColumnOrSuperColumn acosc = amut.column_or_supercolumn; - if (acosc == null) - { - // deletion - assert amut.deletion != null; - Deletion deletion = new Deletion().setTimestamp(amut.deletion.timestamp); - mutation.setDeletion(deletion); - - org.apache.cassandra.hadoop.avro.SlicePredicate apred = amut.deletion.predicate; - if (apred == null && amut.deletion.super_column == null) - { - // leave Deletion alone to delete entire row - } - else if (amut.deletion.super_column != null) - { - // super column - deletion.setSuper_column(ByteBufferUtil.getArray(amut.deletion.super_column)); - } - else if (apred.column_names != null) - { - // column names - List names = new ArrayList(apred.column_names.size()); - for (ByteBuffer name : apred.column_names) - names.add(name); - deletion.setPredicate(new SlicePredicate().setColumn_names(names)); - } - else - { - // range - deletion.setPredicate(new SlicePredicate().setSlice_range(avroToThrift(apred.slice_range))); - } - } - else - { - // creation - ColumnOrSuperColumn cosc = new ColumnOrSuperColumn(); - mutation.setColumn_or_supercolumn(cosc); - if (acosc.column != null) - // standard column - cosc.setColumn(avroToThrift(acosc.column)); - else - { - // super column - ByteBuffer scolname = acosc.super_column.name; - List scolcols = new ArrayList(acosc.super_column.columns.size()); - for (org.apache.cassandra.hadoop.avro.Column acol : acosc.super_column.columns) - scolcols.add(avroToThrift(acol)); - cosc.setSuper_column(new SuperColumn(scolname, scolcols)); - } - } - return mutation; - } - - private SliceRange avroToThrift(org.apache.cassandra.hadoop.avro.SliceRange asr) - { - return new SliceRange(asr.start, asr.finish, asr.reversed, asr.count); - } - - private Column avroToThrift(org.apache.cassandra.hadoop.avro.Column acol) - { - return new Column(acol.name).setValue(acol.value).setTimestamp(acol.timestamp); + for (Mutation amut : value) + client.put(new Pair(keybuff, amut)); } /** diff --git a/src/java/org/apache/cassandra/hadoop/streaming/AvroOutputReader.java b/src/java/org/apache/cassandra/hadoop/streaming/AvroOutputReader.java deleted file mode 100644 index da903db3f8..0000000000 --- a/src/java/org/apache/cassandra/hadoop/streaming/AvroOutputReader.java +++ /dev/null @@ -1,147 +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.hadoop.streaming; - -import java.io.DataInput; -import java.io.EOFException; -import java.io.IOException; -import java.io.InputStream; -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.List; - -import org.apache.avro.io.BinaryDecoder; -import org.apache.avro.io.DecoderFactory; -import org.apache.avro.specific.SpecificDatumReader; -import org.apache.cassandra.hadoop.avro.Mutation; -import org.apache.cassandra.hadoop.avro.StreamingMutation; -import org.apache.cassandra.io.util.FileUtils; -import org.apache.hadoop.streaming.PipeMapRed; -import org.apache.hadoop.streaming.io.OutputReader; - -/** - * An OutputReader that reads sequential StreamingMutations (from Cassandra's Avro client API), and converts them to - * the objects used by CassandraOutputFormat. This allows Hadoop Streaming to output efficiently to Cassandra via - * a familiar API. - * - * Avro requires the reader's and writer's schema: otherwise, it assumes they are the same. - * If the canonical schema that the Cassandra side uses changes, and somebody packaged the {{avpr}} - * up in their application somehow, or generated code, they'd see a runtime failure. - * We could allow specifying an alternate Avro schema using a Configuration property to work around this. - */ -public class AvroOutputReader extends OutputReader> -{ - private BinaryDecoder decoder; - private SpecificDatumReader reader; - - // reusable values - private final StreamingMutation entry = new StreamingMutation(); - private final ArrayList mutations = new ArrayList(1); - - @Override - public void initialize(PipeMapRed pmr) throws IOException - { - super.initialize(pmr); - - // set up decoding around the DataInput (hmm) provided by streaming - InputStream in; - if (pmr.getClientInput() instanceof InputStream) - // let's hope this is the case - in = (InputStream)pmr.getClientInput(); - else - // ...because this is relatively slow - in = new FromDataInputStream(pmr.getClientInput()); - decoder = DecoderFactory.defaultFactory().createBinaryDecoder(in, null); - reader = new SpecificDatumReader(StreamingMutation.SCHEMA$); - } - - @Override - public boolean readKeyValue() throws IOException - { - try - { - reader.read(entry, decoder); - } - catch (EOFException e) - { - return false; - } - mutations.clear(); - mutations.add(entry.mutation); - return true; - } - - @Override - public ByteBuffer getCurrentKey() throws IOException - { - return entry.key; - } - - @Override - public List getCurrentValue() throws IOException - { - return mutations; - } - - @Override - public String getLastOutput() - { - return entry.toString(); - } - - /** - * Wraps a DataInput to extend InputStream. The exception handling in read() is likely to be ridiculous slow. - */ - private static final class FromDataInputStream extends InputStream - { - private final DataInput in; - - public FromDataInputStream(DataInput in) - { - this.in = in; - } - - @Override - public boolean markSupported() - { - return false; - } - - @Override - public int read() throws IOException - { - try - { - return in.readUnsignedByte(); - } - catch (EOFException e) - { - return -1; - } - } - - @Override - public long skip(long n) throws IOException - { - FileUtils.skipBytesFully(in, n); - return n; - } - } -} diff --git a/src/java/org/apache/cassandra/hadoop/streaming/AvroResolver.java b/src/java/org/apache/cassandra/hadoop/streaming/AvroResolver.java deleted file mode 100644 index 293e1b29ad..0000000000 --- a/src/java/org/apache/cassandra/hadoop/streaming/AvroResolver.java +++ /dev/null @@ -1,50 +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.hadoop.streaming; - -import java.nio.ByteBuffer; -import java.util.List; - -import org.apache.hadoop.streaming.io.IdentifierResolver; - -/** - * Resolves AVRO_ID to the appropriate OutputReader and K/V classes for Cassandra output. - * - * TODO: usage explanation - */ -public class AvroResolver extends IdentifierResolver -{ - public static final String AVRO_ID = "cassandra_avro_output"; - - @Override - public void resolve(String identifier) - { - if (!identifier.equalsIgnoreCase(AVRO_ID)) - { - super.resolve(identifier); - return; - } - - setInputWriterClass(null); - setOutputReaderClass(AvroOutputReader.class); - setOutputKeyClass(ByteBuffer.class); - setOutputValueClass(List.class); - } -} diff --git a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java index 84b91e9f4d..7b6f6b9a0e 100644 --- a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java +++ b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java @@ -24,8 +24,6 @@ import java.util.*; import java.util.concurrent.LinkedBlockingQueue; import com.google.common.collect.AbstractIterator; -import com.google.common.collect.Iterables; -import com.google.common.collect.Iterators; import org.apache.commons.collections.iterators.CollatingIterator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -104,7 +102,21 @@ public class RangeSliceResponseResolver implements IResponseResolver 1 + ? RowRepairResolver.resolveSuperset(versions) + : versions.get(0); + if (versions.size() < sources.size()) + { + // add placeholder rows for sources that didn't have any data, so maybeScheduleRepairs sees them + for (InetAddress source : sources) + { + if (!versionSources.contains(source)) + { + versions.add(null); + versionSources.add(source); + } + } + } RowRepairResolver.maybeScheduleRepairs(resolved, table, key, versions, versionSources); versions.clear(); versionSources.clear();