From 177f607057a9d4c4b3746cec51e8e283938a5363 Mon Sep 17 00:00:00 2001 From: Alex Liu Date: Tue, 10 Nov 2015 14:32:55 +0000 Subject: [PATCH 1/2] (Hadoop) ensure that Cluster instances are always closed patch by Alex Liu; reviewed by Aleksey Yeschenko for CASSANDRA-10058 --- CHANGES.txt | 1 + .../AbstractColumnFamilyInputFormat.java | 74 ++++++++------- .../hadoop/cql3/CqlRecordWriter.java | 94 ++++++++++--------- .../hadoop/pig/CqlNativeStorage.java | 4 +- 4 files changed, 90 insertions(+), 83 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 5edad20180..81ceb25cce 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.2.4 + * (Hadoop) ensure that Cluster instances are always closed (CASSANDRA-10058) * (cqlsh) show partial trace if incomplete after max_trace_wait (CASSANDRA-7645) * Use most up-to-date version of schema for system tables (CASSANDRA-10652) * Deprecate memory_allocator in cassandra.yaml (CASSANDRA-10581,10628) diff --git a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java index e531ad1693..d6871832fb 100644 --- a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java +++ b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java @@ -24,6 +24,7 @@ import java.util.concurrent.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.datastax.driver.core.Cluster; import com.datastax.driver.core.Host; import com.datastax.driver.core.Metadata; import com.datastax.driver.core.ResultSet; @@ -58,7 +59,6 @@ public abstract class AbstractColumnFamilyInputFormat extends InputFormat< private String keyspace; private String cfName; private IPartitioner partitioner; - private Session session; protected void validateConfiguration(Configuration conf) { @@ -90,36 +90,36 @@ public abstract class AbstractColumnFamilyInputFormat extends InputFormat< ExecutorService executor = new ThreadPoolExecutor(0, 128, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue()); List splits = new ArrayList<>(); - try + List>> splitfutures = new ArrayList<>(); + KeyRange jobKeyRange = ConfigHelper.getInputKeyRange(conf); + Range jobRange = null; + if (jobKeyRange != null) { - List>> splitfutures = new ArrayList<>(); - KeyRange jobKeyRange = ConfigHelper.getInputKeyRange(conf); - Range jobRange = null; - if (jobKeyRange != null) + if (jobKeyRange.start_key != null) { - if (jobKeyRange.start_key != null) - { - if (!partitioner.preservesOrder()) - throw new UnsupportedOperationException("KeyRange based on keys can only be used with a order preserving partitioner"); - if (jobKeyRange.start_token != null) - throw new IllegalArgumentException("only start_key supported"); - if (jobKeyRange.end_token != null) - throw new IllegalArgumentException("only start_key supported"); - jobRange = new Range<>(partitioner.getToken(jobKeyRange.start_key), - partitioner.getToken(jobKeyRange.end_key)); - } - else if (jobKeyRange.start_token != null) - { - jobRange = new Range<>(partitioner.getTokenFactory().fromString(jobKeyRange.start_token), - partitioner.getTokenFactory().fromString(jobKeyRange.end_token)); - } - else - { - logger.warn("ignoring jobKeyRange specified without start_key or start_token"); - } + if (!partitioner.preservesOrder()) + throw new UnsupportedOperationException("KeyRange based on keys can only be used with a order preserving partitioner"); + if (jobKeyRange.start_token != null) + throw new IllegalArgumentException("only start_key supported"); + if (jobKeyRange.end_token != null) + throw new IllegalArgumentException("only start_key supported"); + jobRange = new Range<>(partitioner.getToken(jobKeyRange.start_key), + partitioner.getToken(jobKeyRange.end_key)); } + else if (jobKeyRange.start_token != null) + { + jobRange = new Range<>(partitioner.getTokenFactory().fromString(jobKeyRange.start_token), + partitioner.getTokenFactory().fromString(jobKeyRange.end_token)); + } + else + { + logger.warn("ignoring jobKeyRange specified without start_key or start_token"); + } + } - session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect(); + try (Cluster cluster = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf)) + { + Session session = cluster.connect(); Metadata metadata = session.getCluster().getMetadata(); for (TokenRange range : masterRangeNodes.keySet()) @@ -127,7 +127,7 @@ public abstract class AbstractColumnFamilyInputFormat extends InputFormat< if (jobRange == null) { // for each tokenRange, pick a live owner and ask it to compute bite-sized splits - splitfutures.add(executor.submit(new SplitCallable(range, masterRangeNodes.get(range), conf))); + splitfutures.add(executor.submit(new SplitCallable(range, masterRangeNodes.get(range), conf, session))); } else { @@ -137,7 +137,7 @@ public abstract class AbstractColumnFamilyInputFormat extends InputFormat< for (TokenRange intersection: range.intersectWith(jobTokenRange)) { // for each tokenRange, pick a live owner and ask it to compute bite-sized splits - splitfutures.add(executor.submit(new SplitCallable(intersection, masterRangeNodes.get(range), conf))); + splitfutures.add(executor.submit(new SplitCallable(intersection, masterRangeNodes.get(range), conf, session))); } } } @@ -182,19 +182,21 @@ public abstract class AbstractColumnFamilyInputFormat extends InputFormat< private final TokenRange tokenRange; private final Set hosts; private final Configuration conf; + private final Session session; - public SplitCallable(TokenRange tr, Set hosts, Configuration conf) + public SplitCallable(TokenRange tr, Set hosts, Configuration conf, Session session) { this.tokenRange = tr; this.hosts = hosts; this.conf = conf; + this.session = session; } public List call() throws Exception { ArrayList splits = new ArrayList<>(); Map subSplits; - subSplits = getSubSplits(keyspace, cfName, tokenRange, conf); + subSplits = getSubSplits(keyspace, cfName, tokenRange, conf, session); // turn the sub-ranges into InputSplits String[] endpoints = new String[hosts.size()]; @@ -225,12 +227,12 @@ public abstract class AbstractColumnFamilyInputFormat extends InputFormat< } } - private Map getSubSplits(String keyspace, String cfName, TokenRange range, Configuration conf) throws IOException + private Map getSubSplits(String keyspace, String cfName, TokenRange range, Configuration conf, Session session) throws IOException { int splitSize = ConfigHelper.getInputSplitSize(conf); try { - return describeSplits(keyspace, cfName, range, splitSize); + return describeSplits(keyspace, cfName, range, splitSize, session); } catch (Exception e) { @@ -240,17 +242,17 @@ public abstract class AbstractColumnFamilyInputFormat extends InputFormat< private Map> getRangeMap(Configuration conf, String keyspace) { - try (Session session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf).connect()) + try (Cluster cluster = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","), conf)) { Map> map = new HashMap<>(); - Metadata metadata = session.getCluster().getMetadata(); + Metadata metadata = cluster.connect().getCluster().getMetadata(); for (TokenRange tokenRange : metadata.getTokenRanges()) map.put(tokenRange, metadata.getReplicas('"' + keyspace + '"', tokenRange)); return map; } } - private Map describeSplits(String keyspace, String table, TokenRange tokenRange, int splitSize) + private Map describeSplits(String keyspace, String table, TokenRange tokenRange, int splitSize, Session session) { String query = String.format("SELECT mean_partition_size, partitions_count " + "FROM %s.%s " + diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java index 6e8ffd9fdc..14e24fb4b1 100644 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java +++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java @@ -113,27 +113,25 @@ class CqlRecordWriter extends RecordWriter, List(); - 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) @@ -235,7 +233,7 @@ class CqlRecordWriter extends RecordWriter, List endpoints; - protected Session client; + protected Cluster cluster = null; // A bounded queue of incoming mutations for this range protected final BlockingQueue> queue = new ArrayBlockingQueue>(queueSize); @@ -281,6 +279,7 @@ class CqlRecordWriter extends RecordWriter, List, List= 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; + } + } } // attempt to connect to a different endpoint @@ -334,7 +336,8 @@ class CqlRecordWriter extends RecordWriter, List, List, List(); metadata = session.getCluster().getMetadata(); Set ranges = metadata.getTokenRanges(); for (TokenRange range : ranges) - { rangeMap.put(range, metadata.getReplicas(keyspace, range)); - } } } diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java index ba0a37d6dc..74058b1de2 100644 --- a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java +++ b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java @@ -28,6 +28,7 @@ import java.net.URLDecoder; import java.nio.ByteBuffer; import java.util.*; +import com.datastax.driver.core.Cluster; import com.datastax.driver.core.ColumnMetadata; import com.datastax.driver.core.Metadata; import com.datastax.driver.core.Row; @@ -723,8 +724,9 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo // Only get the schema if we haven't already gotten it if (!properties.containsKey(signature)) { - try (Session client = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf), conf).connect()) + try (Cluster cluster = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf), conf)) { + Session client = cluster.connect(); client.execute("USE " + keyspace); // compose the CfDef for the columfamily From d766f4fb20af4914b54420e22af0de909eb180ed Mon Sep 17 00:00:00 2001 From: Aleksey Yeschenko Date: Tue, 10 Nov 2015 14:42:01 +0000 Subject: [PATCH 2/2] Fix CQLSSTableWriterTest --- .../org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java index 602a1086c5..557bebaa7d 100644 --- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java @@ -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(); }