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();
}