You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by jd...@apache.org on 2016/07/25 17:15:14 UTC
[09/36] incubator-kudu git commit: [java-client] repackage to
org.apache.kudu (Part 1)
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanPredicate.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanPredicate.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanPredicate.java
new file mode 100644
index 0000000..d67380b
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestScanPredicate.java
@@ -0,0 +1,609 @@
+// 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.kududb.client;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSortedSet;
+import org.junit.Assert;
+import org.junit.Test;
+import org.kududb.ColumnSchema;
+import org.kududb.Schema;
+import org.kududb.Type;
+import org.kududb.client.KuduPredicate.ComparisonOp;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.NavigableSet;
+import java.util.TreeSet;
+
+public class TestScanPredicate extends BaseKuduTest {
+
+ private Schema createTableSchema(Type type) {
+ ColumnSchema key = new ColumnSchema.ColumnSchemaBuilder("key", Type.INT64).key(true).build();
+ ColumnSchema val = new ColumnSchema.ColumnSchemaBuilder("value", type).nullable(true).build();
+ return new Schema(ImmutableList.of(key, val));
+ }
+
+ private static CreateTableOptions createTableOptions() {
+ return new CreateTableOptions().setRangePartitionColumns(ImmutableList.of("key"));
+ }
+
+ private int countRows(KuduTable table, KuduPredicate... predicates) throws Exception {
+ KuduScanner.KuduScannerBuilder scanBuilder = new KuduScanner.KuduScannerBuilder(client, table);
+ for (KuduPredicate predicate : predicates) {
+ scanBuilder.addPredicate(predicate);
+ }
+
+ KuduScanner scanner = scanBuilder.build();
+
+ int count = 0;
+ while (scanner.hasMoreRows()) {
+ count += scanner.nextRows().getNumRows();
+ }
+ return count;
+ }
+
+ private NavigableSet<Long> createIntegerValues(Type type) {
+ NavigableSet<Long> values = new TreeSet<>();
+ for (long i = -50; i < 50; i++) {
+ values.add(i);
+ }
+ values.add(KuduPredicate.minIntValue(type));
+ values.add(KuduPredicate.minIntValue(type) + 1);
+ values.add(KuduPredicate.maxIntValue(type) - 1);
+ values.add(KuduPredicate.maxIntValue(type));
+ return values;
+ }
+
+ private List<Long> createIntegerTestValues(Type type) {
+ return ImmutableList.of(
+ KuduPredicate.minIntValue(type),
+ KuduPredicate.minIntValue(type) + 1,
+ -51L,
+ 50L,
+ 0L,
+ 49L,
+ 50L,
+ KuduPredicate.maxIntValue(type) - 1,
+ KuduPredicate.maxIntValue(type));
+ }
+
+ private NavigableSet<Float> createFloatValues() {
+ NavigableSet<Float> values = new TreeSet<>();
+ for (long i = -50; i < 50; i++) {
+ values.add((float) i + (float) i / 100.0F);
+ }
+
+ values.add(Float.NEGATIVE_INFINITY);
+ values.add(-Float.MAX_VALUE);
+ values.add(-Float.MIN_NORMAL);
+ values.add(-Float.MIN_VALUE);
+ values.add(Float.MIN_VALUE);
+ values.add(Float.MIN_NORMAL);
+ values.add(Float.MAX_VALUE);
+ values.add(Float.POSITIVE_INFINITY);
+
+ // TODO: uncomment after fixing KUDU-1386
+ // values.add(Float.NaN);
+ return values;
+ }
+
+ private List<Float> createFloatTestValues() {
+ return ImmutableList.of(
+ Float.NEGATIVE_INFINITY,
+ -Float.MAX_VALUE,
+ -100.0F,
+ -1.1F,
+ -1.0F,
+ -Float.MIN_NORMAL,
+ -Float.MIN_VALUE,
+ 0.0F,
+ Float.MIN_VALUE,
+ Float.MIN_NORMAL,
+ 1.0F,
+ 1.1F,
+ 100.0F,
+ Float.MAX_VALUE,
+ Float.POSITIVE_INFINITY
+
+ // TODO: uncomment after fixing KUDU-1386
+ // Float.NaN
+ );
+ }
+
+ private NavigableSet<Double> createDoubleValues() {
+ NavigableSet<Double> values = new TreeSet<>();
+ for (long i = -50; i < 50; i++) {
+ values.add((double) i + (double) i / 100.0);
+ }
+
+ values.add(Double.NEGATIVE_INFINITY);
+ values.add(-Double.MAX_VALUE);
+ values.add(-Double.MIN_NORMAL);
+ values.add(-Double.MIN_VALUE);
+ values.add(Double.MIN_VALUE);
+ values.add(Double.MIN_NORMAL);
+ values.add(Double.MAX_VALUE);
+ values.add(Double.POSITIVE_INFINITY);
+
+ // TODO: uncomment after fixing KUDU-1386
+ // values.add(Double.NaN);
+ return values;
+ }
+
+ private List<Double> createDoubleTestValues() {
+ return ImmutableList.of(
+ Double.NEGATIVE_INFINITY,
+ -Double.MAX_VALUE,
+ -100.0,
+ -1.1,
+ -1.0,
+ -Double.MIN_NORMAL,
+ -Double.MIN_VALUE,
+ 0.0,
+ Double.MIN_VALUE,
+ Double.MIN_NORMAL,
+ 1.0,
+ 1.1,
+ 100.0,
+ Double.MAX_VALUE,
+ Double.POSITIVE_INFINITY
+
+ // TODO: uncomment after fixing KUDU-1386
+ // Double.NaN
+ );
+ }
+
+ private NavigableSet<String> createStringValues() {
+ return ImmutableSortedSet.of("", "\0", "\0\0", "a", "a\0", "a\0a", "aa\0");
+ }
+
+ private List<String> createStringTestValues() {
+ List<String> values = new ArrayList<>(createStringValues());
+ values.add("aa");
+ values.add("\1");
+ values.add("a\1");
+ return values;
+ }
+
+ private void checkIntPredicates(KuduTable table,
+ NavigableSet<Long> values,
+ List<Long> testValues) throws Exception {
+ ColumnSchema col = table.getSchema().getColumn("value");
+ Assert.assertEquals(values.size() + 1, countRows(table));
+ for (long v : testValues) {
+ // value = v
+ KuduPredicate equal = KuduPredicate.newComparisonPredicate(col, ComparisonOp.EQUAL, v);
+ Assert.assertEquals(values.contains(v) ? 1 : 0, countRows(table, equal));
+
+ // value >= v
+ KuduPredicate greaterEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER_EQUAL, v);
+ Assert.assertEquals(values.tailSet(v).size(), countRows(table, greaterEqual));
+
+ // value <= v
+ KuduPredicate lessEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS_EQUAL, v);
+ Assert.assertEquals(values.headSet(v, true).size(), countRows(table, lessEqual));
+
+ // value > v
+ KuduPredicate greater =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER, v);
+ Assert.assertEquals(values.tailSet(v, false).size(), countRows(table, greater));
+
+ // value < v
+ KuduPredicate less =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS, v);
+ Assert.assertEquals(values.headSet(v).size(), countRows(table, less));
+ }
+ }
+
+ @Test
+ public void testBoolPredicates() throws Exception {
+ Schema schema = createTableSchema(Type.BOOL);
+ syncClient.createTable("bool-table", schema, createTableOptions());
+ KuduTable table = syncClient.openTable("bool-table");
+
+ NavigableSet<Boolean> values = ImmutableSortedSet.of(false, true);
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ long i = 0;
+ for (boolean value : values) {
+ Insert insert = table.newInsert();
+ insert.getRow().addLong("key", i++);
+ insert.getRow().addBoolean("value", value);
+ session.apply(insert);
+ }
+ Insert nullInsert = table.newInsert();
+ nullInsert.getRow().addLong("key", i++);
+ nullInsert.getRow().setNull("value");
+ session.apply(nullInsert);
+ session.flush();
+
+ ColumnSchema col = table.getSchema().getColumn("value");
+ Assert.assertEquals(values.size() + 1, countRows(table));
+
+ for (boolean v : values) {
+ // value = v
+ KuduPredicate equal = KuduPredicate.newComparisonPredicate(col, ComparisonOp.EQUAL, v);
+ Assert.assertEquals(values.contains(v) ? 1 : 0, countRows(table, equal));
+
+ // value >= v
+ KuduPredicate greaterEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER_EQUAL, v);
+ Assert.assertEquals(values.tailSet(v).size(), countRows(table, greaterEqual));
+
+ // value <= v
+ KuduPredicate lessEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS_EQUAL, v);
+ Assert.assertEquals(values.headSet(v, true).size(), countRows(table, lessEqual));
+
+ // value > v
+ KuduPredicate greater =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER, v);
+ Assert.assertEquals(values.tailSet(v, false).size(), countRows(table, greater));
+
+ // value < v
+ KuduPredicate less =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS, v);
+ Assert.assertEquals(values.headSet(v).size(), countRows(table, less));
+ }
+ }
+
+ @Test
+ public void testBytePredicates() throws Exception {
+ Schema schema = createTableSchema(Type.INT8);
+ syncClient.createTable("byte-table", schema, createTableOptions());
+ KuduTable table = syncClient.openTable("byte-table");
+
+ NavigableSet<Long> values = createIntegerValues(Type.INT8);
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ long i = 0;
+ for (long value : values) {
+ Insert insert = table.newInsert();
+ insert.getRow().addLong("key", i++);
+ insert.getRow().addByte("value", (byte) value);
+ session.apply(insert);
+ }
+ Insert nullInsert = table.newInsert();
+ nullInsert.getRow().addLong("key", i++);
+ nullInsert.getRow().setNull("value");
+ session.apply(nullInsert);
+ session.flush();
+
+ checkIntPredicates(table, values, createIntegerTestValues(Type.INT8));
+ }
+
+ @Test
+ public void testShortPredicates() throws Exception {
+ Schema schema = createTableSchema(Type.INT16);
+ syncClient.createTable("short-table", schema,
+ new CreateTableOptions().setRangePartitionColumns(
+ ImmutableList.<String>of()));
+ KuduTable table = syncClient.openTable("short-table");
+
+ NavigableSet<Long> values = createIntegerValues(Type.INT16);
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ long i = 0;
+ for (long value : values) {
+ Insert insert = table.newInsert();
+ insert.getRow().addLong("key", i++);
+ insert.getRow().addShort("value", (short) value);
+ session.apply(insert);
+ }
+ Insert nullInsert = table.newInsert();
+ nullInsert.getRow().addLong("key", i++);
+ nullInsert.getRow().setNull("value");
+ session.apply(nullInsert);
+ session.flush();
+
+ checkIntPredicates(table, values, createIntegerTestValues(Type.INT16));
+ }
+
+ @Test
+ public void testIntPredicates() throws Exception {
+ Schema schema = createTableSchema(Type.INT32);
+ syncClient.createTable("int-table", schema, createTableOptions());
+ KuduTable table = syncClient.openTable("int-table");
+
+ NavigableSet<Long> values = createIntegerValues(Type.INT32);
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ long i = 0;
+ for (long value : values) {
+ Insert insert = table.newInsert();
+ insert.getRow().addLong("key", i++);
+ insert.getRow().addInt("value", (int) value);
+ session.apply(insert);
+ }
+ Insert nullInsert = table.newInsert();
+ nullInsert.getRow().addLong("key", i++);
+ nullInsert.getRow().setNull("value");
+ session.apply(nullInsert);
+ session.flush();
+
+ checkIntPredicates(table, values, createIntegerTestValues(Type.INT32));
+ }
+
+ @Test
+ public void testLongPredicates() throws Exception {
+ Schema schema = createTableSchema(Type.INT64);
+ syncClient.createTable("long-table", schema,
+ new CreateTableOptions().setRangePartitionColumns(
+ ImmutableList.<String>of()));
+ KuduTable table = syncClient.openTable("long-table");
+
+ NavigableSet<Long> values = createIntegerValues(Type.INT64);
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ long i = 0;
+ for (long value : values) {
+ Insert insert = table.newInsert();
+ insert.getRow().addLong("key", i++);
+ insert.getRow().addLong("value", value);
+ session.apply(insert);
+ }
+ Insert nullInsert = table.newInsert();
+ nullInsert.getRow().addLong("key", i++);
+ nullInsert.getRow().setNull("value");
+ session.apply(nullInsert);
+ session.flush();
+
+ checkIntPredicates(table, values, createIntegerTestValues(Type.INT64));
+ }
+
+ @Test
+ public void testTimestampPredicate() throws Exception {
+ Schema schema = createTableSchema(Type.INT64);
+ syncClient.createTable("timestamp-table", schema, createTableOptions());
+ KuduTable table = syncClient.openTable("timestamp-table");
+
+ NavigableSet<Long> values = createIntegerValues(Type.INT64);
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ long i = 0;
+ for (long value : values) {
+ Insert insert = table.newInsert();
+ insert.getRow().addLong("key", i++);
+ insert.getRow().addLong("value", value);
+ session.apply(insert);
+ }
+ Insert nullInsert = table.newInsert();
+ nullInsert.getRow().addLong("key", i++);
+ nullInsert.getRow().setNull("value");
+ session.apply(nullInsert);
+ session.flush();
+
+ checkIntPredicates(table, values, createIntegerTestValues(Type.INT64));
+ }
+
+ @Test
+ public void testFloatPredicates() throws Exception {
+ Schema schema = createTableSchema(Type.FLOAT);
+ syncClient.createTable("float-table", schema, createTableOptions());
+ KuduTable table = syncClient.openTable("float-table");
+
+ NavigableSet<Float> values = createFloatValues();
+ List<Float> testValues = createFloatTestValues();
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ long i = 0;
+ for (float value : values) {
+ Insert insert = table.newInsert();
+ insert.getRow().addLong("key", i++);
+ insert.getRow().addFloat("value", value);
+ session.apply(insert);
+ }
+ Insert nullInsert = table.newInsert();
+ nullInsert.getRow().addLong("key", i++);
+ nullInsert.getRow().setNull("value");
+ session.apply(nullInsert);
+ session.flush();
+
+ ColumnSchema col = table.getSchema().getColumn("value");
+ Assert.assertEquals(values.size() + 1, countRows(table));
+
+ for (float v : testValues) {
+ // value = v
+ KuduPredicate equal = KuduPredicate.newComparisonPredicate(col, ComparisonOp.EQUAL, v);
+ Assert.assertEquals(values.subSet(v, true, v, true).size(), countRows(table, equal));
+
+ // value >= v
+ KuduPredicate greaterEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER_EQUAL, v);
+ Assert.assertEquals(values.tailSet(v).size(), countRows(table, greaterEqual));
+
+ // value <= v
+ KuduPredicate lessEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS_EQUAL, v);
+ Assert.assertEquals(values.headSet(v, true).size(), countRows(table, lessEqual));
+
+ // value > v
+ KuduPredicate greater =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER, v);
+ Assert.assertEquals(values.tailSet(v, false).size(), countRows(table, greater));
+
+ // value < v
+ KuduPredicate less =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS, v);
+ Assert.assertEquals(values.headSet(v).size(), countRows(table, less));
+ }
+ }
+
+ @Test
+ public void testDoublePredicates() throws Exception {
+ Schema schema = createTableSchema(Type.DOUBLE);
+ syncClient.createTable("double-table", schema, createTableOptions());
+ KuduTable table = syncClient.openTable("double-table");
+
+ NavigableSet<Double> values = createDoubleValues();
+ List<Double> testValues = createDoubleTestValues();
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ long i = 0;
+ for (double value : values) {
+ Insert insert = table.newInsert();
+ insert.getRow().addLong("key", i++);
+ insert.getRow().addDouble("value", value);
+ session.apply(insert);
+ }
+ Insert nullInsert = table.newInsert();
+ nullInsert.getRow().addLong("key", i++);
+ nullInsert.getRow().setNull("value");
+ session.apply(nullInsert);
+ session.flush();
+
+ ColumnSchema col = table.getSchema().getColumn("value");
+ Assert.assertEquals(values.size() + 1, countRows(table));
+
+ for (double v : testValues) {
+ // value = v
+ KuduPredicate equal = KuduPredicate.newComparisonPredicate(col, ComparisonOp.EQUAL, v);
+ Assert.assertEquals(values.subSet(v, true, v, true).size(), countRows(table, equal));
+
+ // value >= v
+ KuduPredicate greaterEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER_EQUAL, v);
+ Assert.assertEquals(values.tailSet(v).size(), countRows(table, greaterEqual));
+
+ // value <= v
+ KuduPredicate lessEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS_EQUAL, v);
+ Assert.assertEquals(values.headSet(v, true).size(), countRows(table, lessEqual));
+
+ // value > v
+ KuduPredicate greater =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER, v);
+ Assert.assertEquals(values.tailSet(v, false).size(), countRows(table, greater));
+
+ // value < v
+ KuduPredicate less =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS, v);
+ Assert.assertEquals(values.headSet(v).size(), countRows(table, less));
+ }
+ }
+
+ @Test
+ public void testStringPredicates() throws Exception {
+ Schema schema = createTableSchema(Type.STRING);
+ syncClient.createTable("string-table", schema, createTableOptions());
+ KuduTable table = syncClient.openTable("string-table");
+
+ NavigableSet<String> values = createStringValues();
+ List<String> testValues = createStringTestValues();
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ long i = 0;
+ for (String value : values) {
+ Insert insert = table.newInsert();
+ insert.getRow().addLong("key", i++);
+ insert.getRow().addString("value", value);
+ session.apply(insert);
+ }
+ Insert nullInsert = table.newInsert();
+ nullInsert.getRow().addLong("key", i++);
+ nullInsert.getRow().setNull("value");
+ session.apply(nullInsert);
+ session.flush();
+
+ ColumnSchema col = table.getSchema().getColumn("value");
+ Assert.assertEquals(values.size() + 1, countRows(table));
+
+ for (String v : testValues) {
+ // value = v
+ KuduPredicate equal = KuduPredicate.newComparisonPredicate(col, ComparisonOp.EQUAL, v);
+ Assert.assertEquals(values.subSet(v, true, v, true).size(), countRows(table, equal));
+
+ // value >= v
+ KuduPredicate greaterEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER_EQUAL, v);
+ Assert.assertEquals(values.tailSet(v).size(), countRows(table, greaterEqual));
+
+ // value <= v
+ KuduPredicate lessEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS_EQUAL, v);
+ Assert.assertEquals(values.headSet(v, true).size(), countRows(table, lessEqual));
+
+ // value > v
+ KuduPredicate greater =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER, v);
+ Assert.assertEquals(values.tailSet(v, false).size(), countRows(table, greater));
+
+ // value < v
+ KuduPredicate less =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS, v);
+ Assert.assertEquals(values.headSet(v).size(), countRows(table, less));
+ }
+ }
+
+ @Test
+ public void testBinaryPredicates() throws Exception {
+ Schema schema = createTableSchema(Type.BINARY);
+ syncClient.createTable("binary-table", schema, createTableOptions());
+ KuduTable table = syncClient.openTable("binary-table");
+
+ NavigableSet<String> values = createStringValues();
+ List<String> testValues = createStringTestValues();
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ long i = 0;
+ for (String value : values) {
+ Insert insert = table.newInsert();
+ insert.getRow().addLong("key", i++);
+ insert.getRow().addBinary("value", Bytes.fromString(value));
+ session.apply(insert);
+ }
+ Insert nullInsert = table.newInsert();
+ nullInsert.getRow().addLong("key", i++);
+ nullInsert.getRow().setNull("value");
+ session.apply(nullInsert);
+ session.flush();
+
+ ColumnSchema col = table.getSchema().getColumn("value");
+ Assert.assertEquals(values.size() + 1, countRows(table));
+
+ for (String s : testValues) {
+ byte[] v = Bytes.fromString(s);
+ // value = v
+ KuduPredicate equal = KuduPredicate.newComparisonPredicate(col, ComparisonOp.EQUAL, v);
+ Assert.assertEquals(values.subSet(s, true, s, true).size(), countRows(table, equal));
+
+ // value >= v
+ KuduPredicate greaterEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER_EQUAL, v);
+ Assert.assertEquals(values.tailSet(s).size(), countRows(table, greaterEqual));
+
+ // value <= v
+ KuduPredicate lessEqual =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS_EQUAL, v);
+ Assert.assertEquals(values.headSet(s, true).size(), countRows(table, lessEqual));
+
+ // value > v
+ KuduPredicate greater =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.GREATER, v);
+ Assert.assertEquals(values.tailSet(s, false).size(), countRows(table, greater));
+
+ // value < v
+ KuduPredicate less =
+ KuduPredicate.newComparisonPredicate(col, ComparisonOp.LESS, v);
+ Assert.assertEquals(values.headSet(s).size(), countRows(table, less));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestScannerMultiTablet.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestScannerMultiTablet.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestScannerMultiTablet.java
new file mode 100644
index 0000000..251057f
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestScannerMultiTablet.java
@@ -0,0 +1,236 @@
+// 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.kududb.client;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.stumbleupon.async.Deferred;
+import org.kududb.ColumnSchema;
+import org.kududb.Schema;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+
+import static org.junit.Assert.assertNull;
+import static org.kududb.Type.STRING;
+import static org.junit.Assert.assertEquals;
+
+public class TestScannerMultiTablet extends BaseKuduTest {
+ // Generate a unique table name
+ private static final String TABLE_NAME =
+ TestScannerMultiTablet.class.getName()+"-"+System.currentTimeMillis();
+
+ private static Schema schema = getSchema();
+ private static KuduTable table;
+
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+ BaseKuduTest.setUpBeforeClass();
+ // create a 4-tablets table for scanning
+ CreateTableOptions builder =
+ new CreateTableOptions().setRangePartitionColumns(ImmutableList.of("key1", "key2"));
+
+ for (int i = 1; i < 4; i++){
+ PartialRow splitRow = schema.newPartialRow();
+ splitRow.addString("key1", "" + i);
+ splitRow.addString("key2", "");
+ builder.addSplitRow(splitRow);
+ }
+
+ createTable(TABLE_NAME, schema, builder);
+
+ table = openTable(TABLE_NAME);
+
+ AsyncKuduSession session = client.newSession();
+ session.setFlushMode(AsyncKuduSession.FlushMode.AUTO_FLUSH_SYNC);
+
+ // The data layout ends up like this:
+ // tablet '', '1': no rows
+ // tablet '1', '2': '111', '122', '133'
+ // tablet '2', '3': '211', '222', '233'
+ // tablet '3', '': '311', '322', '333'
+ String[] keys = new String[] {"1", "2", "3"};
+ for (String key1 : keys) {
+ for (String key2 : keys) {
+ Insert insert = table.newInsert();
+ PartialRow row = insert.getRow();
+ row.addString(0, key1);
+ row.addString(1, key2);
+ row.addString(2, key2);
+ Deferred<OperationResponse> d = session.apply(insert);
+ d.join(DEFAULT_SLEEP);
+ }
+ }
+ }
+
+ // Test various combinations of start/end row keys.
+ @Test(timeout = 100000)
+ public void testKeyStartEnd() throws Exception {
+ assertEquals(0,
+ countRowsInScan(getScanner("", "", "1", ""))); // There's nothing in the 1st tablet
+ assertEquals(1, countRowsInScan(getScanner("", "", "1", "2"))); // Grab the very first row
+ assertEquals(3, countRowsInScan(getScanner("1", "1", "1", "4"))); // Grab the whole 2nd tablet
+ assertEquals(3, countRowsInScan(getScanner("1", "1", "2", ""))); // Same, and peek at the 3rd
+ assertEquals(3, countRowsInScan(getScanner("1", "1", "2", "0"))); // Same, different peek
+ assertEquals(4,
+ countRowsInScan(getScanner("1", "2", "2", "3"))); // Middle of 2nd to middle of 3rd
+ assertEquals(3,
+ countRowsInScan(getScanner("1", "4", "2", "4"))); // Peek at the 2nd then whole 3rd
+ assertEquals(6, countRowsInScan(getScanner("1", "5", "3", "4"))); // Whole 3rd and 4th
+ assertEquals(9, countRowsInScan(getScanner("", "", "4", ""))); // Full table scan
+
+ assertEquals(9,
+ countRowsInScan(getScanner("", "", null, null))); // Full table scan with empty upper
+ assertEquals(9,
+ countRowsInScan(getScanner(null, null, "4", ""))); // Full table scan with empty lower
+ assertEquals(9,
+ countRowsInScan(getScanner(null, null, null, null))); // Full table scan with empty bounds
+
+ // Test that we can close a scanner while in between two tablets. We start on the second
+ // tablet and our first nextRows() will get 3 rows. At that moment we want to close the scanner
+ // before getting on the 3rd tablet.
+ AsyncKuduScanner scanner = getScanner("1", "", null, null);
+ Deferred<RowResultIterator> d = scanner.nextRows();
+ RowResultIterator rri = d.join(DEFAULT_SLEEP);
+ assertEquals(3, rri.getNumRows());
+ d = scanner.close();
+ rri = d.join(DEFAULT_SLEEP);
+ assertNull(rri);
+ }
+
+ // Test mixing start/end row keys with predicates.
+ @Test(timeout = 100000)
+ public void testKeysAndPredicates() throws Exception {
+ // First row from the 2nd tablet.
+ ColumnRangePredicate predicate = new ColumnRangePredicate(schema.getColumnByIndex(2));
+ predicate.setLowerBound("1");
+ predicate.setUpperBound("1");
+ assertEquals(1, countRowsInScan(getScanner("1", "", "2", "", predicate)));
+
+ // All the 2nd tablet.
+ predicate = new ColumnRangePredicate(schema.getColumnByIndex(2));
+ predicate.setLowerBound("1");
+ predicate.setUpperBound("3");
+ assertEquals(3, countRowsInScan(getScanner("1", "", "2", "", predicate)));
+
+ // Value that doesn't exist.
+ predicate = new ColumnRangePredicate(schema.getColumnByIndex(2));
+ predicate.setLowerBound("4");
+ assertEquals(0, countRowsInScan(getScanner("1", "", "2", "", predicate)));
+
+ // First row from every tablet.
+ predicate = new ColumnRangePredicate(schema.getColumnByIndex(2));
+ predicate.setLowerBound("1");
+ predicate.setUpperBound("1");
+ assertEquals(3, countRowsInScan(getScanner(null, null, null, null, predicate)));
+
+ // All the rows.
+ predicate = new ColumnRangePredicate(schema.getColumnByIndex(2));
+ predicate.setLowerBound("1");
+ assertEquals(9, countRowsInScan(getScanner(null, null, null, null, predicate)));
+ }
+
+ @Test(timeout = 100000)
+ public void testProjections() throws Exception {
+ // Test with column names.
+ AsyncKuduScanner.AsyncKuduScannerBuilder builder = client.newScannerBuilder(table);
+ builder.setProjectedColumnNames(Lists.newArrayList(schema.getColumnByIndex(0).getName(),
+ schema.getColumnByIndex(1).getName()));
+ buildScannerAndCheckColumnsCount(builder, 2);
+
+ // Test with column indexes.
+ builder = client.newScannerBuilder(table);
+ builder.setProjectedColumnIndexes(Lists.newArrayList(0, 1));
+ buildScannerAndCheckColumnsCount(builder, 2);
+
+ // Test with column names overriding indexes.
+ builder = client.newScannerBuilder(table);
+ builder.setProjectedColumnIndexes(Lists.newArrayList(0, 1));
+ builder.setProjectedColumnNames(Lists.newArrayList(schema.getColumnByIndex(0).getName()));
+ buildScannerAndCheckColumnsCount(builder, 1);
+
+ // Test with keys last with indexes.
+ builder = client.newScannerBuilder(table);
+ builder.setProjectedColumnIndexes(Lists.newArrayList(2, 1, 0));
+ buildScannerAndCheckColumnsCount(builder, 3);
+
+ // Test with keys last with column names.
+ builder = client.newScannerBuilder(table);
+ builder.setProjectedColumnNames(Lists.newArrayList(schema.getColumnByIndex(2).getName(),
+ schema.getColumnByIndex(0).getName()));
+ buildScannerAndCheckColumnsCount(builder, 2);
+ }
+
+ private AsyncKuduScanner getScanner(String lowerBoundKeyOne,
+ String lowerBoundKeyTwo,
+ String exclusiveUpperBoundKeyOne,
+ String exclusiveUpperBoundKeyTwo) {
+ return getScanner(lowerBoundKeyOne, lowerBoundKeyTwo,
+ exclusiveUpperBoundKeyOne, exclusiveUpperBoundKeyTwo, null);
+ }
+
+ private AsyncKuduScanner getScanner(String lowerBoundKeyOne,
+ String lowerBoundKeyTwo,
+ String exclusiveUpperBoundKeyOne,
+ String exclusiveUpperBoundKeyTwo,
+ ColumnRangePredicate predicate) {
+ AsyncKuduScanner.AsyncKuduScannerBuilder builder = client.newScannerBuilder(table);
+
+ if (lowerBoundKeyOne != null) {
+ PartialRow lowerBoundRow = schema.newPartialRow();
+ lowerBoundRow.addString(0, lowerBoundKeyOne);
+ lowerBoundRow.addString(1, lowerBoundKeyTwo);
+ builder.lowerBound(lowerBoundRow);
+ }
+
+ if (exclusiveUpperBoundKeyOne != null) {
+ PartialRow upperBoundRow = schema.newPartialRow();
+ upperBoundRow.addString(0, exclusiveUpperBoundKeyOne);
+ upperBoundRow.addString(1, exclusiveUpperBoundKeyTwo);
+ builder.exclusiveUpperBound(upperBoundRow);
+ }
+
+ if (predicate != null) {
+ builder.addColumnRangePredicate(predicate);
+ }
+
+ return builder.build();
+ }
+
+ private void buildScannerAndCheckColumnsCount(AsyncKuduScanner.AsyncKuduScannerBuilder builder,
+ int count) throws Exception {
+ AsyncKuduScanner scanner = builder.build();
+ scanner.nextRows().join(DEFAULT_SLEEP);
+ RowResultIterator rri = scanner.nextRows().join(DEFAULT_SLEEP);
+ assertEquals(count, rri.next().getSchema().getColumns().size());
+ }
+
+ private static Schema getSchema() {
+ ArrayList<ColumnSchema> columns = new ArrayList<>(3);
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("key1", STRING)
+ .key(true)
+ .build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("key2", STRING)
+ .key(true)
+ .build());
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("val", STRING)
+ .nullable(true) // Important because we need to make sure it gets passed in projections
+ .build());
+ return new Schema(columns);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestStatistics.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestStatistics.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestStatistics.java
new file mode 100644
index 0000000..6cfbee7
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestStatistics.java
@@ -0,0 +1,74 @@
+// 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.kududb.client;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.kududb.client.Statistics.Statistic;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestStatistics extends BaseKuduTest {
+
+ private static final String TABLE_NAME = TestStatistics.class.getName() + "-"
+ + System.currentTimeMillis();
+ private static KuduTable table;
+
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+ BaseKuduTest.setUpBeforeClass();
+ CreateTableOptions options = getBasicCreateTableOptions().setNumReplicas(1);
+ table = createTable(TABLE_NAME, basicSchema, options);
+ }
+
+ @Test(timeout = 10000)
+ public void test() throws Exception {
+ KuduSession session = syncClient.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+ int rowCount = 20;
+ for (int i = 0; i < rowCount; i++) {
+ Insert insert = createBasicSchemaInsert(table, i);
+ session.apply(insert);
+ if (i % 2 == 1) {
+ session.flush();
+ }
+ }
+ Statistics statistics = syncClient.getStatistics();
+ assertEquals(rowCount / 2, statistics.getClientStatistic(Statistic.WRITE_RPCS));
+ assertEquals(rowCount, statistics.getClientStatistic(Statistic.WRITE_OPS));
+ assertEquals(0, statistics.getClientStatistic(Statistic.RPC_ERRORS));
+ assertEquals(0, statistics.getClientStatistic(Statistic.OPS_ERRORS));
+
+ // Use default flush mode.
+ session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
+ // Insert duplicate rows, expect to get ALREADY_PRESENT error.
+ long byteSize = 0;
+ for (int i = 0; i < rowCount; i++) {
+ Insert insert = createBasicSchemaInsert(table, i);
+ session.apply(insert);
+ byteSize += insert.getRowOperationSizeBytes();
+ }
+ assertEquals(rowCount + rowCount / 2, statistics.getClientStatistic(Statistic.WRITE_RPCS));
+ assertEquals(rowCount, statistics.getClientStatistic(Statistic.WRITE_OPS));
+ assertEquals(0, statistics.getClientStatistic(Statistic.RPC_ERRORS));
+ assertEquals(rowCount, statistics.getClientStatistic(Statistic.OPS_ERRORS));
+ assertEquals(byteSize * 2, statistics.getClientStatistic(Statistic.BYTES_WRITTEN));
+
+ assertEquals(1, statistics.getTableSet().size());
+ assertEquals(1, statistics.getTabletSet().size());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestStatus.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestStatus.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestStatus.java
new file mode 100644
index 0000000..11bd23b
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestStatus.java
@@ -0,0 +1,55 @@
+// 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.kududb.client;
+
+import org.junit.Test;
+import org.kududb.client.Status;
+
+import static org.junit.Assert.*;
+
+public class TestStatus {
+
+ @Test
+ public void testOKStatus() {
+ Status s = Status.OK();
+ assertTrue(s.ok());
+ assertFalse(s.isNotAuthorized());
+ assertEquals(-1, s.getPosixCode());
+ assertEquals("OK", s.toString());
+ }
+
+ @Test
+ public void testStatusNonPosix() {
+ Status s = Status.Aborted("foo");
+ assertFalse(s.ok());
+ assertTrue(s.isAborted());
+ assertEquals("ABORTED", s.getCodeName());
+ assertEquals("foo", s.getMessage());
+ assertEquals(-1, s.getPosixCode());
+ assertEquals("Aborted: foo", s.toString());
+ }
+
+ @Test
+ public void testPosixCode() {
+ Status s = Status.NotFound("File not found", 2);
+ assertFalse(s.ok());
+ assertFalse(s.isAborted());
+ assertTrue(s.isNotFound());
+ assertEquals(2, s.getPosixCode());
+ assertEquals("Not found: File not found (error 2)", s.toString());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestTestUtils.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestTestUtils.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestTestUtils.java
new file mode 100644
index 0000000..b150f8e
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestTestUtils.java
@@ -0,0 +1,114 @@
+// 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.kududb.client;
+
+import org.junit.After;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for non-trivial helper methods in TestUtils.
+ */
+public class TestTestUtils {
+
+ public static final Logger LOG = LoggerFactory.getLogger(TestUtils.class);
+
+ private Process proc;
+
+ @After
+ public void tearDown() {
+ if (proc != null) {
+ proc.destroy();
+ }
+ }
+
+ /**
+ * Starts a process that executes the "yes" command (which prints 'y' in a loop),
+ * sends a SIGSTOP to the process, and ensures that SIGSTOP does indeed pause the process.
+ * Afterwards, sends a SIGCONT to the process and ensures that the process resumes.
+ */
+ @Test(timeout = 2000)
+ public void testPauseAndResume() throws Exception {
+ ProcessBuilder processBuilder = new ProcessBuilder("yes");
+ proc = processBuilder.start();
+ LineCounterRunnable lineCounter = new LineCounterRunnable(proc.getInputStream());
+ Thread thread = new Thread(lineCounter);
+ thread.setDaemon(true);
+ thread.start();
+ TestUtils.pauseProcess(proc);
+ long prevCount;
+ do {
+ prevCount = lineCounter.getCount();
+ Thread.sleep(10);
+ } while (prevCount != lineCounter.getCount());
+ assertEquals(prevCount, lineCounter.getCount());
+ TestUtils.resumeProcess(proc);
+ do {
+ prevCount = lineCounter.getCount();
+ Thread.sleep(10);
+ } while (prevCount == lineCounter.getCount());
+ assertTrue(lineCounter.getCount() > prevCount);
+ }
+
+ /**
+ * Counts the number of lines in a specified input stream.
+ */
+ static class LineCounterRunnable implements Runnable {
+ private final AtomicLong counter;
+ private final InputStream is;
+
+ public LineCounterRunnable(InputStream is) {
+ this.is = is;
+ counter = new AtomicLong(0);
+ }
+
+ @Override
+ public void run() {
+ BufferedReader in = null;
+ try {
+ in = new BufferedReader(new InputStreamReader(is));
+ while (in.readLine() != null) {
+ counter.incrementAndGet();
+ }
+ } catch (Exception e) {
+ LOG.error("Error while reading from the process", e);
+ } finally {
+ if (in != null) {
+ try {
+ in.close();
+ } catch (IOException e) {
+ LOG.error("Error closing the stream", e);
+ }
+ }
+ }
+ }
+
+ public long getCount() {
+ return counter.get();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestTimeouts.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestTimeouts.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestTimeouts.java
new file mode 100644
index 0000000..3e78918
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestTimeouts.java
@@ -0,0 +1,68 @@
+// 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.kududb.client;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.stumbleupon.async.TimeoutException;
+import org.junit.Test;
+
+public class TestTimeouts extends BaseKuduTest {
+
+ private static final String TABLE_NAME =
+ TestTimeouts.class.getName() + "-" + System.currentTimeMillis();
+
+ /**
+ * This test case tries different methods that should all timeout, while relying on the client to
+ * pass down the timeouts to the session and scanner.
+ */
+ @Test(timeout = 100000)
+ public void testLowTimeouts() throws Exception {
+ KuduClient lowTimeoutsClient = new KuduClient.KuduClientBuilder(masterAddresses)
+ .defaultAdminOperationTimeoutMs(1)
+ .defaultOperationTimeoutMs(1)
+ .build();
+
+ try {
+ lowTimeoutsClient.listTabletServers();
+ fail("Should have timed out");
+ } catch (KuduException ex) {
+ // Expected.
+ }
+
+ createTable(TABLE_NAME, basicSchema, getBasicCreateTableOptions());
+ KuduTable table = openTable(TABLE_NAME);
+
+ KuduSession lowTimeoutSession = lowTimeoutsClient.newSession();
+
+ try {
+ lowTimeoutSession.apply(createBasicSchemaInsert(table, 1));
+ fail("Should have timed out");
+ } catch (KuduException ex) {
+ assertTrue(ex.getStatus().isTimedOut());
+ }
+
+ KuduScanner lowTimeoutScanner = lowTimeoutsClient.newScannerBuilder(table).build();
+ try {
+ lowTimeoutScanner.nextRows();
+ fail("Should have timed out");
+ } catch (KuduException ex) {
+ assertTrue(ex.getStatus().isTimedOut());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/client/TestUtils.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestUtils.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestUtils.java
new file mode 100644
index 0000000..392bed9
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestUtils.java
@@ -0,0 +1,289 @@
+// 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.kududb.client;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.sun.security.auth.module.UnixSystem;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.management.VMManagement;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.lang.management.ManagementFactory;
+import java.lang.management.RuntimeMXBean;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.SocketAddress;
+import java.net.URL;
+import java.net.URLDecoder;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * A grouping of methods that help unit testing.
+ */
+public class TestUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(TestUtils.class);
+
+ // Used by pidOfProcess()
+ private static String UNIX_PROCESS_CLS_NAME = "java.lang.UNIXProcess";
+ private static Set<String> VALID_SIGNALS = ImmutableSet.of("STOP", "CONT", "TERM", "KILL");
+
+ private static final String BIN_DIR_PROP = "binDir";
+
+ /**
+ * @return the path of the flags file to pass to daemon processes
+ * started by the tests
+ */
+ public static String getFlagsPath() {
+ URL u = BaseKuduTest.class.getResource("/flags");
+ if (u == null) {
+ throw new RuntimeException("Unable to find 'flags' file");
+ }
+ if (u.getProtocol() == "file") {
+ return urlToPath(u);
+ }
+ // If the flags are inside a JAR, extract them into our temporary
+ // test directory.
+ try {
+ // Somewhat unintuitively, createTempFile() actually creates the file,
+ // not just the path, so we have to use REPLACE_EXISTING below.
+ Path tmpFile = Files.createTempFile(
+ Paths.get(getBaseDir()), "kudu-flags", ".flags");
+ Files.copy(BaseKuduTest.class.getResourceAsStream("/flags"), tmpFile,
+ StandardCopyOption.REPLACE_EXISTING);
+ return tmpFile.toAbsolutePath().toString();
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to extract flags file into tmp", e);
+ }
+ }
+
+ /**
+ * Return the path portion of a file URL, after decoding the escaped
+ * components. This fixes issues when trying to build within a
+ * working directory with special characters.
+ */
+ private static String urlToPath(URL u) {
+ try {
+ return URLDecoder.decode(u.getPath(), "UTF-8");
+ } catch (UnsupportedEncodingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static String findBuildDir() {
+ URL myUrl = BaseKuduTest.class.getProtectionDomain().getCodeSource().getLocation();
+ File myPath = new File(urlToPath(myUrl));
+ while (myPath != null) {
+ if (new File(myPath, ".git").isDirectory()) {
+ return new File(myPath, "build/latest/bin").getAbsolutePath();
+ }
+ myPath = myPath.getParentFile();
+ }
+ LOG.warn("Unable to find build dir! myUrl={}", myUrl);
+ return null;
+ }
+
+ /**
+ * @param binName the binary to look for (eg 'kudu-tserver')
+ * @return the absolute path of that binary
+ * @throws FileNotFoundException if no such binary is found
+ */
+ public static String findBinary(String binName) throws FileNotFoundException {
+ String binDir = System.getProperty(BIN_DIR_PROP);
+ if (binDir != null) {
+ LOG.info("Using binary directory specified by property: {}",
+ binDir);
+ } else {
+ binDir = findBuildDir();
+ }
+
+ File candidate = new File(binDir, binName);
+ if (candidate.canExecute()) {
+ return candidate.getAbsolutePath();
+ }
+ throw new FileNotFoundException("Cannot find binary " + binName +
+ " in binary directory " + binDir);
+ }
+
+ /**
+ * @return the base directory within which we will store server data
+ */
+ public static String getBaseDir() {
+ String s = System.getenv("TEST_TMPDIR");
+ if (s == null) {
+ s = String.format("/tmp/kudutest-%d", new UnixSystem().getUid());
+ }
+ File f = new File(s);
+ f.mkdirs();
+ return f.getAbsolutePath();
+ }
+
+ /**
+ * Finds the next free port, starting with the one passed. Keep in mind the
+ * time-of-check-time-of-use nature of this method, the returned port might become occupied
+ * after it was checked for availability.
+ * @param startPort first port to be probed
+ * @return a currently usable port
+ * @throws IOException IOE is thrown if we can't close a socket we tried to open or if we run
+ * out of ports to try
+ */
+ public static int findFreePort(int startPort) throws IOException {
+ ServerSocket ss;
+ for(int i = startPort; i < 65536; i++) {
+ try {
+ ss = new ServerSocket();
+ SocketAddress address = new InetSocketAddress(getUniqueLocalhost(), i);
+ ss.bind(address);
+ } catch (IOException e) {
+ continue;
+ }
+ ss.close();
+ return i;
+ }
+ throw new IOException("Ran out of ports.");
+ }
+
+ /**
+ * Finds a specified number of parts, starting with one passed. Keep in mind the
+ * time-of-check-time-of-use nature of this method.
+ * @see {@link #findFreePort(int)}
+ * @param startPort First port to be probed.
+ * @param numPorts Number of ports to reserve.
+ * @return A list of currently usable ports.
+ * @throws IOException IOE Is thrown if we can't close a socket we tried to open or if run
+ * out of ports to try.
+ */
+ public static List<Integer> findFreePorts(int startPort, int numPorts) throws IOException {
+ List<Integer> ports = Lists.newArrayListWithCapacity(numPorts);
+ for (int i = 0; i < numPorts; i++) {
+ startPort = findFreePort(startPort);
+ ports.add(startPort++);
+ }
+ return ports;
+ }
+
+ /**
+ * Gets the pid of a specified process. Relies on reflection and only works on
+ * UNIX process, not guaranteed to work on JDKs other than Oracle and OpenJDK.
+ * @param proc The specified process.
+ * @return The process UNIX pid.
+ * @throws IllegalArgumentException If the process is not a UNIXProcess.
+ * @throws Exception If there are other getting the pid via reflection.
+ */
+ static int pidOfProcess(Process proc) throws Exception {
+ Class<?> procCls = proc.getClass();
+ if (!procCls.getName().equals(UNIX_PROCESS_CLS_NAME)) {
+ throw new IllegalArgumentException("stopProcess() expects objects of class " +
+ UNIX_PROCESS_CLS_NAME + ", but " + procCls.getName() + " was passed in instead!");
+ }
+ Field pidField = procCls.getDeclaredField("pid");
+ pidField.setAccessible(true);
+ return (Integer) pidField.get(proc);
+ }
+
+ /**
+ * Send a code specified by its string representation to the specified process.
+ * TODO: Use a JNR/JNR-Posix instead of forking the JVM to exec "kill".
+ * @param proc The specified process.
+ * @param sig The string representation of the process (e.g., STOP for SIGSTOP).
+ * @throws IllegalArgumentException If the signal type is not supported.
+ * @throws IllegalStateException If we are unable to send the specified signal.
+ */
+ static void signalProcess(Process proc, String sig) throws Exception {
+ if (!VALID_SIGNALS.contains(sig)) {
+ throw new IllegalArgumentException(sig + " is not a supported signal, only " +
+ Joiner.on(",").join(VALID_SIGNALS) + " are supported");
+ }
+ int pid = pidOfProcess(proc);
+ int rv = Runtime.getRuntime()
+ .exec(String.format("kill -%s %d", sig, pid))
+ .waitFor();
+ if (rv != 0) {
+ throw new IllegalStateException(String.format("unable to send SIG%s to process %s(pid=%d): " +
+ "expected return code from kill, but got %d instead", sig, proc, pid, rv));
+ }
+ }
+
+ /**
+ * Pause the specified process by sending a SIGSTOP using the kill command.
+ * @param proc The specified process.
+ * @throws Exception If error prevents us from pausing the process.
+ */
+ static void pauseProcess(Process proc) throws Exception {
+ signalProcess(proc, "STOP");
+ }
+
+ /**
+ * Resumes the specified process by sending a SIGCONT using the kill command.
+ * @param proc The specified process.
+ * @throws Exception If error prevents us from resuming the process.
+ */
+ static void resumeProcess(Process proc) throws Exception {
+ signalProcess(proc, "CONT");
+ }
+
+ /**
+ * This is used to generate unique loopback IPs for parallel test running.
+ * @return the local PID of this process
+ */
+ static int getPid() {
+ try {
+ RuntimeMXBean runtime = ManagementFactory.getRuntimeMXBean();
+ java.lang.reflect.Field jvm = runtime.getClass().getDeclaredField("jvm");
+ jvm.setAccessible(true);
+ VMManagement mgmt = (VMManagement)jvm.get(runtime);
+ Method pid_method = mgmt.getClass().getDeclaredMethod("getProcessId");
+ pid_method.setAccessible(true);
+
+ return (Integer)pid_method.invoke(mgmt);
+ } catch (Exception e) {
+ LOG.warn("Cannot get PID", e);
+ return 1;
+ }
+ }
+
+ /**
+ * The generated IP is based on pid, so this requires that the parallel tests
+ * run in separate VMs.
+ *
+ * On OSX, the above trick doesn't work, so we can't run parallel tests on OSX.
+ * Given that, we just return the normal localhost IP.
+ *
+ * @return a unique loopback IP address for this PID. This allows running
+ * tests in parallel, since 127.0.0.0/8 all act as loopbacks on Linux.
+ */
+ static String getUniqueLocalhost() {
+ if ("Mac OS X".equals(System.getProperty("os.name"))) {
+ return "127.0.0.1";
+ }
+
+ int pid = getPid();
+ return "127." + ((pid & 0xff00) >> 8) + "." + (pid & 0xff) + ".1";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/util/TestAsyncUtil.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/util/TestAsyncUtil.java b/java/kudu-client/src/test/java/org/apache/kudu/util/TestAsyncUtil.java
new file mode 100644
index 0000000..fce7ddc
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/util/TestAsyncUtil.java
@@ -0,0 +1,75 @@
+// 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.kududb.util;
+
+import com.stumbleupon.async.Callback;
+import com.stumbleupon.async.Deferred;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for {@link AsyncUtil}.
+ */
+public class TestAsyncUtil {
+
+ @Rule
+ public ExpectedException exception = ExpectedException.none();
+
+ @Test
+ public void testAddCallbacksDeferring() throws Exception {
+ Deferred<String> d = new Deferred<String>();
+ TestCallback cb = new TestCallback();
+ TestErrback eb = new TestErrback();
+
+ // Test normal callbacks.
+ AsyncUtil.addCallbacksDeferring(d, cb, eb);
+ final String testStr = "hello world";
+ d.callback(testStr);
+ assertEquals(d.join(), "callback: " + testStr);
+
+ d = new Deferred<String>();
+ AsyncUtil.addCallbacksDeferring(d, cb, eb);
+ d.callback(new IllegalArgumentException());
+ assertEquals(d.join(), "illegal arg");
+
+ d = new Deferred<String>();
+ AsyncUtil.addCallbacksDeferring(d, cb, eb);
+ d.callback(new IllegalStateException());
+ exception.expect(IllegalStateException.class);
+ d.join();
+ }
+
+ final class TestCallback implements Callback<Deferred<String>, String> {
+ @Override
+ public Deferred<String> call(String arg) throws Exception {
+ return Deferred.fromResult("callback: " + arg);
+ }
+ }
+
+ final class TestErrback implements Callback<Deferred<String>, Exception> {
+ @Override
+ public Deferred<String> call(Exception arg) {
+ if (arg instanceof IllegalArgumentException) {
+ return Deferred.fromResult("illegal arg");
+ }
+ return Deferred.fromError(arg);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/util/TestMurmurHash.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/util/TestMurmurHash.java b/java/kudu-client/src/test/java/org/apache/kudu/util/TestMurmurHash.java
new file mode 100644
index 0000000..051107c
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/util/TestMurmurHash.java
@@ -0,0 +1,46 @@
+// 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.kududb.util;
+
+import com.google.common.primitives.UnsignedLongs;
+import com.sangupta.murmur.Murmur2;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test Murmur2 Hash64 returns the expected values for inputs.
+ *
+ * These tests are duplicated on the C++ side to ensure that hash computations
+ * are stable across both platforms.
+ */
+public class TestMurmurHash {
+
+ @Test
+ public void testMurmur2Hash64() throws Exception {
+ long hash;
+
+ hash = Murmur2.hash64("ab".getBytes("UTF-8"), 2, 0);
+ assertEquals(UnsignedLongs.parseUnsignedLong("7115271465109541368"), hash);
+
+ hash = Murmur2.hash64("abcdefg".getBytes("UTF-8"), 7, 0);
+ assertEquals(UnsignedLongs.parseUnsignedLong("2601573339036254301"), hash);
+
+ hash = Murmur2.hash64("quick brown fox".getBytes("UTF-8"), 15, 42);
+ assertEquals(UnsignedLongs.parseUnsignedLong("3575930248840144026"), hash);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/apache/kudu/util/TestNetUtil.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/util/TestNetUtil.java b/java/kudu-client/src/test/java/org/apache/kudu/util/TestNetUtil.java
new file mode 100644
index 0000000..5c003ae
--- /dev/null
+++ b/java/kudu-client/src/test/java/org/apache/kudu/util/TestNetUtil.java
@@ -0,0 +1,73 @@
+// 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.kududb.util;
+
+import com.google.common.net.HostAndPort;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for {@link NetUtil}.
+ */
+public class TestNetUtil {
+
+ /**
+ * Tests parsing strings into {@link HostAndPort} objects with and without specifying
+ * the port in the string.
+ */
+ @Test
+ public void testParseString() {
+ String aStringWithPort = "1.2.3.4:1234";
+ HostAndPort hostAndPortForAStringWithPort = NetUtil.parseString(aStringWithPort, 0);
+ assertEquals(hostAndPortForAStringWithPort.getHostText(), "1.2.3.4");
+ assertEquals(hostAndPortForAStringWithPort.getPort(), 1234);
+
+ String aStringWithoutPort = "1.2.3.4";
+ HostAndPort hostAndPortForAStringWithoutPort = NetUtil.parseString(aStringWithoutPort, 12345);
+ assertEquals(hostAndPortForAStringWithoutPort.getHostText(), aStringWithoutPort);
+ assertEquals(hostAndPortForAStringWithoutPort.getPort(), 12345);
+ }
+
+ /**
+ * Tests parsing comma separated list of "host:port" pairs and hosts into a list of
+ * {@link HostAndPort} objects.
+ */
+ @Test
+ public void testParseStrings() {
+ String testAddrs = "1.2.3.4.5,10.0.0.1:5555,127.0.0.1:7777";
+ List<HostAndPort> hostsAndPorts = NetUtil.parseStrings(testAddrs, 3333);
+ assertArrayEquals(hostsAndPorts.toArray(),
+ new HostAndPort[] { HostAndPort.fromParts("1.2.3.4.5", 3333),
+ HostAndPort.fromParts("10.0.0.1", 5555),
+ HostAndPort.fromParts("127.0.0.1", 7777) }
+ );
+ }
+
+ @Test
+ public void testHostsAndPortsToString() {
+ List<HostAndPort> hostsAndPorts = Arrays.asList(
+ HostAndPort.fromParts("127.0.0.1", 1111),
+ HostAndPort.fromParts("1.2.3.4.5", 0)
+ );
+ assertEquals(NetUtil.hostsAndPortsToString(hostsAndPorts), "127.0.0.1:1111,1.2.3.4.5:0");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/5c305689/java/kudu-client/src/test/java/org/kududb/client/BaseKuduTest.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/kududb/client/BaseKuduTest.java b/java/kudu-client/src/test/java/org/kududb/client/BaseKuduTest.java
deleted file mode 100644
index 1464fa4..0000000
--- a/java/kudu-client/src/test/java/org/kududb/client/BaseKuduTest.java
+++ /dev/null
@@ -1,438 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-package org.kududb.client;
-
-import com.google.common.base.Stopwatch;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
-import com.google.common.net.HostAndPort;
-import com.stumbleupon.async.Callback;
-import com.stumbleupon.async.Deferred;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.kududb.ColumnSchema;
-import org.kududb.Schema;
-import org.kududb.Type;
-import org.kududb.master.Master;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Random;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import static org.junit.Assert.fail;
-
-public class BaseKuduTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(BaseKuduTest.class);
-
- private static final String NUM_MASTERS_PROP = "NUM_MASTERS";
- private static final int NUM_TABLET_SERVERS = 3;
- private static final int DEFAULT_NUM_MASTERS = 1;
-
- // Number of masters that will be started for this test if we're starting
- // a cluster.
- private static final int NUM_MASTERS =
- Integer.getInteger(NUM_MASTERS_PROP, DEFAULT_NUM_MASTERS);
-
- protected static final int DEFAULT_SLEEP = 50000;
-
- // Currently not specifying a seed since we want a random behavior when running tests that
- // restart tablet servers. Would be nice to have the same kind of facility that C++ has that dumps
- // the seed it picks so that you can re-run tests with it.
- private static final Random randomForTSRestart = new Random();
-
- private static MiniKuduCluster miniCluster;
-
- // Comma separate describing the master addresses and ports.
- protected static String masterAddresses;
- protected static List<HostAndPort> masterHostPorts;
-
- // We create both versions of the client for ease of use.
- protected static AsyncKuduClient client;
- protected static KuduClient syncClient;
- protected static final Schema basicSchema = getBasicSchema();
- protected static final Schema allTypesSchema = getSchemaWithAllTypes();
-
- @BeforeClass
- public static void setUpBeforeClass() throws Exception {
- LOG.info("Setting up before class...");
-
- miniCluster = new MiniKuduCluster.MiniKuduClusterBuilder()
- .numMasters(NUM_MASTERS)
- .numTservers(NUM_TABLET_SERVERS)
- .defaultTimeoutMs(DEFAULT_SLEEP)
- .build();
- masterAddresses = miniCluster.getMasterAddresses();
- masterHostPorts = miniCluster.getMasterHostPorts();
-
- LOG.info("Creating new Kudu client...");
- client = new AsyncKuduClient.AsyncKuduClientBuilder(masterAddresses)
- .defaultAdminOperationTimeoutMs(DEFAULT_SLEEP)
- .build();
- syncClient = new KuduClient(client);
- LOG.info("Waiting for tablet servers...");
- if (!miniCluster.waitForTabletServers(NUM_TABLET_SERVERS)) {
- fail("Couldn't get " + NUM_TABLET_SERVERS + " tablet servers running, aborting");
- }
- }
-
- @AfterClass
- public static void tearDownAfterClass() throws Exception {
- try {
- if (client != null) {
- Deferred<ArrayList<Void>> d = client.shutdown();
- d.addErrback(defaultErrorCB);
- d.join(DEFAULT_SLEEP);
- // No need to explicitly shutdown the sync client,
- // shutting down the async client effectively does that.
- }
- } finally {
- if (miniCluster != null) {
- miniCluster.shutdown();
- }
- }
- }
-
- protected static KuduTable createTable(String tableName, Schema schema,
- CreateTableOptions builder) throws Exception {
- LOG.info("Creating table: {}", tableName);
- return client.syncClient().createTable(tableName, schema, builder);
- }
-
- /**
- * Counts the rows from the {@code scanner} until exhaustion. It doesn't require the scanner to
- * be new, so it can be used to finish scanning a previously-started scan.
- */
- protected static int countRowsInScan(AsyncKuduScanner scanner)
- throws Exception {
- final AtomicInteger counter = new AtomicInteger();
-
- Callback<Object, RowResultIterator> cb = new Callback<Object, RowResultIterator>() {
- @Override
- public Object call(RowResultIterator arg) throws Exception {
- if (arg == null) return null;
- counter.addAndGet(arg.getNumRows());
- return null;
- }
- };
-
- while (scanner.hasMoreRows()) {
- Deferred<RowResultIterator> data = scanner.nextRows();
- data.addCallbacks(cb, defaultErrorCB);
- data.join(DEFAULT_SLEEP);
- }
-
- Deferred<RowResultIterator> closer = scanner.close();
- closer.addCallbacks(cb, defaultErrorCB);
- closer.join(DEFAULT_SLEEP);
- return counter.get();
- }
-
- protected List<String> scanTableToStrings(KuduTable table,
- KuduPredicate... predicates) throws Exception {
- List<String> rowStrings = Lists.newArrayList();
- KuduScanner.KuduScannerBuilder scanBuilder = syncClient.newScannerBuilder(table);
- for (KuduPredicate predicate : predicates) {
- scanBuilder.addPredicate(predicate);
- }
- KuduScanner scanner = scanBuilder.build();
- while (scanner.hasMoreRows()) {
- RowResultIterator rows = scanner.nextRows();
- for (RowResult r : rows) {
- rowStrings.add(r.rowToString());
- }
- }
- Collections.sort(rowStrings);
- return rowStrings;
- }
-
- private static final int[] KEYS = new int[] {10, 20, 30};
- protected static KuduTable createFourTabletsTableWithNineRows(String tableName) throws
- Exception {
- CreateTableOptions builder = getBasicCreateTableOptions();
- for (int i : KEYS) {
- PartialRow splitRow = basicSchema.newPartialRow();
- splitRow.addInt(0, i);
- builder.addSplitRow(splitRow);
- }
- KuduTable table = createTable(tableName, basicSchema, builder);
- AsyncKuduSession session = client.newSession();
-
- // create a table with on empty tablet and 3 tablets of 3 rows each
- for (int key1 : KEYS) {
- for (int key2 = 1; key2 <= 3; key2++) {
- Insert insert = table.newInsert();
- PartialRow row = insert.getRow();
- row.addInt(0, key1 + key2);
- row.addInt(1, key1);
- row.addInt(2, key2);
- row.addString(3, "a string");
- row.addBoolean(4, true);
- session.apply(insert).join(DEFAULT_SLEEP);
- }
- }
- session.close().join(DEFAULT_SLEEP);
- return table;
- }
-
- public static Schema getSchemaWithAllTypes() {
- List<ColumnSchema> columns =
- ImmutableList.of(
- new ColumnSchema.ColumnSchemaBuilder("int8", Type.INT8).key(true).build(),
- new ColumnSchema.ColumnSchemaBuilder("int16", Type.INT16).build(),
- new ColumnSchema.ColumnSchemaBuilder("int32", Type.INT32).build(),
- new ColumnSchema.ColumnSchemaBuilder("int64", Type.INT64).build(),
- new ColumnSchema.ColumnSchemaBuilder("bool", Type.BOOL).build(),
- new ColumnSchema.ColumnSchemaBuilder("float", Type.FLOAT).build(),
- new ColumnSchema.ColumnSchemaBuilder("double", Type.DOUBLE).build(),
- new ColumnSchema.ColumnSchemaBuilder("string", Type.STRING).build(),
- new ColumnSchema.ColumnSchemaBuilder("binary-array", Type.BINARY).build(),
- new ColumnSchema.ColumnSchemaBuilder("binary-bytebuffer", Type.BINARY).build(),
- new ColumnSchema.ColumnSchemaBuilder("null", Type.STRING).nullable(true).build(),
- new ColumnSchema.ColumnSchemaBuilder("timestamp", Type.TIMESTAMP).build());
-
- return new Schema(columns);
- }
-
- public static CreateTableOptions getAllTypesCreateTableOptions() {
- return new CreateTableOptions().setRangePartitionColumns(ImmutableList.of("int8"));
- }
-
- public static Schema getBasicSchema() {
- ArrayList<ColumnSchema> columns = new ArrayList<ColumnSchema>(5);
- columns.add(new ColumnSchema.ColumnSchemaBuilder("key", Type.INT32).key(true).build());
- columns.add(new ColumnSchema.ColumnSchemaBuilder("column1_i", Type.INT32).build());
- columns.add(new ColumnSchema.ColumnSchemaBuilder("column2_i", Type.INT32).build());
- columns.add(new ColumnSchema.ColumnSchemaBuilder("column3_s", Type.STRING)
- .nullable(true)
- .desiredBlockSize(4096)
- .encoding(ColumnSchema.Encoding.DICT_ENCODING)
- .compressionAlgorithm(ColumnSchema.CompressionAlgorithm.LZ4)
- .build());
- columns.add(new ColumnSchema.ColumnSchemaBuilder("column4_b", Type.BOOL).build());
- return new Schema(columns);
- }
-
- public static CreateTableOptions getBasicCreateTableOptions() {
- return new CreateTableOptions().setRangePartitionColumns(ImmutableList.of("key"));
- }
-
- /**
- * Creates table options with non-covering range partitioning for a table with
- * the basic schema. Range partition key ranges fall between the following values:
- *
- * [ 0, 50)
- * [ 50, 100)
- * [200, 300)
- */
- public static CreateTableOptions getBasicTableOptionsWithNonCoveredRange() {
- Schema schema = basicSchema;
- CreateTableOptions option = new CreateTableOptions();
- option.setRangePartitionColumns(ImmutableList.of("key"));
-
- PartialRow aLowerBound = schema.newPartialRow();
- aLowerBound.addInt("key", 0);
- PartialRow aUpperBound = schema.newPartialRow();
- aUpperBound.addInt("key", 100);
- option.addRangeBound(aLowerBound, aUpperBound);
-
- PartialRow bLowerBound = schema.newPartialRow();
- bLowerBound.addInt("key", 200);
- PartialRow bUpperBound = schema.newPartialRow();
- bUpperBound.addInt("key", 300);
- option.addRangeBound(bLowerBound, bUpperBound);
-
- PartialRow split = schema.newPartialRow();
- split.addInt("key", 50);
- option.addSplitRow(split);
- return option;
- }
-
- protected Insert createBasicSchemaInsert(KuduTable table, int key) {
- Insert insert = table.newInsert();
- PartialRow row = insert.getRow();
- row.addInt(0, key);
- row.addInt(1, 2);
- row.addInt(2, 3);
- row.addString(3, "a string");
- row.addBoolean(4, true);
- return insert;
- }
-
- static Callback<Object, Object> defaultErrorCB = new Callback<Object, Object>() {
- @Override
- public Object call(Object arg) throws Exception {
- if (arg == null) return null;
- if (arg instanceof Exception) {
- LOG.warn("Got exception", (Exception) arg);
- } else {
- LOG.warn("Got an error response back " + arg);
- }
- return new Exception("Can't recover from error, see previous WARN");
- }
- };
-
- /**
- * Helper method to open a table. It sets the default sleep time when joining on the Deferred.
- * @param name Name of the table
- * @return A KuduTable
- * @throws Exception MasterErrorException if the table doesn't exist
- */
- protected static KuduTable openTable(String name) throws Exception {
- Deferred<KuduTable> d = client.openTable(name);
- return d.join(DEFAULT_SLEEP);
- }
-
- /**
- * Helper method to easily kill a tablet server that serves the given table's only tablet's
- * leader. The currently running test case will be failed if there's more than one tablet,
- * if the tablet has no leader after some retries, or if the tablet server was already killed.
- *
- * This method is thread-safe.
- * @param table a KuduTable which will get its single tablet's leader killed.
- * @throws Exception
- */
- protected static void killTabletLeader(KuduTable table) throws Exception {
- LocatedTablet.Replica leader = null;
- DeadlineTracker deadlineTracker = new DeadlineTracker();
- deadlineTracker.setDeadline(DEFAULT_SLEEP);
- while (leader == null) {
- if (deadlineTracker.timedOut()) {
- fail("Timed out while trying to find a leader for this table: " + table.getName());
- }
- List<LocatedTablet> tablets = table.getTabletsLocations(DEFAULT_SLEEP);
- if (tablets.isEmpty() || tablets.size() > 1) {
- fail("Currently only support killing leaders for tables containing 1 tablet, table " +
- table.getName() + " has " + tablets.size());
- }
- LocatedTablet tablet = tablets.get(0);
- if (tablet.getReplicas().size() == 1) {
- fail("Table " + table.getName() + " only has 1 tablet, please enable replication");
- }
- leader = tablet.getLeaderReplica();
- if (leader == null) {
- LOG.info("Sleeping while waiting for a tablet LEADER to arise, currently slept " +
- deadlineTracker.getElapsedMillis() + "ms");
- Thread.sleep(50);
- }
- }
-
- Integer port = leader.getRpcPort();
- miniCluster.killTabletServerOnPort(port);
- }
-
- /**
- * Helper method to easily kill the leader master.
- *
- * This method is thread-safe.
- * @throws Exception If there is an error finding or killing the leader master.
- */
- protected static void killMasterLeader() throws Exception {
- int leaderPort = findLeaderMasterPort();
- miniCluster.killMasterOnPort(leaderPort);
- }
-
- /**
- * Find the port of the leader master in order to retrieve it from the port to process map.
- * @return The port of the leader master.
- * @throws Exception If we are unable to find the leader master.
- */
- protected static int findLeaderMasterPort() throws Exception {
- Stopwatch sw = Stopwatch.createStarted();
- int leaderPort = -1;
- while (leaderPort == -1 && sw.elapsed(TimeUnit.MILLISECONDS) < DEFAULT_SLEEP) {
- Deferred<Master.GetTableLocationsResponsePB> masterLocD = client.getMasterTableLocationsPB();
- Master.GetTableLocationsResponsePB r = masterLocD.join(DEFAULT_SLEEP);
- leaderPort = r.getTabletLocations(0)
- .getReplicas(0)
- .getTsInfo()
- .getRpcAddresses(0)
- .getPort();
- }
- if (leaderPort == -1) {
- fail("No leader master found after " + DEFAULT_SLEEP + " ms.");
- }
- return leaderPort;
- }
-
- /**
- * Picks at random a tablet server that serves tablets from the passed table and restarts it.
- * Waits between killing and restarting the process.
- * @param table table to query for a TS to restart
- * @throws Exception
- */
- protected static void restartTabletServer(KuduTable table) throws Exception {
- List<LocatedTablet> tablets = table.getTabletsLocations(DEFAULT_SLEEP);
- if (tablets.isEmpty()) {
- fail("Table " + table.getName() + " doesn't have any tablets");
- }
-
- LocatedTablet tablet = tablets.get(0);
-
- int port = tablet.getReplicas().get(
- randomForTSRestart.nextInt(tablet.getReplicas().size())).getRpcPort();
-
- miniCluster.killTabletServerOnPort(port);
-
- Thread.sleep(1000);
-
- miniCluster.restartDeadTabletServerOnPort(port);
- }
-
- /**
- * Kills, sleeps, then restarts the leader master.
- * @throws Exception
- */
- protected static void restartLeaderMaster() throws Exception {
- int master = findLeaderMasterPort();
- miniCluster.killMasterOnPort(master);
-
- Thread.sleep(1000);
-
- miniCluster.restartDeadMasterOnPort(master);
- }
-
- /**
- * Return the comma-separated list of "host:port" pairs that describes the master
- * config for this cluster.
- * @return The master config string.
- */
- protected static String getMasterAddresses() {
- return masterAddresses;
- }
-
- /**
- * Kills all tablet servers in the cluster.
- * @throws InterruptedException
- */
- protected void killTabletServers() throws InterruptedException {
- miniCluster.killTabletServers();
- }
-
- /**
- * Restarts killed tablet servers in the cluster.
- * @throws Exception
- */
- protected void restartTabletServers() throws Exception {
- miniCluster.restartDeadTabletServers();
- }
-}