From 592cbeaab09e7e959dfaea591d4afe7c1613e3db Mon Sep 17 00:00:00 2001 From: Stefan Miklosovic Date: Mon, 13 Mar 2023 12:25:46 +0100 Subject: [PATCH] Remove org.apache.cassandra.hadoop code patch by Stefan Miklosovic; reviewed by David Capwell and Brandon Williams for CASSANDRA-18323 --- .build/cassandra-deps-template.xml | 8 - .build/parent-pom-template.xml | 60 -- CHANGES.txt | 1 + NEWS.txt | 2 + .../cassandra/hadoop/ColumnFamilySplit.java | 126 --- .../apache/cassandra/hadoop/ConfigHelper.java | 408 --------- .../apache/cassandra/hadoop/HadoopCompat.java | 350 -------- .../cassandra/hadoop/ReporterWrapper.java | 85 -- .../hadoop/cql3/CqlBulkOutputFormat.java | 209 ----- .../hadoop/cql3/CqlBulkRecordWriter.java | 333 -------- .../hadoop/cql3/CqlClientHelper.java | 109 --- .../hadoop/cql3/CqlConfigHelper.java | 654 --------------- .../cassandra/hadoop/cql3/CqlInputFormat.java | 502 ----------- .../hadoop/cql3/CqlOutputFormat.java | 143 ---- .../hadoop/cql3/CqlRecordReader.java | 784 ------------------ .../hadoop/cql3/CqlRecordWriter.java | 536 ------------ ...tedLocalNodeFirstLocalBalancingPolicy.java | 216 ----- .../apache/cassandra/hadoop/package-info.java | 23 - .../transport/SimpleClientPerfTest.java | 2 +- .../debug/SelfReconcilingRandom.java | 20 +- 20 files changed, 17 insertions(+), 4554 deletions(-) delete mode 100644 src/java/org/apache/cassandra/hadoop/ColumnFamilySplit.java delete mode 100644 src/java/org/apache/cassandra/hadoop/ConfigHelper.java delete mode 100644 src/java/org/apache/cassandra/hadoop/HadoopCompat.java delete mode 100644 src/java/org/apache/cassandra/hadoop/ReporterWrapper.java delete mode 100644 src/java/org/apache/cassandra/hadoop/cql3/CqlBulkOutputFormat.java delete mode 100644 src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java delete mode 100644 src/java/org/apache/cassandra/hadoop/cql3/CqlClientHelper.java delete mode 100644 src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java delete mode 100644 src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java delete mode 100644 src/java/org/apache/cassandra/hadoop/cql3/CqlOutputFormat.java delete mode 100644 src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java delete mode 100644 src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java delete mode 100644 src/java/org/apache/cassandra/hadoop/cql3/LimitedLocalNodeFirstLocalBalancingPolicy.java delete mode 100644 src/java/org/apache/cassandra/hadoop/package-info.java diff --git a/.build/cassandra-deps-template.xml b/.build/cassandra-deps-template.xml index 44f9f35ab3..b1db0fd0b7 100644 --- a/.build/cassandra-deps-template.xml +++ b/.build/cassandra-deps-template.xml @@ -161,14 +161,6 @@ ch.qos.logback logback-classic - - org.apache.hadoop - hadoop-core - - - org.apache.hadoop - hadoop-minicluster - com.datastax.cassandra cassandra-driver-core diff --git a/.build/parent-pom-template.xml b/.build/parent-pom-template.xml index 4edd92e65c..6809fcd74a 100644 --- a/.build/parent-pom-template.xml +++ b/.build/parent-pom-template.xml @@ -503,66 +503,6 @@ 8.40 test - - org.apache.hadoop - hadoop-core - 1.0.3 - provided - - - servlet-api - org.mortbay.jetty - - - commons-logging - commons-logging - - - commons-lang - commons-lang - - - core - org.eclipse.jdt - - - ant - ant - - - junit - junit - - - jackson-mapper-asl - org.codehaus.jackson - - - slf4j-api - org.slf4j - - - - - org.apache.hadoop - hadoop-minicluster - 1.0.3 - provided - - - asm - asm - - - jackson-mapper-asl - org.codehaus.jackson - - - slf4j-api - org.slf4j - - - net.java.dev.jna jna diff --git a/CHANGES.txt b/CHANGES.txt index cc277dbcc6..149c9b0992 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.0 + * Remove org.apache.cassandra.hadoop code (CASSANDRA-18323) * Remove deprecated CQL functions dateOf and unixTimestampOf (CASSANDRA-18328) * Remove DateTieredCompactionStrategy (CASSANDRA-18043) * Add system_views.max_sstable_size and system_views.max_sstable_duration tables (CASSANDRA-18333) diff --git a/NEWS.txt b/NEWS.txt index e414d66500..c1fe5427f2 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -150,6 +150,8 @@ Upgrading to TimeWindowCompactionStrategy before upgrading to this version. - The deprecated functions `dateOf` and `unixTimestampOf` have been removed. They were deprecated and replaced by `toTimestamp` and `toUnixTimestamp` in Cassandra 2.2. + - Hadoop integration is no longer available (CASSANDRA-18323). If you want to process Cassandra data by big data frameworks, + please upgrade your infrastructure to use Cassandra Spark connector. Deprecation ----------- diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilySplit.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilySplit.java deleted file mode 100644 index 36256852f2..0000000000 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilySplit.java +++ /dev/null @@ -1,126 +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; - -import org.apache.hadoop.io.Writable; -import org.apache.hadoop.mapreduce.InputSplit; - -import java.io.DataInput; -import java.io.DataOutput; -import java.io.EOFException; -import java.io.IOException; -import java.util.Arrays; - -public class ColumnFamilySplit extends InputSplit implements Writable, org.apache.hadoop.mapred.InputSplit -{ - private String startToken; - private String endToken; - private long length; - private String[] dataNodes; - - @Deprecated - public ColumnFamilySplit(String startToken, String endToken, String[] dataNodes) - { - this(startToken, endToken, Long.MAX_VALUE, dataNodes); - } - - public ColumnFamilySplit(String startToken, String endToken, long length, String[] dataNodes) - { - assert startToken != null; - assert endToken != null; - this.startToken = startToken; - this.endToken = endToken; - this.length = length; - this.dataNodes = dataNodes; - } - - public String getStartToken() - { - return startToken; - } - - public String getEndToken() - { - return endToken; - } - - // getLength and getLocations satisfy the InputSplit abstraction - - public long getLength() - { - return length; - } - - public String[] getLocations() - { - return dataNodes; - } - - // This should only be used by KeyspaceSplit.read(); - protected ColumnFamilySplit() {} - - // These three methods are for serializing and deserializing - // KeyspaceSplits as needed by the Writable interface. - public void write(DataOutput out) throws IOException - { - out.writeUTF(startToken); - out.writeUTF(endToken); - out.writeInt(dataNodes.length); - for (String endpoint : dataNodes) - { - out.writeUTF(endpoint); - } - out.writeLong(length); - } - - public void readFields(DataInput in) throws IOException - { - startToken = in.readUTF(); - endToken = in.readUTF(); - int numOfEndpoints = in.readInt(); - dataNodes = new String[numOfEndpoints]; - for(int i = 0; i < numOfEndpoints; i++) - { - dataNodes[i] = in.readUTF(); - } - try - { - length = in.readLong(); - } - catch (EOFException e) - { - //We must be deserializing in a mixed-version cluster. - } - } - - @Override - public String toString() - { - return "ColumnFamilySplit(" + - "(" + startToken - + ", '" + endToken + ']' - + " @" + (dataNodes == null ? null : Arrays.asList(dataNodes)) + ')'; - } - - public static ColumnFamilySplit read(DataInput in) throws IOException - { - ColumnFamilySplit w = new ColumnFamilySplit(); - w.readFields(in); - return w; - } -} diff --git a/src/java/org/apache/cassandra/hadoop/ConfigHelper.java b/src/java/org/apache/cassandra/hadoop/ConfigHelper.java deleted file mode 100644 index cc539b1a70..0000000000 --- a/src/java/org/apache/cassandra/hadoop/ConfigHelper.java +++ /dev/null @@ -1,408 +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; - - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.schema.CompressionParams; -import org.apache.cassandra.utils.FBUtilities; -import org.apache.cassandra.utils.Pair; -import org.apache.hadoop.conf.Configuration; - -public class ConfigHelper -{ - private static final String INPUT_PARTITIONER_CONFIG = "cassandra.input.partitioner.class"; - private static final String OUTPUT_PARTITIONER_CONFIG = "cassandra.output.partitioner.class"; - private static final String INPUT_KEYSPACE_CONFIG = "cassandra.input.keyspace"; - private static final String OUTPUT_KEYSPACE_CONFIG = "cassandra.output.keyspace"; - private static final String INPUT_KEYSPACE_USERNAME_CONFIG = "cassandra.input.keyspace.username"; - private static final String INPUT_KEYSPACE_PASSWD_CONFIG = "cassandra.input.keyspace.passwd"; - private static final String OUTPUT_KEYSPACE_USERNAME_CONFIG = "cassandra.output.keyspace.username"; - private static final String OUTPUT_KEYSPACE_PASSWD_CONFIG = "cassandra.output.keyspace.passwd"; - private static final String INPUT_COLUMNFAMILY_CONFIG = "cassandra.input.columnfamily"; - private static final String OUTPUT_COLUMNFAMILY_CONFIG = "mapreduce.output.basename"; //this must == OutputFormat.BASE_OUTPUT_NAME - private static final String INPUT_PREDICATE_CONFIG = "cassandra.input.predicate"; - private static final String INPUT_KEYRANGE_CONFIG = "cassandra.input.keyRange"; - private static final String INPUT_SPLIT_SIZE_CONFIG = "cassandra.input.split.size"; - private static final String INPUT_SPLIT_SIZE_IN_MIB_CONFIG = "cassandra.input.split.size_mb"; - private static final String INPUT_WIDEROWS_CONFIG = "cassandra.input.widerows"; - private static final int DEFAULT_SPLIT_SIZE = 64 * 1024; - private static final String RANGE_BATCH_SIZE_CONFIG = "cassandra.range.batch.size"; - private static final int DEFAULT_RANGE_BATCH_SIZE = 4096; - private static final String INPUT_INITIAL_ADDRESS = "cassandra.input.address"; - private static final String OUTPUT_INITIAL_ADDRESS = "cassandra.output.address"; - private static final String OUTPUT_INITIAL_PORT = "cassandra.output.port"; - private static final String READ_CONSISTENCY_LEVEL = "cassandra.consistencylevel.read"; - private static final String WRITE_CONSISTENCY_LEVEL = "cassandra.consistencylevel.write"; - private static final String OUTPUT_COMPRESSION_CLASS = "cassandra.output.compression.class"; - private static final String OUTPUT_COMPRESSION_CHUNK_LENGTH = "cassandra.output.compression.length"; - private static final String OUTPUT_LOCAL_DC_ONLY = "cassandra.output.local.dc.only"; - private static final String DEFAULT_CASSANDRA_NATIVE_PORT = "7000"; - - private static final Logger logger = LoggerFactory.getLogger(ConfigHelper.class); - - /** - * Set the keyspace and column family for the input of this job. - * - * @param conf Job configuration you are about to run - * @param keyspace - * @param columnFamily - * @param widerows - */ - public static void setInputColumnFamily(Configuration conf, String keyspace, String columnFamily, boolean widerows) - { - if (keyspace == null) - throw new UnsupportedOperationException("keyspace may not be null"); - - if (columnFamily == null) - throw new UnsupportedOperationException("table may not be null"); - - conf.set(INPUT_KEYSPACE_CONFIG, keyspace); - conf.set(INPUT_COLUMNFAMILY_CONFIG, columnFamily); - conf.set(INPUT_WIDEROWS_CONFIG, String.valueOf(widerows)); - } - - /** - * Set the keyspace and column family for the input of this job. - * - * @param conf Job configuration you are about to run - * @param keyspace - * @param columnFamily - */ - public static void setInputColumnFamily(Configuration conf, String keyspace, String columnFamily) - { - setInputColumnFamily(conf, keyspace, columnFamily, false); - } - - /** - * Set the keyspace for the output of this job. - * - * @param conf Job configuration you are about to run - * @param keyspace - */ - public static void setOutputKeyspace(Configuration conf, String keyspace) - { - if (keyspace == null) - throw new UnsupportedOperationException("keyspace may not be null"); - - conf.set(OUTPUT_KEYSPACE_CONFIG, keyspace); - } - - /** - * Set the column family for the output of this job. - * - * @param conf Job configuration you are about to run - * @param columnFamily - */ - public static void setOutputColumnFamily(Configuration conf, String columnFamily) - { - conf.set(OUTPUT_COLUMNFAMILY_CONFIG, columnFamily); - } - - /** - * Set the column family for the output of this job. - * - * @param conf Job configuration you are about to run - * @param keyspace - * @param columnFamily - */ - public static void setOutputColumnFamily(Configuration conf, String keyspace, String columnFamily) - { - setOutputKeyspace(conf, keyspace); - setOutputColumnFamily(conf, columnFamily); - } - - /** - * The number of rows to request with each get range slices request. - * Too big and you can either get timeouts when it takes Cassandra too - * long to fetch all the data. Too small and the performance - * will be eaten up by the overhead of each request. - * - * @param conf Job configuration you are about to run - * @param batchsize Number of rows to request each time - */ - public static void setRangeBatchSize(Configuration conf, int batchsize) - { - conf.setInt(RANGE_BATCH_SIZE_CONFIG, batchsize); - } - - /** - * The number of rows to request with each get range slices request. - * Too big and you can either get timeouts when it takes Cassandra too - * long to fetch all the data. Too small and the performance - * will be eaten up by the overhead of each request. - * - * @param conf Job configuration you are about to run - * @return Number of rows to request each time - */ - public static int getRangeBatchSize(Configuration conf) - { - return conf.getInt(RANGE_BATCH_SIZE_CONFIG, DEFAULT_RANGE_BATCH_SIZE); - } - - /** - * Set the size of the input split. - * This affects the number of maps created, if the number is too small - * the overhead of each map will take up the bulk of the job time. - * - * @param conf Job configuration you are about to run - * @param splitsize Number of partitions in the input split - */ - public static void setInputSplitSize(Configuration conf, int splitsize) - { - conf.setInt(INPUT_SPLIT_SIZE_CONFIG, splitsize); - } - - public static int getInputSplitSize(Configuration conf) - { - return conf.getInt(INPUT_SPLIT_SIZE_CONFIG, DEFAULT_SPLIT_SIZE); - } - - /** - * Set the size of the input split. setInputSplitSize value is used if this is not set. - * This affects the number of maps created, if the number is too small - * the overhead of each map will take up the bulk of the job time. - * - * @param conf Job configuration you are about to run - * @param splitSizeMb Input split size in MiB - */ - public static void setInputSplitSizeInMb(Configuration conf, int splitSizeMb) - { - conf.setInt(INPUT_SPLIT_SIZE_IN_MIB_CONFIG, splitSizeMb); - } - - /** - * cassandra.input.split.size will be used if the value is undefined or negative. - * @param conf Job configuration you are about to run - * @return split size in MiB or -1 if it is undefined. - */ - public static int getInputSplitSizeInMb(Configuration conf) - { - return conf.getInt(INPUT_SPLIT_SIZE_IN_MIB_CONFIG, -1); - } - - /** - * Set the KeyRange to limit the rows. - * @param conf Job configuration you are about to run - */ - public static void setInputRange(Configuration conf, String startToken, String endToken) - { - conf.set(INPUT_KEYRANGE_CONFIG, startToken + "," + endToken); - } - - /** - * The start and end token of the input key range as a pair. - * - * may be null if unset. - */ - public static Pair getInputKeyRange(Configuration conf) - { - String str = conf.get(INPUT_KEYRANGE_CONFIG); - if (str == null) - return null; - - String[] parts = str.split(","); - assert parts.length == 2; - return Pair.create(parts[0], parts[1]); - } - - public static String getInputKeyspace(Configuration conf) - { - return conf.get(INPUT_KEYSPACE_CONFIG); - } - - public static String getOutputKeyspace(Configuration conf) - { - return conf.get(OUTPUT_KEYSPACE_CONFIG); - } - - public static void setInputKeyspaceUserNameAndPassword(Configuration conf, String username, String password) - { - setInputKeyspaceUserName(conf, username); - setInputKeyspacePassword(conf, password); - } - - public static void setInputKeyspaceUserName(Configuration conf, String username) - { - conf.set(INPUT_KEYSPACE_USERNAME_CONFIG, username); - } - - public static String getInputKeyspaceUserName(Configuration conf) - { - return conf.get(INPUT_KEYSPACE_USERNAME_CONFIG); - } - - public static void setInputKeyspacePassword(Configuration conf, String password) - { - conf.set(INPUT_KEYSPACE_PASSWD_CONFIG, password); - } - - public static String getInputKeyspacePassword(Configuration conf) - { - return conf.get(INPUT_KEYSPACE_PASSWD_CONFIG); - } - - public static void setOutputKeyspaceUserNameAndPassword(Configuration conf, String username, String password) - { - setOutputKeyspaceUserName(conf, username); - setOutputKeyspacePassword(conf, password); - } - - public static void setOutputKeyspaceUserName(Configuration conf, String username) - { - conf.set(OUTPUT_KEYSPACE_USERNAME_CONFIG, username); - } - - public static String getOutputKeyspaceUserName(Configuration conf) - { - return conf.get(OUTPUT_KEYSPACE_USERNAME_CONFIG); - } - - public static void setOutputKeyspacePassword(Configuration conf, String password) - { - conf.set(OUTPUT_KEYSPACE_PASSWD_CONFIG, password); - } - - public static String getOutputKeyspacePassword(Configuration conf) - { - return conf.get(OUTPUT_KEYSPACE_PASSWD_CONFIG); - } - - public static String getInputColumnFamily(Configuration conf) - { - return conf.get(INPUT_COLUMNFAMILY_CONFIG); - } - - public static String getOutputColumnFamily(Configuration conf) - { - if (conf.get(OUTPUT_COLUMNFAMILY_CONFIG) != null) - return conf.get(OUTPUT_COLUMNFAMILY_CONFIG); - else - throw new UnsupportedOperationException("You must set the output column family using either setOutputColumnFamily or by adding a named output with MultipleOutputs"); - } - - public static boolean getInputIsWide(Configuration conf) - { - return Boolean.parseBoolean(conf.get(INPUT_WIDEROWS_CONFIG)); - } - - public static String getReadConsistencyLevel(Configuration conf) - { - return conf.get(READ_CONSISTENCY_LEVEL, "LOCAL_ONE"); - } - - public static void setReadConsistencyLevel(Configuration conf, String consistencyLevel) - { - conf.set(READ_CONSISTENCY_LEVEL, consistencyLevel); - } - - public static String getWriteConsistencyLevel(Configuration conf) - { - return conf.get(WRITE_CONSISTENCY_LEVEL, "LOCAL_ONE"); - } - - public static void setWriteConsistencyLevel(Configuration conf, String consistencyLevel) - { - conf.set(WRITE_CONSISTENCY_LEVEL, consistencyLevel); - } - - public static String getInputInitialAddress(Configuration conf) - { - return conf.get(INPUT_INITIAL_ADDRESS); - } - - public static void setInputInitialAddress(Configuration conf, String address) - { - conf.set(INPUT_INITIAL_ADDRESS, address); - } - public static void setInputPartitioner(Configuration conf, String classname) - { - conf.set(INPUT_PARTITIONER_CONFIG, classname); - } - - public static IPartitioner getInputPartitioner(Configuration conf) - { - return FBUtilities.newPartitioner(conf.get(INPUT_PARTITIONER_CONFIG)); - } - - public static String getOutputInitialAddress(Configuration conf) - { - return conf.get(OUTPUT_INITIAL_ADDRESS); - } - - public static void setOutputInitialPort(Configuration conf, Integer port) - { - conf.set(OUTPUT_INITIAL_PORT, port.toString()); - } - - public static Integer getOutputInitialPort(Configuration conf) - { - return Integer.valueOf(conf.get(OUTPUT_INITIAL_PORT, DEFAULT_CASSANDRA_NATIVE_PORT)); - } - - public static void setOutputInitialAddress(Configuration conf, String address) - { - conf.set(OUTPUT_INITIAL_ADDRESS, address); - } - - public static void setOutputPartitioner(Configuration conf, String classname) - { - conf.set(OUTPUT_PARTITIONER_CONFIG, classname); - } - - public static IPartitioner getOutputPartitioner(Configuration conf) - { - return FBUtilities.newPartitioner(conf.get(OUTPUT_PARTITIONER_CONFIG)); - } - - public static String getOutputCompressionClass(Configuration conf) - { - return conf.get(OUTPUT_COMPRESSION_CLASS); - } - - public static String getOutputCompressionChunkLength(Configuration conf) - { - return conf.get(OUTPUT_COMPRESSION_CHUNK_LENGTH, String.valueOf(CompressionParams.DEFAULT_CHUNK_LENGTH)); - } - - public static void setOutputCompressionClass(Configuration conf, String classname) - { - conf.set(OUTPUT_COMPRESSION_CLASS, classname); - } - - public static void setOutputCompressionChunkLength(Configuration conf, String length) - { - conf.set(OUTPUT_COMPRESSION_CHUNK_LENGTH, length); - } - - public static boolean getOutputLocalDCOnly(Configuration conf) - { - return Boolean.parseBoolean(conf.get(OUTPUT_LOCAL_DC_ONLY, "false")); - } - - public static void setOutputLocalDCOnly(Configuration conf, boolean localDCOnly) - { - conf.set(OUTPUT_LOCAL_DC_ONLY, Boolean.toString(localDCOnly)); - } -} diff --git a/src/java/org/apache/cassandra/hadoop/HadoopCompat.java b/src/java/org/apache/cassandra/hadoop/HadoopCompat.java deleted file mode 100644 index 479f948e60..0000000000 --- a/src/java/org/apache/cassandra/hadoop/HadoopCompat.java +++ /dev/null @@ -1,350 +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; - -import java.lang.reflect.Constructor; -import java.lang.reflect.Field; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; - -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.mapreduce.Counter; -import org.apache.hadoop.mapreduce.InputSplit; -import org.apache.hadoop.mapreduce.JobContext; -import org.apache.hadoop.mapreduce.JobID; -import org.apache.hadoop.mapreduce.MapContext; -import org.apache.hadoop.mapreduce.OutputCommitter; -import org.apache.hadoop.mapreduce.RecordReader; -import org.apache.hadoop.mapreduce.RecordWriter; -import org.apache.hadoop.mapreduce.StatusReporter; -import org.apache.hadoop.mapreduce.TaskAttemptContext; -import org.apache.hadoop.mapreduce.TaskAttemptID; -import org.apache.hadoop.mapreduce.TaskInputOutputContext; - -/* - * This is based on ContextFactory.java from hadoop-2.0.x sources. - */ - -/** - * Utility methods to allow applications to deal with inconsistencies between - * MapReduce Context Objects API between Hadoop 1.x and 2.x. - */ -public class HadoopCompat -{ - - private static final boolean useV21; - - private static final Constructor JOB_CONTEXT_CONSTRUCTOR; - private static final Constructor TASK_CONTEXT_CONSTRUCTOR; - private static final Constructor MAP_CONTEXT_CONSTRUCTOR; - private static final Constructor GENERIC_COUNTER_CONSTRUCTOR; - - private static final Field READER_FIELD; - private static final Field WRITER_FIELD; - - private static final Method GET_CONFIGURATION_METHOD; - private static final Method SET_STATUS_METHOD; - private static final Method GET_COUNTER_METHOD; - private static final Method INCREMENT_COUNTER_METHOD; - private static final Method GET_TASK_ATTEMPT_ID; - private static final Method PROGRESS_METHOD; - - static - { - boolean v21 = true; - final String PACKAGE = "org.apache.hadoop.mapreduce"; - try - { - Class.forName(PACKAGE + ".task.JobContextImpl"); - } catch (ClassNotFoundException cnfe) - { - v21 = false; - } - useV21 = v21; - Class jobContextCls; - Class taskContextCls; - Class taskIOContextCls; - Class mapContextCls; - Class genericCounterCls; - try - { - if (v21) - { - jobContextCls = - Class.forName(PACKAGE+".task.JobContextImpl"); - taskContextCls = - Class.forName(PACKAGE+".task.TaskAttemptContextImpl"); - taskIOContextCls = - Class.forName(PACKAGE+".task.TaskInputOutputContextImpl"); - mapContextCls = Class.forName(PACKAGE + ".task.MapContextImpl"); - genericCounterCls = Class.forName(PACKAGE+".counters.GenericCounter"); - } - else - { - jobContextCls = - Class.forName(PACKAGE+".JobContext"); - taskContextCls = - Class.forName(PACKAGE+".TaskAttemptContext"); - taskIOContextCls = - Class.forName(PACKAGE+".TaskInputOutputContext"); - mapContextCls = Class.forName(PACKAGE + ".MapContext"); - genericCounterCls = - Class.forName("org.apache.hadoop.mapred.Counters$Counter"); - - } - } catch (ClassNotFoundException e) - { - throw new IllegalArgumentException("Can't find class", e); - } - try - { - JOB_CONTEXT_CONSTRUCTOR = - jobContextCls.getConstructor(Configuration.class, JobID.class); - JOB_CONTEXT_CONSTRUCTOR.setAccessible(true); - TASK_CONTEXT_CONSTRUCTOR = - taskContextCls.getConstructor(Configuration.class, - TaskAttemptID.class); - TASK_CONTEXT_CONSTRUCTOR.setAccessible(true); - GENERIC_COUNTER_CONSTRUCTOR = - genericCounterCls.getDeclaredConstructor(String.class, - String.class, - Long.TYPE); - GENERIC_COUNTER_CONSTRUCTOR.setAccessible(true); - - if (useV21) - { - MAP_CONTEXT_CONSTRUCTOR = - mapContextCls.getDeclaredConstructor(Configuration.class, - TaskAttemptID.class, - RecordReader.class, - RecordWriter.class, - OutputCommitter.class, - StatusReporter.class, - InputSplit.class); - Method get_counter; - try - { - get_counter = Class.forName(PACKAGE + ".TaskAttemptContext").getMethod("getCounter", String.class, - String.class); - } - catch (Exception e) - { - get_counter = Class.forName(PACKAGE + ".TaskInputOutputContext").getMethod("getCounter", - String.class, String.class); - } - GET_COUNTER_METHOD = get_counter; - } - else - { - MAP_CONTEXT_CONSTRUCTOR = - mapContextCls.getConstructor(Configuration.class, - TaskAttemptID.class, - RecordReader.class, - RecordWriter.class, - OutputCommitter.class, - StatusReporter.class, - InputSplit.class); - GET_COUNTER_METHOD = Class.forName(PACKAGE+".TaskInputOutputContext") - .getMethod("getCounter", String.class, String.class); - } - MAP_CONTEXT_CONSTRUCTOR.setAccessible(true); - READER_FIELD = mapContextCls.getDeclaredField("reader"); - READER_FIELD.setAccessible(true); - WRITER_FIELD = taskIOContextCls.getDeclaredField("output"); - WRITER_FIELD.setAccessible(true); - GET_CONFIGURATION_METHOD = Class.forName(PACKAGE+".JobContext") - .getMethod("getConfiguration"); - SET_STATUS_METHOD = Class.forName(PACKAGE+".TaskAttemptContext") - .getMethod("setStatus", String.class); - GET_TASK_ATTEMPT_ID = Class.forName(PACKAGE+".TaskAttemptContext") - .getMethod("getTaskAttemptID"); - INCREMENT_COUNTER_METHOD = Class.forName(PACKAGE+".Counter") - .getMethod("increment", Long.TYPE); - PROGRESS_METHOD = Class.forName(PACKAGE+".TaskAttemptContext") - .getMethod("progress"); - - } - catch (SecurityException e) - { - throw new IllegalArgumentException("Can't run constructor ", e); - } - catch (NoSuchMethodException e) - { - throw new IllegalArgumentException("Can't find constructor ", e); - } - catch (NoSuchFieldException e) - { - throw new IllegalArgumentException("Can't find field ", e); - } - catch (ClassNotFoundException e) - { - throw new IllegalArgumentException("Can't find class", e); - } - } - - /** - * True if runtime Hadoop version is 2.x, false otherwise. - */ - public static boolean isVersion2x() - { - return useV21; - } - - private static Object newInstance(Constructor constructor, Object...args) - { - try - { - return constructor.newInstance(args); - } - catch (InstantiationException e) - { - throw new IllegalArgumentException("Can't instantiate " + constructor, e); - } - catch (IllegalAccessException e) - { - throw new IllegalArgumentException("Can't instantiate " + constructor, e); - } - catch (InvocationTargetException e) - { - throw new IllegalArgumentException("Can't instantiate " + constructor, e); - } - } - - /** - * Creates JobContext from a JobConf and jobId using the correct constructor - * for based on Hadoop version. jobId could be null. - */ - public static JobContext newJobContext(Configuration conf, JobID jobId) { - return (JobContext) newInstance(JOB_CONTEXT_CONSTRUCTOR, conf, jobId); - } - - /** - * Creates TaskAttempContext from a JobConf and jobId using the correct - * constructor for based on Hadoop version. - */ - public static TaskAttemptContext newTaskAttemptContext( - Configuration conf, TaskAttemptID taskAttemptId) { - return (TaskAttemptContext) - newInstance(TASK_CONTEXT_CONSTRUCTOR, conf, taskAttemptId); - } - - /** - * Instantiates MapContext under Hadoop 1 and MapContextImpl under Hadoop 2. - */ - public static MapContext newMapContext(Configuration conf, - TaskAttemptID taskAttemptID, - RecordReader recordReader, - RecordWriter recordWriter, - OutputCommitter outputCommitter, - StatusReporter statusReporter, - InputSplit inputSplit) { - return (MapContext) newInstance(MAP_CONTEXT_CONSTRUCTOR, - conf, taskAttemptID, recordReader, recordWriter, outputCommitter, - statusReporter, inputSplit); - } - - /** - * @return with Hadoop 2 : new GenericCounter(args),
- * with Hadoop 1 : new Counter(args) - */ - public static Counter newGenericCounter(String name, String displayName, long value) - { - try - { - return (Counter) - GENERIC_COUNTER_CONSTRUCTOR.newInstance(name, displayName, value); - } - catch (InstantiationException | IllegalAccessException | InvocationTargetException e) - { - throw new IllegalArgumentException("Can't instantiate Counter", e); - } - } - - /** - * Invokes a method and rethrows any exception as runtime excetpions. - */ - private static Object invoke(Method method, Object obj, Object... args) - { - try - { - return method.invoke(obj, args); - } - catch (IllegalAccessException | InvocationTargetException e) - { - throw new IllegalArgumentException("Can't invoke method " + method.getName(), e); - } - } - - /** - * Invoke getConfiguration() on JobContext. Works with both - * Hadoop 1 and 2. - */ - public static Configuration getConfiguration(JobContext context) - { - return (Configuration) invoke(GET_CONFIGURATION_METHOD, context); - } - - /** - * Invoke setStatus() on TaskAttemptContext. Works with both - * Hadoop 1 and 2. - */ - public static void setStatus(TaskAttemptContext context, String status) - { - invoke(SET_STATUS_METHOD, context, status); - } - - /** - * returns TaskAttemptContext.getTaskAttemptID(). Works with both - * Hadoop 1 and 2. - */ - public static TaskAttemptID getTaskAttemptID(TaskAttemptContext taskContext) - { - return (TaskAttemptID) invoke(GET_TASK_ATTEMPT_ID, taskContext); - } - - /** - * Invoke getCounter() on TaskInputOutputContext. Works with both - * Hadoop 1 and 2. - */ - public static Counter getCounter(TaskInputOutputContext context, - String groupName, String counterName) - { - return (Counter) invoke(GET_COUNTER_METHOD, context, groupName, counterName); - } - - /** - * Invoke TaskAttemptContext.progress(). Works with both - * Hadoop 1 and 2. - */ - public static void progress(TaskAttemptContext context) - { - invoke(PROGRESS_METHOD, context); - } - - /** - * Increment the counter. Works with both Hadoop 1 and 2 - */ - public static void incrementCounter(Counter counter, long increment) - { - // incrementing a count might be called often. Might be affected by - // cost of invoke(). might be good candidate to handle in a shim. - // (TODO Raghu) figure out how achieve such a build with maven - invoke(INCREMENT_COUNTER_METHOD, counter, increment); - } -} diff --git a/src/java/org/apache/cassandra/hadoop/ReporterWrapper.java b/src/java/org/apache/cassandra/hadoop/ReporterWrapper.java deleted file mode 100644 index d2cc7699ad..0000000000 --- a/src/java/org/apache/cassandra/hadoop/ReporterWrapper.java +++ /dev/null @@ -1,85 +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; - -import org.apache.hadoop.mapred.Counters; -import org.apache.hadoop.mapred.InputSplit; -import org.apache.hadoop.mapred.Reporter; -import org.apache.hadoop.mapreduce.StatusReporter; - -/** - * A reporter that works with both mapred and mapreduce APIs. - */ -public class ReporterWrapper extends StatusReporter implements Reporter -{ - private Reporter wrappedReporter; - - public ReporterWrapper(Reporter reporter) - { - wrappedReporter = reporter; - } - - @Override - public Counters.Counter getCounter(Enum anEnum) - { - return wrappedReporter.getCounter(anEnum); - } - - @Override - public Counters.Counter getCounter(String s, String s1) - { - return wrappedReporter.getCounter(s, s1); - } - - @Override - public void incrCounter(Enum anEnum, long l) - { - wrappedReporter.incrCounter(anEnum, l); - } - - @Override - public void incrCounter(String s, String s1, long l) - { - wrappedReporter.incrCounter(s, s1, l); - } - - @Override - public InputSplit getInputSplit() throws UnsupportedOperationException - { - return wrappedReporter.getInputSplit(); - } - - @Override - public void progress() - { - wrappedReporter.progress(); - } - - // @Override - public float getProgress() - { - throw new UnsupportedOperationException(); - } - - @Override - public void setStatus(String s) - { - wrappedReporter.setStatus(s); - } -} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkOutputFormat.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkOutputFormat.java deleted file mode 100644 index dfdf855e57..0000000000 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkOutputFormat.java +++ /dev/null @@ -1,209 +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.cql3; - - -import java.io.IOException; -import java.nio.ByteBuffer; -import java.util.Collection; -import java.util.List; - -import org.apache.cassandra.hadoop.ConfigHelper; -import org.apache.cassandra.hadoop.HadoopCompat; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.mapred.JobConf; -import org.apache.hadoop.mapreduce.JobContext; -import org.apache.hadoop.mapreduce.OutputCommitter; -import org.apache.hadoop.mapreduce.OutputFormat; -import org.apache.hadoop.mapreduce.RecordWriter; -import org.apache.hadoop.mapreduce.TaskAttemptContext; -import org.apache.hadoop.util.Progressable; - -/** - * The CqlBulkOutputFormat acts as a Hadoop-specific - * OutputFormat that allows reduce tasks to store keys (and corresponding - * bound variable values) as CQL rows (and respective columns) in a given - * table. - * - *

- * As is the case with the {@link org.apache.cassandra.hadoop.cql3.CqlOutputFormat}, - * you need to set the prepared statement in your - * Hadoop job Configuration. The {@link CqlConfigHelper} class, through its - * {@link org.apache.cassandra.hadoop.ConfigHelper#setOutputPreparedStatement} method, is provided to make this - * simple. - * you need to set the Keyspace. The {@link ConfigHelper} class, through its - * {@link org.apache.cassandra.hadoop.ConfigHelper#setOutputColumnFamily} method, is provided to make this - * simple. - *

- */ -public class CqlBulkOutputFormat extends OutputFormat> - implements org.apache.hadoop.mapred.OutputFormat> -{ - - private static final String OUTPUT_CQL_SCHEMA_PREFIX = "cassandra.table.schema."; - private static final String OUTPUT_CQL_INSERT_PREFIX = "cassandra.table.insert."; - private static final String DELETE_SOURCE = "cassandra.output.delete.source"; - private static final String TABLE_ALIAS_PREFIX = "cqlbulkoutputformat.table.alias."; - - /** Fills the deprecated OutputFormat interface for streaming. */ - @Deprecated - public CqlBulkRecordWriter getRecordWriter(FileSystem filesystem, JobConf job, String name, Progressable progress) throws IOException - { - return new CqlBulkRecordWriter(job, progress); - } - - /** - * Get the {@link RecordWriter} for the given task. - * - * @param context - * the information about the current task. - * @return a {@link RecordWriter} to write the output for the job. - * @throws IOException - */ - public CqlBulkRecordWriter getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException - { - return new CqlBulkRecordWriter(context); - } - - @Override - public void checkOutputSpecs(JobContext context) - { - checkOutputSpecs(HadoopCompat.getConfiguration(context)); - } - - private void checkOutputSpecs(Configuration conf) - { - if (ConfigHelper.getOutputKeyspace(conf) == null) - { - throw new UnsupportedOperationException("you must set the keyspace with setTable()"); - } - } - - /** Fills the deprecated OutputFormat interface for streaming. */ - @Deprecated - public void checkOutputSpecs(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job) throws IOException - { - checkOutputSpecs(job); - } - - @Override - public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException - { - return new NullOutputCommitter(); - } - - public static void setTableSchema(Configuration conf, String columnFamily, String schema) - { - conf.set(OUTPUT_CQL_SCHEMA_PREFIX + columnFamily, schema); - } - - public static void setTableInsertStatement(Configuration conf, String columnFamily, String insertStatement) - { - conf.set(OUTPUT_CQL_INSERT_PREFIX + columnFamily, insertStatement); - } - - public static String getTableSchema(Configuration conf, String columnFamily) - { - String schema = conf.get(OUTPUT_CQL_SCHEMA_PREFIX + columnFamily); - if (schema == null) - { - throw new UnsupportedOperationException("You must set the Table schema using setTableSchema."); - } - return schema; - } - - public static String getTableInsertStatement(Configuration conf, String columnFamily) - { - String insert = conf.get(OUTPUT_CQL_INSERT_PREFIX + columnFamily); - if (insert == null) - { - throw new UnsupportedOperationException("You must set the Table insert statement using setTableSchema."); - } - return insert; - } - - public static void setDeleteSourceOnSuccess(Configuration conf, boolean deleteSrc) - { - conf.setBoolean(DELETE_SOURCE, deleteSrc); - } - - public static boolean getDeleteSourceOnSuccess(Configuration conf) - { - return conf.getBoolean(DELETE_SOURCE, false); - } - - public static void setTableAlias(Configuration conf, String alias, String columnFamily) - { - conf.set(TABLE_ALIAS_PREFIX + alias, columnFamily); - } - - public static String getTableForAlias(Configuration conf, String alias) - { - return conf.get(TABLE_ALIAS_PREFIX + alias); - } - - /** - * Set the hosts to ignore as comma delimited values. - * Data will not be bulk loaded onto the ignored nodes. - * @param conf job configuration - * @param ignoreNodesCsv a comma delimited list of nodes to ignore - */ - public static void setIgnoreHosts(Configuration conf, String ignoreNodesCsv) - { - conf.set(CqlBulkRecordWriter.IGNORE_HOSTS, ignoreNodesCsv); - } - - /** - * Set the hosts to ignore. Data will not be bulk loaded onto the ignored nodes. - * @param conf job configuration - * @param ignoreNodes the nodes to ignore - */ - public static void setIgnoreHosts(Configuration conf, String... ignoreNodes) - { - conf.setStrings(CqlBulkRecordWriter.IGNORE_HOSTS, ignoreNodes); - } - - /** - * Get the hosts to ignore as a collection of strings - * @param conf job configuration - * @return the nodes to ignore as a collection of stirngs - */ - public static Collection getIgnoreHosts(Configuration conf) - { - return conf.getStringCollection(CqlBulkRecordWriter.IGNORE_HOSTS); - } - - public static class NullOutputCommitter extends OutputCommitter - { - public void abortTask(TaskAttemptContext taskContext) { } - - public void cleanupJob(JobContext jobContext) { } - - public void commitTask(TaskAttemptContext taskContext) { } - - public boolean needsTaskCommit(TaskAttemptContext taskContext) - { - return false; - } - - public void setupJob(JobContext jobContext) { } - - public void setupTask(TaskAttemptContext taskContext) { } - } -} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java deleted file mode 100644 index 9b154e9241..0000000000 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java +++ /dev/null @@ -1,333 +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.cql3; - -import java.io.Closeable; -import java.io.IOException; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.net.UnknownHostException; -import java.nio.ByteBuffer; -import java.util.Collection; -import java.util.HashSet; -import java.util.List; -import java.util.Set; -import java.util.UUID; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; - -import com.google.common.net.HostAndPort; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; -import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.dht.Murmur3Partitioner; -import org.apache.cassandra.exceptions.InvalidRequestException; -import org.apache.cassandra.hadoop.ConfigHelper; -import org.apache.cassandra.hadoop.HadoopCompat; -import org.apache.cassandra.io.sstable.CQLSSTableWriter; -import org.apache.cassandra.io.sstable.SSTableLoader; -import org.apache.cassandra.io.util.File; -import org.apache.cassandra.io.util.FileUtils; -import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.schema.TableMetadataRef; -import org.apache.cassandra.streaming.StreamState; -import org.apache.cassandra.utils.NativeSSTableLoaderClient; -import org.apache.cassandra.utils.OutputHandler; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.mapreduce.RecordWriter; -import org.apache.hadoop.mapreduce.TaskAttemptContext; -import org.apache.hadoop.util.Progressable; - -import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_IO_TMPDIR; - -/** - * The CqlBulkRecordWriter maps the output <key, value> - * pairs to a Cassandra column family. In particular, it applies the binded variables - * in the value to the prepared statement, which it associates with the key, and in - * turn the responsible endpoint. - * - *

- * Furthermore, this writer groups the cql queries by the endpoint responsible for - * the rows being affected. This allows the cql queries to be executed in parallel, - * directly to a responsible endpoint. - *

- * - * @see CqlBulkOutputFormat - */ -public class CqlBulkRecordWriter extends RecordWriter> - implements org.apache.hadoop.mapred.RecordWriter> -{ - public final static String OUTPUT_LOCATION = "mapreduce.output.bulkoutputformat.localdir"; - public final static String BUFFER_SIZE_IN_MB = "mapreduce.output.bulkoutputformat.buffersize"; - public final static String STREAM_THROTTLE_MBITS = "mapreduce.output.bulkoutputformat.streamthrottlembits"; - public final static String MAX_FAILED_HOSTS = "mapreduce.output.bulkoutputformat.maxfailedhosts"; - public final static String IGNORE_HOSTS = "mapreduce.output.bulkoutputformat.ignorehosts"; - - private final Logger logger = LoggerFactory.getLogger(CqlBulkRecordWriter.class); - - protected final Configuration conf; - protected final int maxFailures; - protected final int bufferSize; - protected Closeable writer; - protected SSTableLoader loader; - protected Progressable progress; - protected TaskAttemptContext context; - protected final Set ignores = new HashSet<>(); - - private String keyspace; - private String table; - private String schema; - private String insertStatement; - private File outputDir; - private boolean deleteSrc; - private IPartitioner partitioner; - - CqlBulkRecordWriter(TaskAttemptContext context) throws IOException - { - this(HadoopCompat.getConfiguration(context)); - this.context = context; - setConfigs(); - } - - CqlBulkRecordWriter(Configuration conf, Progressable progress) throws IOException - { - this(conf); - this.progress = progress; - setConfigs(); - } - - CqlBulkRecordWriter(Configuration conf) throws IOException - { - this.conf = conf; - DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(Integer.parseInt(conf.get(STREAM_THROTTLE_MBITS, "0"))); - maxFailures = Integer.parseInt(conf.get(MAX_FAILED_HOSTS, "0")); - bufferSize = Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB, "64")); - setConfigs(); - } - - private void setConfigs() throws IOException - { - // if anything is missing, exceptions will be thrown here, instead of on write() - keyspace = ConfigHelper.getOutputKeyspace(conf); - table = ConfigHelper.getOutputColumnFamily(conf); - - // check if table is aliased - String aliasedCf = CqlBulkOutputFormat.getTableForAlias(conf, table); - if (aliasedCf != null) - table = aliasedCf; - - schema = CqlBulkOutputFormat.getTableSchema(conf, table); - insertStatement = CqlBulkOutputFormat.getTableInsertStatement(conf, table); - outputDir = getTableDirectory(); - deleteSrc = CqlBulkOutputFormat.getDeleteSourceOnSuccess(conf); - try - { - partitioner = ConfigHelper.getInputPartitioner(conf); - } - catch (Exception e) - { - partitioner = Murmur3Partitioner.instance; - } - try - { - for (String hostToIgnore : CqlBulkOutputFormat.getIgnoreHosts(conf)) - ignores.add(InetAddressAndPort.getByName(hostToIgnore)); - } - catch (UnknownHostException e) - { - throw new RuntimeException(("Unknown host: " + e.getMessage())); - } - } - - protected String getOutputLocation() throws IOException - { - String dir = conf.get(OUTPUT_LOCATION, JAVA_IO_TMPDIR.getString()); - if (dir == null) - throw new IOException("Output directory not defined, if hadoop is not setting java.io.tmpdir then define " + OUTPUT_LOCATION); - return dir; - } - - private void prepareWriter() throws IOException - { - if (writer == null) - { - writer = CQLSSTableWriter.builder() - .forTable(schema) - .using(insertStatement) - .withPartitioner(ConfigHelper.getOutputPartitioner(conf)) - .inDirectory(outputDir) - .withBufferSizeInMiB(Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB, "64"))) - .withPartitioner(partitioner) - .build(); - } - - if (loader == null) - { - ExternalClient externalClient = new ExternalClient(conf); - externalClient.setTableMetadata(TableMetadataRef.forOfflineTools(CreateTableStatement.parse(schema, keyspace).build())); - - loader = new SSTableLoader(outputDir, externalClient, new NullOutputHandler()) - { - @Override - public void onSuccess(StreamState finalState) - { - if (deleteSrc) - FileUtils.deleteRecursive(outputDir); - } - }; - } - } - - /** - *

- * The column values must correspond to the order in which - * they appear in the insert stored procedure. - * - * Key is not used, so it can be null or any object. - *

- * - * @param key - * any object or null. - * @param values - * the values to write. - * @throws IOException - */ - @Override - public void write(Object key, List values) throws IOException - { - prepareWriter(); - try - { - ((CQLSSTableWriter) writer).rawAddRow(values); - - if (null != progress) - progress.progress(); - if (null != context) - HadoopCompat.progress(context); - } - catch (InvalidRequestException e) - { - throw new IOException("Error adding row with key: " + key, e); - } - } - - private File getTableDirectory() throws IOException - { - File dir = new File(String.format("%s%s%s%s%s-%s", getOutputLocation(), File.pathSeparator(), keyspace, File.pathSeparator(), table, UUID.randomUUID().toString())); - - if (!dir.exists() && !dir.tryCreateDirectories()) - { - throw new IOException("Failed to created output directory: " + dir); - } - - return dir; - } - - @Override - public void close(TaskAttemptContext context) throws IOException, InterruptedException - { - close(); - } - - /** Fills the deprecated RecordWriter interface for streaming. */ - @Deprecated - public void close(org.apache.hadoop.mapred.Reporter reporter) throws IOException - { - close(); - } - - private void close() throws IOException - { - if (writer != null) - { - writer.close(); - Future future = loader.stream(ignores); - while (true) - { - try - { - future.get(1000, TimeUnit.MILLISECONDS); - break; - } - catch (ExecutionException | TimeoutException te) - { - if (null != progress) - progress.progress(); - if (null != context) - HadoopCompat.progress(context); - } - catch (InterruptedException e) - { - throw new IOException(e); - } - } - if (loader.getFailedHosts().size() > 0) - { - if (loader.getFailedHosts().size() > maxFailures) - throw new IOException("Too many hosts failed: " + loader.getFailedHosts()); - else - logger.warn("Some hosts failed: {}", loader.getFailedHosts()); - } - } - } - - public static class ExternalClient extends NativeSSTableLoaderClient - { - public ExternalClient(Configuration conf) - { - super(resolveHostAddresses(conf), - ConfigHelper.getOutputInitialPort(conf), - ConfigHelper.getOutputKeyspaceUserName(conf), - ConfigHelper.getOutputKeyspacePassword(conf), - CqlConfigHelper.getSSLOptions(conf).orNull()); - } - - private static Collection resolveHostAddresses(Configuration conf) - { - Set addresses = new HashSet<>(); - int port = CqlConfigHelper.getOutputNativePort(conf); - for (String host : ConfigHelper.getOutputInitialAddress(conf).split(",")) - { - try - { - HostAndPort hap = HostAndPort.fromString(host); - addresses.add(new InetSocketAddress(InetAddress.getByName(hap.getHost()), hap.getPortOrDefault(port))); - } - catch (UnknownHostException e) - { - throw new RuntimeException(e); - } - } - - return addresses; - } - } - - public static class NullOutputHandler implements OutputHandler - { - public void output(String msg) {} - public void debug(String msg) {} - public void warn(String msg) {} - public void warn(Throwable th, String msg) {} - } -} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlClientHelper.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlClientHelper.java deleted file mode 100644 index e0a6384251..0000000000 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlClientHelper.java +++ /dev/null @@ -1,109 +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.cql3; - -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Objects; - -import com.datastax.driver.core.Host; -import com.datastax.driver.core.Metadata; -import com.datastax.driver.core.Token; -import com.datastax.driver.core.TokenRange; - -public class CqlClientHelper -{ - private CqlClientHelper() - { - } - - public static Map> getLocalPrimaryRangeForDC(String keyspace, Metadata metadata, String targetDC) - { - Objects.requireNonNull(keyspace, "keyspace"); - Objects.requireNonNull(metadata, "metadata"); - Objects.requireNonNull(targetDC, "targetDC"); - - // In 2.1 the logic was to have a set of nodes used as a seed, they were used to query - // client.describe_local_ring(keyspace) -> List; this should include all nodes in the local dc. - // TokenRange contained the endpoints in order, so .endpoints.get(0) is the primary owner - // Client does not have a similar API, instead it returns Set. To replicate this we first need - // to compute the primary owners, then add in the replicas - - List tokens = new ArrayList<>(); - Map tokenToHost = new HashMap<>(); - for (Host host : metadata.getAllHosts()) - { - if (!targetDC.equals(host.getDatacenter())) - continue; - - for (Token token : host.getTokens()) - { - Host previous = tokenToHost.putIfAbsent(token, host); - if (previous != null) - throw new IllegalStateException("Two hosts share the same token; hosts " + host.getHostId() + ":" - + host.getTokens() + ", " + previous.getHostId() + ":" + previous.getTokens()); - tokens.add(token); - } - } - Collections.sort(tokens); - - Map> rangeToReplicas = new HashMap<>(); - - // The first token in the ring uses the last token as its 'start', handle this here to simplify the loop - Token start = tokens.get(tokens.size() - 1); - Token end = tokens.get(0); - - addRange(keyspace, metadata, tokenToHost, rangeToReplicas, start, end); - for (int i = 1; i < tokens.size(); i++) - { - start = tokens.get(i - 1); - end = tokens.get(i); - - addRange(keyspace, metadata, tokenToHost, rangeToReplicas, start, end); - } - - return rangeToReplicas; - } - - private static void addRange(String keyspace, - Metadata metadata, - Map tokenToHost, - Map> rangeToReplicas, - Token start, Token end) - { - Host host = tokenToHost.get(end); - String dc = host.getDatacenter(); - - TokenRange range = metadata.newTokenRange(start, end); - List replicas = new ArrayList<>(); - replicas.add(host); - // get all the replicas for the specific DC - for (Host replica : metadata.getReplicas(keyspace, range)) - { - if (dc.equals(replica.getDatacenter()) && !host.equals(replica)) - replicas.add(replica); - } - List previous = rangeToReplicas.put(range, replicas); - if (previous != null) - throw new IllegalStateException("Two hosts (" + host + ", " + previous + ") map to the same token range: " + range); - } -} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java deleted file mode 100644 index 998a754692..0000000000 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java +++ /dev/null @@ -1,654 +0,0 @@ -package org.apache.cassandra.hadoop.cql3; -/* -* -* 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. -* -*/ -import java.nio.file.Files; -import java.io.InputStream; -import java.io.IOException; -import java.security.KeyManagementException; -import java.security.KeyStore; -import java.security.KeyStoreException; -import java.security.NoSuchAlgorithmException; -import java.security.SecureRandom; -import java.security.UnrecoverableKeyException; -import java.security.cert.CertificateException; -import java.util.Arrays; - -import javax.net.ssl.KeyManagerFactory; -import javax.net.ssl.SSLContext; -import javax.net.ssl.TrustManagerFactory; - -import com.google.common.base.Optional; -import org.apache.commons.lang3.StringUtils; - -import com.datastax.driver.core.AuthProvider; -import com.datastax.driver.core.Cluster; -import com.datastax.driver.core.HostDistance; -import com.datastax.driver.core.JdkSSLOptions; -import com.datastax.driver.core.PlainTextAuthProvider; -import com.datastax.driver.core.ProtocolVersion; -import com.datastax.driver.core.policies.LoadBalancingPolicy; -import com.datastax.driver.core.PoolingOptions; -import com.datastax.driver.core.ProtocolOptions; -import com.datastax.driver.core.QueryOptions; -import com.datastax.driver.core.SSLOptions; -import com.datastax.driver.core.SocketOptions; -import org.apache.cassandra.hadoop.ConfigHelper; -import org.apache.cassandra.io.util.File; -import org.apache.hadoop.conf.Configuration; - - -public class CqlConfigHelper -{ - private static final String INPUT_CQL_COLUMNS_CONFIG = "cassandra.input.columnfamily.columns"; - private static final String INPUT_CQL_PAGE_ROW_SIZE_CONFIG = "cassandra.input.page.row.size"; - private static final String INPUT_CQL_WHERE_CLAUSE_CONFIG = "cassandra.input.where.clause"; - private static final String INPUT_CQL = "cassandra.input.cql"; - - private static final String USERNAME = "cassandra.username"; - private static final String PASSWORD = "cassandra.password"; - - private static final String INPUT_NATIVE_PORT = "cassandra.input.native.port"; - private static final String INPUT_NATIVE_CORE_CONNECTIONS_PER_HOST = "cassandra.input.native.core.connections.per.host"; - private static final String INPUT_NATIVE_MAX_CONNECTIONS_PER_HOST = "cassandra.input.native.max.connections.per.host"; - private static final String INPUT_NATIVE_MAX_SIMULT_REQ_PER_CONNECTION = "cassandra.input.native.max.simult.reqs.per.connection"; - private static final String INPUT_NATIVE_CONNECTION_TIMEOUT = "cassandra.input.native.connection.timeout"; - private static final String INPUT_NATIVE_READ_CONNECTION_TIMEOUT = "cassandra.input.native.read.connection.timeout"; - private static final String INPUT_NATIVE_RECEIVE_BUFFER_SIZE = "cassandra.input.native.receive.buffer.size"; - private static final String INPUT_NATIVE_SEND_BUFFER_SIZE = "cassandra.input.native.send.buffer.size"; - private static final String INPUT_NATIVE_SOLINGER = "cassandra.input.native.solinger"; - private static final String INPUT_NATIVE_TCP_NODELAY = "cassandra.input.native.tcp.nodelay"; - private static final String INPUT_NATIVE_REUSE_ADDRESS = "cassandra.input.native.reuse.address"; - private static final String INPUT_NATIVE_KEEP_ALIVE = "cassandra.input.native.keep.alive"; - private static final String INPUT_NATIVE_AUTH_PROVIDER = "cassandra.input.native.auth.provider"; - private static final String INPUT_NATIVE_SSL_TRUST_STORE_PATH = "cassandra.input.native.ssl.trust.store.path"; - private static final String INPUT_NATIVE_SSL_KEY_STORE_PATH = "cassandra.input.native.ssl.key.store.path"; - private static final String INPUT_NATIVE_SSL_TRUST_STORE_PASSWARD = "cassandra.input.native.ssl.trust.store.password"; - private static final String INPUT_NATIVE_SSL_KEY_STORE_PASSWARD = "cassandra.input.native.ssl.key.store.password"; - private static final String INPUT_NATIVE_SSL_CIPHER_SUITES = "cassandra.input.native.ssl.cipher.suites"; - - private static final String INPUT_NATIVE_PROTOCOL_VERSION = "cassandra.input.native.protocol.version"; - - private static final String OUTPUT_CQL = "cassandra.output.cql"; - private static final String OUTPUT_NATIVE_PORT = "cassandra.output.native.port"; - - /** - * Set the CQL columns for the input of this job. - * - * @param conf Job configuration you are about to run - * @param columns - */ - public static void setInputColumns(Configuration conf, String columns) - { - if (columns == null || columns.isEmpty()) - return; - - conf.set(INPUT_CQL_COLUMNS_CONFIG, columns); - } - - /** - * Set the CQL query Limit for the input of this job. - * - * @param conf Job configuration you are about to run - * @param cqlPageRowSize - */ - public static void setInputCQLPageRowSize(Configuration conf, String cqlPageRowSize) - { - if (cqlPageRowSize == null) - { - throw new UnsupportedOperationException("cql page row size may not be null"); - } - - conf.set(INPUT_CQL_PAGE_ROW_SIZE_CONFIG, cqlPageRowSize); - } - - /** - * Set the CQL user defined where clauses for the input of this job. - * - * @param conf Job configuration you are about to run - * @param clauses - */ - public static void setInputWhereClauses(Configuration conf, String clauses) - { - if (clauses == null || clauses.isEmpty()) - return; - - conf.set(INPUT_CQL_WHERE_CLAUSE_CONFIG, clauses); - } - - /** - * Set the CQL prepared statement for the output of this job. - * - * @param conf Job configuration you are about to run - * @param cql - */ - public static void setOutputCql(Configuration conf, String cql) - { - if (cql == null || cql.isEmpty()) - return; - - conf.set(OUTPUT_CQL, cql); - } - - public static void setInputCql(Configuration conf, String cql) - { - if (cql == null || cql.isEmpty()) - return; - - conf.set(INPUT_CQL, cql); - } - - public static void setUserNameAndPassword(Configuration conf, String username, String password) - { - if (StringUtils.isNotBlank(username)) - { - conf.set(INPUT_NATIVE_AUTH_PROVIDER, PlainTextAuthProvider.class.getName()); - conf.set(USERNAME, username); - conf.set(PASSWORD, password); - } - } - - public static Optional getInputCoreConnections(Configuration conf) - { - return getIntSetting(INPUT_NATIVE_CORE_CONNECTIONS_PER_HOST, conf); - } - - public static Optional getInputMaxConnections(Configuration conf) - { - return getIntSetting(INPUT_NATIVE_MAX_CONNECTIONS_PER_HOST, conf); - } - - public static int getInputNativePort(Configuration conf) - { - return Integer.parseInt(conf.get(INPUT_NATIVE_PORT, "9042")); - } - - public static int getOutputNativePort(Configuration conf) - { - return Integer.parseInt(conf.get(OUTPUT_NATIVE_PORT, "9042")); - } - - public static Optional getInputMaxSimultReqPerConnections(Configuration conf) - { - return getIntSetting(INPUT_NATIVE_MAX_SIMULT_REQ_PER_CONNECTION, conf); - } - - public static Optional getInputNativeConnectionTimeout(Configuration conf) - { - return getIntSetting(INPUT_NATIVE_CONNECTION_TIMEOUT, conf); - } - - public static Optional getInputNativeReadConnectionTimeout(Configuration conf) - { - return getIntSetting(INPUT_NATIVE_READ_CONNECTION_TIMEOUT, conf); - } - - public static Optional getInputNativeReceiveBufferSize(Configuration conf) - { - return getIntSetting(INPUT_NATIVE_RECEIVE_BUFFER_SIZE, conf); - } - - public static Optional getInputNativeSendBufferSize(Configuration conf) - { - return getIntSetting(INPUT_NATIVE_SEND_BUFFER_SIZE, conf); - } - - public static Optional getInputNativeSolinger(Configuration conf) - { - return getIntSetting(INPUT_NATIVE_SOLINGER, conf); - } - - public static Optional getInputNativeTcpNodelay(Configuration conf) - { - return getBooleanSetting(INPUT_NATIVE_TCP_NODELAY, conf); - } - - public static Optional getInputNativeReuseAddress(Configuration conf) - { - return getBooleanSetting(INPUT_NATIVE_REUSE_ADDRESS, conf); - } - - public static Optional getInputNativeAuthProvider(Configuration conf) - { - return getStringSetting(INPUT_NATIVE_AUTH_PROVIDER, conf); - } - - public static Optional getInputNativeSSLTruststorePath(Configuration conf) - { - return getStringSetting(INPUT_NATIVE_SSL_TRUST_STORE_PATH, conf); - } - - public static Optional getInputNativeSSLKeystorePath(Configuration conf) - { - return getStringSetting(INPUT_NATIVE_SSL_KEY_STORE_PATH, conf); - } - - public static Optional getInputNativeSSLKeystorePassword(Configuration conf) - { - return getStringSetting(INPUT_NATIVE_SSL_KEY_STORE_PASSWARD, conf); - } - - public static Optional getInputNativeSSLTruststorePassword(Configuration conf) - { - return getStringSetting(INPUT_NATIVE_SSL_TRUST_STORE_PASSWARD, conf); - } - - public static Optional getInputNativeSSLCipherSuites(Configuration conf) - { - return getStringSetting(INPUT_NATIVE_SSL_CIPHER_SUITES, conf); - } - - public static Optional getInputNativeKeepAlive(Configuration conf) - { - return getBooleanSetting(INPUT_NATIVE_KEEP_ALIVE, conf); - } - - public static String getInputcolumns(Configuration conf) - { - return conf.get(INPUT_CQL_COLUMNS_CONFIG); - } - - public static Optional getInputPageRowSize(Configuration conf) - { - return getIntSetting(INPUT_CQL_PAGE_ROW_SIZE_CONFIG, conf); - } - - public static String getInputWhereClauses(Configuration conf) - { - return conf.get(INPUT_CQL_WHERE_CLAUSE_CONFIG); - } - - public static String getInputCql(Configuration conf) - { - return conf.get(INPUT_CQL); - } - - public static String getOutputCql(Configuration conf) - { - return conf.get(OUTPUT_CQL); - } - - private static Optional getProtocolVersion(Configuration conf) - { - return getIntSetting(INPUT_NATIVE_PROTOCOL_VERSION, conf); - } - - public static Cluster getInputCluster(String host, Configuration conf) - { - // this method has been left for backward compatibility - return getInputCluster(new String[] {host}, conf); - } - - public static Cluster getInputCluster(String[] hosts, Configuration conf) - { - int port = getInputNativePort(conf); - return getCluster(hosts, conf, port); - } - - public static Cluster getOutputCluster(String host, Configuration conf) - { - return getOutputCluster(new String[]{host}, conf); - } - - public static Cluster getOutputCluster(String[] hosts, Configuration conf) - { - int port = getOutputNativePort(conf); - return getCluster(hosts, conf, port); - } - - public static Cluster getCluster(String[] hosts, Configuration conf, int port) - { - Optional authProvider = getAuthProvider(conf); - Optional sslOptions = getSSLOptions(conf); - Optional protocolVersion = getProtocolVersion(conf); - LoadBalancingPolicy loadBalancingPolicy = getReadLoadBalancingPolicy(hosts); - SocketOptions socketOptions = getReadSocketOptions(conf); - QueryOptions queryOptions = getReadQueryOptions(conf); - PoolingOptions poolingOptions = getReadPoolingOptions(conf); - - Cluster.Builder builder = Cluster.builder() - .addContactPoints(hosts) - .withPort(port) - .withCompression(ProtocolOptions.Compression.NONE); - - if (authProvider.isPresent()) - builder.withAuthProvider(authProvider.get()); - if (sslOptions.isPresent()) - builder.withSSL(sslOptions.get()); - - if (protocolVersion.isPresent()) - { - builder.withProtocolVersion(ProtocolVersion.fromInt(protocolVersion.get())); - } - builder.withLoadBalancingPolicy(loadBalancingPolicy) - .withSocketOptions(socketOptions) - .withQueryOptions(queryOptions) - .withPoolingOptions(poolingOptions); - - return builder.build(); - } - - public static void setInputCoreConnections(Configuration conf, String connections) - { - conf.set(INPUT_NATIVE_CORE_CONNECTIONS_PER_HOST, connections); - } - - public static void setInputMaxConnections(Configuration conf, String connections) - { - conf.set(INPUT_NATIVE_MAX_CONNECTIONS_PER_HOST, connections); - } - - public static void setInputMaxSimultReqPerConnections(Configuration conf, String reqs) - { - conf.set(INPUT_NATIVE_MAX_SIMULT_REQ_PER_CONNECTION, reqs); - } - - public static void setInputNativeConnectionTimeout(Configuration conf, String timeout) - { - conf.set(INPUT_NATIVE_CONNECTION_TIMEOUT, timeout); - } - - public static void setInputNativeReadConnectionTimeout(Configuration conf, String timeout) - { - conf.set(INPUT_NATIVE_READ_CONNECTION_TIMEOUT, timeout); - } - - public static void setInputNativeReceiveBufferSize(Configuration conf, String size) - { - conf.set(INPUT_NATIVE_RECEIVE_BUFFER_SIZE, size); - } - - public static void setInputNativeSendBufferSize(Configuration conf, String size) - { - conf.set(INPUT_NATIVE_SEND_BUFFER_SIZE, size); - } - - public static void setInputNativeSolinger(Configuration conf, String solinger) - { - conf.set(INPUT_NATIVE_SOLINGER, solinger); - } - - public static void setInputNativeTcpNodelay(Configuration conf, String tcpNodelay) - { - conf.set(INPUT_NATIVE_TCP_NODELAY, tcpNodelay); - } - - public static void setInputNativeAuthProvider(Configuration conf, String authProvider) - { - conf.set(INPUT_NATIVE_AUTH_PROVIDER, authProvider); - } - - public static void setInputNativeSSLTruststorePath(Configuration conf, String path) - { - conf.set(INPUT_NATIVE_SSL_TRUST_STORE_PATH, path); - } - - public static void setInputNativeSSLKeystorePath(Configuration conf, String path) - { - conf.set(INPUT_NATIVE_SSL_KEY_STORE_PATH, path); - } - - public static void setInputNativeSSLKeystorePassword(Configuration conf, String pass) - { - conf.set(INPUT_NATIVE_SSL_KEY_STORE_PASSWARD, pass); - } - - public static void setInputNativeSSLTruststorePassword(Configuration conf, String pass) - { - conf.set(INPUT_NATIVE_SSL_TRUST_STORE_PASSWARD, pass); - } - - public static void setInputNativeSSLCipherSuites(Configuration conf, String suites) - { - conf.set(INPUT_NATIVE_SSL_CIPHER_SUITES, suites); - } - - public static void setInputNativeReuseAddress(Configuration conf, String reuseAddress) - { - conf.set(INPUT_NATIVE_REUSE_ADDRESS, reuseAddress); - } - - public static void setInputNativeKeepAlive(Configuration conf, String keepAlive) - { - conf.set(INPUT_NATIVE_KEEP_ALIVE, keepAlive); - } - - public static void setInputNativePort(Configuration conf, String port) - { - conf.set(INPUT_NATIVE_PORT, port); - } - - private static PoolingOptions getReadPoolingOptions(Configuration conf) - { - Optional coreConnections = getInputCoreConnections(conf); - Optional maxConnections = getInputMaxConnections(conf); - Optional maxSimultaneousRequests = getInputMaxSimultReqPerConnections(conf); - - PoolingOptions poolingOptions = new PoolingOptions(); - - for (HostDistance hostDistance : Arrays.asList(HostDistance.LOCAL, HostDistance.REMOTE)) - { - if (coreConnections.isPresent()) - poolingOptions.setCoreConnectionsPerHost(hostDistance, coreConnections.get()); - if (maxConnections.isPresent()) - poolingOptions.setMaxConnectionsPerHost(hostDistance, maxConnections.get()); - if (maxSimultaneousRequests.isPresent()) - poolingOptions.setNewConnectionThreshold(hostDistance, maxSimultaneousRequests.get()); - } - - return poolingOptions; - } - - private static QueryOptions getReadQueryOptions(Configuration conf) - { - String CL = ConfigHelper.getReadConsistencyLevel(conf); - Optional fetchSize = getInputPageRowSize(conf); - QueryOptions queryOptions = new QueryOptions(); - if (CL != null && !CL.isEmpty()) - queryOptions.setConsistencyLevel(com.datastax.driver.core.ConsistencyLevel.valueOf(CL)); - - if (fetchSize.isPresent()) - queryOptions.setFetchSize(fetchSize.get()); - return queryOptions; - } - - private static SocketOptions getReadSocketOptions(Configuration conf) - { - SocketOptions socketOptions = new SocketOptions(); - Optional connectTimeoutMillis = getInputNativeConnectionTimeout(conf); - Optional readTimeoutMillis = getInputNativeReadConnectionTimeout(conf); - Optional receiveBufferSize = getInputNativeReceiveBufferSize(conf); - Optional sendBufferSize = getInputNativeSendBufferSize(conf); - Optional soLinger = getInputNativeSolinger(conf); - Optional tcpNoDelay = getInputNativeTcpNodelay(conf); - Optional reuseAddress = getInputNativeReuseAddress(conf); - Optional keepAlive = getInputNativeKeepAlive(conf); - - if (connectTimeoutMillis.isPresent()) - socketOptions.setConnectTimeoutMillis(connectTimeoutMillis.get()); - if (readTimeoutMillis.isPresent()) - socketOptions.setReadTimeoutMillis(readTimeoutMillis.get()); - if (receiveBufferSize.isPresent()) - socketOptions.setReceiveBufferSize(receiveBufferSize.get()); - if (sendBufferSize.isPresent()) - socketOptions.setSendBufferSize(sendBufferSize.get()); - if (soLinger.isPresent()) - socketOptions.setSoLinger(soLinger.get()); - if (tcpNoDelay.isPresent()) - socketOptions.setTcpNoDelay(tcpNoDelay.get()); - if (reuseAddress.isPresent()) - socketOptions.setReuseAddress(reuseAddress.get()); - if (keepAlive.isPresent()) - socketOptions.setKeepAlive(keepAlive.get()); - - return socketOptions; - } - - private static LoadBalancingPolicy getReadLoadBalancingPolicy(final String[] stickHosts) - { - return new LimitedLocalNodeFirstLocalBalancingPolicy(stickHosts); - } - - private static Optional getDefaultAuthProvider(Configuration conf) - { - Optional username = getStringSetting(USERNAME, conf); - Optional password = getStringSetting(PASSWORD, conf); - - if (username.isPresent() && password.isPresent()) - { - return Optional.of(new PlainTextAuthProvider(username.get(), password.get())); - } - else - { - return Optional.absent(); - } - } - - private static Optional getAuthProvider(Configuration conf) - { - Optional authProvider = getInputNativeAuthProvider(conf); - if (!authProvider.isPresent()) - return getDefaultAuthProvider(conf); - - return Optional.of(getClientAuthProvider(authProvider.get(), conf)); - } - - public static Optional getSSLOptions(Configuration conf) - { - Optional truststorePath = getInputNativeSSLTruststorePath(conf); - - if (truststorePath.isPresent()) - { - Optional keystorePath = getInputNativeSSLKeystorePath(conf); - Optional truststorePassword = getInputNativeSSLTruststorePassword(conf); - Optional keystorePassword = getInputNativeSSLKeystorePassword(conf); - Optional cipherSuites = getInputNativeSSLCipherSuites(conf); - - SSLContext context; - try - { - context = getSSLContext(truststorePath, truststorePassword, keystorePath, keystorePassword); - } - catch (UnrecoverableKeyException | KeyManagementException | - NoSuchAlgorithmException | KeyStoreException | CertificateException | IOException e) - { - throw new RuntimeException(e); - } - String[] css = null; - if (cipherSuites.isPresent()) - css = cipherSuites.get().split(","); - return Optional.of(JdkSSLOptions.builder() - .withSSLContext(context) - .withCipherSuites(css) - .build()); - } - return Optional.absent(); - } - - private static Optional getIntSetting(String parameter, Configuration conf) - { - String setting = conf.get(parameter); - if (setting == null) - return Optional.absent(); - return Optional.of(Integer.valueOf(setting)); - } - - private static Optional getBooleanSetting(String parameter, Configuration conf) - { - String setting = conf.get(parameter); - if (setting == null) - return Optional.absent(); - return Optional.of(Boolean.valueOf(setting)); - } - - private static Optional getStringSetting(String parameter, Configuration conf) - { - String setting = conf.get(parameter); - if (setting == null) - return Optional.absent(); - return Optional.of(setting); - } - - private static AuthProvider getClientAuthProvider(String factoryClassName, Configuration conf) - { - try - { - Class c = Class.forName(factoryClassName); - if (PlainTextAuthProvider.class.equals(c)) - { - String username = getStringSetting(USERNAME, conf).or(""); - String password = getStringSetting(PASSWORD, conf).or(""); - return (AuthProvider) c.getConstructor(String.class, String.class) - .newInstance(username, password); - } - else - { - return (AuthProvider) c.newInstance(); - } - } - catch (Exception e) - { - throw new RuntimeException("Failed to instantiate auth provider:" + factoryClassName, e); - } - } - - private static SSLContext getSSLContext(Optional truststorePath, - Optional truststorePassword, - Optional keystorePath, - Optional keystorePassword) - throws NoSuchAlgorithmException, - KeyStoreException, - CertificateException, - IOException, - UnrecoverableKeyException, - KeyManagementException - { - SSLContext ctx = SSLContext.getInstance("SSL"); - - TrustManagerFactory tmf = null; - if (truststorePath.isPresent()) - { - try (InputStream tsf = Files.newInputStream(File.getPath(truststorePath.get()))) - { - KeyStore ts = KeyStore.getInstance("JKS"); - ts.load(tsf, truststorePassword.isPresent() ? truststorePassword.get().toCharArray() : null); - tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); - tmf.init(ts); - } - } - - KeyManagerFactory kmf = null; - if (keystorePath.isPresent()) - { - try (InputStream ksf = Files.newInputStream(File.getPath(keystorePath.get()))) - { - KeyStore ks = KeyStore.getInstance("JKS"); - ks.load(ksf, keystorePassword.isPresent() ? keystorePassword.get().toCharArray() : null); - kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); - kmf.init(ks, keystorePassword.isPresent() ? keystorePassword.get().toCharArray() : null); - } - } - - ctx.init(kmf != null ? kmf.getKeyManagers() : null, - tmf != null ? tmf.getTrustManagers() : null, - new SecureRandom()); - return ctx; - } -} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java deleted file mode 100644 index 07556846b9..0000000000 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java +++ /dev/null @@ -1,502 +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.cql3; - -import java.io.IOException; -import java.net.InetAddress; -import java.util.*; -import java.util.concurrent.*; - -import com.datastax.driver.core.Cluster; -import com.datastax.driver.core.Host; -import com.datastax.driver.core.Metadata; -import com.datastax.driver.core.ResultSet; -import com.datastax.driver.core.Row; -import com.datastax.driver.core.Session; -import com.datastax.driver.core.SimpleStatement; -import com.datastax.driver.core.Statement; -import com.datastax.driver.core.TokenRange; - -import com.google.common.base.Preconditions; -import com.google.common.collect.Iterables; -import com.google.common.collect.Maps; -import com.google.common.collect.Multimap; - -import com.datastax.driver.core.exceptions.InvalidQueryException; -import org.apache.cassandra.schema.SchemaConstants; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.mapred.InputSplit; -import org.apache.hadoop.mapred.JobConf; -import org.apache.hadoop.mapred.RecordReader; -import org.apache.hadoop.mapred.Reporter; -import org.apache.hadoop.mapreduce.JobContext; -import org.apache.hadoop.mapreduce.TaskAttemptContext; -import org.apache.hadoop.mapreduce.TaskAttemptID; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.apache.cassandra.db.SystemKeyspace; -import org.apache.cassandra.dht.*; -import org.apache.cassandra.hadoop.*; -import org.apache.cassandra.utils.*; - -import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; -import static org.apache.cassandra.utils.Clock.Global.nanoTime; - -/** - * Hadoop InputFormat allowing map/reduce against Cassandra rows within one ColumnFamily. - * - * At minimum, you need to set the KS and CF in your Hadoop job Configuration. - * The ConfigHelper class is provided to make this - * simple: - * ConfigHelper.setInputColumnFamily - * - * You can also configure the number of rows per InputSplit with - * 1: ConfigHelper.setInputSplitSize. The default split size is 64k rows. - * or - * 2: ConfigHelper.setInputSplitSizeInMb. InputSplit size in MB with new, more precise method - * If no value is provided for InputSplitSizeInMb, we default to using InputSplitSize. - * - * CQLConfigHelper.setInputCQLPageRowSize. The default page row size is 1000. You - * should set it to "as big as possible, but no bigger." It set the LIMIT for the CQL - * query, so you need set it big enough to minimize the network overhead, and also - * not too big to avoid out of memory issue. - * - * other native protocol connection parameters in CqlConfigHelper - */ -public class CqlInputFormat extends org.apache.hadoop.mapreduce.InputFormat implements org.apache.hadoop.mapred.InputFormat -{ - public static final String MAPRED_TASK_ID = "mapred.task.id"; - private static final Logger logger = LoggerFactory.getLogger(CqlInputFormat.class); - private String keyspace; - private String cfName; - private IPartitioner partitioner; - - public RecordReader getRecordReader(InputSplit split, JobConf jobConf, final Reporter reporter) - throws IOException - { - TaskAttemptContext tac = HadoopCompat.newMapContext( - jobConf, - TaskAttemptID.forName(jobConf.get(MAPRED_TASK_ID)), - null, - null, - null, - new ReporterWrapper(reporter), - null); - - - CqlRecordReader recordReader = new CqlRecordReader(); - recordReader.initialize((org.apache.hadoop.mapreduce.InputSplit)split, tac); - return recordReader; - } - - @Override - public org.apache.hadoop.mapreduce.RecordReader createRecordReader( - org.apache.hadoop.mapreduce.InputSplit arg0, TaskAttemptContext arg1) throws IOException, - InterruptedException - { - return new CqlRecordReader(); - } - - protected void validateConfiguration(Configuration conf) - { - if (ConfigHelper.getInputKeyspace(conf) == null || ConfigHelper.getInputColumnFamily(conf) == null) - { - throw new UnsupportedOperationException("you must set the keyspace and table with setInputColumnFamily()"); - } - if (ConfigHelper.getInputInitialAddress(conf) == null) - throw new UnsupportedOperationException("You must set the initial output address to a Cassandra node with setInputInitialAddress"); - if (ConfigHelper.getInputPartitioner(conf) == null) - throw new UnsupportedOperationException("You must set the Cassandra partitioner class with setInputPartitioner"); - } - - public List getSplits(JobContext context) throws IOException - { - Configuration conf = HadoopCompat.getConfiguration(context); - - validateConfiguration(conf); - - keyspace = ConfigHelper.getInputKeyspace(conf); - cfName = ConfigHelper.getInputColumnFamily(conf); - partitioner = ConfigHelper.getInputPartitioner(conf); - logger.trace("partitioner is {}", partitioner); - - // canonical ranges, split into pieces, fetching the splits in parallel - ExecutorService executor = executorFactory().pooled("HadoopInput", 128); - List splits = new ArrayList<>(); - - String[] inputInitialAddress = ConfigHelper.getInputInitialAddress(conf).split(","); - try (Cluster cluster = CqlConfigHelper.getInputCluster(inputInitialAddress, conf); - Session session = cluster.connect()) - { - List splitfutures = new ArrayList<>(); - //TODO if the job range is defined and does perfectly match tokens, then the logic will be unable to get estimates since they are pre-computed - // tokens: [0, 10, 20] - // job range: [0, 10) - able to get estimate - // job range: [5, 15) - unable to get estimate - Pair jobKeyRange = ConfigHelper.getInputKeyRange(conf); - Range jobRange = null; - if (jobKeyRange != null) - { - jobRange = new Range<>(partitioner.getTokenFactory().fromString(jobKeyRange.left), - partitioner.getTokenFactory().fromString(jobKeyRange.right)); - } - - Metadata metadata = cluster.getMetadata(); - - // canonical ranges and nodes holding replicas - Map> masterRangeNodes = getRangeMap(keyspace, metadata, getTargetDC(metadata, inputInitialAddress)); - for (TokenRange range : masterRangeNodes.keySet()) - { - if (jobRange == null) - { - for (TokenRange unwrapped : range.unwrap()) - { - // for each tokenRange, pick a live owner and ask it for the byte-sized splits - SplitFuture task = new SplitFuture(new SplitCallable(unwrapped, masterRangeNodes.get(range), conf, session)); - executor.submit(task); - splitfutures.add(task); - } - } - else - { - TokenRange jobTokenRange = rangeToTokenRange(metadata, jobRange); - if (range.intersects(jobTokenRange)) - { - for (TokenRange intersection: range.intersectWith(jobTokenRange)) - { - for (TokenRange unwrapped : intersection.unwrap()) - { - // for each tokenRange, pick a live owner and ask it for the byte-sized splits - SplitFuture task = new SplitFuture(new SplitCallable(unwrapped, masterRangeNodes.get(range), conf, session)); - executor.submit(task); - splitfutures.add(task); - } - } - } - } - } - - // wait until we have all the results back - List failedTasks = new ArrayList<>(); - int maxSplits = 0; - long expectedPartionsForFailedRanges = 0; - for (SplitFuture task : splitfutures) - { - try - { - List tokenRangeSplits = task.get(); - if (tokenRangeSplits.size() > maxSplits) - { - maxSplits = tokenRangeSplits.size(); - expectedPartionsForFailedRanges = tokenRangeSplits.get(0).getLength(); - } - splits.addAll(tokenRangeSplits); - } - catch (Exception e) - { - failedTasks.add(task); - } - } - // The estimate is only stored on a single host, if that host is down then can not get the estimate - // its more than likely that a single host could be "too large" for one split but there is no way of - // knowning! - // This logic attempts to guess the estimate from all the successful ranges - if (!failedTasks.isEmpty()) - { - // if every split failed this will be 0 - if (maxSplits == 0) - throwAllSplitsFailed(failedTasks); - for (SplitFuture task : failedTasks) - { - try - { - // the task failed, so this should throw - task.get(); - } - catch (Exception cause) - { - logger.warn("Unable to get estimate for {}, the host {} had a exception; falling back to default estimate", task.splitCallable.tokenRange, task.splitCallable.hosts.get(0), cause); - } - } - for (SplitFuture task : failedTasks) - splits.addAll(toSplit(task.splitCallable.hosts, splitTokenRange(task.splitCallable.tokenRange, maxSplits, expectedPartionsForFailedRanges))); - } - } - finally - { - executor.shutdownNow(); - } - - assert splits.size() > 0; - Collections.shuffle(splits, new Random(nanoTime())); - return splits; - } - - private static IllegalStateException throwAllSplitsFailed(List failedTasks) - { - IllegalStateException exception = new IllegalStateException("No successful tasks found"); - for (SplitFuture task : failedTasks) - { - try - { - // the task failed, so this should throw - task.get(); - } - catch (Exception cause) - { - exception.addSuppressed(cause); - } - } - throw exception; - } - - private static String getTargetDC(Metadata metadata, String[] inputInitialAddress) - { - BiMultiValMap addressToDc = new BiMultiValMap<>(); - Multimap dcToAddresses = addressToDc.inverse(); - - // only way to match is off the broadcast addresses, so for all hosts do a existence check - Set addresses = new HashSet<>(inputInitialAddress.length); - for (String inputAddress : inputInitialAddress) - addresses.addAll(parseAddress(inputAddress)); - - for (Host host : metadata.getAllHosts()) - { - InetAddress address = host.getBroadcastAddress(); - if (addresses.contains(address)) - addressToDc.put(address, host.getDatacenter()); - } - - switch (dcToAddresses.keySet().size()) - { - case 1: - return Iterables.getOnlyElement(dcToAddresses.keySet()); - case 0: - throw new IllegalStateException("Input addresses could not be used to find DC; non match client metadata"); - default: - // Mutliple DCs found, attempt to pick the first based off address list. This is to mimic the 2.1 - // behavior which would connect in order and the first node successfully able to connect to was the - // local DC to use; since client abstracts this, we rely on existence as a proxy for connect. - for (String inputAddress : inputInitialAddress) - { - for (InetAddress add : parseAddress(inputAddress)) - { - String dc = addressToDc.get(add); - // possible the address isn't in the cluster and the client dropped, so ignore null - if (dc != null) - return dc; - } - } - // some how we were able to connect to the cluster, find multiple DCs using matching, and yet couldn't - // match again... - throw new AssertionError("Unable to infer datacenter from initial addresses; multiple datacenters found " - + dcToAddresses.keySet() + ", should only use addresses from one datacenter"); - } - } - - private static List parseAddress(String str) - { - try - { - return Arrays.asList(InetAddress.getAllByName(str)); - } - catch (Exception e) - { - throw new RuntimeException(e); - } - } - - private TokenRange rangeToTokenRange(Metadata metadata, Range range) - { - return metadata.newTokenRange(metadata.newToken(partitioner.getTokenFactory().toString(range.left)), - metadata.newToken(partitioner.getTokenFactory().toString(range.right))); - } - - private Map getSubSplits(String keyspace, String cfName, TokenRange range, Host host, Configuration conf, Session session) - { - int splitSize = ConfigHelper.getInputSplitSize(conf); - int splitSizeMiB = ConfigHelper.getInputSplitSizeInMb(conf); - return describeSplits(keyspace, cfName, range, host, splitSize, splitSizeMiB, session); - } - - private static Map> getRangeMap(String keyspace, Metadata metadata, String targetDC) - { - return CqlClientHelper.getLocalPrimaryRangeForDC(keyspace, metadata, targetDC); - } - - private Map describeSplits(String keyspace, String table, TokenRange tokenRange, Host host, int splitSize, int splitSizeMb, Session session) - { - // In 2.1 the host list was walked in-order (only move to next if IOException) and calls - // org.apache.cassandra.service.StorageService.getSplits(java.lang.String, java.lang.String, org.apache.cassandra.dht.Range, int) - // that call computes totalRowCountEstimate (used to compute #splits) then splits the ring based off those estimates - // - // The main difference is that the estimates in 2.1 were computed based off the data, so replicas could answer the estimates - // In 3.0 we rely on the below CQL query which is local and only computes estimates for the primary range; this - // puts us in a sticky spot to answer, if the node fails what do we do? 3.0 behavior only matches 2.1 IFF all - // nodes are up and healthy - ResultSet resultSet = queryTableEstimates(session, host, keyspace, table, tokenRange); - - Row row = resultSet.one(); - - long meanPartitionSize = 0; - long partitionCount = 0; - int splitCount = 0; - - if (row != null) - { - meanPartitionSize = row.getLong("mean_partition_size"); - partitionCount = row.getLong("partitions_count"); - - splitCount = splitSizeMb > 0 - ? (int)(meanPartitionSize * partitionCount / splitSizeMb / 1024 / 1024) - : (int)(partitionCount / splitSize); - } - - // If we have no data on this split or the size estimate is 0, - // return the full split i.e., do not sub-split - // Assume smallest granularity of partition count available from CASSANDRA-7688 - if (splitCount == 0) - { - Map wrappedTokenRange = new HashMap<>(); - wrappedTokenRange.put(tokenRange, partitionCount == 0 ? 128L : partitionCount); - return wrappedTokenRange; - } - - return splitTokenRange(tokenRange, splitCount, partitionCount / splitCount); - } - - private static ResultSet queryTableEstimates(Session session, Host host, String keyspace, String table, TokenRange tokenRange) - { - try - { - String query = String.format("SELECT mean_partition_size, partitions_count " + - "FROM %s.%s " + - "WHERE keyspace_name = ? AND table_name = ? AND range_type = '%s' AND range_start = ? AND range_end = ?", - SchemaConstants.SYSTEM_KEYSPACE_NAME, - SystemKeyspace.TABLE_ESTIMATES, - SystemKeyspace.TABLE_ESTIMATES_TYPE_LOCAL_PRIMARY); - Statement stmt = new SimpleStatement(query, keyspace, table, tokenRange.getStart().toString(), tokenRange.getEnd().toString()).setHost(host); - return session.execute(stmt); - } - catch (InvalidQueryException e) - { - // if the table doesn't exist, fall back to old table. This is likely to return no records in a multi - // DC setup, but should work fine in a single DC setup. - String query = String.format("SELECT mean_partition_size, partitions_count " + - "FROM %s.%s " + - "WHERE keyspace_name = ? AND table_name = ? AND range_start = ? AND range_end = ?", - SchemaConstants.SYSTEM_KEYSPACE_NAME, - SystemKeyspace.LEGACY_SIZE_ESTIMATES); - - Statement stmt = new SimpleStatement(query, keyspace, table, tokenRange.getStart().toString(), tokenRange.getEnd().toString()).setHost(host); - return session.execute(stmt); - } - } - - private static Map splitTokenRange(TokenRange tokenRange, int splitCount, long partitionCount) - { - List splitRanges = tokenRange.splitEvenly(splitCount); - Map rangesWithLength = Maps.newHashMapWithExpectedSize(splitRanges.size()); - for (TokenRange range : splitRanges) - rangesWithLength.put(range, partitionCount); - - return rangesWithLength; - } - - // Old Hadoop API - public InputSplit[] getSplits(JobConf jobConf, int numSplits) throws IOException - { - TaskAttemptContext tac = HadoopCompat.newTaskAttemptContext(jobConf, new TaskAttemptID()); - List newInputSplits = this.getSplits(tac); - InputSplit[] oldInputSplits = new InputSplit[newInputSplits.size()]; - for (int i = 0; i < newInputSplits.size(); i++) - oldInputSplits[i] = (ColumnFamilySplit)newInputSplits.get(i); - return oldInputSplits; - } - - /** - * Gets a token tokenRange and splits it up according to the suggested - * size into input splits that Hadoop can use. - */ - class SplitCallable implements Callable> - { - - private final TokenRange tokenRange; - private final List hosts; - private final Configuration conf; - private final Session session; - - public SplitCallable(TokenRange tokenRange, List hosts, Configuration conf, Session session) - { - Preconditions.checkArgument(!hosts.isEmpty(), "hosts list requires at least 1 host but was empty"); - this.tokenRange = tokenRange; - this.hosts = hosts; - this.conf = conf; - this.session = session; - } - - public List call() throws Exception - { - Map subSplits = getSubSplits(keyspace, cfName, tokenRange, hosts.get(0), conf, session); - return toSplit(hosts, subSplits); - } - - } - - private static class SplitFuture extends FutureTask> - { - private final SplitCallable splitCallable; - - SplitFuture(SplitCallable splitCallable) - { - super(splitCallable); - this.splitCallable = splitCallable; - } - } - - private List toSplit(List hosts, Map subSplits) - { - // turn the sub-ranges into InputSplits - String[] endpoints = new String[hosts.size()]; - - // hadoop needs hostname, not ip - int endpointIndex = 0; - for (Host endpoint : hosts) - endpoints[endpointIndex++] = endpoint.getAddress().getHostName(); - - boolean partitionerIsOpp = partitioner instanceof OrderPreservingPartitioner || partitioner instanceof ByteOrderedPartitioner; - - ArrayList splits = new ArrayList<>(); - for (Map.Entry subSplitEntry : subSplits.entrySet()) - { - TokenRange subrange = subSplitEntry.getKey(); - ColumnFamilySplit split = - new ColumnFamilySplit( - partitionerIsOpp ? - subrange.getStart().toString().substring(2) : subrange.getStart().toString(), - partitionerIsOpp ? - subrange.getEnd().toString().substring(2) : subrange.getEnd().toString(), - subSplitEntry.getValue(), - endpoints); - - logger.trace("adding {}", split); - splits.add(split); - } - return splits; - } -} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlOutputFormat.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlOutputFormat.java deleted file mode 100644 index cc0a6b1ce8..0000000000 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlOutputFormat.java +++ /dev/null @@ -1,143 +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.cql3; - - -import java.io.IOException; -import java.nio.ByteBuffer; -import java.util.List; -import java.util.Map; - -import org.apache.cassandra.hadoop.*; -import org.apache.hadoop.conf.*; -import org.apache.hadoop.mapreduce.*; - -/** - * The CqlOutputFormat acts as a Hadoop-specific - * OutputFormat that allows reduce tasks to store keys (and corresponding - * bound variable values) as CQL rows (and respective columns) in a given - * table. - * - *

- * As is the case with the {@link org.apache.cassandra.hadoop.ColumnFamilyInputFormat}, - * you need to set the prepared statement in your - * Hadoop job Configuration. The {@link CqlConfigHelper} class, through its - * {@link CqlConfigHelper#setOutputCql} method, is provided to make this - * simple. - * you need to set the Keyspace. The {@link ConfigHelper} class, through its - * {@link ConfigHelper#setOutputColumnFamily} method, is provided to make this - * simple. - *

- * - *

- * For the sake of performance, this class employs a lazy write-back caching - * mechanism, where its record writer prepared statement binded variable values - * created based on the reduce's inputs (in a task-specific map), and periodically - * makes the changes official by sending a execution of prepared statement request - * to Cassandra. - *

- */ -public class CqlOutputFormat extends OutputFormat, List> - implements org.apache.hadoop.mapred.OutputFormat, List> -{ - public static final String BATCH_THRESHOLD = "mapreduce.output.columnfamilyoutputformat.batch.threshold"; - public static final String QUEUE_SIZE = "mapreduce.output.columnfamilyoutputformat.queue.size"; - - /** - * Check for validity of the output-specification for the job. - * - * @param context - * information about the job - */ - public void checkOutputSpecs(JobContext context) - { - checkOutputSpecs(HadoopCompat.getConfiguration(context)); - } - - protected void checkOutputSpecs(Configuration conf) - { - if (ConfigHelper.getOutputKeyspace(conf) == null) - throw new UnsupportedOperationException("You must set the keyspace with setOutputKeyspace()"); - if (ConfigHelper.getOutputPartitioner(conf) == null) - throw new UnsupportedOperationException("You must set the output partitioner to the one used by your Cassandra cluster"); - if (ConfigHelper.getOutputInitialAddress(conf) == null) - throw new UnsupportedOperationException("You must set the initial output address to a Cassandra node"); - } - - /** Fills the deprecated OutputFormat interface for streaming. */ - @Deprecated - public void checkOutputSpecs(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job) throws IOException - { - checkOutputSpecs(job); - } - - /** - * The OutputCommitter for this format does not write any data to the DFS. - * - * @param context - * the task context - * @return an output committer - * @throws IOException - * @throws InterruptedException - */ - public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException - { - return new NullOutputCommitter(); - } - - /** Fills the deprecated OutputFormat interface for streaming. */ - @Deprecated - public CqlRecordWriter getRecordWriter(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job, String name, org.apache.hadoop.util.Progressable progress) throws IOException - { - return new CqlRecordWriter(job, progress); - } - - /** - * Get the {@link RecordWriter} for the given task. - * - * @param context - * the information about the current task. - * @return a {@link RecordWriter} to write the output for the job. - * @throws IOException - */ - public CqlRecordWriter getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException - { - return new CqlRecordWriter(context); - } - - /** - * An {@link OutputCommitter} that does nothing. - */ - private static class NullOutputCommitter extends OutputCommitter - { - public void abortTask(TaskAttemptContext taskContext) { } - - public void cleanupJob(JobContext jobContext) { } - - public void commitTask(TaskAttemptContext taskContext) { } - - public boolean needsTaskCommit(TaskAttemptContext taskContext) - { - return false; - } - - public void setupJob(JobContext jobContext) { } - - public void setupTask(TaskAttemptContext taskContext) { } - } -} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java deleted file mode 100644 index d9aad1952d..0000000000 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java +++ /dev/null @@ -1,784 +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.cql3; - -import java.io.IOException; -import java.math.BigDecimal; -import java.math.BigInteger; -import java.net.InetAddress; -import java.nio.ByteBuffer; -import java.util.*; - -import com.google.common.base.Function; -import com.google.common.base.Joiner; -import com.google.common.base.Splitter; - -import com.datastax.driver.core.TypeCodec; -import org.apache.cassandra.utils.AbstractIterator; -import com.google.common.collect.Iterables; -import com.google.common.collect.Maps; -import org.apache.commons.lang3.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.datastax.driver.core.Cluster; -import com.datastax.driver.core.ColumnDefinitions; -import com.datastax.driver.core.ColumnMetadata; -import com.datastax.driver.core.LocalDate; -import com.datastax.driver.core.Metadata; -import com.datastax.driver.core.ResultSet; -import com.datastax.driver.core.Row; -import com.datastax.driver.core.Session; -import com.datastax.driver.core.TableMetadata; -import com.datastax.driver.core.Token; -import com.datastax.driver.core.TupleValue; -import com.datastax.driver.core.UDTValue; -import com.google.common.reflect.TypeToken; -import org.apache.cassandra.db.marshal.AbstractType; -import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.hadoop.ColumnFamilySplit; -import org.apache.cassandra.hadoop.ConfigHelper; -import org.apache.cassandra.hadoop.HadoopCompat; -import org.apache.cassandra.utils.ByteBufferUtil; -import org.apache.cassandra.utils.Pair; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.mapreduce.InputSplit; -import org.apache.hadoop.mapreduce.RecordReader; -import org.apache.hadoop.mapreduce.TaskAttemptContext; - -/** - *

- * CqlRecordReader reads the rows return from the CQL query - * It uses CQL auto-paging. - *

- *

- * Return a Long as a local CQL row key starts from 0; - *

- * {@code - * Row as C* java driver CQL result set row - * 1) select clause must include partition key columns (to calculate the progress based on the actual CF row processed) - * 2) where clause must include token(partition_key1, ... , partition_keyn) > ? and - * token(partition_key1, ... , partition_keyn) <= ? (in the right order) - * } - */ -public class CqlRecordReader extends RecordReader - implements org.apache.hadoop.mapred.RecordReader, AutoCloseable -{ - private static final Logger logger = LoggerFactory.getLogger(CqlRecordReader.class); - - private ColumnFamilySplit split; - private RowIterator rowIterator; - - private Pair currentRow; - private int totalRowCount; // total number of rows to fetch - private String keyspace; - private String cfName; - private String cqlQuery; - private Cluster cluster; - private Session session; - private IPartitioner partitioner; - private String inputColumns; - private String userDefinedWhereClauses; - - private List partitionKeys = new ArrayList<>(); - - // partition keys -- key aliases - private LinkedHashMap partitionBoundColumns = Maps.newLinkedHashMap(); - protected int nativeProtocolVersion = 1; - - public CqlRecordReader() - { - super(); - } - - @Override - public void initialize(InputSplit split, TaskAttemptContext context) throws IOException - { - this.split = (ColumnFamilySplit) split; - Configuration conf = HadoopCompat.getConfiguration(context); - totalRowCount = (this.split.getLength() < Long.MAX_VALUE) - ? (int) this.split.getLength() - : ConfigHelper.getInputSplitSize(conf); - cfName = ConfigHelper.getInputColumnFamily(conf); - keyspace = ConfigHelper.getInputKeyspace(conf); - partitioner = ConfigHelper.getInputPartitioner(conf); - inputColumns = CqlConfigHelper.getInputcolumns(conf); - userDefinedWhereClauses = CqlConfigHelper.getInputWhereClauses(conf); - - try - { - if (cluster != null) - return; - - // create a Cluster instance - String[] locations = split.getLocations(); - cluster = CqlConfigHelper.getInputCluster(locations, conf); - } - catch (Exception e) - { - throw new RuntimeException(e); - } - - if (cluster != null) - session = cluster.connect(quote(keyspace)); - - if (session == null) - throw new RuntimeException("Can't create connection session"); - - //get negotiated serialization protocol - nativeProtocolVersion = cluster.getConfiguration().getProtocolOptions().getProtocolVersion().toInt(); - - // If the user provides a CQL query then we will use it without validation - // otherwise we will fall back to building a query using the: - // inputColumns - // whereClauses - cqlQuery = CqlConfigHelper.getInputCql(conf); - // validate that the user hasn't tried to give us a custom query along with input columns - // and where clauses - if (StringUtils.isNotEmpty(cqlQuery) && (StringUtils.isNotEmpty(inputColumns) || - StringUtils.isNotEmpty(userDefinedWhereClauses))) - { - throw new AssertionError("Cannot define a custom query with input columns and / or where clauses"); - } - - if (StringUtils.isEmpty(cqlQuery)) - cqlQuery = buildQuery(); - logger.trace("cqlQuery {}", cqlQuery); - - rowIterator = new RowIterator(); - logger.trace("created {}", rowIterator); - } - - public void close() - { - if (session != null) - session.close(); - if (cluster != null) - cluster.close(); - } - - public Long getCurrentKey() - { - return currentRow.left; - } - - public Row getCurrentValue() - { - return currentRow.right; - } - - public float getProgress() - { - if (!rowIterator.hasNext()) - return 1.0F; - - // the progress is likely to be reported slightly off the actual but close enough - float progress = ((float) rowIterator.totalRead / totalRowCount); - return progress > 1.0F ? 1.0F : progress; - } - - public boolean nextKeyValue() throws IOException - { - if (!rowIterator.hasNext()) - { - logger.trace("Finished scanning {} rows (estimate was: {})", rowIterator.totalRead, totalRowCount); - return false; - } - - try - { - currentRow = rowIterator.next(); - } - catch (Exception e) - { - // throw it as IOException, so client can catch it and handle it at client side - IOException ioe = new IOException(e.getMessage()); - ioe.initCause(ioe.getCause()); - throw ioe; - } - return true; - } - - // Because the old Hadoop API wants us to write to the key and value - // and the new asks for them, we need to copy the output of the new API - // to the old. Thus, expect a small performance hit. - // And obviously this wouldn't work for wide rows. But since ColumnFamilyInputFormat - // and ColumnFamilyRecordReader don't support them, it should be fine for now. - public boolean next(Long key, Row value) throws IOException - { - if (nextKeyValue()) - { - ((WrappedRow)value).setRow(getCurrentValue()); - return true; - } - return false; - } - - public long getPos() throws IOException - { - return rowIterator.totalRead; - } - - public Long createKey() - { - return Long.valueOf(0L); - } - - public Row createValue() - { - return new WrappedRow(); - } - - /** - * Return native version protocol of the cluster connection - * @return serialization protocol version. - */ - public int getNativeProtocolVersion() - { - return nativeProtocolVersion; - } - - /** CQL row iterator - * Input cql query - * 1) select clause must include key columns (if we use partition key based row count) - * 2) where clause must include token(partition_key1 ... partition_keyn) > ? and - * token(partition_key1 ... partition_keyn) <= ? - */ - private class RowIterator extends AbstractIterator> - { - private long keyId = 0L; - protected int totalRead = 0; // total number of cf rows read - protected Iterator rows; - private Map previousRowKey = new HashMap(); // previous CF row key - - public RowIterator() - { - AbstractType type = partitioner.getTokenValidator(); - ResultSet rs = session.execute(cqlQuery, type.compose(type.fromString(split.getStartToken())), type.compose(type.fromString(split.getEndToken())) ); - for (ColumnMetadata meta : cluster.getMetadata().getKeyspace(quote(keyspace)).getTable(quote(cfName)).getPartitionKey()) - partitionBoundColumns.put(meta.getName(), Boolean.TRUE); - rows = rs.iterator(); - } - - protected Pair computeNext() - { - if (rows == null || !rows.hasNext()) - return endOfData(); - - Row row = rows.next(); - Map keyColumns = new HashMap(partitionBoundColumns.size()); - for (String column : partitionBoundColumns.keySet()) - keyColumns.put(column, row.getBytesUnsafe(column)); - - // increase total CF row read - if (previousRowKey.isEmpty() && !keyColumns.isEmpty()) - { - previousRowKey = keyColumns; - totalRead++; - } - else - { - for (String column : partitionBoundColumns.keySet()) - { - // this is not correct - but we don't seem to have easy access to better type information here - if (ByteBufferUtil.compareUnsigned(keyColumns.get(column), previousRowKey.get(column)) != 0) - { - previousRowKey = keyColumns; - totalRead++; - break; - } - } - } - keyId ++; - return Pair.create(keyId, row); - } - } - - private static class WrappedRow implements Row - { - private Row row; - - public void setRow(Row row) - { - this.row = row; - } - - @Override - public ColumnDefinitions getColumnDefinitions() - { - return row.getColumnDefinitions(); - } - - @Override - public boolean isNull(int i) - { - return row.isNull(i); - } - - @Override - public boolean isNull(String name) - { - return row.isNull(name); - } - - @Override - public Object getObject(int i) - { - return row.getObject(i); - } - - @Override - public T get(int i, Class aClass) - { - return row.get(i, aClass); - } - - @Override - public T get(int i, TypeToken typeToken) - { - return row.get(i, typeToken); - } - - @Override - public T get(int i, TypeCodec typeCodec) - { - return row.get(i, typeCodec); - } - - @Override - public Object getObject(String s) - { - return row.getObject(s); - } - - @Override - public T get(String s, Class aClass) - { - return row.get(s, aClass); - } - - @Override - public T get(String s, TypeToken typeToken) - { - return row.get(s, typeToken); - } - - @Override - public T get(String s, TypeCodec typeCodec) - { - return row.get(s, typeCodec); - } - - @Override - public boolean getBool(int i) - { - return row.getBool(i); - } - - @Override - public boolean getBool(String name) - { - return row.getBool(name); - } - - @Override - public short getShort(int i) - { - return row.getShort(i); - } - - @Override - public short getShort(String s) - { - return row.getShort(s); - } - - @Override - public byte getByte(int i) - { - return row.getByte(i); - } - - @Override - public byte getByte(String s) - { - return row.getByte(s); - } - - @Override - public int getInt(int i) - { - return row.getInt(i); - } - - @Override - public int getInt(String name) - { - return row.getInt(name); - } - - @Override - public long getLong(int i) - { - return row.getLong(i); - } - - @Override - public long getLong(String name) - { - return row.getLong(name); - } - - @Override - public Date getTimestamp(int i) - { - return row.getTimestamp(i); - } - - @Override - public Date getTimestamp(String s) - { - return row.getTimestamp(s); - } - - @Override - public LocalDate getDate(int i) - { - return row.getDate(i); - } - - @Override - public LocalDate getDate(String s) - { - return row.getDate(s); - } - - @Override - public long getTime(int i) - { - return row.getTime(i); - } - - @Override - public long getTime(String s) - { - return row.getTime(s); - } - - @Override - public float getFloat(int i) - { - return row.getFloat(i); - } - - @Override - public float getFloat(String name) - { - return row.getFloat(name); - } - - @Override - public double getDouble(int i) - { - return row.getDouble(i); - } - - @Override - public double getDouble(String name) - { - return row.getDouble(name); - } - - @Override - public ByteBuffer getBytesUnsafe(int i) - { - return row.getBytesUnsafe(i); - } - - @Override - public ByteBuffer getBytesUnsafe(String name) - { - return row.getBytesUnsafe(name); - } - - @Override - public ByteBuffer getBytes(int i) - { - return row.getBytes(i); - } - - @Override - public ByteBuffer getBytes(String name) - { - return row.getBytes(name); - } - - @Override - public String getString(int i) - { - return row.getString(i); - } - - @Override - public String getString(String name) - { - return row.getString(name); - } - - @Override - public BigInteger getVarint(int i) - { - return row.getVarint(i); - } - - @Override - public BigInteger getVarint(String name) - { - return row.getVarint(name); - } - - @Override - public BigDecimal getDecimal(int i) - { - return row.getDecimal(i); - } - - @Override - public BigDecimal getDecimal(String name) - { - return row.getDecimal(name); - } - - @Override - public UUID getUUID(int i) - { - return row.getUUID(i); - } - - @Override - public UUID getUUID(String name) - { - return row.getUUID(name); - } - - @Override - public InetAddress getInet(int i) - { - return row.getInet(i); - } - - @Override - public InetAddress getInet(String name) - { - return row.getInet(name); - } - - @Override - public List getList(int i, Class elementsClass) - { - return row.getList(i, elementsClass); - } - - @Override - public List getList(int i, TypeToken typeToken) - { - return row.getList(i, typeToken); - } - - @Override - public List getList(String name, Class elementsClass) - { - return row.getList(name, elementsClass); - } - - @Override - public List getList(String s, TypeToken typeToken) - { - return row.getList(s, typeToken); - } - - @Override - public Set getSet(int i, Class elementsClass) - { - return row.getSet(i, elementsClass); - } - - @Override - public Set getSet(int i, TypeToken typeToken) - { - return row.getSet(i, typeToken); - } - - @Override - public Set getSet(String name, Class elementsClass) - { - return row.getSet(name, elementsClass); - } - - @Override - public Set getSet(String s, TypeToken typeToken) - { - return row.getSet(s, typeToken); - } - - @Override - public Map getMap(int i, Class keysClass, Class valuesClass) - { - return row.getMap(i, keysClass, valuesClass); - } - - @Override - public Map getMap(int i, TypeToken typeToken, TypeToken typeToken1) - { - return row.getMap(i, typeToken, typeToken1); - } - - @Override - public Map getMap(String name, Class keysClass, Class valuesClass) - { - return row.getMap(name, keysClass, valuesClass); - } - - @Override - public Map getMap(String s, TypeToken typeToken, TypeToken typeToken1) - { - return row.getMap(s, typeToken, typeToken1); - } - - @Override - public UDTValue getUDTValue(int i) - { - return row.getUDTValue(i); - } - - @Override - public UDTValue getUDTValue(String name) - { - return row.getUDTValue(name); - } - - @Override - public TupleValue getTupleValue(int i) - { - return row.getTupleValue(i); - } - - @Override - public TupleValue getTupleValue(String name) - { - return row.getTupleValue(name); - } - - @Override - public Token getToken(int i) - { - return row.getToken(i); - } - - @Override - public Token getToken(String name) - { - return row.getToken(name); - } - - @Override - public Token getPartitionKeyToken() - { - return row.getPartitionKeyToken(); - } - } - - /** - * Build a query for the reader of the form: - * - * SELECT * FROM ks>cf token(pk1,...pkn)>? AND token(pk1,...pkn)<=? [AND user where clauses] [ALLOW FILTERING] - */ - private String buildQuery() - { - fetchKeys(); - - List columns = getSelectColumns(); - String selectColumnList = columns.size() == 0 ? "*" : makeColumnList(columns); - String partitionKeyList = makeColumnList(partitionKeys); - - return String.format("SELECT %s FROM %s.%s WHERE token(%s)>? AND token(%s)<=?" + getAdditionalWhereClauses(), - selectColumnList, quote(keyspace), quote(cfName), partitionKeyList, partitionKeyList); - } - - private String getAdditionalWhereClauses() - { - String whereClause = ""; - if (StringUtils.isNotEmpty(userDefinedWhereClauses)) - whereClause += " AND " + userDefinedWhereClauses; - if (StringUtils.isNotEmpty(userDefinedWhereClauses)) - whereClause += " ALLOW FILTERING"; - return whereClause; - } - - private List getSelectColumns() - { - List selectColumns = new ArrayList<>(); - - if (StringUtils.isNotEmpty(inputColumns)) - { - // We must select all the partition keys plus any other columns the user wants - selectColumns.addAll(partitionKeys); - for (String column : Splitter.on(',').split(inputColumns)) - { - if (!partitionKeys.contains(column)) - selectColumns.add(column); - } - } - return selectColumns; - } - - private String makeColumnList(Collection columns) - { - return Joiner.on(',').join(Iterables.transform(columns, new Function() - { - public String apply(String column) - { - return quote(column); - } - })); - } - - private void fetchKeys() - { - // get CF meta data - TableMetadata tableMetadata = session.getCluster() - .getMetadata() - .getKeyspace(Metadata.quote(keyspace)) - .getTable(Metadata.quote(cfName)); - if (tableMetadata == null) - { - throw new RuntimeException("No table metadata found for " + keyspace + "." + cfName); - } - //Here we assume that tableMetadata.getPartitionKey() always - //returns the list of columns in order of component_index - for (ColumnMetadata partitionKey : tableMetadata.getPartitionKey()) - { - partitionKeys.add(partitionKey.getName()); - } - } - - private String quote(String identifier) - { - return "\"" + identifier.replaceAll("\"", "\"\"") + "\""; - } -} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java deleted file mode 100644 index 4bce989602..0000000000 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java +++ /dev/null @@ -1,536 +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.cql3; - -import java.io.IOException; -import java.net.InetAddress; -import java.nio.ByteBuffer; -import java.util.*; -import java.util.concurrent.*; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.datastax.driver.core.*; -import com.datastax.driver.core.exceptions.*; - -import org.apache.cassandra.db.marshal.ByteBufferAccessor; -import org.apache.cassandra.db.marshal.CompositeType; -import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.dht.Token; -import org.apache.cassandra.hadoop.ConfigHelper; -import org.apache.cassandra.hadoop.HadoopCompat; -import org.apache.cassandra.utils.FBUtilities; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.mapreduce.RecordWriter; -import org.apache.hadoop.mapreduce.TaskAttemptContext; -import org.apache.hadoop.util.Progressable; - -import static java.util.stream.Collectors.toMap; - -/** - * The CqlRecordWriter maps the output <key, value> - * pairs to a Cassandra table. In particular, it applies the binded variables - * in the value to the prepared statement, which it associates with the key, and in - * turn the responsible endpoint. - * - *

- * Furthermore, this writer groups the cql queries by the endpoint responsible for - * the rows being affected. This allows the cql queries to be executed in parallel, - * directly to a responsible endpoint. - *

- * - * @see CqlOutputFormat - */ -class CqlRecordWriter extends RecordWriter, List> implements - org.apache.hadoop.mapred.RecordWriter, List>, AutoCloseable -{ - private static final Logger logger = LoggerFactory.getLogger(CqlRecordWriter.class); - - // The configuration this writer is associated with. - protected final Configuration conf; - // The number of mutations to buffer per endpoint - protected final int queueSize; - - protected final long batchThreshold; - - protected Progressable progressable; - protected TaskAttemptContext context; - - // The ring cache that describes the token ranges each node in the ring is - // responsible for. This is what allows us to group the mutations by - // the endpoints they should be targeted at. The targeted endpoint - // essentially - // acts as the primary replica for the rows being affected by the mutations. - private final NativeRingCache ringCache; - - // handles for clients for each range running in the threadpool - protected final Map clients; - - // host to prepared statement id mappings - protected final ConcurrentHashMap preparedStatements = new ConcurrentHashMap(); - - protected final String cql; - - protected List partitionKeyColumns; - protected List clusterColumns; - - /** - * Upon construction, obtain the map that this writer will use to collect - * mutations, and the ring cache for the given keyspace. - * - * @param context the task attempt context - * @throws IOException - */ - CqlRecordWriter(TaskAttemptContext context) throws IOException - { - this(HadoopCompat.getConfiguration(context)); - this.context = context; - } - - CqlRecordWriter(Configuration conf, Progressable progressable) - { - this(conf); - this.progressable = progressable; - } - - CqlRecordWriter(Configuration conf) - { - this.conf = conf; - this.queueSize = conf.getInt(CqlOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors()); - batchThreshold = conf.getLong(CqlOutputFormat.BATCH_THRESHOLD, 32); - this.clients = new HashMap<>(); - String keyspace = ConfigHelper.getOutputKeyspace(conf); - - try (Cluster cluster = CqlConfigHelper.getOutputCluster(ConfigHelper.getOutputInitialAddress(conf), conf)) - { - Metadata metadata = cluster.getMetadata(); - ringCache = new NativeRingCache(conf, metadata); - TableMetadata tableMetadata = metadata.getKeyspace(Metadata.quote(keyspace)).getTable(ConfigHelper.getOutputColumnFamily(conf)); - clusterColumns = tableMetadata.getClusteringColumns(); - partitionKeyColumns = tableMetadata.getPartitionKey(); - - String cqlQuery = CqlConfigHelper.getOutputCql(conf).trim(); - if (cqlQuery.toLowerCase(Locale.ENGLISH).startsWith("insert")) - throw new UnsupportedOperationException("INSERT with CqlRecordWriter is not supported, please use UPDATE/DELETE statement"); - cql = appendKeyWhereClauses(cqlQuery); - } - catch (Exception e) - { - throw new RuntimeException(e); - } - } - - /** - * Close this RecordWriter to future operations, but not before - * flushing out the batched mutations. - * - * @param context the context of the task - * @throws IOException - */ - public void close(TaskAttemptContext context) throws IOException, InterruptedException - { - close(); - } - - /** Fills the deprecated RecordWriter interface for streaming. */ - @Deprecated - public void close(org.apache.hadoop.mapred.Reporter reporter) throws IOException - { - close(); - } - - @Override - public void close() throws IOException - { - // close all the clients before throwing anything - IOException clientException = null; - for (RangeClient client : clients.values()) - { - try - { - client.close(); - } - catch (IOException e) - { - clientException = e; - } - } - - if (clientException != null) - throw clientException; - } - - /** - * If the key is to be associated with a valid value, a mutation is created - * for it with the given table and columns. In the event the value - * in the column is missing (i.e., null), then it is marked for - * {@link Deletion}. Similarly, if the entire value for a key is missing - * (i.e., null), then the entire key is marked for {@link Deletion}. - *

- * - * @param keyColumns - * the key to write. - * @param values - * the values to write. - * @throws IOException - */ - @Override - public void write(Map keyColumns, List values) throws IOException - { - TokenRange range = ringCache.getRange(getPartitionKey(keyColumns)); - - // get the client for the given range, or create a new one - final InetAddress address = ringCache.getEndpoints(range).get(0); - RangeClient client = clients.get(address); - if (client == null) - { - // haven't seen keys for this range: create new client - client = new RangeClient(ringCache.getEndpoints(range)); - client.start(); - clients.put(address, client); - } - - // add primary key columns to the bind variables - List allValues = new ArrayList(values); - for (ColumnMetadata column : partitionKeyColumns) - allValues.add(keyColumns.get(column.getName())); - for (ColumnMetadata column : clusterColumns) - allValues.add(keyColumns.get(column.getName())); - - client.put(allValues); - - if (progressable != null) - progressable.progress(); - if (context != null) - HadoopCompat.progress(context); - } - - private static void closeSession(Session session) - { - //Close the session to satisfy to avoid warnings for the resource not being closed - try - { - if (session != null) - session.getCluster().closeAsync(); - } - catch (Throwable t) - { - logger.warn("Error closing connection", t); - } - } - - /** - * A client that runs in a threadpool and connects to the list of endpoints for a particular - * range. Bound variables for keys in that range are sent to this client via a queue. - */ - public class RangeClient extends Thread - { - // The list of endpoints for this range - protected final List endpoints; - protected Cluster cluster = null; - // A bounded queue of incoming mutations for this range - protected final BlockingQueue> queue = new ArrayBlockingQueue>(queueSize); - - protected volatile boolean run = true; - // we want the caller to know if something went wrong, so we record any unrecoverable exception while writing - // so we can throw it on the caller's stack when he calls put() again, or if there are no more put calls, - // when the client is closed. - protected volatile IOException lastException; - - /** - * Constructs an {@link RangeClient} for the given endpoints. - * @param endpoints the possible endpoints to execute the mutations on - */ - public RangeClient(List endpoints) - { - super("client-" + endpoints); - this.endpoints = endpoints; - } - - /** - * enqueues the given value to Cassandra - */ - public void put(List value) throws IOException - { - while (true) - { - if (lastException != null) - throw lastException; - try - { - if (queue.offer(value, 100, TimeUnit.MILLISECONDS)) - break; - } - catch (InterruptedException e) - { - throw new AssertionError(e); - } - } - } - - /** - * Loops collecting cql binded variable values from the queue and sending to Cassandra - */ - @SuppressWarnings("resource") - public void run() - { - Session session = null; - - try - { - outer: - while (run || !queue.isEmpty()) - { - List bindVariables; - try - { - bindVariables = queue.take(); - } - catch (InterruptedException e) - { - // re-check loop condition after interrupt - continue; - } - - ListIterator iter = endpoints.listIterator(); - while (true) - { - // send the mutation to the last-used endpoint. first time through, this will NPE harmlessly. - if (session != null) - { - try - { - int i = 0; - PreparedStatement statement = preparedStatement(session); - while (bindVariables != null) - { - 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(); - } - break; - } - catch (Exception e) - { - closeInternal(); - if (!iter.hasNext()) - { - lastException = new IOException(e); - break outer; - } - } - } - - // attempt to connect to a different endpoint - try - { - InetAddress address = iter.next(); - String host = address.getHostName(); - cluster = CqlConfigHelper.getOutputCluster(host, conf); - closeSession(session); - session = cluster.connect(); - } - catch (Exception e) - { - //If connection died due to Interrupt, just try connecting to the endpoint again. - //There are too many ways for the Thread.interrupted() state to be cleared, so - //we can't rely on that here. Until the java driver gives us a better way of knowing - //that this exception came from an InterruptedException, this is the best solution. - if (canRetryDriverConnection(e)) - { - iter.previous(); - } - closeInternal(); - - // Most exceptions mean something unexpected went wrong to that endpoint, so - // we should try again to another. Other exceptions (auth or invalid request) are fatal. - if ((e instanceof AuthenticationException || e instanceof InvalidQueryException) || !iter.hasNext()) - { - lastException = new IOException(e); - break outer; - } - } - } - } - } - finally - { - closeSession(session); - // close all our connections once we are done. - closeInternal(); - } - } - - /** get prepared statement id from cache, otherwise prepare it from Cassandra server*/ - private PreparedStatement preparedStatement(Session client) - { - PreparedStatement statement = preparedStatements.get(client); - if (statement == null) - { - PreparedStatement result; - try - { - result = client.prepare(cql); - } - catch (NoHostAvailableException e) - { - throw new RuntimeException("failed to prepare cql query " + cql, e); - } - - PreparedStatement previousId = preparedStatements.putIfAbsent(client, result); - statement = previousId == null ? result : previousId; - } - return statement; - } - - public void close() throws IOException - { - // stop the run loop. this will result in closeInternal being called by the time join() finishes. - run = false; - interrupt(); - try - { - this.join(); - } - catch (InterruptedException e) - { - throw new AssertionError(e); - } - - if (lastException != null) - throw lastException; - } - - protected void closeInternal() - { - if (cluster != null) - { - cluster.close(); - } - } - - private boolean canRetryDriverConnection(Exception e) - { - if (e instanceof DriverException && e.getMessage().contains("Connection thread interrupted")) - return true; - if (e instanceof NoHostAvailableException) - { - if (((NoHostAvailableException) e).getErrors().size() == 1) - { - Throwable cause = ((NoHostAvailableException) e).getErrors().values().iterator().next(); - if (cause != null && cause.getCause() instanceof java.nio.channels.ClosedByInterruptException) - { - return true; - } - } - } - return false; - } - } - - private ByteBuffer getPartitionKey(Map keyColumns) - { - ByteBuffer partitionKey; - if (partitionKeyColumns.size() > 1) - { - ByteBuffer[] keys = new ByteBuffer[partitionKeyColumns.size()]; - for (int i = 0; i< keys.length; i++) - keys[i] = keyColumns.get(partitionKeyColumns.get(i).getName()); - - partitionKey = CompositeType.build(ByteBufferAccessor.instance, keys); - } - else - { - partitionKey = keyColumns.get(partitionKeyColumns.get(0).getName()); - } - return partitionKey; - } - - /** - * add where clauses for partition keys and cluster columns - */ - private String appendKeyWhereClauses(String cqlQuery) - { - String keyWhereClause = ""; - - for (ColumnMetadata partitionKey : partitionKeyColumns) - keyWhereClause += String.format("%s = ?", keyWhereClause.isEmpty() ? quote(partitionKey.getName()) : (" AND " + quote(partitionKey.getName()))); - for (ColumnMetadata clusterColumn : clusterColumns) - keyWhereClause += " AND " + quote(clusterColumn.getName()) + " = ?"; - - return cqlQuery + " WHERE " + keyWhereClause; - } - - /** Quoting for working with uppercase */ - private String quote(String identifier) - { - return "\"" + identifier.replaceAll("\"", "\"\"") + "\""; - } - - static class NativeRingCache - { - private final Map> rangeMap; - private final Metadata metadata; - private final IPartitioner partitioner; - - public NativeRingCache(Configuration conf, Metadata metadata) - { - this.partitioner = ConfigHelper.getOutputPartitioner(conf); - this.metadata = metadata; - String keyspace = ConfigHelper.getOutputKeyspace(conf); - this.rangeMap = metadata.getTokenRanges() - .stream() - .collect(toMap(p -> p, p -> metadata.getReplicas('"' + keyspace + '"', p))); - } - - public TokenRange getRange(ByteBuffer key) - { - Token t = partitioner.getToken(key); - com.datastax.driver.core.Token driverToken = metadata.newToken(partitioner.getTokenFactory().toString(t)); - for (TokenRange range : rangeMap.keySet()) - { - if (range.contains(driverToken)) - { - return range; - } - } - - throw new RuntimeException("Invalid token information returned by describe_ring: " + rangeMap); - } - - public List getEndpoints(TokenRange range) - { - Set hostSet = rangeMap.get(range); - List addresses = new ArrayList<>(hostSet.size()); - for (Host host: hostSet) - { - addresses.add(host.getAddress()); - } - return addresses; - } - } -} diff --git a/src/java/org/apache/cassandra/hadoop/cql3/LimitedLocalNodeFirstLocalBalancingPolicy.java b/src/java/org/apache/cassandra/hadoop/cql3/LimitedLocalNodeFirstLocalBalancingPolicy.java deleted file mode 100644 index 59b4ecab11..0000000000 --- a/src/java/org/apache/cassandra/hadoop/cql3/LimitedLocalNodeFirstLocalBalancingPolicy.java +++ /dev/null @@ -1,216 +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.cql3; - -import com.datastax.driver.core.Cluster; -import com.datastax.driver.core.Host; -import com.datastax.driver.core.HostDistance; -import com.datastax.driver.core.Statement; -import com.datastax.driver.core.policies.LoadBalancingPolicy; -import com.google.common.base.Function; -import com.google.common.collect.Iterators; -import com.google.common.collect.Sets; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.net.InetAddress; -import java.net.NetworkInterface; -import java.net.SocketException; -import java.net.UnknownHostException; -import java.util.*; -import java.util.concurrent.CopyOnWriteArraySet; - -/** - * This load balancing policy is intended to be used only for CqlRecordReader when it fetches a particular split. - *

- * It chooses alive hosts only from the set of the given replicas - because the connection is used to load the data from - * the particular split, with a strictly defined list of the replicas, it is pointless to try the other nodes. - * The policy tracks which of the replicas are alive, and when a new query plan is requested, it returns those replicas - * in the following order: - *

    - *
  • the local node
    • - *
  • the collection of the remaining hosts (which is shuffled on each request)
    • - *
- */ -class LimitedLocalNodeFirstLocalBalancingPolicy implements LoadBalancingPolicy -{ - private final static Logger logger = LoggerFactory.getLogger(LimitedLocalNodeFirstLocalBalancingPolicy.class); - - private final static Set localAddresses = Collections.unmodifiableSet(getLocalInetAddresses()); - - private final CopyOnWriteArraySet liveReplicaHosts = new CopyOnWriteArraySet<>(); - - private final Set replicaAddresses = new HashSet<>(); - private final Set allowedDCs = new CopyOnWriteArraySet<>(); - - public LimitedLocalNodeFirstLocalBalancingPolicy(String[] replicas) - { - for (String replica : replicas) - { - try - { - InetAddress[] addresses = InetAddress.getAllByName(replica); - Collections.addAll(replicaAddresses, addresses); - } - catch (UnknownHostException e) - { - logger.warn("Invalid replica host name: {}, skipping it", replica); - } - } - if (logger.isTraceEnabled()) - logger.trace("Created instance with the following replicas: {}", Arrays.asList(replicas)); - } - - @Override - public void init(Cluster cluster, Collection hosts) - { - // first find which DCs the user defined - Set dcs = new HashSet<>(); - for (Host host : hosts) - { - if (replicaAddresses.contains(host.getAddress())) - dcs.add(host.getDatacenter()); - } - // filter to all nodes within the targeted DCs - List replicaHosts = new ArrayList<>(); - for (Host host : hosts) - { - if (dcs.contains(host.getDatacenter())) - replicaHosts.add(host); - } - liveReplicaHosts.addAll(replicaHosts); - allowedDCs.addAll(dcs); - logger.trace("Initialized with replica hosts: {}", replicaHosts); - } - - @Override - public void close() - { - // - } - - @Override - public HostDistance distance(Host host) - { - if (isLocalHost(host)) - { - return HostDistance.LOCAL; - } - else - { - return HostDistance.REMOTE; - } - } - - @Override - public Iterator newQueryPlan(String keyspace, Statement statement) - { - List local = new ArrayList<>(1); - List remote = new ArrayList<>(liveReplicaHosts.size()); - for (Host liveReplicaHost : liveReplicaHosts) - { - if (isLocalHost(liveReplicaHost)) - { - local.add(liveReplicaHost); - } - else - { - remote.add(liveReplicaHost); - } - } - - Collections.shuffle(remote); - - logger.trace("Using the following hosts order for the new query plan: {} | {}", local, remote); - - return Iterators.concat(local.iterator(), remote.iterator()); - } - - @Override - public void onAdd(Host host) - { - if (liveReplicaHosts.contains(host)) - { - liveReplicaHosts.add(host); - logger.trace("Added a new host {}", host); - } - } - - @Override - public void onUp(Host host) - { - if (liveReplicaHosts.contains(host)) - { - liveReplicaHosts.add(host); - logger.trace("The host {} is now up", host); - } - } - - @Override - public void onDown(Host host) - { - if (liveReplicaHosts.remove(host)) - { - logger.trace("The host {} is now down", host); - } - } - - - @Override - public void onRemove(Host host) - { - if (liveReplicaHosts.remove(host)) - { - logger.trace("Removed the host {}", host); - } - } - - public void onSuspected(Host host) - { - // not supported by this load balancing policy - } - - private static boolean isLocalHost(Host host) - { - InetAddress hostAddress = host.getAddress(); - return hostAddress.isLoopbackAddress() || localAddresses.contains(hostAddress); - } - - private static Set getLocalInetAddresses() - { - try - { - return Sets.newHashSet(Iterators.concat( - Iterators.transform( - Iterators.forEnumeration(NetworkInterface.getNetworkInterfaces()), - new Function>() - { - @Override - public Iterator apply(NetworkInterface netIface) - { - return Iterators.forEnumeration(netIface.getInetAddresses()); - } - }))); - } - catch (SocketException e) - { - logger.warn("Could not retrieve local network interfaces.", e); - return Collections.emptySet(); - } - } -} diff --git a/src/java/org/apache/cassandra/hadoop/package-info.java b/src/java/org/apache/cassandra/hadoop/package-info.java deleted file mode 100644 index 835165b738..0000000000 --- a/src/java/org/apache/cassandra/hadoop/package-info.java +++ /dev/null @@ -1,23 +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. - */ - -/** - * This package was deprecated. See CASSANDRA-16984. - */ -@Deprecated -package org.apache.cassandra.hadoop; \ No newline at end of file diff --git a/test/burn/org/apache/cassandra/transport/SimpleClientPerfTest.java b/test/burn/org/apache/cassandra/transport/SimpleClientPerfTest.java index a050245d04..e6046979ff 100644 --- a/test/burn/org/apache/cassandra/transport/SimpleClientPerfTest.java +++ b/test/burn/org/apache/cassandra/transport/SimpleClientPerfTest.java @@ -30,7 +30,7 @@ import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; import com.google.common.util.concurrent.RateLimiter; -import org.apache.commons.math.stat.descriptive.DescriptiveStatistics; +import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics; import org.junit.Assert; import org.junit.Before; import org.junit.Test; diff --git a/test/simulator/main/org/apache/cassandra/simulator/debug/SelfReconcilingRandom.java b/test/simulator/main/org/apache/cassandra/simulator/debug/SelfReconcilingRandom.java index 444ec44344..99df4b987e 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/debug/SelfReconcilingRandom.java +++ b/test/simulator/main/org/apache/cassandra/simulator/debug/SelfReconcilingRandom.java @@ -20,24 +20,29 @@ package org.apache.cassandra.simulator.debug; import java.util.function.Supplier; +import org.agrona.collections.Long2LongHashMap; import org.apache.cassandra.simulator.RandomSource; -import org.hsqldb.lib.IntKeyLongValueHashMap; import static org.apache.cassandra.simulator.SimulatorUtils.failWithOOM; public class SelfReconcilingRandom implements Supplier { - static class Map extends IntKeyLongValueHashMap + static class Map extends Long2LongHashMap { + public Map(long missingValue) + { + super(missingValue); + } + public boolean put(int i, long v) { int size = this.size(); - super.addOrRemove((long)i, (long)v, (Object)null, (Object)null, false); + super.put(i, v); return size != this.size(); } } - final Map map = new Map(); - long[] tmp = new long[1]; + + final Map map = new Map(Long.MIN_VALUE); boolean isNextPrimary = true; static abstract class AbstractVerifying extends RandomSource.Abstract @@ -113,10 +118,11 @@ public class SelfReconcilingRandom implements Supplier { void next(long v) { - if (!map.get(++cur, tmp)) + long value = map.get(++cur); + if (value == Long.MIN_VALUE) throw failWithOOM(); map.remove(cur); - if (tmp[0] != v) + if (value != v) throw failWithOOM(); }