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

Merge branch 'cassandra-2.0' into cassandra-2.1

This commit is contained in:
Brandon Williams 2014-06-09 13:43:20 -05:00
parent acb4cf62db 4b8bb86e27
commit 60c7c73c45
2 changed files with 20 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyOutputFormat.java
View File

@ -124,18 +124,24 @@ public abstract class AbstractColumnFamilyOutputFormat<K, Y> extends OutputForma
TProtocol binaryProtocol = new TBinaryProtocol(transport, true, true);
Cassandra.Client client = new Cassandra.Client(binaryProtocol);
client.set_keyspace(ConfigHelper.getOutputKeyspace(conf));
if ((ConfigHelper.getOutputKeyspaceUserName(conf) != null) && (ConfigHelper.getOutputKeyspacePassword(conf) != null))
{
Map<String, String> creds = new HashMap<String, String>();
creds.put(IAuthenticator.USERNAME_KEY, ConfigHelper.getOutputKeyspaceUserName(conf));
creds.put(IAuthenticator.PASSWORD_KEY, ConfigHelper.getOutputKeyspacePassword(conf));
AuthenticationRequest authRequest = new AuthenticationRequest(creds);
client.login(authRequest);
}
String user = ConfigHelper.getOutputKeyspaceUserName(conf);
String password = ConfigHelper.getOutputKeyspacePassword(conf);
if ((user != null) && (password != null))
login(user, password, client);
logger.debug("Authenticated client for CF output format created successfully");
return client;
}
public static void login(String user, String password, Cassandra.Client client) throws Exception
{
Map<String, String> creds = new HashMap<String, String>();
creds.put(IAuthenticator.USERNAME_KEY, user);
creds.put(IAuthenticator.PASSWORD_KEY, password);
AuthenticationRequest authRequest = new AuthenticationRequest(creds);
client.login(authRequest);
}
/**
* An {@link OutputCommitter} that does nothing.
*/

7
src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java
View File

@ -27,7 +27,6 @@ import org.apache.cassandra.hadoop.HadoopCompat;
import org.apache.hadoop.util.Progressable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.db.marshal.LongType;
@ -36,6 +35,7 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.SyntaxException;
import org.apache.cassandra.hadoop.AbstractColumnFamilyOutputFormat;
import org.apache.cassandra.hadoop.AbstractColumnFamilyRecordWriter;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.thrift.*;
@ -103,6 +103,11 @@ class CqlRecordWriter extends AbstractColumnFamilyRecordWriter<Map<String, ByteB
try
{
Cassandra.Client client = ConfigHelper.getClientFromOutputAddressList(conf);
client.set_keyspace(ConfigHelper.getOutputKeyspace(conf));
String user = ConfigHelper.getOutputKeyspaceUserName(conf);
String password = ConfigHelper.getOutputKeyspacePassword(conf);
if ((user != null) && (password != null))
AbstractColumnFamilyOutputFormat.login(user, password, client);
retrievePartitionKeyValidator(client);
String cqlQuery = CqlConfigHelper.getOutputCql(conf).trim();
if (cqlQuery.toLowerCase().startsWith("insert"))