From cf39b740d0fadd042ba5aefd824885db6b43ca9f Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Mon, 26 Jul 2010 21:00:40 +0000 Subject: [PATCH] finish CASSANDRA-1280 for trunk (lots of new code that the merge from 0.6 didn't finish). bonus: contrib/word_count actually tested and working again. patch by jbellis git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@979440 13f79535-47bb-0310-9956-ffa450edef68 --- contrib/word_count/src/WordCount.java | 1 - .../cassandra/config/DatabaseDescriptor.java | 4 +- .../hadoop/ColumnFamilyInputFormat.java | 33 ++-- .../hadoop/ColumnFamilyOutputFormat.java | 17 +- .../hadoop/ColumnFamilyRecordReader.java | 154 ++++++++---------- .../apache/cassandra/hadoop/ConfigHelper.java | 2 +- 6 files changed, 98 insertions(+), 113 deletions(-) diff --git a/contrib/word_count/src/WordCount.java b/contrib/word_count/src/WordCount.java index 5d807196e0..f30a2294a4 100644 --- a/contrib/word_count/src/WordCount.java +++ b/contrib/word_count/src/WordCount.java @@ -54,7 +54,6 @@ public class WordCount extends Configured implements Tool static final String COLUMN_FAMILY = "Standard1"; private static final String CONF_COLUMN_NAME = "columnname"; private static final String OUTPUT_PATH_PREFIX = "/tmp/word_count"; - static final int RING_DELAY = 3000; // this is enough for testing a single server node; may need more for a real cluster public static void main(String[] args) throws Exception { diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 22ef597f63..545b0d40a9 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -113,9 +113,8 @@ public class DatabaseDescriptor { try { - configFileName = getStorageConfigPath(); - + if (logger.isDebugEnabled()) logger.info("Loading settings from " + configFileName); @@ -647,7 +646,6 @@ public class DatabaseDescriptor public static AbstractType getComparator(String compareWith) throws ConfigurationException { - logger.info(compareWith); Class typeClass; if (compareWith == null) diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java index e1e6857bf2..50c4e3650f 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java @@ -41,7 +41,10 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.mapreduce.*; import org.apache.thrift.TException; import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.transport.TFramedTransport; import org.apache.thrift.transport.TSocket; +import org.apache.thrift.transport.TTransport; import org.apache.thrift.transport.TTransportException; /** @@ -170,18 +173,8 @@ public class ColumnFamilyInputFormat extends InputFormat getSubSplits(String keyspace, String cfName, TokenRange range, Configuration conf) throws IOException { // TODO handle failure of range replicas & retry - TSocket socket = new TSocket(range.endpoints.get(0), ConfigHelper.getThriftPort(conf)); - TBinaryProtocol binaryProtocol = new TBinaryProtocol(socket, false, false); - Cassandra.Client client = new Cassandra.Client(binaryProtocol); + Cassandra.Client client = createConnection(range.endpoints.get(0), ConfigHelper.getRpcPort(conf), true); int splitsize = ConfigHelper.getInputSplitSize(conf); - try - { - socket.open(); - } - catch (TTransportException e) - { - throw new IOException(e); - } List splits; try { @@ -194,19 +187,25 @@ public class ColumnFamilyInputFormat extends InputFormat getRangeMap(Configuration conf) throws IOException + private static Cassandra.Client createConnection(String host, Integer port, boolean framed) throws IOException { - TSocket socket = new TSocket(ConfigHelper.getInitialAddress(conf), ConfigHelper.getThriftPort(conf)); - TBinaryProtocol binaryProtocol = new TBinaryProtocol(socket, false, false); - Cassandra.Client client = new Cassandra.Client(binaryProtocol); + TSocket socket = new TSocket(host, port); + TTransport trans = framed ? new TFramedTransport(socket) : socket; try { - socket.open(); + trans.open(); } catch (TTransportException e) { - throw new IOException(e); + throw new IOException("unable to connect to server", e); } + return new Cassandra.Client(new TBinaryProtocol(trans)); + } + + private List getRangeMap(Configuration conf) throws IOException + { + Cassandra.Client client = createConnection(ConfigHelper.getInitialAddress(conf), ConfigHelper.getRpcPort(conf), true); + List map; try { diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java index 088d4db067..9731571ab2 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java @@ -26,9 +26,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import org.apache.cassandra.auth.AllowAllAuthenticator; import org.apache.cassandra.auth.SimpleAuthenticator; -import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamily; import org.apache.cassandra.db.IColumn; import org.apache.cassandra.thrift.AuthenticationException; @@ -44,6 +42,7 @@ import org.apache.hadoop.mapreduce.RecordWriter; import org.apache.hadoop.mapreduce.TaskAttemptContext; import org.apache.thrift.TException; import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.transport.TFramedTransport; import org.apache.thrift.transport.TSocket; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -151,16 +150,18 @@ public class ColumnFamilyOutputFormat extends OutputFormat> public static Cassandra.Client createAuthenticatedClient(TSocket socket, JobContext context) throws InvalidRequestException, TException, AuthenticationException, AuthorizationException { - TBinaryProtocol binaryProtocol = new TBinaryProtocol(socket, false, false); + TBinaryProtocol binaryProtocol = new TBinaryProtocol(new TFramedTransport(socket)); Cassandra.Client client = new Cassandra.Client(binaryProtocol); socket.open(); client.set_keyspace(ConfigHelper.getOutputKeyspace(context.getConfiguration())); - Map creds = new HashMap(); - creds.put(SimpleAuthenticator.USERNAME_KEY, ConfigHelper.getOutputKeyspaceUserName(context.getConfiguration())); - creds.put(SimpleAuthenticator.PASSWORD_KEY, ConfigHelper.getOutputKeyspacePassword(context.getConfiguration())); - AuthenticationRequest authRequest = new AuthenticationRequest(creds); - if (!(DatabaseDescriptor.getAuthenticator() instanceof AllowAllAuthenticator)) + if (ConfigHelper.getOutputKeyspaceUserName(context.getConfiguration()) != null) + { + Map creds = new HashMap(); + creds.put(SimpleAuthenticator.USERNAME_KEY, ConfigHelper.getOutputKeyspaceUserName(context.getConfiguration())); + creds.put(SimpleAuthenticator.PASSWORD_KEY, ConfigHelper.getOutputKeyspacePassword(context.getConfiguration())); + AuthenticationRequest authRequest = new AuthenticationRequest(creds); client.login(authRequest); + } return client; } diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java index b070527e85..5bf835d98a 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java @@ -35,6 +35,7 @@ import org.apache.cassandra.config.ConfigurationException; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.*; import org.apache.cassandra.db.clock.AbstractReconciler; +import org.apache.cassandra.db.clock.TimestampReconciler; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.thrift.*; @@ -48,6 +49,7 @@ import org.apache.hadoop.mapreduce.RecordReader; import org.apache.hadoop.mapreduce.TaskAttemptContext; import org.apache.thrift.TException; import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.transport.TFramedTransport; import org.apache.thrift.transport.TSocket; public class ColumnFamilyRecordReader extends RecordReader> @@ -60,8 +62,6 @@ public class ColumnFamilyRecordReader extends RecordReader creds = new HashMap(); - creds.put(SimpleAuthenticator.USERNAME_KEY, ConfigHelper.getInputKeyspaceUserName(conf)); - creds.put(SimpleAuthenticator.PASSWORD_KEY, ConfigHelper.getInputKeyspacePassword(conf)); - authRequest = new AuthenticationRequest(creds); - + try + { + // only need to connect once + if (socket != null && socket.isOpen()) + return; + + // create connection using thrift + String location = getLocation(); + socket = new TSocket(location, ConfigHelper.getRpcPort(conf)); + TBinaryProtocol binaryProtocol = new TBinaryProtocol(new TFramedTransport(socket)); + client = new Cassandra.Client(binaryProtocol); + socket.open(); + + // log in + client.set_keyspace(keyspace); + if (ConfigHelper.getInputKeyspaceUserName(conf) != null) + { + Map creds = new HashMap(); + creds.put(SimpleAuthenticator.USERNAME_KEY, ConfigHelper.getInputKeyspaceUserName(conf)); + creds.put(SimpleAuthenticator.PASSWORD_KEY, ConfigHelper.getInputKeyspacePassword(conf)); + AuthenticationRequest authRequest = new AuthenticationRequest(creds); + client.login(authRequest); + } + } + catch (Exception e) + { + throw new RuntimeException(e); + } + iter = new RowIterator(); } @@ -117,6 +141,41 @@ public class ColumnFamilyRecordReader extends RecordReader>> { private List rows; @@ -134,7 +193,7 @@ public class ColumnFamilyRecordReader extends RecordReader info = client.describe_keyspace(keyspace).get(cfName); comparator = FBUtilities.getComparator(info.get("CompareWith")); - subComparator = FBUtilities.getComparator(info.get("CompareSubcolumnsWith")); + subComparator = info.get("CompareSubcolumnsWith") == null ? null : FBUtilities.getComparator(info.get("CompareSubcolumnsWith")); } catch (ConfigurationException e) { @@ -158,16 +217,7 @@ public class ColumnFamilyRecordReader extends RecordReader