You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2015/08/06 14:37:37 UTC
[1/9] cassandra git commit: Back Columns by a BTree, not an array
Repository: cassandra
Updated Branches:
refs/heads/cassandra-3.0 c3ed25b0a -> ace28c928
refs/heads/trunk f512995e0 -> 9cd4f8293
Back Columns by a BTree, not an array
patch by benedict; reviewed by branimir for CASSANDRA-9471
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ace28c92
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ace28c92
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ace28c92
Branch: refs/heads/cassandra-3.0
Commit: ace28c9283358da75538c4e2250f8437efb7168f
Parents: f54580d
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Mon Jul 27 17:28:02 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Aug 6 14:36:07 2015 +0200
----------------------------------------------------------------------
src/java/org/apache/cassandra/db/Columns.java | 266 +++++--------------
.../cassandra/db/view/MaterializedView.java | 2 +-
.../org/apache/cassandra/utils/btree/BTree.java | 13 +-
.../org/apache/cassandra/db/ColumnsTest.java | 244 +++++++++++++++++
4 files changed, 320 insertions(+), 205 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ace28c92/src/java/org/apache/cassandra/db/Columns.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Columns.java b/src/java/org/apache/cassandra/db/Columns.java
index 03d2e14..231b529 100644
--- a/src/java/org/apache/cassandra/db/Columns.java
+++ b/src/java/org/apache/cassandra/db/Columns.java
@@ -23,16 +23,20 @@ import java.util.function.Predicate;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
-import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterators;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.db.marshal.UTF8Type;
-import org.apache.cassandra.db.marshal.MapType;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.SearchIterator;
+import org.apache.cassandra.utils.btree.BTree;
+import org.apache.cassandra.utils.btree.BTreeSearchIterator;
/**
* An immutable and sorted list of (non-PK) columns for a given table.
@@ -43,19 +47,21 @@ import org.apache.cassandra.utils.ByteBufferUtil;
public class Columns implements Iterable<ColumnDefinition>
{
public static final Serializer serializer = new Serializer();
- public static final Columns NONE = new Columns(new ColumnDefinition[0], 0);
+ public static final Columns NONE = new Columns(BTree.empty(), 0);
+ public static final ColumnDefinition FIRST_COMPLEX = new ColumnDefinition("", "", ColumnIdentifier.getInterned(ByteBufferUtil.EMPTY_BYTE_BUFFER, UTF8Type.instance),
+ SetType.getInstance(UTF8Type.instance, true), null, ColumnDefinition.Kind.REGULAR);
- public final ColumnDefinition[] columns;
- public final int complexIdx; // Index of the first complex column
+ private final Object[] columns;
+ private final int complexIdx; // Index of the first complex column
- private Columns(ColumnDefinition[] columns, int complexIdx)
+ private Columns(Object[] columns, int complexIdx)
{
- assert complexIdx <= columns.length;
+ assert complexIdx <= BTree.size(columns);
this.columns = columns;
this.complexIdx = complexIdx;
}
- private Columns(ColumnDefinition[] columns)
+ private Columns(Object[] columns)
{
this(columns, findFirstComplexIdx(columns));
}
@@ -69,30 +75,28 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public static Columns of(ColumnDefinition c)
{
- ColumnDefinition[] columns = new ColumnDefinition[]{ c };
- return new Columns(columns, c.isComplex() ? 0 : 1);
+ return new Columns(BTree.singleton(c), c.isComplex() ? 0 : 1);
}
/**
* Returns a new {@code Columns} object holing the same columns than the provided set.
*
- * @param param s the set from which to create the new {@code Columns}.
- *
+ * @param s the set from which to create the new {@code Columns}.
* @return the newly created {@code Columns} containing the columns from {@code s}.
*/
public static Columns from(Set<ColumnDefinition> s)
{
- ColumnDefinition[] columns = s.toArray(new ColumnDefinition[s.size()]);
- Arrays.sort(columns);
- return new Columns(columns, findFirstComplexIdx(columns));
+ Object[] tree = BTree.<ColumnDefinition>builder(Comparator.naturalOrder()).addAll(s).build();
+ return new Columns(tree, findFirstComplexIdx(tree));
}
- private static int findFirstComplexIdx(ColumnDefinition[] columns)
+ private static int findFirstComplexIdx(Object[] tree)
{
- for (int i = 0; i < columns.length; i++)
- if (columns[i].isComplex())
- return i;
- return columns.length;
+ // have fast path for common no-complex case
+ int size = BTree.size(tree);
+ if (!BTree.isEmpty(tree) && BTree.<ColumnDefinition>findByIndex(tree, size - 1).isSimple())
+ return size;
+ return BTree.ceilIndex(tree, Comparator.naturalOrder(), FIRST_COMPLEX);
}
/**
@@ -102,7 +106,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public boolean isEmpty()
{
- return columns.length == 0;
+ return BTree.isEmpty(columns);
}
/**
@@ -122,7 +126,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public int complexColumnCount()
{
- return columns.length - complexIdx;
+ return BTree.size(columns) - complexIdx;
}
/**
@@ -132,7 +136,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public int columnCount()
{
- return columns.length;
+ return BTree.size(columns);
}
/**
@@ -152,7 +156,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public boolean hasComplex()
{
- return complexIdx < columns.length;
+ return complexIdx < BTree.size(columns);
}
/**
@@ -165,7 +169,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public ColumnDefinition getSimple(int i)
{
- return columns[i];
+ return BTree.findByIndex(columns, i);
}
/**
@@ -178,7 +182,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public ColumnDefinition getComplex(int i)
{
- return columns[complexIdx + i];
+ return BTree.findByIndex(columns, complexIdx + i);
}
/**
@@ -186,19 +190,13 @@ public class Columns implements Iterable<ColumnDefinition>
* the provided column).
*
* @param c the simple column for which to return the index of.
- * @param from the index to start the search from.
*
* @return the index for simple column {@code c} if it is contains in this
- * object (starting from index {@code from}), {@code -1} otherwise.
+ * object
*/
- public int simpleIdx(ColumnDefinition c, int from)
+ public int simpleIdx(ColumnDefinition c)
{
- assert !c.isComplex();
- for (int i = from; i < complexIdx; i++)
- // We know we only use "interned" ColumnIdentifier so == is ok.
- if (columns[i].name == c.name)
- return i;
- return -1;
+ return BTree.findIndex(columns, Comparator.naturalOrder(), c);
}
/**
@@ -206,19 +204,13 @@ public class Columns implements Iterable<ColumnDefinition>
* the provided column).
*
* @param c the complex column for which to return the index of.
- * @param from the index to start the search from.
*
* @return the index for complex column {@code c} if it is contains in this
- * object (starting from index {@code from}), {@code -1} otherwise.
+ * object
*/
- public int complexIdx(ColumnDefinition c, int from)
+ public int complexIdx(ColumnDefinition c)
{
- assert c.isComplex();
- for (int i = complexIdx + from; i < columns.length; i++)
- // We know we only use "interned" ColumnIdentifier so == is ok.
- if (columns[i].name == c.name)
- return i - complexIdx;
- return -1;
+ return BTree.findIndex(columns, Comparator.naturalOrder(), c) - complexIdx;
}
/**
@@ -230,30 +222,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public boolean contains(ColumnDefinition c)
{
- return c.isComplex() ? complexIdx(c, 0) >= 0 : simpleIdx(c, 0) >= 0;
- }
-
- /**
- * Whether or not there is some counter columns within those columns.
- *
- * @return whether or not there is some counter columns within those columns.
- */
- public boolean hasCounters()
- {
- for (int i = 0; i < complexIdx; i++)
- {
- if (columns[i].type.isCounter())
- return true;
- }
-
- for (int i = complexIdx; i < columns.length; i++)
- {
- // We only support counter in maps because that's all we need for now (and we need it for the sake of thrift super columns of counter)
- if (columns[i].type instanceof MapType && (((MapType)columns[i].type).valueComparator().isCounter()))
- return true;
- }
-
- return false;
+ return BTree.findIndex(columns, Comparator.naturalOrder(), c) >= 0;
}
/**
@@ -273,60 +242,13 @@ public class Columns implements Iterable<ColumnDefinition>
if (this == NONE)
return other;
- int i = 0, j = 0;
- int size = 0;
- while (i < columns.length && j < other.columns.length)
- {
- ++size;
- int cmp = columns[i].compareTo(other.columns[j]);
- if (cmp == 0)
- {
- ++i;
- ++j;
- }
- else if (cmp < 0)
- {
- ++i;
- }
- else
- {
- ++j;
- }
- }
-
- // If every element was always counted on both array, we have the same
- // arrays for the first min elements
- if (i == size && j == size)
- {
- // We've exited because of either c1 or c2 (or both). The array that
- // made us stop is thus a subset of the 2nd one, return that array.
- return i == columns.length ? other : this;
- }
+ Object[] tree = BTree.<ColumnDefinition>merge(this.columns, other.columns, Comparator.naturalOrder());
+ if (tree == this.columns)
+ return this;
+ if (tree == other.columns)
+ return other;
- size += i == columns.length ? other.columns.length - j : columns.length - i;
- ColumnDefinition[] result = new ColumnDefinition[size];
- i = 0;
- j = 0;
- for (int k = 0; k < size; k++)
- {
- int cmp = i >= columns.length ? 1
- : (j >= other.columns.length ? -1 : columns[i].compareTo(other.columns[j]));
- if (cmp == 0)
- {
- result[k] = columns[i];
- ++i;
- ++j;
- }
- else if (cmp < 0)
- {
- result[k] = columns[i++];
- }
- else
- {
- result[k] = other.columns[j++];
- }
- }
- return new Columns(result, findFirstComplexIdx(result));
+ return new Columns(tree, findFirstComplexIdx(tree));
}
/**
@@ -341,20 +263,10 @@ public class Columns implements Iterable<ColumnDefinition>
if (other.columns.length > columns.length)
return false;
- int j = 0;
- int cmp = 0;
- for (ColumnDefinition def : other.columns)
- {
- while (j < columns.length && (cmp = columns[j].compareTo(def)) < 0)
- j++;
-
- if (j >= columns.length || cmp > 0)
+ BTreeSearchIterator<ColumnDefinition, ColumnDefinition> iter = BTree.slice(columns, Comparator.naturalOrder(), BTree.Dir.ASC);
+ for (ColumnDefinition def : BTree.<ColumnDefinition>iterable(other.columns))
+ if (iter.next(def) == null)
return false;
-
- // cmp == 0, we've found the definition. Ce can bump j once more since
- // we know we won't need to compare that element again
- j++;
- }
return true;
}
@@ -365,7 +277,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public Iterator<ColumnDefinition> simpleColumns()
{
- return new ColumnIterator(0, complexIdx);
+ return BTree.iterator(columns, 0, complexIdx - 1, BTree.Dir.ASC);
}
/**
@@ -375,7 +287,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public Iterator<ColumnDefinition> complexColumns()
{
- return new ColumnIterator(complexIdx, columns.length);
+ return BTree.iterator(columns, complexIdx, BTree.size(columns) - 1, BTree.Dir.ASC);
}
/**
@@ -385,7 +297,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public Iterator<ColumnDefinition> iterator()
{
- return Iterators.forArray(columns);
+ return BTree.iterator(columns);
}
/**
@@ -399,23 +311,8 @@ public class Columns implements Iterable<ColumnDefinition>
{
// In wildcard selection, we want to return all columns in alphabetical order,
// irregarding of whether they are complex or not
- return new AbstractIterator<ColumnDefinition>()
- {
- private int regular;
- private int complex = complexIdx;
-
- protected ColumnDefinition computeNext()
- {
- if (complex >= columns.length)
- return regular >= complexIdx ? endOfData() : columns[regular++];
- if (regular >= complexIdx)
- return columns[complex++];
-
- return columns[regular].name.compareTo(columns[complex].name) < 0
- ? columns[regular++]
- : columns[complex++];
- }
- };
+ return Iterators.<ColumnDefinition>mergeSorted(ImmutableList.of(simpleColumns(), complexColumns()),
+ (s, c) -> s.name.compareTo(c.name));
}
/**
@@ -428,15 +325,10 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public Columns without(ColumnDefinition column)
{
- int idx = column.isComplex() ? complexIdx(column, 0) : simpleIdx(column, 0);
- if (idx < 0)
+ if (!contains(column))
return this;
- int realIdx = column.isComplex() ? complexIdx + idx : idx;
-
- ColumnDefinition[] newColumns = new ColumnDefinition[columns.length - 1];
- System.arraycopy(columns, 0, newColumns, 0, realIdx);
- System.arraycopy(columns, realIdx + 1, newColumns, realIdx, newColumns.length - realIdx);
+ Object[] newColumns = BTree.<ColumnDefinition>transformAndFilter(columns, (c) -> c.equals(column) ? null : c);
return new Columns(newColumns);
}
@@ -448,24 +340,8 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public Predicate<ColumnDefinition> inOrderInclusionTester()
{
- return new Predicate<ColumnDefinition>()
- {
- private int i = 0;
-
- public boolean test(ColumnDefinition column)
- {
- while (i < columns.length)
- {
- int cmp = column.compareTo(columns[i]);
- if (cmp < 0)
- return false;
- i++;
- if (cmp == 0)
- return true;
- }
- return false;
- }
- };
+ SearchIterator<ColumnDefinition, ColumnDefinition> iter = BTree.slice(columns, Comparator.naturalOrder(), BTree.Dir.ASC);
+ return column -> iter.next(column) != null;
}
public void digest(MessageDigest digest)
@@ -477,17 +353,19 @@ public class Columns implements Iterable<ColumnDefinition>
@Override
public boolean equals(Object other)
{
+ if (other == this)
+ return true;
if (!(other instanceof Columns))
return false;
Columns that = (Columns)other;
- return this.complexIdx == that.complexIdx && Arrays.equals(this.columns, that.columns);
+ return this.complexIdx == that.complexIdx && BTree.equals(this.columns, that.columns);
}
@Override
public int hashCode()
{
- return Objects.hash(complexIdx, Arrays.hashCode(columns));
+ return Objects.hash(complexIdx, BTree.hashCode(columns));
}
@Override
@@ -503,25 +381,6 @@ public class Columns implements Iterable<ColumnDefinition>
return sb.toString();
}
- private class ColumnIterator extends AbstractIterator<ColumnDefinition>
- {
- private final int to;
- private int idx;
-
- private ColumnIterator(int from, int to)
- {
- this.idx = from;
- this.to = to;
- }
-
- protected ColumnDefinition computeNext()
- {
- if (idx >= to)
- return endOfData();
- return columns[idx++];
- }
- }
-
public static class Serializer
{
public void serialize(Columns columns, DataOutputPlus out) throws IOException
@@ -542,7 +401,8 @@ public class Columns implements Iterable<ColumnDefinition>
public Columns deserialize(DataInputPlus in, CFMetaData metadata) throws IOException
{
int length = (int)in.readVInt();
- ColumnDefinition[] columns = new ColumnDefinition[length];
+ BTree.Builder<ColumnDefinition> builder = BTree.builder(Comparator.naturalOrder());
+ builder.auto(false);
for (int i = 0; i < length; i++)
{
ByteBuffer name = ByteBufferUtil.readWithVIntLength(in);
@@ -556,9 +416,9 @@ public class Columns implements Iterable<ColumnDefinition>
if (column == null)
throw new RuntimeException("Unknown column " + UTF8Type.instance.getString(name) + " during deserialization");
}
- columns[i] = column;
+ builder.add(column);
}
- return new Columns(columns);
+ return new Columns(builder.build());
}
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ace28c92/src/java/org/apache/cassandra/db/view/MaterializedView.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/view/MaterializedView.java b/src/java/org/apache/cassandra/db/view/MaterializedView.java
index 988bfc5..06c4dc2 100644
--- a/src/java/org/apache/cassandra/db/view/MaterializedView.java
+++ b/src/java/org/apache/cassandra/db/view/MaterializedView.java
@@ -666,7 +666,7 @@ public class MaterializedView
viewBuilder.addClusteringColumn(ident, properties.getReversableType(ident, column.type));
}
- for (ColumnDefinition column : baseCf.partitionColumns().regulars.columns)
+ for (ColumnDefinition column : baseCf.partitionColumns().regulars)
{
if (column != nonPkTarget && (includeAll || included.contains(column)))
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ace28c92/src/java/org/apache/cassandra/utils/btree/BTree.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/btree/BTree.java b/src/java/org/apache/cassandra/utils/btree/BTree.java
index 62942b4..353e7a5 100644
--- a/src/java/org/apache/cassandra/utils/btree/BTree.java
+++ b/src/java/org/apache/cassandra/utils/btree/BTree.java
@@ -669,7 +669,18 @@ public class BTree
public static boolean equals(Object[] a, Object[] b)
{
- return Iterators.elementsEqual(iterator(a), iterator(b));
+ return size(a) == size(b) && Iterators.elementsEqual(iterator(a), iterator(b));
+ }
+
+ public static int hashCode(Object[] btree)
+ {
+ // we can't just delegate to Arrays.deepHashCode(),
+ // because two equivalent trees may be represented by differently shaped trees
+ int result = 1;
+ for (Object v : iterable(btree))
+ result = 31 * result + Objects.hashCode(v);
+ return result;
+
}
/**
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ace28c92/test/unit/org/apache/cassandra/db/ColumnsTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/ColumnsTest.java b/test/unit/org/apache/cassandra/db/ColumnsTest.java
new file mode 100644
index 0000000..5447fcc
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/ColumnsTest.java
@@ -0,0 +1,244 @@
+/*
+* 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.db;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Predicate;
+
+import com.google.common.collect.Lists;
+
+import org.junit.AfterClass;
+import org.junit.Test;
+
+import junit.framework.Assert;
+import org.apache.cassandra.MockSchema;
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.db.marshal.SetType;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.btree.BTreeSet;
+
+public class ColumnsTest
+{
+
+ private static CFMetaData cfMetaData = MockSchema.newCFS().metadata;
+
+ @Test
+ public void testContainsWithoutAndMergeTo()
+ {
+ for (RandomColumns randomColumns : random())
+ testContainsWithoutAndMergeTo(randomColumns.columns, randomColumns.definitions);
+ }
+
+ private void testContainsWithoutAndMergeTo(Columns columns, List<ColumnDefinition> definitions)
+ {
+ // pick some arbitrary groupings of columns to remove at-once (to avoid factorial complexity)
+ // whatever is left after each removal, we perform this logic on again, recursively
+ List<List<ColumnDefinition>> removeGroups = shuffleAndGroup(Lists.newArrayList(definitions));
+ for (List<ColumnDefinition> defs : removeGroups)
+ {
+ Columns subset = columns;
+ for (ColumnDefinition def : defs)
+ subset = subset.without(def);
+ Assert.assertEquals(columns.columnCount() - defs.size(), subset.columnCount());
+ List<ColumnDefinition> remainingDefs = Lists.newArrayList(columns);
+ remainingDefs.removeAll(defs);
+
+ // test contents after .without
+ assertContents(subset, remainingDefs);
+
+ // test .contains
+ assertSubset(columns, subset);
+
+ // test .mergeTo
+ Columns otherSubset = columns;
+ for (ColumnDefinition def : remainingDefs)
+ {
+ otherSubset = otherSubset.without(def);
+ assertContents(otherSubset.mergeTo(subset), definitions);
+ }
+
+ testContainsWithoutAndMergeTo(subset, remainingDefs);
+ }
+ }
+
+ private void assertSubset(Columns superset, Columns subset)
+ {
+ Assert.assertTrue(superset.contains(superset));
+ Assert.assertTrue(superset.contains(subset));
+ Assert.assertFalse(subset.contains(superset));
+ }
+
+ private static void assertContents(Columns columns, List<ColumnDefinition> defs)
+ {
+ Assert.assertEquals(defs, Lists.newArrayList(columns));
+ boolean hasSimple = false, hasComplex = false;
+ int firstComplexIdx = 0;
+ int i = 0;
+ Iterator<ColumnDefinition> simple = columns.simpleColumns();
+ Iterator<ColumnDefinition> complex = columns.complexColumns();
+ Iterator<ColumnDefinition> all = columns.iterator();
+ Predicate<ColumnDefinition> predicate = columns.inOrderInclusionTester();
+ for (ColumnDefinition def : defs)
+ {
+ Assert.assertEquals(def, all.next());
+ Assert.assertTrue(columns.contains(def));
+ Assert.assertTrue(predicate.test(def));
+ if (def.isSimple())
+ {
+ hasSimple = true;
+ Assert.assertEquals(i, columns.simpleIdx(def));
+ Assert.assertEquals(def, simple.next());
+ ++firstComplexIdx;
+ }
+ else
+ {
+ Assert.assertFalse(simple.hasNext());
+ hasComplex = true;
+ Assert.assertEquals(i - firstComplexIdx, columns.complexIdx(def));
+ Assert.assertEquals(def, complex.next());
+ }
+ i++;
+ }
+ Assert.assertEquals(defs.isEmpty(), columns.isEmpty());
+ Assert.assertFalse(simple.hasNext());
+ Assert.assertFalse(complex.hasNext());
+ Assert.assertFalse(all.hasNext());
+ Assert.assertEquals(hasSimple, columns.hasSimple());
+ Assert.assertEquals(hasComplex, columns.hasComplex());
+ }
+
+ private static <V> List<List<V>> shuffleAndGroup(List<V> list)
+ {
+ // first shuffle
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ for (int i = 0 ; i < list.size() - 1 ; i++)
+ {
+ int j = random.nextInt(i, list.size());
+ V v = list.get(i);
+ list.set(i, list.get(j));
+ list.set(j, v);
+ }
+
+ // then group
+ List<List<V>> result = new ArrayList<>();
+ for (int i = 0 ; i < list.size() ;)
+ {
+ List<V> group = new ArrayList<>();
+ int maxCount = list.size() - i;
+ int count = maxCount <= 2 ? maxCount : random.nextInt(1, maxCount);
+ for (int j = 0 ; j < count ; j++)
+ group.add(list.get(i + j));
+ i += count;
+ result.add(group);
+ }
+ return result;
+ }
+
+ @AfterClass
+ public static void cleanup()
+ {
+ MockSchema.cleanup();
+ }
+
+ private static class RandomColumns
+ {
+ final Columns columns;
+ final List<ColumnDefinition> definitions;
+
+ private RandomColumns(List<ColumnDefinition> definitions)
+ {
+ this.columns = Columns.from(BTreeSet.of(definitions));
+ this.definitions = definitions;
+ }
+ }
+
+ private static List<RandomColumns> random()
+ {
+ List<RandomColumns> random = new ArrayList<>();
+ for (int i = 1 ; i <= 3 ; i++)
+ {
+ random.add(random(i, i - 1, i - 1, i - 1));
+ random.add(random(i - 1, i, i - 1, i - 1));
+ random.add(random(i - 1, i - 1, i, i - 1));
+ random.add(random(i - 1, i - 1, i - 1, i));
+ }
+ return random;
+ }
+
+ private static RandomColumns random(int pkCount, int clCount, int regularCount, int complexCount)
+ {
+ List<Character> chars = new ArrayList<>();
+ for (char c = 'a' ; c <= 'z' ; c++)
+ chars.add(c);
+
+ List<ColumnDefinition> result = new ArrayList<>();
+ addPartition(select(chars, pkCount), result);
+ addClustering(select(chars, clCount), result);
+ addRegular(select(chars, regularCount), result);
+ addComplex(select(chars, complexCount), result);
+ Collections.sort(result);
+ return new RandomColumns(result);
+ }
+
+ private static List<Character> select(List<Character> chars, int count)
+ {
+ List<Character> result = new ArrayList<>();
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ for (int i = 0 ; i < count ; i++)
+ {
+ int v = random.nextInt(chars.size());
+ result.add(chars.get(v));
+ chars.remove(v);
+ }
+ return result;
+ }
+
+ private static void addPartition(List<Character> chars, List<ColumnDefinition> results)
+ {
+ addSimple(ColumnDefinition.Kind.PARTITION_KEY, chars, results);
+ }
+
+ private static void addClustering(List<Character> chars, List<ColumnDefinition> results)
+ {
+ addSimple(ColumnDefinition.Kind.CLUSTERING, chars, results);
+ }
+
+ private static void addRegular(List<Character> chars, List<ColumnDefinition> results)
+ {
+ addSimple(ColumnDefinition.Kind.REGULAR, chars, results);
+ }
+
+ private static void addSimple(ColumnDefinition.Kind kind, List<Character> chars, List<ColumnDefinition> results)
+ {
+ for (Character c : chars)
+ results.add(new ColumnDefinition(cfMetaData, ByteBufferUtil.bytes(c.toString()), UTF8Type.instance, null, kind));
+ }
+
+ private static void addComplex(List<Character> chars, List<ColumnDefinition> results)
+ {
+ for (Character c : chars)
+ results.add(new ColumnDefinition(cfMetaData, ByteBufferUtil.bytes(c.toString()), SetType.getInstance(UTF8Type.instance, true), null, ColumnDefinition.Kind.REGULAR));
+ }
+}
[5/9] cassandra git commit: Back Columns by a BTree, not an array
Posted by be...@apache.org.
Back Columns by a BTree, not an array
patch by benedict; reviewed by branimir for CASSANDRA-9471
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ace28c92
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ace28c92
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ace28c92
Branch: refs/heads/trunk
Commit: ace28c9283358da75538c4e2250f8437efb7168f
Parents: f54580d
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Mon Jul 27 17:28:02 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Aug 6 14:36:07 2015 +0200
----------------------------------------------------------------------
src/java/org/apache/cassandra/db/Columns.java | 266 +++++--------------
.../cassandra/db/view/MaterializedView.java | 2 +-
.../org/apache/cassandra/utils/btree/BTree.java | 13 +-
.../org/apache/cassandra/db/ColumnsTest.java | 244 +++++++++++++++++
4 files changed, 320 insertions(+), 205 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ace28c92/src/java/org/apache/cassandra/db/Columns.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Columns.java b/src/java/org/apache/cassandra/db/Columns.java
index 03d2e14..231b529 100644
--- a/src/java/org/apache/cassandra/db/Columns.java
+++ b/src/java/org/apache/cassandra/db/Columns.java
@@ -23,16 +23,20 @@ import java.util.function.Predicate;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
-import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterators;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.db.marshal.UTF8Type;
-import org.apache.cassandra.db.marshal.MapType;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.SearchIterator;
+import org.apache.cassandra.utils.btree.BTree;
+import org.apache.cassandra.utils.btree.BTreeSearchIterator;
/**
* An immutable and sorted list of (non-PK) columns for a given table.
@@ -43,19 +47,21 @@ import org.apache.cassandra.utils.ByteBufferUtil;
public class Columns implements Iterable<ColumnDefinition>
{
public static final Serializer serializer = new Serializer();
- public static final Columns NONE = new Columns(new ColumnDefinition[0], 0);
+ public static final Columns NONE = new Columns(BTree.empty(), 0);
+ public static final ColumnDefinition FIRST_COMPLEX = new ColumnDefinition("", "", ColumnIdentifier.getInterned(ByteBufferUtil.EMPTY_BYTE_BUFFER, UTF8Type.instance),
+ SetType.getInstance(UTF8Type.instance, true), null, ColumnDefinition.Kind.REGULAR);
- public final ColumnDefinition[] columns;
- public final int complexIdx; // Index of the first complex column
+ private final Object[] columns;
+ private final int complexIdx; // Index of the first complex column
- private Columns(ColumnDefinition[] columns, int complexIdx)
+ private Columns(Object[] columns, int complexIdx)
{
- assert complexIdx <= columns.length;
+ assert complexIdx <= BTree.size(columns);
this.columns = columns;
this.complexIdx = complexIdx;
}
- private Columns(ColumnDefinition[] columns)
+ private Columns(Object[] columns)
{
this(columns, findFirstComplexIdx(columns));
}
@@ -69,30 +75,28 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public static Columns of(ColumnDefinition c)
{
- ColumnDefinition[] columns = new ColumnDefinition[]{ c };
- return new Columns(columns, c.isComplex() ? 0 : 1);
+ return new Columns(BTree.singleton(c), c.isComplex() ? 0 : 1);
}
/**
* Returns a new {@code Columns} object holing the same columns than the provided set.
*
- * @param param s the set from which to create the new {@code Columns}.
- *
+ * @param s the set from which to create the new {@code Columns}.
* @return the newly created {@code Columns} containing the columns from {@code s}.
*/
public static Columns from(Set<ColumnDefinition> s)
{
- ColumnDefinition[] columns = s.toArray(new ColumnDefinition[s.size()]);
- Arrays.sort(columns);
- return new Columns(columns, findFirstComplexIdx(columns));
+ Object[] tree = BTree.<ColumnDefinition>builder(Comparator.naturalOrder()).addAll(s).build();
+ return new Columns(tree, findFirstComplexIdx(tree));
}
- private static int findFirstComplexIdx(ColumnDefinition[] columns)
+ private static int findFirstComplexIdx(Object[] tree)
{
- for (int i = 0; i < columns.length; i++)
- if (columns[i].isComplex())
- return i;
- return columns.length;
+ // have fast path for common no-complex case
+ int size = BTree.size(tree);
+ if (!BTree.isEmpty(tree) && BTree.<ColumnDefinition>findByIndex(tree, size - 1).isSimple())
+ return size;
+ return BTree.ceilIndex(tree, Comparator.naturalOrder(), FIRST_COMPLEX);
}
/**
@@ -102,7 +106,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public boolean isEmpty()
{
- return columns.length == 0;
+ return BTree.isEmpty(columns);
}
/**
@@ -122,7 +126,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public int complexColumnCount()
{
- return columns.length - complexIdx;
+ return BTree.size(columns) - complexIdx;
}
/**
@@ -132,7 +136,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public int columnCount()
{
- return columns.length;
+ return BTree.size(columns);
}
/**
@@ -152,7 +156,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public boolean hasComplex()
{
- return complexIdx < columns.length;
+ return complexIdx < BTree.size(columns);
}
/**
@@ -165,7 +169,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public ColumnDefinition getSimple(int i)
{
- return columns[i];
+ return BTree.findByIndex(columns, i);
}
/**
@@ -178,7 +182,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public ColumnDefinition getComplex(int i)
{
- return columns[complexIdx + i];
+ return BTree.findByIndex(columns, complexIdx + i);
}
/**
@@ -186,19 +190,13 @@ public class Columns implements Iterable<ColumnDefinition>
* the provided column).
*
* @param c the simple column for which to return the index of.
- * @param from the index to start the search from.
*
* @return the index for simple column {@code c} if it is contains in this
- * object (starting from index {@code from}), {@code -1} otherwise.
+ * object
*/
- public int simpleIdx(ColumnDefinition c, int from)
+ public int simpleIdx(ColumnDefinition c)
{
- assert !c.isComplex();
- for (int i = from; i < complexIdx; i++)
- // We know we only use "interned" ColumnIdentifier so == is ok.
- if (columns[i].name == c.name)
- return i;
- return -1;
+ return BTree.findIndex(columns, Comparator.naturalOrder(), c);
}
/**
@@ -206,19 +204,13 @@ public class Columns implements Iterable<ColumnDefinition>
* the provided column).
*
* @param c the complex column for which to return the index of.
- * @param from the index to start the search from.
*
* @return the index for complex column {@code c} if it is contains in this
- * object (starting from index {@code from}), {@code -1} otherwise.
+ * object
*/
- public int complexIdx(ColumnDefinition c, int from)
+ public int complexIdx(ColumnDefinition c)
{
- assert c.isComplex();
- for (int i = complexIdx + from; i < columns.length; i++)
- // We know we only use "interned" ColumnIdentifier so == is ok.
- if (columns[i].name == c.name)
- return i - complexIdx;
- return -1;
+ return BTree.findIndex(columns, Comparator.naturalOrder(), c) - complexIdx;
}
/**
@@ -230,30 +222,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public boolean contains(ColumnDefinition c)
{
- return c.isComplex() ? complexIdx(c, 0) >= 0 : simpleIdx(c, 0) >= 0;
- }
-
- /**
- * Whether or not there is some counter columns within those columns.
- *
- * @return whether or not there is some counter columns within those columns.
- */
- public boolean hasCounters()
- {
- for (int i = 0; i < complexIdx; i++)
- {
- if (columns[i].type.isCounter())
- return true;
- }
-
- for (int i = complexIdx; i < columns.length; i++)
- {
- // We only support counter in maps because that's all we need for now (and we need it for the sake of thrift super columns of counter)
- if (columns[i].type instanceof MapType && (((MapType)columns[i].type).valueComparator().isCounter()))
- return true;
- }
-
- return false;
+ return BTree.findIndex(columns, Comparator.naturalOrder(), c) >= 0;
}
/**
@@ -273,60 +242,13 @@ public class Columns implements Iterable<ColumnDefinition>
if (this == NONE)
return other;
- int i = 0, j = 0;
- int size = 0;
- while (i < columns.length && j < other.columns.length)
- {
- ++size;
- int cmp = columns[i].compareTo(other.columns[j]);
- if (cmp == 0)
- {
- ++i;
- ++j;
- }
- else if (cmp < 0)
- {
- ++i;
- }
- else
- {
- ++j;
- }
- }
-
- // If every element was always counted on both array, we have the same
- // arrays for the first min elements
- if (i == size && j == size)
- {
- // We've exited because of either c1 or c2 (or both). The array that
- // made us stop is thus a subset of the 2nd one, return that array.
- return i == columns.length ? other : this;
- }
+ Object[] tree = BTree.<ColumnDefinition>merge(this.columns, other.columns, Comparator.naturalOrder());
+ if (tree == this.columns)
+ return this;
+ if (tree == other.columns)
+ return other;
- size += i == columns.length ? other.columns.length - j : columns.length - i;
- ColumnDefinition[] result = new ColumnDefinition[size];
- i = 0;
- j = 0;
- for (int k = 0; k < size; k++)
- {
- int cmp = i >= columns.length ? 1
- : (j >= other.columns.length ? -1 : columns[i].compareTo(other.columns[j]));
- if (cmp == 0)
- {
- result[k] = columns[i];
- ++i;
- ++j;
- }
- else if (cmp < 0)
- {
- result[k] = columns[i++];
- }
- else
- {
- result[k] = other.columns[j++];
- }
- }
- return new Columns(result, findFirstComplexIdx(result));
+ return new Columns(tree, findFirstComplexIdx(tree));
}
/**
@@ -341,20 +263,10 @@ public class Columns implements Iterable<ColumnDefinition>
if (other.columns.length > columns.length)
return false;
- int j = 0;
- int cmp = 0;
- for (ColumnDefinition def : other.columns)
- {
- while (j < columns.length && (cmp = columns[j].compareTo(def)) < 0)
- j++;
-
- if (j >= columns.length || cmp > 0)
+ BTreeSearchIterator<ColumnDefinition, ColumnDefinition> iter = BTree.slice(columns, Comparator.naturalOrder(), BTree.Dir.ASC);
+ for (ColumnDefinition def : BTree.<ColumnDefinition>iterable(other.columns))
+ if (iter.next(def) == null)
return false;
-
- // cmp == 0, we've found the definition. Ce can bump j once more since
- // we know we won't need to compare that element again
- j++;
- }
return true;
}
@@ -365,7 +277,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public Iterator<ColumnDefinition> simpleColumns()
{
- return new ColumnIterator(0, complexIdx);
+ return BTree.iterator(columns, 0, complexIdx - 1, BTree.Dir.ASC);
}
/**
@@ -375,7 +287,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public Iterator<ColumnDefinition> complexColumns()
{
- return new ColumnIterator(complexIdx, columns.length);
+ return BTree.iterator(columns, complexIdx, BTree.size(columns) - 1, BTree.Dir.ASC);
}
/**
@@ -385,7 +297,7 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public Iterator<ColumnDefinition> iterator()
{
- return Iterators.forArray(columns);
+ return BTree.iterator(columns);
}
/**
@@ -399,23 +311,8 @@ public class Columns implements Iterable<ColumnDefinition>
{
// In wildcard selection, we want to return all columns in alphabetical order,
// irregarding of whether they are complex or not
- return new AbstractIterator<ColumnDefinition>()
- {
- private int regular;
- private int complex = complexIdx;
-
- protected ColumnDefinition computeNext()
- {
- if (complex >= columns.length)
- return regular >= complexIdx ? endOfData() : columns[regular++];
- if (regular >= complexIdx)
- return columns[complex++];
-
- return columns[regular].name.compareTo(columns[complex].name) < 0
- ? columns[regular++]
- : columns[complex++];
- }
- };
+ return Iterators.<ColumnDefinition>mergeSorted(ImmutableList.of(simpleColumns(), complexColumns()),
+ (s, c) -> s.name.compareTo(c.name));
}
/**
@@ -428,15 +325,10 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public Columns without(ColumnDefinition column)
{
- int idx = column.isComplex() ? complexIdx(column, 0) : simpleIdx(column, 0);
- if (idx < 0)
+ if (!contains(column))
return this;
- int realIdx = column.isComplex() ? complexIdx + idx : idx;
-
- ColumnDefinition[] newColumns = new ColumnDefinition[columns.length - 1];
- System.arraycopy(columns, 0, newColumns, 0, realIdx);
- System.arraycopy(columns, realIdx + 1, newColumns, realIdx, newColumns.length - realIdx);
+ Object[] newColumns = BTree.<ColumnDefinition>transformAndFilter(columns, (c) -> c.equals(column) ? null : c);
return new Columns(newColumns);
}
@@ -448,24 +340,8 @@ public class Columns implements Iterable<ColumnDefinition>
*/
public Predicate<ColumnDefinition> inOrderInclusionTester()
{
- return new Predicate<ColumnDefinition>()
- {
- private int i = 0;
-
- public boolean test(ColumnDefinition column)
- {
- while (i < columns.length)
- {
- int cmp = column.compareTo(columns[i]);
- if (cmp < 0)
- return false;
- i++;
- if (cmp == 0)
- return true;
- }
- return false;
- }
- };
+ SearchIterator<ColumnDefinition, ColumnDefinition> iter = BTree.slice(columns, Comparator.naturalOrder(), BTree.Dir.ASC);
+ return column -> iter.next(column) != null;
}
public void digest(MessageDigest digest)
@@ -477,17 +353,19 @@ public class Columns implements Iterable<ColumnDefinition>
@Override
public boolean equals(Object other)
{
+ if (other == this)
+ return true;
if (!(other instanceof Columns))
return false;
Columns that = (Columns)other;
- return this.complexIdx == that.complexIdx && Arrays.equals(this.columns, that.columns);
+ return this.complexIdx == that.complexIdx && BTree.equals(this.columns, that.columns);
}
@Override
public int hashCode()
{
- return Objects.hash(complexIdx, Arrays.hashCode(columns));
+ return Objects.hash(complexIdx, BTree.hashCode(columns));
}
@Override
@@ -503,25 +381,6 @@ public class Columns implements Iterable<ColumnDefinition>
return sb.toString();
}
- private class ColumnIterator extends AbstractIterator<ColumnDefinition>
- {
- private final int to;
- private int idx;
-
- private ColumnIterator(int from, int to)
- {
- this.idx = from;
- this.to = to;
- }
-
- protected ColumnDefinition computeNext()
- {
- if (idx >= to)
- return endOfData();
- return columns[idx++];
- }
- }
-
public static class Serializer
{
public void serialize(Columns columns, DataOutputPlus out) throws IOException
@@ -542,7 +401,8 @@ public class Columns implements Iterable<ColumnDefinition>
public Columns deserialize(DataInputPlus in, CFMetaData metadata) throws IOException
{
int length = (int)in.readVInt();
- ColumnDefinition[] columns = new ColumnDefinition[length];
+ BTree.Builder<ColumnDefinition> builder = BTree.builder(Comparator.naturalOrder());
+ builder.auto(false);
for (int i = 0; i < length; i++)
{
ByteBuffer name = ByteBufferUtil.readWithVIntLength(in);
@@ -556,9 +416,9 @@ public class Columns implements Iterable<ColumnDefinition>
if (column == null)
throw new RuntimeException("Unknown column " + UTF8Type.instance.getString(name) + " during deserialization");
}
- columns[i] = column;
+ builder.add(column);
}
- return new Columns(columns);
+ return new Columns(builder.build());
}
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ace28c92/src/java/org/apache/cassandra/db/view/MaterializedView.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/view/MaterializedView.java b/src/java/org/apache/cassandra/db/view/MaterializedView.java
index 988bfc5..06c4dc2 100644
--- a/src/java/org/apache/cassandra/db/view/MaterializedView.java
+++ b/src/java/org/apache/cassandra/db/view/MaterializedView.java
@@ -666,7 +666,7 @@ public class MaterializedView
viewBuilder.addClusteringColumn(ident, properties.getReversableType(ident, column.type));
}
- for (ColumnDefinition column : baseCf.partitionColumns().regulars.columns)
+ for (ColumnDefinition column : baseCf.partitionColumns().regulars)
{
if (column != nonPkTarget && (includeAll || included.contains(column)))
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ace28c92/src/java/org/apache/cassandra/utils/btree/BTree.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/btree/BTree.java b/src/java/org/apache/cassandra/utils/btree/BTree.java
index 62942b4..353e7a5 100644
--- a/src/java/org/apache/cassandra/utils/btree/BTree.java
+++ b/src/java/org/apache/cassandra/utils/btree/BTree.java
@@ -669,7 +669,18 @@ public class BTree
public static boolean equals(Object[] a, Object[] b)
{
- return Iterators.elementsEqual(iterator(a), iterator(b));
+ return size(a) == size(b) && Iterators.elementsEqual(iterator(a), iterator(b));
+ }
+
+ public static int hashCode(Object[] btree)
+ {
+ // we can't just delegate to Arrays.deepHashCode(),
+ // because two equivalent trees may be represented by differently shaped trees
+ int result = 1;
+ for (Object v : iterable(btree))
+ result = 31 * result + Objects.hashCode(v);
+ return result;
+
}
/**
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ace28c92/test/unit/org/apache/cassandra/db/ColumnsTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/ColumnsTest.java b/test/unit/org/apache/cassandra/db/ColumnsTest.java
new file mode 100644
index 0000000..5447fcc
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/ColumnsTest.java
@@ -0,0 +1,244 @@
+/*
+* 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.db;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Predicate;
+
+import com.google.common.collect.Lists;
+
+import org.junit.AfterClass;
+import org.junit.Test;
+
+import junit.framework.Assert;
+import org.apache.cassandra.MockSchema;
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.db.marshal.SetType;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.btree.BTreeSet;
+
+public class ColumnsTest
+{
+
+ private static CFMetaData cfMetaData = MockSchema.newCFS().metadata;
+
+ @Test
+ public void testContainsWithoutAndMergeTo()
+ {
+ for (RandomColumns randomColumns : random())
+ testContainsWithoutAndMergeTo(randomColumns.columns, randomColumns.definitions);
+ }
+
+ private void testContainsWithoutAndMergeTo(Columns columns, List<ColumnDefinition> definitions)
+ {
+ // pick some arbitrary groupings of columns to remove at-once (to avoid factorial complexity)
+ // whatever is left after each removal, we perform this logic on again, recursively
+ List<List<ColumnDefinition>> removeGroups = shuffleAndGroup(Lists.newArrayList(definitions));
+ for (List<ColumnDefinition> defs : removeGroups)
+ {
+ Columns subset = columns;
+ for (ColumnDefinition def : defs)
+ subset = subset.without(def);
+ Assert.assertEquals(columns.columnCount() - defs.size(), subset.columnCount());
+ List<ColumnDefinition> remainingDefs = Lists.newArrayList(columns);
+ remainingDefs.removeAll(defs);
+
+ // test contents after .without
+ assertContents(subset, remainingDefs);
+
+ // test .contains
+ assertSubset(columns, subset);
+
+ // test .mergeTo
+ Columns otherSubset = columns;
+ for (ColumnDefinition def : remainingDefs)
+ {
+ otherSubset = otherSubset.without(def);
+ assertContents(otherSubset.mergeTo(subset), definitions);
+ }
+
+ testContainsWithoutAndMergeTo(subset, remainingDefs);
+ }
+ }
+
+ private void assertSubset(Columns superset, Columns subset)
+ {
+ Assert.assertTrue(superset.contains(superset));
+ Assert.assertTrue(superset.contains(subset));
+ Assert.assertFalse(subset.contains(superset));
+ }
+
+ private static void assertContents(Columns columns, List<ColumnDefinition> defs)
+ {
+ Assert.assertEquals(defs, Lists.newArrayList(columns));
+ boolean hasSimple = false, hasComplex = false;
+ int firstComplexIdx = 0;
+ int i = 0;
+ Iterator<ColumnDefinition> simple = columns.simpleColumns();
+ Iterator<ColumnDefinition> complex = columns.complexColumns();
+ Iterator<ColumnDefinition> all = columns.iterator();
+ Predicate<ColumnDefinition> predicate = columns.inOrderInclusionTester();
+ for (ColumnDefinition def : defs)
+ {
+ Assert.assertEquals(def, all.next());
+ Assert.assertTrue(columns.contains(def));
+ Assert.assertTrue(predicate.test(def));
+ if (def.isSimple())
+ {
+ hasSimple = true;
+ Assert.assertEquals(i, columns.simpleIdx(def));
+ Assert.assertEquals(def, simple.next());
+ ++firstComplexIdx;
+ }
+ else
+ {
+ Assert.assertFalse(simple.hasNext());
+ hasComplex = true;
+ Assert.assertEquals(i - firstComplexIdx, columns.complexIdx(def));
+ Assert.assertEquals(def, complex.next());
+ }
+ i++;
+ }
+ Assert.assertEquals(defs.isEmpty(), columns.isEmpty());
+ Assert.assertFalse(simple.hasNext());
+ Assert.assertFalse(complex.hasNext());
+ Assert.assertFalse(all.hasNext());
+ Assert.assertEquals(hasSimple, columns.hasSimple());
+ Assert.assertEquals(hasComplex, columns.hasComplex());
+ }
+
+ private static <V> List<List<V>> shuffleAndGroup(List<V> list)
+ {
+ // first shuffle
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ for (int i = 0 ; i < list.size() - 1 ; i++)
+ {
+ int j = random.nextInt(i, list.size());
+ V v = list.get(i);
+ list.set(i, list.get(j));
+ list.set(j, v);
+ }
+
+ // then group
+ List<List<V>> result = new ArrayList<>();
+ for (int i = 0 ; i < list.size() ;)
+ {
+ List<V> group = new ArrayList<>();
+ int maxCount = list.size() - i;
+ int count = maxCount <= 2 ? maxCount : random.nextInt(1, maxCount);
+ for (int j = 0 ; j < count ; j++)
+ group.add(list.get(i + j));
+ i += count;
+ result.add(group);
+ }
+ return result;
+ }
+
+ @AfterClass
+ public static void cleanup()
+ {
+ MockSchema.cleanup();
+ }
+
+ private static class RandomColumns
+ {
+ final Columns columns;
+ final List<ColumnDefinition> definitions;
+
+ private RandomColumns(List<ColumnDefinition> definitions)
+ {
+ this.columns = Columns.from(BTreeSet.of(definitions));
+ this.definitions = definitions;
+ }
+ }
+
+ private static List<RandomColumns> random()
+ {
+ List<RandomColumns> random = new ArrayList<>();
+ for (int i = 1 ; i <= 3 ; i++)
+ {
+ random.add(random(i, i - 1, i - 1, i - 1));
+ random.add(random(i - 1, i, i - 1, i - 1));
+ random.add(random(i - 1, i - 1, i, i - 1));
+ random.add(random(i - 1, i - 1, i - 1, i));
+ }
+ return random;
+ }
+
+ private static RandomColumns random(int pkCount, int clCount, int regularCount, int complexCount)
+ {
+ List<Character> chars = new ArrayList<>();
+ for (char c = 'a' ; c <= 'z' ; c++)
+ chars.add(c);
+
+ List<ColumnDefinition> result = new ArrayList<>();
+ addPartition(select(chars, pkCount), result);
+ addClustering(select(chars, clCount), result);
+ addRegular(select(chars, regularCount), result);
+ addComplex(select(chars, complexCount), result);
+ Collections.sort(result);
+ return new RandomColumns(result);
+ }
+
+ private static List<Character> select(List<Character> chars, int count)
+ {
+ List<Character> result = new ArrayList<>();
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ for (int i = 0 ; i < count ; i++)
+ {
+ int v = random.nextInt(chars.size());
+ result.add(chars.get(v));
+ chars.remove(v);
+ }
+ return result;
+ }
+
+ private static void addPartition(List<Character> chars, List<ColumnDefinition> results)
+ {
+ addSimple(ColumnDefinition.Kind.PARTITION_KEY, chars, results);
+ }
+
+ private static void addClustering(List<Character> chars, List<ColumnDefinition> results)
+ {
+ addSimple(ColumnDefinition.Kind.CLUSTERING, chars, results);
+ }
+
+ private static void addRegular(List<Character> chars, List<ColumnDefinition> results)
+ {
+ addSimple(ColumnDefinition.Kind.REGULAR, chars, results);
+ }
+
+ private static void addSimple(ColumnDefinition.Kind kind, List<Character> chars, List<ColumnDefinition> results)
+ {
+ for (Character c : chars)
+ results.add(new ColumnDefinition(cfMetaData, ByteBufferUtil.bytes(c.toString()), UTF8Type.instance, null, kind));
+ }
+
+ private static void addComplex(List<Character> chars, List<ColumnDefinition> results)
+ {
+ for (Character c : chars)
+ results.add(new ColumnDefinition(cfMetaData, ByteBufferUtil.bytes(c.toString()), SetType.getInstance(UTF8Type.instance, true), null, ColumnDefinition.Kind.REGULAR));
+ }
+}
[6/9] cassandra git commit: Fix WaitQueueTest flakiness
Posted by be...@apache.org.
Fix WaitQueueTest flakiness
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ecb2b4b0
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ecb2b4b0
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ecb2b4b0
Branch: refs/heads/trunk
Commit: ecb2b4b0473c019d1132c89887734792c75e0895
Parents: 765ab3f
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Mon Aug 3 16:34:29 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Aug 6 14:36:07 2015 +0200
----------------------------------------------------------------------
test/unit/org/apache/cassandra/Util.java | 5 ++
.../cassandra/concurrent/WaitQueueTest.java | 91 ++++++--------------
2 files changed, 32 insertions(+), 64 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecb2b4b0/test/unit/org/apache/cassandra/Util.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/Util.java b/test/unit/org/apache/cassandra/Util.java
index 254c21c..7efe6f4 100644
--- a/test/unit/org/apache/cassandra/Util.java
+++ b/test/unit/org/apache/cassandra/Util.java
@@ -526,4 +526,9 @@ public class Util
assert p == newP;
}
}
+
+ public static void joinThread(Thread thread) throws InterruptedException
+ {
+ thread.join(10000);
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecb2b4b0/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java b/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java
index 3e7cb7b..8e092c5 100644
--- a/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java
+++ b/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java
@@ -21,10 +21,13 @@ package org.apache.cassandra.concurrent;
*/
+import org.apache.cassandra.Util;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import org.junit.*;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.*;
@@ -38,6 +41,7 @@ public class WaitQueueTest
}
public void testSerial(final WaitQueue queue) throws InterruptedException
{
+ AtomicInteger ready = new AtomicInteger();
Thread[] ts = new Thread[4];
for (int i = 0 ; i < ts.length ; i++)
ts[i] = new Thread(new Runnable()
@@ -46,6 +50,7 @@ public class WaitQueueTest
public void run()
{
WaitQueue.Signal wait = queue.register();
+ ready.incrementAndGet();
try
{
wait.await();
@@ -55,68 +60,28 @@ public class WaitQueueTest
}
}
});
- for (int i = 0 ; i < ts.length ; i++)
- ts[i].start();
- Thread.sleep(100);
- queue.signal();
- queue.signal();
- queue.signal();
- queue.signal();
- for (int i = 0 ; i < ts.length ; i++)
+ for (Thread t : ts)
+ t.start();
+ final ThreadLocalRandom random = ThreadLocalRandom.current();
+ while (ready.get() < ts.length)
+ random.nextLong();
+ for (Thread t : ts)
+ queue.signal();
+ for (Thread t : ts)
{
- ts[i].join(100);
- assertFalse(queue.getClass().getName(), ts[i].isAlive());
+ Util.joinThread(t);
+ assertFalse(queue.getClass().getName(), t.isAlive());
}
}
-
- @Test
- public void testCondition1() throws InterruptedException
- {
- testCondition1(new WaitQueue());
- }
-
- public void testCondition1(final WaitQueue queue) throws InterruptedException
- {
- final AtomicBoolean cond1 = new AtomicBoolean(false);
- final AtomicBoolean fail = new AtomicBoolean(false);
- Thread t1 = new Thread(new Runnable()
- {
- @Override
- public void run()
- {
- try
- {
- Thread.sleep(200);
- } catch (InterruptedException e)
- {
- e.printStackTrace();
- }
- WaitQueue.Signal wait = queue.register();
- if (!cond1.get())
- {
- System.err.println("Condition should have already been met");
- fail.set(true);
- }
- }
- });
- t1.start();
- Thread.sleep(50);
- cond1.set(true);
- Thread.sleep(300);
- queue.signal();
- t1.join(300);
- assertFalse(queue.getClass().getName(), t1.isAlive());
- assertFalse(fail.get());
- }
-
@Test
- public void testCondition2() throws InterruptedException
+ public void testCondition() throws InterruptedException
{
- testCondition2(new WaitQueue());
+ testCondition(new WaitQueue());
}
- public void testCondition2(final WaitQueue queue) throws InterruptedException
+ public void testCondition(final WaitQueue queue) throws InterruptedException
{
+ final AtomicBoolean ready = new AtomicBoolean(false);
final AtomicBoolean condition = new AtomicBoolean(false);
final AtomicBoolean fail = new AtomicBoolean(false);
Thread t = new Thread(new Runnable()
@@ -129,16 +94,12 @@ public class WaitQueueTest
{
System.err.println("");
fail.set(true);
+ ready.set(true);
+ return;
}
- try
- {
- Thread.sleep(200);
- wait.await();
- } catch (InterruptedException e)
- {
- e.printStackTrace();
- }
+ ready.set(true);
+ wait.awaitUninterruptibly();
if (!condition.get())
{
System.err.println("Woke up when condition not met");
@@ -147,10 +108,12 @@ public class WaitQueueTest
}
});
t.start();
- Thread.sleep(50);
+ final ThreadLocalRandom random = ThreadLocalRandom.current();
+ while (!ready.get())
+ random.nextLong();
condition.set(true);
queue.signal();
- t.join(300);
+ Util.joinThread(t);
assertFalse(queue.getClass().getName(), t.isAlive());
assertFalse(fail.get());
}
[7/9] cassandra git commit: ninja-fix an assert in CipherFactory,
and added tests for it
Posted by be...@apache.org.
ninja-fix an assert in CipherFactory, and added tests for it
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/f54580d0
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/f54580d0
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/f54580d0
Branch: refs/heads/cassandra-3.0
Commit: f54580d0715a9189ba1658ad036e7d19cecdc3c8
Parents: ecb2b4b
Author: Jason Brown <ja...@gmail.com>
Authored: Thu Aug 6 05:14:15 2015 -0700
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Aug 6 14:36:07 2015 +0200
----------------------------------------------------------------------
.../org/apache/cassandra/security/CipherFactory.java | 2 +-
.../apache/cassandra/security/CipherFactoryTest.java | 12 ++++++++++++
2 files changed, 13 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f54580d0/src/java/org/apache/cassandra/security/CipherFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/security/CipherFactory.java b/src/java/org/apache/cassandra/security/CipherFactory.java
index 0ff9867..7c1495a 100644
--- a/src/java/org/apache/cassandra/security/CipherFactory.java
+++ b/src/java/org/apache/cassandra/security/CipherFactory.java
@@ -109,7 +109,7 @@ public class CipherFactory
public Cipher getDecryptor(String transformation, String keyAlias, byte[] iv) throws IOException
{
- assert iv != null || iv.length > 0 : "trying to decrypt, but the initialization vector is empty";
+ assert iv != null && iv.length > 0 : "trying to decrypt, but the initialization vector is empty";
return buildCipher(transformation, keyAlias, iv, Cipher.DECRYPT_MODE);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f54580d0/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/security/CipherFactoryTest.java b/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
index 4239973..bb6f7ce 100644
--- a/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
+++ b/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
@@ -84,4 +84,16 @@ public class CipherFactoryTest
Cipher c2 = cipherFactory.buildCipher(encryptionOptions.cipher, EncryptionContextGenerator.KEY_ALIAS_2, nextIV(), Cipher.DECRYPT_MODE);
Assert.assertFalse(c1 == c2);
}
+
+ @Test(expected = AssertionError.class)
+ public void getDecryptor_NullIv() throws IOException
+ {
+ cipherFactory.getDecryptor(encryptionOptions.cipher, encryptionOptions.key_alias, null);
+ }
+
+ @Test(expected = AssertionError.class)
+ public void getDecryptor_EmptyIv() throws IOException
+ {
+ cipherFactory.getDecryptor(encryptionOptions.cipher, encryptionOptions.key_alias, new byte[0]);
+ }
}
[3/9] cassandra git commit: ninja-fix an assert in CipherFactory,
and added tests for it
Posted by be...@apache.org.
ninja-fix an assert in CipherFactory, and added tests for it
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/f54580d0
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/f54580d0
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/f54580d0
Branch: refs/heads/trunk
Commit: f54580d0715a9189ba1658ad036e7d19cecdc3c8
Parents: ecb2b4b
Author: Jason Brown <ja...@gmail.com>
Authored: Thu Aug 6 05:14:15 2015 -0700
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Aug 6 14:36:07 2015 +0200
----------------------------------------------------------------------
.../org/apache/cassandra/security/CipherFactory.java | 2 +-
.../apache/cassandra/security/CipherFactoryTest.java | 12 ++++++++++++
2 files changed, 13 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f54580d0/src/java/org/apache/cassandra/security/CipherFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/security/CipherFactory.java b/src/java/org/apache/cassandra/security/CipherFactory.java
index 0ff9867..7c1495a 100644
--- a/src/java/org/apache/cassandra/security/CipherFactory.java
+++ b/src/java/org/apache/cassandra/security/CipherFactory.java
@@ -109,7 +109,7 @@ public class CipherFactory
public Cipher getDecryptor(String transformation, String keyAlias, byte[] iv) throws IOException
{
- assert iv != null || iv.length > 0 : "trying to decrypt, but the initialization vector is empty";
+ assert iv != null && iv.length > 0 : "trying to decrypt, but the initialization vector is empty";
return buildCipher(transformation, keyAlias, iv, Cipher.DECRYPT_MODE);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f54580d0/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/security/CipherFactoryTest.java b/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
index 4239973..bb6f7ce 100644
--- a/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
+++ b/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
@@ -84,4 +84,16 @@ public class CipherFactoryTest
Cipher c2 = cipherFactory.buildCipher(encryptionOptions.cipher, EncryptionContextGenerator.KEY_ALIAS_2, nextIV(), Cipher.DECRYPT_MODE);
Assert.assertFalse(c1 == c2);
}
+
+ @Test(expected = AssertionError.class)
+ public void getDecryptor_NullIv() throws IOException
+ {
+ cipherFactory.getDecryptor(encryptionOptions.cipher, encryptionOptions.key_alias, null);
+ }
+
+ @Test(expected = AssertionError.class)
+ public void getDecryptor_EmptyIv() throws IOException
+ {
+ cipherFactory.getDecryptor(encryptionOptions.cipher, encryptionOptions.key_alias, new byte[0]);
+ }
}
[9/9] cassandra git commit: Merge branch 'cassandra-3.0' into trunk
Posted by be...@apache.org.
Merge branch 'cassandra-3.0' into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/9cd4f829
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/9cd4f829
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/9cd4f829
Branch: refs/heads/trunk
Commit: 9cd4f8293927f5e67a06c6f3d4e6dd25ab20b35b
Parents: f512995 ace28c9
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Thu Aug 6 14:37:01 2015 +0200
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Aug 6 14:37:01 2015 +0200
----------------------------------------------------------------------
src/java/org/apache/cassandra/db/Columns.java | 266 +++++--------------
.../cassandra/db/view/MaterializedView.java | 2 +-
.../org/apache/cassandra/utils/btree/BTree.java | 13 +-
.../org/apache/cassandra/db/ColumnsTest.java | 244 +++++++++++++++++
4 files changed, 320 insertions(+), 205 deletions(-)
----------------------------------------------------------------------
[8/9] cassandra git commit: Add transparent data encryption core
classes (CASSANDRA-9945)
Posted by be...@apache.org.
Add transparent data encryption core classes (CASSANDRA-9945)
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/765ab3fc
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/765ab3fc
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/765ab3fc
Branch: refs/heads/cassandra-3.0
Commit: 765ab3fc3b9f4e891c257f36f117ddf4232da6be
Parents: c3ed25b
Author: Jason Brown <ja...@gmail.com>
Authored: Thu Jul 30 14:03:17 2015 -0700
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Aug 6 14:36:07 2015 +0200
----------------------------------------------------------------------
.../org/apache/cassandra/config/Config.java | 3 +-
.../cassandra/config/DatabaseDescriptor.java | 17 ++
.../TransparentDataEncryptionOptions.java | 76 ++++++++
.../cassandra/security/CipherFactory.java | 175 +++++++++++++++++++
.../cassandra/security/EncryptionContext.java | 122 +++++++++++++
.../cassandra/security/JKSKeyProvider.java | 90 ++++++++++
.../apache/cassandra/security/KeyProvider.java | 33 ++++
test/conf/cassandra.keystore | Bin 0 -> 1004 bytes
test/conf/cassandra_encryption.yaml | 14 ++
.../cassandra/security/CipherFactoryTest.java | 87 +++++++++
.../security/EncryptionContextGenerator.java | 54 ++++++
.../cassandra/security/JKSKeyProviderTest.java | 52 ++++++
12 files changed, 722 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/config/Config.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index e93d090..f8f34e0 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -169,7 +169,8 @@ public class Config
public int commitlog_segment_size_in_mb = 32;
public ParameterizedClass commitlog_compression;
public int commitlog_max_compression_buffers_in_pool = 3;
-
+ public TransparentDataEncryptionOptions transparent_data_encryption_options = new TransparentDataEncryptionOptions();
+
@Deprecated
public int commitlog_periodic_queue_size = -1;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 02f3c17..e7b9455 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -44,6 +44,7 @@ import org.apache.cassandra.locator.*;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.scheduler.IRequestScheduler;
import org.apache.cassandra.scheduler.NoScheduler;
+import org.apache.cassandra.security.EncryptionContext;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.thrift.ThriftServer;
import org.apache.cassandra.utils.ByteBufferUtil;
@@ -92,6 +93,7 @@ public class DatabaseDescriptor
private static String localDC;
private static Comparator<InetAddress> localComparator;
+ private static EncryptionContext encryptionContext;
public static void forceStaticInitialization() {}
static
@@ -613,6 +615,10 @@ public class DatabaseDescriptor
if (conf.user_defined_function_fail_timeout < conf.user_defined_function_warn_timeout)
throw new ConfigurationException("user_defined_function_warn_timeout must less than user_defined_function_fail_timeout", false);
+
+ // always attempt to load the cipher factory, as we could be in the situation where the user has disabled encryption,
+ // but has existing commitlogs and sstables on disk that are still encrypted (and still need to be read)
+ encryptionContext = new EncryptionContext(config.transparent_data_encryption_options);
}
private static IEndpointSnitch createEndpointSnitch(String snitchClassName) throws ConfigurationException
@@ -1792,4 +1798,15 @@ public class DatabaseDescriptor
{
conf.user_function_timeout_policy = userFunctionTimeoutPolicy;
}
+
+ public static EncryptionContext getEncryptionContext()
+ {
+ return encryptionContext;
+ }
+
+ @VisibleForTesting
+ public static void setEncryptionContext(EncryptionContext ec)
+ {
+ encryptionContext = ec;
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/config/TransparentDataEncryptionOptions.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/TransparentDataEncryptionOptions.java b/src/java/org/apache/cassandra/config/TransparentDataEncryptionOptions.java
new file mode 100644
index 0000000..4ad0305
--- /dev/null
+++ b/src/java/org/apache/cassandra/config/TransparentDataEncryptionOptions.java
@@ -0,0 +1,76 @@
+/*
+ * 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.config;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Objects;
+
+public class TransparentDataEncryptionOptions
+{
+ public boolean enabled = false;
+ public int chunk_length_kb = 64;
+ public String cipher = "AES/CBC/PKCS5Padding";
+ public String key_alias;
+ public int iv_length = 16;
+
+ public ParameterizedClass key_provider;
+
+ public TransparentDataEncryptionOptions()
+ { }
+
+ public TransparentDataEncryptionOptions(boolean enabled)
+ {
+ this.enabled = enabled;
+ }
+
+ public TransparentDataEncryptionOptions(String cipher, String keyAlias, ParameterizedClass keyProvider)
+ {
+ this(true, cipher, keyAlias, keyProvider);
+ }
+
+ public TransparentDataEncryptionOptions(boolean enabled, String cipher, String keyAlias, ParameterizedClass keyProvider)
+ {
+ this.enabled = enabled;
+ this.cipher = cipher;
+ key_alias = keyAlias;
+ key_provider = keyProvider;
+ }
+
+ public String get(String key)
+ {
+ return key_provider.parameters.get(key);
+ }
+
+ @VisibleForTesting
+ public void remove(String key)
+ {
+ key_provider.parameters.remove(key);
+ }
+
+ public boolean equals(Object o)
+ {
+ return o instanceof TransparentDataEncryptionOptions && equals((TransparentDataEncryptionOptions) o);
+ }
+
+ public boolean equals(TransparentDataEncryptionOptions other)
+ {
+ // not sure if this is a great equals() impl....
+ return Objects.equal(cipher, other.cipher) &&
+ Objects.equal(key_alias, other.key_alias);
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/security/CipherFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/security/CipherFactory.java b/src/java/org/apache/cassandra/security/CipherFactory.java
new file mode 100644
index 0000000..0ff9867
--- /dev/null
+++ b/src/java/org/apache/cassandra/security/CipherFactory.java
@@ -0,0 +1,175 @@
+/*
+ * 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.security;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.security.InvalidAlgorithmParameterException;
+import java.security.InvalidKeyException;
+import java.security.Key;
+import java.security.NoSuchAlgorithmException;
+import java.security.SecureRandom;
+import java.util.Arrays;
+import java.util.concurrent.ExecutionException;
+import javax.crypto.Cipher;
+import javax.crypto.NoSuchPaddingException;
+import javax.crypto.spec.IvParameterSpec;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+
+/**
+ * A factory for loading encryption keys from {@link KeyProvider} instances.
+ * Maintains a cache of loaded keys to avoid invoking the key provider on every call.
+ */
+public class CipherFactory
+{
+ private final Logger logger = LoggerFactory.getLogger(CipherFactory.class);
+
+ /**
+ * Keep around thread local instances of Cipher as they are quite expensive to instantiate (@code Cipher#getInstance).
+ * Bonus points if you can avoid calling (@code Cipher#init); hence, the point of the supporting struct
+ * for caching Cipher instances.
+ */
+ private static final ThreadLocal<CachedCipher> cipherThreadLocal = new ThreadLocal<>();
+
+ private final SecureRandom secureRandom;
+ private final LoadingCache<String, Key> cache;
+ private final int ivLength;
+ private final KeyProvider keyProvider;
+
+ public CipherFactory(TransparentDataEncryptionOptions options)
+ {
+ logger.info("initializing CipherFactory");
+ ivLength = options.iv_length;
+
+ try
+ {
+ secureRandom = SecureRandom.getInstance("SHA1PRNG");
+ Class<KeyProvider> keyProviderClass = (Class<KeyProvider>)Class.forName(options.key_provider.class_name);
+ Constructor ctor = keyProviderClass.getConstructor(TransparentDataEncryptionOptions.class);
+ keyProvider = (KeyProvider)ctor.newInstance(options);
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException("couldn't load cipher factory", e);
+ }
+
+ cache = CacheBuilder.newBuilder() // by default cache is unbounded
+ .maximumSize(64) // a value large enough that we should never even get close (so nothing gets evicted)
+ .concurrencyLevel(Runtime.getRuntime().availableProcessors())
+ .removalListener(new RemovalListener<String, Key>()
+ {
+ public void onRemoval(RemovalNotification<String, Key> notice)
+ {
+ // maybe reload the key? (to avoid the reload being on the user's dime)
+ logger.info("key {} removed from cipher key cache", notice.getKey());
+ }
+ })
+ .build(new CacheLoader<String, Key>()
+ {
+ @Override
+ public Key load(String alias) throws Exception
+ {
+ logger.info("loading secret key for alias {}", alias);
+ return keyProvider.getSecretKey(alias);
+ }
+ });
+ }
+
+ public Cipher getEncryptor(String transformation, String keyAlias) throws IOException
+ {
+ byte[] iv = new byte[ivLength];
+ secureRandom.nextBytes(iv);
+ return buildCipher(transformation, keyAlias, iv, Cipher.ENCRYPT_MODE);
+ }
+
+ public Cipher getDecryptor(String transformation, String keyAlias, byte[] iv) throws IOException
+ {
+ assert iv != null || iv.length > 0 : "trying to decrypt, but the initialization vector is empty";
+ return buildCipher(transformation, keyAlias, iv, Cipher.DECRYPT_MODE);
+ }
+
+ @VisibleForTesting
+ Cipher buildCipher(String transformation, String keyAlias, byte[] iv, int cipherMode) throws IOException
+ {
+ try
+ {
+ CachedCipher cachedCipher = cipherThreadLocal.get();
+ if (cachedCipher != null)
+ {
+ Cipher cipher = cachedCipher.cipher;
+ // rigorous checks to make sure we've absolutely got the correct instance (with correct alg/key/iv/...)
+ if (cachedCipher.mode == cipherMode && cipher.getAlgorithm().equals(transformation)
+ && cachedCipher.keyAlias.equals(keyAlias) && Arrays.equals(cipher.getIV(), iv))
+ return cipher;
+ }
+
+ Key key = retrieveKey(keyAlias);
+ Cipher cipher = Cipher.getInstance(transformation);
+ cipher.init(cipherMode, key, new IvParameterSpec(iv));
+ cipherThreadLocal.set(new CachedCipher(cipherMode, keyAlias, cipher));
+ return cipher;
+ }
+ catch (NoSuchAlgorithmException | NoSuchPaddingException | InvalidAlgorithmParameterException | InvalidKeyException e)
+ {
+ logger.error("could not build cipher", e);
+ throw new IOException("cannot load cipher", e);
+ }
+ }
+
+ private Key retrieveKey(String keyAlias) throws IOException
+ {
+ try
+ {
+ return cache.get(keyAlias);
+ }
+ catch (ExecutionException e)
+ {
+ if (e.getCause() instanceof IOException)
+ throw (IOException)e.getCause();
+ throw new IOException("failed to load key from cache: " + keyAlias, e);
+ }
+ }
+
+ /**
+ * A simple struct to use with the thread local caching of Cipher as we can't get the mode (encrypt/decrypt) nor
+ * key_alias (or key!) from the Cipher itself to use for comparisons
+ */
+ private static class CachedCipher
+ {
+ public final int mode;
+ public final String keyAlias;
+ public final Cipher cipher;
+
+ private CachedCipher(int mode, String keyAlias, Cipher cipher)
+ {
+ this.mode = mode;
+ this.keyAlias = keyAlias;
+ this.cipher = cipher;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/security/EncryptionContext.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/security/EncryptionContext.java b/src/java/org/apache/cassandra/security/EncryptionContext.java
new file mode 100644
index 0000000..dff6894
--- /dev/null
+++ b/src/java/org/apache/cassandra/security/EncryptionContext.java
@@ -0,0 +1,122 @@
+/*
+ * 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.security;
+
+import java.io.IOException;
+import java.util.Collections;
+import javax.crypto.Cipher;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Objects;
+
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.io.compress.ICompressor;
+import org.apache.cassandra.io.compress.LZ4Compressor;
+
+/**
+ * A (largely) immutable wrapper for the application-wide file-level encryption settings.
+ */
+public class EncryptionContext
+{
+ public static final String ENCRYPTION_CIPHER = "encCipher";
+ public static final String ENCRYPTION_KEY_ALIAS = "encKeyAlias";
+ public static final String ENCRYPTION_IV = "encIV";
+
+ private final TransparentDataEncryptionOptions tdeOptions;
+ private final ICompressor compressor;
+ private final CipherFactory cipherFactory;
+
+ private final int chunkLength;
+
+ public EncryptionContext()
+ {
+ this(new TransparentDataEncryptionOptions());
+ }
+
+ public EncryptionContext(TransparentDataEncryptionOptions tdeOptions)
+ {
+ this(tdeOptions, true);
+ }
+
+ @VisibleForTesting
+ public EncryptionContext(TransparentDataEncryptionOptions tdeOptions, boolean init)
+ {
+ this.tdeOptions = tdeOptions;
+ compressor = LZ4Compressor.create(Collections.<String, String>emptyMap());
+ chunkLength = tdeOptions.chunk_length_kb * 1024;
+
+ // always attempt to load the cipher factory, as we could be in the situation where the user has disabled encryption,
+ // but has existing commitlogs and sstables on disk that are still git addencrypted (and still need to be read)
+ CipherFactory factory = null;
+
+ if (tdeOptions.enabled && init)
+ {
+ try
+ {
+ factory = new CipherFactory(tdeOptions);
+ }
+ catch (Exception e)
+ {
+ throw new ConfigurationException("failed to load key provider for transparent data encryption", e);
+ }
+ }
+
+ cipherFactory = factory;
+ }
+
+ public ICompressor getCompressor()
+ {
+ return compressor;
+ }
+
+ public Cipher getEncryptor() throws IOException
+ {
+ return cipherFactory.getEncryptor(tdeOptions.cipher, tdeOptions.key_alias);
+ }
+
+ public Cipher getDecryptor(byte[] IV) throws IOException
+ {
+ return cipherFactory.getDecryptor(tdeOptions.cipher, tdeOptions.key_alias, IV);
+ }
+
+ public boolean isEnabled()
+ {
+ return tdeOptions.enabled;
+ }
+
+ public int getChunkLength()
+ {
+ return chunkLength;
+ }
+
+ public TransparentDataEncryptionOptions getTransparentDataEncryptionOptions()
+ {
+ return tdeOptions;
+ }
+
+ public boolean equals(Object o)
+ {
+ return o instanceof EncryptionContext && equals((EncryptionContext) o);
+ }
+
+ public boolean equals(EncryptionContext other)
+ {
+ return Objects.equal(tdeOptions, other.tdeOptions) && Objects.equal(compressor, other.compressor);
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/security/JKSKeyProvider.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/security/JKSKeyProvider.java b/src/java/org/apache/cassandra/security/JKSKeyProvider.java
new file mode 100644
index 0000000..8d7f1c6
--- /dev/null
+++ b/src/java/org/apache/cassandra/security/JKSKeyProvider.java
@@ -0,0 +1,90 @@
+/*
+ * 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.security;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.security.Key;
+import java.security.KeyStore;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+import org.apache.cassandra.io.util.FileUtils;
+
+/**
+ * A {@code KeyProvider} that retrieves keys from a java keystore.
+ */
+public class JKSKeyProvider implements KeyProvider
+{
+ private final Logger logger = LoggerFactory.getLogger(JKSKeyProvider.class);
+ static final String PROP_KEYSTORE = "keystore";
+ static final String PROP_KEYSTORE_PW = "keystore_password";
+ static final String PROP_KEYSTORE_TYPE = "store_type";
+ static final String PROP_KEY_PW = "key_password";
+
+ private final KeyStore store;
+ private final boolean isJceks;
+ private final TransparentDataEncryptionOptions options;
+
+ public JKSKeyProvider(TransparentDataEncryptionOptions options)
+ {
+ this.options = options;
+ logger.info("initializing keystore from file {}", options.get(PROP_KEYSTORE));
+ FileInputStream inputStream = null;
+ try
+ {
+ inputStream = new FileInputStream(options.get(PROP_KEYSTORE));
+ store = KeyStore.getInstance(options.get(PROP_KEYSTORE_TYPE));
+ store.load(inputStream, options.get(PROP_KEYSTORE_PW).toCharArray());
+ isJceks = store.getType().equalsIgnoreCase("jceks");
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException("couldn't load keystore", e);
+ }
+ finally
+ {
+ FileUtils.closeQuietly(inputStream);
+ }
+ }
+
+ public Key getSecretKey(String keyAlias) throws IOException
+ {
+ // there's a lovely behavior with jceks files that all aliases are lower-cased
+ if (isJceks)
+ keyAlias = keyAlias.toLowerCase();
+
+ Key key;
+ try
+ {
+ String password = options.get(PROP_KEY_PW);
+ if (password == null || password.isEmpty())
+ password = options.get(PROP_KEYSTORE_PW);
+ key = store.getKey(keyAlias, password.toCharArray());
+ }
+ catch (Exception e)
+ {
+ throw new IOException("unable to load key from keystore");
+ }
+ if (key == null)
+ throw new IOException(String.format("key %s was not found in keystore", keyAlias));
+ return key;
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/security/KeyProvider.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/security/KeyProvider.java b/src/java/org/apache/cassandra/security/KeyProvider.java
new file mode 100644
index 0000000..f380aed
--- /dev/null
+++ b/src/java/org/apache/cassandra/security/KeyProvider.java
@@ -0,0 +1,33 @@
+/*
+ * 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.security;
+
+import java.io.IOException;
+import java.security.Key;
+
+/**
+ * Customizable key retrieval mechanism. Implementations should expect that retrieved keys will be cached.
+ * Further, each key will be requested non-concurrently (that is, no stampeding herds for the same key), although
+ * unique keys may be requested concurrently (unless you mark {@code getSecretKey} synchronized).
+ *
+ * Implementations must provide a constructor that accepts {@code TransparentDataEncryptionOptions} as the sole parameter.
+ */
+public interface KeyProvider
+{
+ Key getSecretKey(String alias) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/test/conf/cassandra.keystore
----------------------------------------------------------------------
diff --git a/test/conf/cassandra.keystore b/test/conf/cassandra.keystore
new file mode 100644
index 0000000..9a704ca
Binary files /dev/null and b/test/conf/cassandra.keystore differ
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/test/conf/cassandra_encryption.yaml
----------------------------------------------------------------------
diff --git a/test/conf/cassandra_encryption.yaml b/test/conf/cassandra_encryption.yaml
new file mode 100644
index 0000000..47e1312
--- /dev/null
+++ b/test/conf/cassandra_encryption.yaml
@@ -0,0 +1,14 @@
+transparent_data_encryption_options:
+ enabled: true
+ chunk_length_kb: 2
+ cipher: AES/CBC/PKCS5Padding
+ key_alias: testing:1
+ # CBC requires iv length to be 16 bytes
+ # iv_length: 16
+ key_provider:
+ - class_name: org.apache.cassandra.security.JKSKeyProvider
+ parameters:
+ - keystore: test/conf/cassandra.keystore
+ keystore_password: cassandra
+ store_type: JCEKS
+ key_password: cassandra
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/security/CipherFactoryTest.java b/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
new file mode 100644
index 0000000..4239973
--- /dev/null
+++ b/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
@@ -0,0 +1,87 @@
+package org.apache.cassandra.security;
+
+import java.io.IOException;
+import java.security.SecureRandom;
+
+import javax.crypto.BadPaddingException;
+import javax.crypto.Cipher;
+import javax.crypto.IllegalBlockSizeException;
+
+import com.google.common.base.Charsets;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+
+public class CipherFactoryTest
+{
+ // http://www.gutenberg.org/files/4300/4300-h/4300-h.htm
+ static final String ULYSSEUS = "Stately, plump Buck Mulligan came from the stairhead, bearing a bowl of lather on which a mirror and a razor lay crossed. " +
+ "A yellow dressinggown, ungirdled, was sustained gently behind him on the mild morning air. He held the bowl aloft and intoned: " +
+ "—Introibo ad altare Dei.";
+ TransparentDataEncryptionOptions encryptionOptions;
+ CipherFactory cipherFactory;
+ SecureRandom secureRandom;
+
+ @Before
+ public void setup()
+ {
+ secureRandom = new SecureRandom(new byte[] {0,1,2,3,4,5,6,7,8,9} );
+ encryptionOptions = EncryptionContextGenerator.createEncryptionOptions();
+ cipherFactory = new CipherFactory(encryptionOptions);
+ }
+
+ @Test
+ public void roundTrip() throws IOException, BadPaddingException, IllegalBlockSizeException
+ {
+ Cipher encryptor = cipherFactory.getEncryptor(encryptionOptions.cipher, encryptionOptions.key_alias);
+ byte[] original = ULYSSEUS.getBytes(Charsets.UTF_8);
+ byte[] encrypted = encryptor.doFinal(original);
+
+ Cipher decryptor = cipherFactory.getDecryptor(encryptionOptions.cipher, encryptionOptions.key_alias, encryptor.getIV());
+ byte[] decrypted = decryptor.doFinal(encrypted);
+ Assert.assertEquals(ULYSSEUS, new String(decrypted, Charsets.UTF_8));
+ }
+
+ private byte[] nextIV()
+ {
+ byte[] b = new byte[16];
+ secureRandom.nextBytes(b);
+ return b;
+ }
+
+ @Test
+ public void buildCipher_SameParams() throws Exception
+ {
+ byte[] iv = nextIV();
+ Cipher c1 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, iv, Cipher.ENCRYPT_MODE);
+ Cipher c2 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, iv, Cipher.ENCRYPT_MODE);
+ Assert.assertTrue(c1 == c2);
+ }
+
+ @Test
+ public void buildCipher_DifferentModes() throws Exception
+ {
+ byte[] iv = nextIV();
+ Cipher c1 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, iv, Cipher.ENCRYPT_MODE);
+ Cipher c2 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, iv, Cipher.DECRYPT_MODE);
+ Assert.assertFalse(c1 == c2);
+ }
+
+ @Test
+ public void buildCipher_DifferentIVs() throws Exception
+ {
+ Cipher c1 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, nextIV(), Cipher.ENCRYPT_MODE);
+ Cipher c2 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, nextIV(), Cipher.DECRYPT_MODE);
+ Assert.assertFalse(c1 == c2);
+ }
+
+ @Test
+ public void buildCipher_DifferentAliases() throws Exception
+ {
+ Cipher c1 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, nextIV(), Cipher.ENCRYPT_MODE);
+ Cipher c2 = cipherFactory.buildCipher(encryptionOptions.cipher, EncryptionContextGenerator.KEY_ALIAS_2, nextIV(), Cipher.DECRYPT_MODE);
+ Assert.assertFalse(c1 == c2);
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/test/unit/org/apache/cassandra/security/EncryptionContextGenerator.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/security/EncryptionContextGenerator.java b/test/unit/org/apache/cassandra/security/EncryptionContextGenerator.java
new file mode 100644
index 0000000..635889b
--- /dev/null
+++ b/test/unit/org/apache/cassandra/security/EncryptionContextGenerator.java
@@ -0,0 +1,54 @@
+/*
+ *
+ * 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.security;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+
+public class EncryptionContextGenerator
+{
+ public static final String KEY_ALIAS_1 = "testing:1";
+ public static final String KEY_ALIAS_2 = "testing:2";
+
+ public static EncryptionContext createContext(boolean init)
+ {
+ return new EncryptionContext(createEncryptionOptions(), init);
+ }
+
+ public static TransparentDataEncryptionOptions createEncryptionOptions()
+ {
+ Map<String,String> params = new HashMap<>();
+ params.put("keystore", "test/conf/cassandra.keystore");
+ params.put("keystore_password", "cassandra");
+ params.put("store_type", "JCEKS");
+ ParameterizedClass keyProvider = new ParameterizedClass(JKSKeyProvider.class.getName(), params);
+
+ return new TransparentDataEncryptionOptions("AES/CBC/PKCS5Padding", KEY_ALIAS_1, keyProvider);
+ }
+
+ public static EncryptionContext createDisabledContext()
+ {
+ return new EncryptionContext();
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/test/unit/org/apache/cassandra/security/JKSKeyProviderTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/security/JKSKeyProviderTest.java b/test/unit/org/apache/cassandra/security/JKSKeyProviderTest.java
new file mode 100644
index 0000000..081f688
--- /dev/null
+++ b/test/unit/org/apache/cassandra/security/JKSKeyProviderTest.java
@@ -0,0 +1,52 @@
+/*
+ * 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.security;
+
+import java.io.IOException;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+
+public class JKSKeyProviderTest
+{
+ JKSKeyProvider jksKeyProvider;
+ TransparentDataEncryptionOptions tdeOptions;
+
+ @Before
+ public void setup()
+ {
+ tdeOptions = EncryptionContextGenerator.createEncryptionOptions();
+ jksKeyProvider = new JKSKeyProvider(tdeOptions);
+ }
+
+ @Test
+ public void getSecretKey_WithKeyPassword() throws IOException
+ {
+ Assert.assertNotNull(jksKeyProvider.getSecretKey(tdeOptions.key_alias));
+ }
+
+ @Test
+ public void getSecretKey_WithoutKeyPassword() throws IOException
+ {
+ tdeOptions.remove("key_password");
+ Assert.assertNotNull(jksKeyProvider.getSecretKey(tdeOptions.key_alias));
+ }
+}
[4/9] cassandra git commit: Add transparent data encryption core
classes (CASSANDRA-9945)
Posted by be...@apache.org.
Add transparent data encryption core classes (CASSANDRA-9945)
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/765ab3fc
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/765ab3fc
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/765ab3fc
Branch: refs/heads/trunk
Commit: 765ab3fc3b9f4e891c257f36f117ddf4232da6be
Parents: c3ed25b
Author: Jason Brown <ja...@gmail.com>
Authored: Thu Jul 30 14:03:17 2015 -0700
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Aug 6 14:36:07 2015 +0200
----------------------------------------------------------------------
.../org/apache/cassandra/config/Config.java | 3 +-
.../cassandra/config/DatabaseDescriptor.java | 17 ++
.../TransparentDataEncryptionOptions.java | 76 ++++++++
.../cassandra/security/CipherFactory.java | 175 +++++++++++++++++++
.../cassandra/security/EncryptionContext.java | 122 +++++++++++++
.../cassandra/security/JKSKeyProvider.java | 90 ++++++++++
.../apache/cassandra/security/KeyProvider.java | 33 ++++
test/conf/cassandra.keystore | Bin 0 -> 1004 bytes
test/conf/cassandra_encryption.yaml | 14 ++
.../cassandra/security/CipherFactoryTest.java | 87 +++++++++
.../security/EncryptionContextGenerator.java | 54 ++++++
.../cassandra/security/JKSKeyProviderTest.java | 52 ++++++
12 files changed, 722 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/config/Config.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index e93d090..f8f34e0 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -169,7 +169,8 @@ public class Config
public int commitlog_segment_size_in_mb = 32;
public ParameterizedClass commitlog_compression;
public int commitlog_max_compression_buffers_in_pool = 3;
-
+ public TransparentDataEncryptionOptions transparent_data_encryption_options = new TransparentDataEncryptionOptions();
+
@Deprecated
public int commitlog_periodic_queue_size = -1;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 02f3c17..e7b9455 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -44,6 +44,7 @@ import org.apache.cassandra.locator.*;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.scheduler.IRequestScheduler;
import org.apache.cassandra.scheduler.NoScheduler;
+import org.apache.cassandra.security.EncryptionContext;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.thrift.ThriftServer;
import org.apache.cassandra.utils.ByteBufferUtil;
@@ -92,6 +93,7 @@ public class DatabaseDescriptor
private static String localDC;
private static Comparator<InetAddress> localComparator;
+ private static EncryptionContext encryptionContext;
public static void forceStaticInitialization() {}
static
@@ -613,6 +615,10 @@ public class DatabaseDescriptor
if (conf.user_defined_function_fail_timeout < conf.user_defined_function_warn_timeout)
throw new ConfigurationException("user_defined_function_warn_timeout must less than user_defined_function_fail_timeout", false);
+
+ // always attempt to load the cipher factory, as we could be in the situation where the user has disabled encryption,
+ // but has existing commitlogs and sstables on disk that are still encrypted (and still need to be read)
+ encryptionContext = new EncryptionContext(config.transparent_data_encryption_options);
}
private static IEndpointSnitch createEndpointSnitch(String snitchClassName) throws ConfigurationException
@@ -1792,4 +1798,15 @@ public class DatabaseDescriptor
{
conf.user_function_timeout_policy = userFunctionTimeoutPolicy;
}
+
+ public static EncryptionContext getEncryptionContext()
+ {
+ return encryptionContext;
+ }
+
+ @VisibleForTesting
+ public static void setEncryptionContext(EncryptionContext ec)
+ {
+ encryptionContext = ec;
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/config/TransparentDataEncryptionOptions.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/TransparentDataEncryptionOptions.java b/src/java/org/apache/cassandra/config/TransparentDataEncryptionOptions.java
new file mode 100644
index 0000000..4ad0305
--- /dev/null
+++ b/src/java/org/apache/cassandra/config/TransparentDataEncryptionOptions.java
@@ -0,0 +1,76 @@
+/*
+ * 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.config;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Objects;
+
+public class TransparentDataEncryptionOptions
+{
+ public boolean enabled = false;
+ public int chunk_length_kb = 64;
+ public String cipher = "AES/CBC/PKCS5Padding";
+ public String key_alias;
+ public int iv_length = 16;
+
+ public ParameterizedClass key_provider;
+
+ public TransparentDataEncryptionOptions()
+ { }
+
+ public TransparentDataEncryptionOptions(boolean enabled)
+ {
+ this.enabled = enabled;
+ }
+
+ public TransparentDataEncryptionOptions(String cipher, String keyAlias, ParameterizedClass keyProvider)
+ {
+ this(true, cipher, keyAlias, keyProvider);
+ }
+
+ public TransparentDataEncryptionOptions(boolean enabled, String cipher, String keyAlias, ParameterizedClass keyProvider)
+ {
+ this.enabled = enabled;
+ this.cipher = cipher;
+ key_alias = keyAlias;
+ key_provider = keyProvider;
+ }
+
+ public String get(String key)
+ {
+ return key_provider.parameters.get(key);
+ }
+
+ @VisibleForTesting
+ public void remove(String key)
+ {
+ key_provider.parameters.remove(key);
+ }
+
+ public boolean equals(Object o)
+ {
+ return o instanceof TransparentDataEncryptionOptions && equals((TransparentDataEncryptionOptions) o);
+ }
+
+ public boolean equals(TransparentDataEncryptionOptions other)
+ {
+ // not sure if this is a great equals() impl....
+ return Objects.equal(cipher, other.cipher) &&
+ Objects.equal(key_alias, other.key_alias);
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/security/CipherFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/security/CipherFactory.java b/src/java/org/apache/cassandra/security/CipherFactory.java
new file mode 100644
index 0000000..0ff9867
--- /dev/null
+++ b/src/java/org/apache/cassandra/security/CipherFactory.java
@@ -0,0 +1,175 @@
+/*
+ * 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.security;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.security.InvalidAlgorithmParameterException;
+import java.security.InvalidKeyException;
+import java.security.Key;
+import java.security.NoSuchAlgorithmException;
+import java.security.SecureRandom;
+import java.util.Arrays;
+import java.util.concurrent.ExecutionException;
+import javax.crypto.Cipher;
+import javax.crypto.NoSuchPaddingException;
+import javax.crypto.spec.IvParameterSpec;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+
+/**
+ * A factory for loading encryption keys from {@link KeyProvider} instances.
+ * Maintains a cache of loaded keys to avoid invoking the key provider on every call.
+ */
+public class CipherFactory
+{
+ private final Logger logger = LoggerFactory.getLogger(CipherFactory.class);
+
+ /**
+ * Keep around thread local instances of Cipher as they are quite expensive to instantiate (@code Cipher#getInstance).
+ * Bonus points if you can avoid calling (@code Cipher#init); hence, the point of the supporting struct
+ * for caching Cipher instances.
+ */
+ private static final ThreadLocal<CachedCipher> cipherThreadLocal = new ThreadLocal<>();
+
+ private final SecureRandom secureRandom;
+ private final LoadingCache<String, Key> cache;
+ private final int ivLength;
+ private final KeyProvider keyProvider;
+
+ public CipherFactory(TransparentDataEncryptionOptions options)
+ {
+ logger.info("initializing CipherFactory");
+ ivLength = options.iv_length;
+
+ try
+ {
+ secureRandom = SecureRandom.getInstance("SHA1PRNG");
+ Class<KeyProvider> keyProviderClass = (Class<KeyProvider>)Class.forName(options.key_provider.class_name);
+ Constructor ctor = keyProviderClass.getConstructor(TransparentDataEncryptionOptions.class);
+ keyProvider = (KeyProvider)ctor.newInstance(options);
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException("couldn't load cipher factory", e);
+ }
+
+ cache = CacheBuilder.newBuilder() // by default cache is unbounded
+ .maximumSize(64) // a value large enough that we should never even get close (so nothing gets evicted)
+ .concurrencyLevel(Runtime.getRuntime().availableProcessors())
+ .removalListener(new RemovalListener<String, Key>()
+ {
+ public void onRemoval(RemovalNotification<String, Key> notice)
+ {
+ // maybe reload the key? (to avoid the reload being on the user's dime)
+ logger.info("key {} removed from cipher key cache", notice.getKey());
+ }
+ })
+ .build(new CacheLoader<String, Key>()
+ {
+ @Override
+ public Key load(String alias) throws Exception
+ {
+ logger.info("loading secret key for alias {}", alias);
+ return keyProvider.getSecretKey(alias);
+ }
+ });
+ }
+
+ public Cipher getEncryptor(String transformation, String keyAlias) throws IOException
+ {
+ byte[] iv = new byte[ivLength];
+ secureRandom.nextBytes(iv);
+ return buildCipher(transformation, keyAlias, iv, Cipher.ENCRYPT_MODE);
+ }
+
+ public Cipher getDecryptor(String transformation, String keyAlias, byte[] iv) throws IOException
+ {
+ assert iv != null || iv.length > 0 : "trying to decrypt, but the initialization vector is empty";
+ return buildCipher(transformation, keyAlias, iv, Cipher.DECRYPT_MODE);
+ }
+
+ @VisibleForTesting
+ Cipher buildCipher(String transformation, String keyAlias, byte[] iv, int cipherMode) throws IOException
+ {
+ try
+ {
+ CachedCipher cachedCipher = cipherThreadLocal.get();
+ if (cachedCipher != null)
+ {
+ Cipher cipher = cachedCipher.cipher;
+ // rigorous checks to make sure we've absolutely got the correct instance (with correct alg/key/iv/...)
+ if (cachedCipher.mode == cipherMode && cipher.getAlgorithm().equals(transformation)
+ && cachedCipher.keyAlias.equals(keyAlias) && Arrays.equals(cipher.getIV(), iv))
+ return cipher;
+ }
+
+ Key key = retrieveKey(keyAlias);
+ Cipher cipher = Cipher.getInstance(transformation);
+ cipher.init(cipherMode, key, new IvParameterSpec(iv));
+ cipherThreadLocal.set(new CachedCipher(cipherMode, keyAlias, cipher));
+ return cipher;
+ }
+ catch (NoSuchAlgorithmException | NoSuchPaddingException | InvalidAlgorithmParameterException | InvalidKeyException e)
+ {
+ logger.error("could not build cipher", e);
+ throw new IOException("cannot load cipher", e);
+ }
+ }
+
+ private Key retrieveKey(String keyAlias) throws IOException
+ {
+ try
+ {
+ return cache.get(keyAlias);
+ }
+ catch (ExecutionException e)
+ {
+ if (e.getCause() instanceof IOException)
+ throw (IOException)e.getCause();
+ throw new IOException("failed to load key from cache: " + keyAlias, e);
+ }
+ }
+
+ /**
+ * A simple struct to use with the thread local caching of Cipher as we can't get the mode (encrypt/decrypt) nor
+ * key_alias (or key!) from the Cipher itself to use for comparisons
+ */
+ private static class CachedCipher
+ {
+ public final int mode;
+ public final String keyAlias;
+ public final Cipher cipher;
+
+ private CachedCipher(int mode, String keyAlias, Cipher cipher)
+ {
+ this.mode = mode;
+ this.keyAlias = keyAlias;
+ this.cipher = cipher;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/security/EncryptionContext.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/security/EncryptionContext.java b/src/java/org/apache/cassandra/security/EncryptionContext.java
new file mode 100644
index 0000000..dff6894
--- /dev/null
+++ b/src/java/org/apache/cassandra/security/EncryptionContext.java
@@ -0,0 +1,122 @@
+/*
+ * 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.security;
+
+import java.io.IOException;
+import java.util.Collections;
+import javax.crypto.Cipher;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Objects;
+
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.io.compress.ICompressor;
+import org.apache.cassandra.io.compress.LZ4Compressor;
+
+/**
+ * A (largely) immutable wrapper for the application-wide file-level encryption settings.
+ */
+public class EncryptionContext
+{
+ public static final String ENCRYPTION_CIPHER = "encCipher";
+ public static final String ENCRYPTION_KEY_ALIAS = "encKeyAlias";
+ public static final String ENCRYPTION_IV = "encIV";
+
+ private final TransparentDataEncryptionOptions tdeOptions;
+ private final ICompressor compressor;
+ private final CipherFactory cipherFactory;
+
+ private final int chunkLength;
+
+ public EncryptionContext()
+ {
+ this(new TransparentDataEncryptionOptions());
+ }
+
+ public EncryptionContext(TransparentDataEncryptionOptions tdeOptions)
+ {
+ this(tdeOptions, true);
+ }
+
+ @VisibleForTesting
+ public EncryptionContext(TransparentDataEncryptionOptions tdeOptions, boolean init)
+ {
+ this.tdeOptions = tdeOptions;
+ compressor = LZ4Compressor.create(Collections.<String, String>emptyMap());
+ chunkLength = tdeOptions.chunk_length_kb * 1024;
+
+ // always attempt to load the cipher factory, as we could be in the situation where the user has disabled encryption,
+ // but has existing commitlogs and sstables on disk that are still git addencrypted (and still need to be read)
+ CipherFactory factory = null;
+
+ if (tdeOptions.enabled && init)
+ {
+ try
+ {
+ factory = new CipherFactory(tdeOptions);
+ }
+ catch (Exception e)
+ {
+ throw new ConfigurationException("failed to load key provider for transparent data encryption", e);
+ }
+ }
+
+ cipherFactory = factory;
+ }
+
+ public ICompressor getCompressor()
+ {
+ return compressor;
+ }
+
+ public Cipher getEncryptor() throws IOException
+ {
+ return cipherFactory.getEncryptor(tdeOptions.cipher, tdeOptions.key_alias);
+ }
+
+ public Cipher getDecryptor(byte[] IV) throws IOException
+ {
+ return cipherFactory.getDecryptor(tdeOptions.cipher, tdeOptions.key_alias, IV);
+ }
+
+ public boolean isEnabled()
+ {
+ return tdeOptions.enabled;
+ }
+
+ public int getChunkLength()
+ {
+ return chunkLength;
+ }
+
+ public TransparentDataEncryptionOptions getTransparentDataEncryptionOptions()
+ {
+ return tdeOptions;
+ }
+
+ public boolean equals(Object o)
+ {
+ return o instanceof EncryptionContext && equals((EncryptionContext) o);
+ }
+
+ public boolean equals(EncryptionContext other)
+ {
+ return Objects.equal(tdeOptions, other.tdeOptions) && Objects.equal(compressor, other.compressor);
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/security/JKSKeyProvider.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/security/JKSKeyProvider.java b/src/java/org/apache/cassandra/security/JKSKeyProvider.java
new file mode 100644
index 0000000..8d7f1c6
--- /dev/null
+++ b/src/java/org/apache/cassandra/security/JKSKeyProvider.java
@@ -0,0 +1,90 @@
+/*
+ * 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.security;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.security.Key;
+import java.security.KeyStore;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+import org.apache.cassandra.io.util.FileUtils;
+
+/**
+ * A {@code KeyProvider} that retrieves keys from a java keystore.
+ */
+public class JKSKeyProvider implements KeyProvider
+{
+ private final Logger logger = LoggerFactory.getLogger(JKSKeyProvider.class);
+ static final String PROP_KEYSTORE = "keystore";
+ static final String PROP_KEYSTORE_PW = "keystore_password";
+ static final String PROP_KEYSTORE_TYPE = "store_type";
+ static final String PROP_KEY_PW = "key_password";
+
+ private final KeyStore store;
+ private final boolean isJceks;
+ private final TransparentDataEncryptionOptions options;
+
+ public JKSKeyProvider(TransparentDataEncryptionOptions options)
+ {
+ this.options = options;
+ logger.info("initializing keystore from file {}", options.get(PROP_KEYSTORE));
+ FileInputStream inputStream = null;
+ try
+ {
+ inputStream = new FileInputStream(options.get(PROP_KEYSTORE));
+ store = KeyStore.getInstance(options.get(PROP_KEYSTORE_TYPE));
+ store.load(inputStream, options.get(PROP_KEYSTORE_PW).toCharArray());
+ isJceks = store.getType().equalsIgnoreCase("jceks");
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException("couldn't load keystore", e);
+ }
+ finally
+ {
+ FileUtils.closeQuietly(inputStream);
+ }
+ }
+
+ public Key getSecretKey(String keyAlias) throws IOException
+ {
+ // there's a lovely behavior with jceks files that all aliases are lower-cased
+ if (isJceks)
+ keyAlias = keyAlias.toLowerCase();
+
+ Key key;
+ try
+ {
+ String password = options.get(PROP_KEY_PW);
+ if (password == null || password.isEmpty())
+ password = options.get(PROP_KEYSTORE_PW);
+ key = store.getKey(keyAlias, password.toCharArray());
+ }
+ catch (Exception e)
+ {
+ throw new IOException("unable to load key from keystore");
+ }
+ if (key == null)
+ throw new IOException(String.format("key %s was not found in keystore", keyAlias));
+ return key;
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/src/java/org/apache/cassandra/security/KeyProvider.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/security/KeyProvider.java b/src/java/org/apache/cassandra/security/KeyProvider.java
new file mode 100644
index 0000000..f380aed
--- /dev/null
+++ b/src/java/org/apache/cassandra/security/KeyProvider.java
@@ -0,0 +1,33 @@
+/*
+ * 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.security;
+
+import java.io.IOException;
+import java.security.Key;
+
+/**
+ * Customizable key retrieval mechanism. Implementations should expect that retrieved keys will be cached.
+ * Further, each key will be requested non-concurrently (that is, no stampeding herds for the same key), although
+ * unique keys may be requested concurrently (unless you mark {@code getSecretKey} synchronized).
+ *
+ * Implementations must provide a constructor that accepts {@code TransparentDataEncryptionOptions} as the sole parameter.
+ */
+public interface KeyProvider
+{
+ Key getSecretKey(String alias) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/test/conf/cassandra.keystore
----------------------------------------------------------------------
diff --git a/test/conf/cassandra.keystore b/test/conf/cassandra.keystore
new file mode 100644
index 0000000..9a704ca
Binary files /dev/null and b/test/conf/cassandra.keystore differ
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/test/conf/cassandra_encryption.yaml
----------------------------------------------------------------------
diff --git a/test/conf/cassandra_encryption.yaml b/test/conf/cassandra_encryption.yaml
new file mode 100644
index 0000000..47e1312
--- /dev/null
+++ b/test/conf/cassandra_encryption.yaml
@@ -0,0 +1,14 @@
+transparent_data_encryption_options:
+ enabled: true
+ chunk_length_kb: 2
+ cipher: AES/CBC/PKCS5Padding
+ key_alias: testing:1
+ # CBC requires iv length to be 16 bytes
+ # iv_length: 16
+ key_provider:
+ - class_name: org.apache.cassandra.security.JKSKeyProvider
+ parameters:
+ - keystore: test/conf/cassandra.keystore
+ keystore_password: cassandra
+ store_type: JCEKS
+ key_password: cassandra
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/security/CipherFactoryTest.java b/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
new file mode 100644
index 0000000..4239973
--- /dev/null
+++ b/test/unit/org/apache/cassandra/security/CipherFactoryTest.java
@@ -0,0 +1,87 @@
+package org.apache.cassandra.security;
+
+import java.io.IOException;
+import java.security.SecureRandom;
+
+import javax.crypto.BadPaddingException;
+import javax.crypto.Cipher;
+import javax.crypto.IllegalBlockSizeException;
+
+import com.google.common.base.Charsets;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+
+public class CipherFactoryTest
+{
+ // http://www.gutenberg.org/files/4300/4300-h/4300-h.htm
+ static final String ULYSSEUS = "Stately, plump Buck Mulligan came from the stairhead, bearing a bowl of lather on which a mirror and a razor lay crossed. " +
+ "A yellow dressinggown, ungirdled, was sustained gently behind him on the mild morning air. He held the bowl aloft and intoned: " +
+ "—Introibo ad altare Dei.";
+ TransparentDataEncryptionOptions encryptionOptions;
+ CipherFactory cipherFactory;
+ SecureRandom secureRandom;
+
+ @Before
+ public void setup()
+ {
+ secureRandom = new SecureRandom(new byte[] {0,1,2,3,4,5,6,7,8,9} );
+ encryptionOptions = EncryptionContextGenerator.createEncryptionOptions();
+ cipherFactory = new CipherFactory(encryptionOptions);
+ }
+
+ @Test
+ public void roundTrip() throws IOException, BadPaddingException, IllegalBlockSizeException
+ {
+ Cipher encryptor = cipherFactory.getEncryptor(encryptionOptions.cipher, encryptionOptions.key_alias);
+ byte[] original = ULYSSEUS.getBytes(Charsets.UTF_8);
+ byte[] encrypted = encryptor.doFinal(original);
+
+ Cipher decryptor = cipherFactory.getDecryptor(encryptionOptions.cipher, encryptionOptions.key_alias, encryptor.getIV());
+ byte[] decrypted = decryptor.doFinal(encrypted);
+ Assert.assertEquals(ULYSSEUS, new String(decrypted, Charsets.UTF_8));
+ }
+
+ private byte[] nextIV()
+ {
+ byte[] b = new byte[16];
+ secureRandom.nextBytes(b);
+ return b;
+ }
+
+ @Test
+ public void buildCipher_SameParams() throws Exception
+ {
+ byte[] iv = nextIV();
+ Cipher c1 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, iv, Cipher.ENCRYPT_MODE);
+ Cipher c2 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, iv, Cipher.ENCRYPT_MODE);
+ Assert.assertTrue(c1 == c2);
+ }
+
+ @Test
+ public void buildCipher_DifferentModes() throws Exception
+ {
+ byte[] iv = nextIV();
+ Cipher c1 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, iv, Cipher.ENCRYPT_MODE);
+ Cipher c2 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, iv, Cipher.DECRYPT_MODE);
+ Assert.assertFalse(c1 == c2);
+ }
+
+ @Test
+ public void buildCipher_DifferentIVs() throws Exception
+ {
+ Cipher c1 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, nextIV(), Cipher.ENCRYPT_MODE);
+ Cipher c2 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, nextIV(), Cipher.DECRYPT_MODE);
+ Assert.assertFalse(c1 == c2);
+ }
+
+ @Test
+ public void buildCipher_DifferentAliases() throws Exception
+ {
+ Cipher c1 = cipherFactory.buildCipher(encryptionOptions.cipher, encryptionOptions.key_alias, nextIV(), Cipher.ENCRYPT_MODE);
+ Cipher c2 = cipherFactory.buildCipher(encryptionOptions.cipher, EncryptionContextGenerator.KEY_ALIAS_2, nextIV(), Cipher.DECRYPT_MODE);
+ Assert.assertFalse(c1 == c2);
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/test/unit/org/apache/cassandra/security/EncryptionContextGenerator.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/security/EncryptionContextGenerator.java b/test/unit/org/apache/cassandra/security/EncryptionContextGenerator.java
new file mode 100644
index 0000000..635889b
--- /dev/null
+++ b/test/unit/org/apache/cassandra/security/EncryptionContextGenerator.java
@@ -0,0 +1,54 @@
+/*
+ *
+ * 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.security;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+
+public class EncryptionContextGenerator
+{
+ public static final String KEY_ALIAS_1 = "testing:1";
+ public static final String KEY_ALIAS_2 = "testing:2";
+
+ public static EncryptionContext createContext(boolean init)
+ {
+ return new EncryptionContext(createEncryptionOptions(), init);
+ }
+
+ public static TransparentDataEncryptionOptions createEncryptionOptions()
+ {
+ Map<String,String> params = new HashMap<>();
+ params.put("keystore", "test/conf/cassandra.keystore");
+ params.put("keystore_password", "cassandra");
+ params.put("store_type", "JCEKS");
+ ParameterizedClass keyProvider = new ParameterizedClass(JKSKeyProvider.class.getName(), params);
+
+ return new TransparentDataEncryptionOptions("AES/CBC/PKCS5Padding", KEY_ALIAS_1, keyProvider);
+ }
+
+ public static EncryptionContext createDisabledContext()
+ {
+ return new EncryptionContext();
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/765ab3fc/test/unit/org/apache/cassandra/security/JKSKeyProviderTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/security/JKSKeyProviderTest.java b/test/unit/org/apache/cassandra/security/JKSKeyProviderTest.java
new file mode 100644
index 0000000..081f688
--- /dev/null
+++ b/test/unit/org/apache/cassandra/security/JKSKeyProviderTest.java
@@ -0,0 +1,52 @@
+/*
+ * 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.security;
+
+import java.io.IOException;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+
+public class JKSKeyProviderTest
+{
+ JKSKeyProvider jksKeyProvider;
+ TransparentDataEncryptionOptions tdeOptions;
+
+ @Before
+ public void setup()
+ {
+ tdeOptions = EncryptionContextGenerator.createEncryptionOptions();
+ jksKeyProvider = new JKSKeyProvider(tdeOptions);
+ }
+
+ @Test
+ public void getSecretKey_WithKeyPassword() throws IOException
+ {
+ Assert.assertNotNull(jksKeyProvider.getSecretKey(tdeOptions.key_alias));
+ }
+
+ @Test
+ public void getSecretKey_WithoutKeyPassword() throws IOException
+ {
+ tdeOptions.remove("key_password");
+ Assert.assertNotNull(jksKeyProvider.getSecretKey(tdeOptions.key_alias));
+ }
+}
[2/9] cassandra git commit: Fix WaitQueueTest flakiness
Posted by be...@apache.org.
Fix WaitQueueTest flakiness
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ecb2b4b0
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ecb2b4b0
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ecb2b4b0
Branch: refs/heads/cassandra-3.0
Commit: ecb2b4b0473c019d1132c89887734792c75e0895
Parents: 765ab3f
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Mon Aug 3 16:34:29 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Aug 6 14:36:07 2015 +0200
----------------------------------------------------------------------
test/unit/org/apache/cassandra/Util.java | 5 ++
.../cassandra/concurrent/WaitQueueTest.java | 91 ++++++--------------
2 files changed, 32 insertions(+), 64 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecb2b4b0/test/unit/org/apache/cassandra/Util.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/Util.java b/test/unit/org/apache/cassandra/Util.java
index 254c21c..7efe6f4 100644
--- a/test/unit/org/apache/cassandra/Util.java
+++ b/test/unit/org/apache/cassandra/Util.java
@@ -526,4 +526,9 @@ public class Util
assert p == newP;
}
}
+
+ public static void joinThread(Thread thread) throws InterruptedException
+ {
+ thread.join(10000);
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ecb2b4b0/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java b/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java
index 3e7cb7b..8e092c5 100644
--- a/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java
+++ b/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java
@@ -21,10 +21,13 @@ package org.apache.cassandra.concurrent;
*/
+import org.apache.cassandra.Util;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import org.junit.*;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.*;
@@ -38,6 +41,7 @@ public class WaitQueueTest
}
public void testSerial(final WaitQueue queue) throws InterruptedException
{
+ AtomicInteger ready = new AtomicInteger();
Thread[] ts = new Thread[4];
for (int i = 0 ; i < ts.length ; i++)
ts[i] = new Thread(new Runnable()
@@ -46,6 +50,7 @@ public class WaitQueueTest
public void run()
{
WaitQueue.Signal wait = queue.register();
+ ready.incrementAndGet();
try
{
wait.await();
@@ -55,68 +60,28 @@ public class WaitQueueTest
}
}
});
- for (int i = 0 ; i < ts.length ; i++)
- ts[i].start();
- Thread.sleep(100);
- queue.signal();
- queue.signal();
- queue.signal();
- queue.signal();
- for (int i = 0 ; i < ts.length ; i++)
+ for (Thread t : ts)
+ t.start();
+ final ThreadLocalRandom random = ThreadLocalRandom.current();
+ while (ready.get() < ts.length)
+ random.nextLong();
+ for (Thread t : ts)
+ queue.signal();
+ for (Thread t : ts)
{
- ts[i].join(100);
- assertFalse(queue.getClass().getName(), ts[i].isAlive());
+ Util.joinThread(t);
+ assertFalse(queue.getClass().getName(), t.isAlive());
}
}
-
- @Test
- public void testCondition1() throws InterruptedException
- {
- testCondition1(new WaitQueue());
- }
-
- public void testCondition1(final WaitQueue queue) throws InterruptedException
- {
- final AtomicBoolean cond1 = new AtomicBoolean(false);
- final AtomicBoolean fail = new AtomicBoolean(false);
- Thread t1 = new Thread(new Runnable()
- {
- @Override
- public void run()
- {
- try
- {
- Thread.sleep(200);
- } catch (InterruptedException e)
- {
- e.printStackTrace();
- }
- WaitQueue.Signal wait = queue.register();
- if (!cond1.get())
- {
- System.err.println("Condition should have already been met");
- fail.set(true);
- }
- }
- });
- t1.start();
- Thread.sleep(50);
- cond1.set(true);
- Thread.sleep(300);
- queue.signal();
- t1.join(300);
- assertFalse(queue.getClass().getName(), t1.isAlive());
- assertFalse(fail.get());
- }
-
@Test
- public void testCondition2() throws InterruptedException
+ public void testCondition() throws InterruptedException
{
- testCondition2(new WaitQueue());
+ testCondition(new WaitQueue());
}
- public void testCondition2(final WaitQueue queue) throws InterruptedException
+ public void testCondition(final WaitQueue queue) throws InterruptedException
{
+ final AtomicBoolean ready = new AtomicBoolean(false);
final AtomicBoolean condition = new AtomicBoolean(false);
final AtomicBoolean fail = new AtomicBoolean(false);
Thread t = new Thread(new Runnable()
@@ -129,16 +94,12 @@ public class WaitQueueTest
{
System.err.println("");
fail.set(true);
+ ready.set(true);
+ return;
}
- try
- {
- Thread.sleep(200);
- wait.await();
- } catch (InterruptedException e)
- {
- e.printStackTrace();
- }
+ ready.set(true);
+ wait.awaitUninterruptibly();
if (!condition.get())
{
System.err.println("Woke up when condition not met");
@@ -147,10 +108,12 @@ public class WaitQueueTest
}
});
t.start();
- Thread.sleep(50);
+ final ThreadLocalRandom random = ThreadLocalRandom.current();
+ while (!ready.get())
+ random.nextLong();
condition.set(true);
queue.signal();
- t.join(300);
+ Util.joinThread(t);
assertFalse(queue.getClass().getName(), t.isAlive());
assertFalse(fail.get());
}