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

Merge branch 'cassandra-3.1' into trunk

This commit is contained in:
Aleksey Yeschenko 2015-11-10 14:43:04 +00:00
parent 41c69ac556 84fe374d29
commit 3ea35bda6e
3 changed files with 50 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -5,6 +5,8 @@
3.1
Merged from 2.2:
* (Hadoop) ensure that Cluster instances are always closed (CASSANDRA-10058)
Merged from 2.1:
* Reject counter writes in CQLSSTableWriter (CASSANDRA-10258)
* Remove superfluous COUNTER_MUTATION stage mapping (CASSANDRA-10605)

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

@ -112,27 +112,25 @@ class CqlRecordWriter extends RecordWriter<Map<String, ByteBuffer>, List<ByteBuf
batchThreshold = conf.getLong(CqlOutputFormat.BATCH_THRESHOLD, 32);
this.clients = new HashMap<>();
try
try (Cluster cluster = CqlConfigHelper.getOutputCluster(ConfigHelper.getOutputInitialAddress(conf), conf))
{
String keyspace = ConfigHelper.getOutputKeyspace(conf);
try (Session client = CqlConfigHelper.getOutputCluster(ConfigHelper.getOutputInitialAddress(conf), conf).connect(keyspace))
Session client = cluster.connect(keyspace);
ringCache = new NativeRingCache(conf);
if (client != null)
{
ringCache = new NativeRingCache(conf);
if (client != null)
{
TableMetadata tableMetadata = client.getCluster().getMetadata().getKeyspace(client.getLoggedKeyspace()).getTable(ConfigHelper.getOutputColumnFamily(conf));
clusterColumns = tableMetadata.getClusteringColumns();
partitionKeyColumns = tableMetadata.getPartitionKey();
TableMetadata tableMetadata = client.getCluster().getMetadata().getKeyspace(client.getLoggedKeyspace()).getTable(ConfigHelper.getOutputColumnFamily(conf));
clusterColumns = tableMetadata.getClusteringColumns();
partitionKeyColumns = tableMetadata.getPartitionKey();
String cqlQuery = CqlConfigHelper.getOutputCql(conf).trim();
if (cqlQuery.toLowerCase().startsWith("insert"))
throw new UnsupportedOperationException("INSERT with CqlRecordWriter is not supported, please use UPDATE/DELETE statement");
cql = appendKeyWhereClauses(cqlQuery);
}
else
{
throw new IllegalArgumentException("Invalid configuration specified " + conf);
}
String cqlQuery = CqlConfigHelper.getOutputCql(conf).trim();
if (cqlQuery.toLowerCase().startsWith("insert"))
throw new UnsupportedOperationException("INSERT with CqlRecordWriter is not supported, please use UPDATE/DELETE statement");
cql = appendKeyWhereClauses(cqlQuery);
}
else
{
throw new IllegalArgumentException("Invalid configuration specified " + conf);
}
}
catch (Exception e)
@ -234,7 +232,7 @@ class CqlRecordWriter extends RecordWriter<Map<String, ByteBuffer>, List<ByteBuf
{
// The list of endpoints for this range
protected final List<InetAddress> endpoints;
protected Session client;
protected Cluster cluster = null;
// A bounded queue of incoming mutations for this range
protected final BlockingQueue<List<ByteBuffer>> queue = new ArrayBlockingQueue<List<ByteBuffer>>(queueSize);
@ -280,6 +278,7 @@ class CqlRecordWriter extends RecordWriter<Map<String, ByteBuffer>, List<ByteBuf
*/
public void run()
{
Session session = null;
outer:
while (run || !queue.isEmpty())
{
@ -298,33 +297,36 @@ class CqlRecordWriter extends RecordWriter<Map<String, ByteBuffer>, List<ByteBuf
while (true)
{
// send the mutation to the last-used endpoint. first time through, this will NPE harmlessly.
try
if (session != null)
{
int i = 0;
PreparedStatement statement = preparedStatement(client);
while (bindVariables != null)
try
{
BoundStatement boundStatement = new BoundStatement(statement);
for (int columnPosition = 0; columnPosition < bindVariables.size(); columnPosition++)
int i = 0;
PreparedStatement statement = preparedStatement(session);
while (bindVariables != null)
{
boundStatement.setBytesUnsafe(columnPosition, bindVariables.get(columnPosition));
}
client.execute(boundStatement);
i++;
BoundStatement boundStatement = new BoundStatement(statement);
for (int columnPosition = 0; columnPosition < bindVariables.size(); columnPosition++)
{
boundStatement.setBytesUnsafe(columnPosition, bindVariables.get(columnPosition));
}
session.execute(boundStatement);
i++;
if (i >= batchThreshold)
break;
bindVariables = queue.poll();
if (i >= batchThreshold)
break;
bindVariables = queue.poll();
}
break;
}
break;
}
catch (Exception e)
{
closeInternal();
if (!iter.hasNext())
catch (Exception e)
{
lastException = new IOException(e);
break outer;
closeInternal();
if (!iter.hasNext())
{
lastException = new IOException(e);
break outer;
}
}
}
@ -333,7 +335,8 @@ class CqlRecordWriter extends RecordWriter<Map<String, ByteBuffer>, List<ByteBuf
{
InetAddress address = iter.next();
String host = address.getHostName();
client = CqlConfigHelper.getOutputCluster(host, conf).connect();
cluster = CqlConfigHelper.getOutputCluster(host, conf);
session = cluster.connect();
}
catch (Exception e)
{
@ -403,9 +406,9 @@ class CqlRecordWriter extends RecordWriter<Map<String, ByteBuffer>, List<ByteBuf
protected void closeInternal()
{
if (client != null)
if (cluster != null)
{
client.close();
cluster.close();
}
}
@ -485,15 +488,14 @@ class CqlRecordWriter extends RecordWriter<Map<String, ByteBuffer>, List<ByteBuf
private void refreshEndpointMap()
{
String keyspace = ConfigHelper.getOutputKeyspace(conf);
try (Session session = CqlConfigHelper.getOutputCluster(ConfigHelper.getOutputInitialAddress(conf), conf).connect(keyspace))
try (Cluster cluster = CqlConfigHelper.getOutputCluster(ConfigHelper.getOutputInitialAddress(conf), conf))
{
Session session = cluster.connect(keyspace);
rangeMap = new HashMap<>();
metadata = session.getCluster().getMetadata();
Set<TokenRange> ranges = metadata.getTokenRanges();
for (TokenRange range : ranges)
{
rangeMap.put(range, metadata.getReplicas(keyspace, range));
}
}
}

2
test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
View File

@ -158,7 +158,7 @@ public class CQLSSTableWriterTest
String insert = String.format("UPDATE cql_keyspace.counter1 SET my_counter = my_counter - ? WHERE my_id = ?");
CQLSSTableWriter.builder().inDirectory(dataDir)
.forTable(schema)
.withPartitioner(StorageService.instance.getPartitioner())
.withPartitioner(Murmur3Partitioner.instance)
.using(insert).build();
}